(lock: ThreadLockType, ctxid: str, create_if_not_exists: bool = True)
| 7 | |
| 8 | |
| 9 | def use_context(lock: ThreadLockType, ctxid: str, create_if_not_exists: bool = True): |
| 10 | from agent import AgentContext |
| 11 | from initialize import initialize_agent |
| 12 | |
| 13 | with lock: |
| 14 | if not ctxid: |
| 15 | first = AgentContext.first() |
| 16 | if first: |
| 17 | AgentContext.use(first.id) |
| 18 | return first |
| 19 | context = AgentContext(config=initialize_agent(), set_current=True) |
| 20 | return context |
| 21 | got = AgentContext.use(ctxid) |
| 22 | if got: |
| 23 | return got |
| 24 | if create_if_not_exists: |
| 25 | context = AgentContext( |
| 26 | config=initialize_agent(), id=ctxid, set_current=True |
| 27 | ) |
| 28 | return context |
| 29 | else: |
| 30 | raise Exception(f"Context {ctxid} not found") |
nothing calls this directly
no test coverage detected