kiln_ai.run_context

Agent run context management using contextvars.

The agent run ID propagates automatically through async call chains, including asyncio.gather and sub-agent calls via KilnTaskTool.

This is a general-purpose ID for scoping work to a single agent run, usable for logging, caching, metrics, or any run-scoped operations.

 1"""Agent run context management using contextvars.
 2
 3The agent run ID propagates automatically through async call chains,
 4including asyncio.gather and sub-agent calls via KilnTaskTool.
 5
 6This is a general-purpose ID for scoping work to a single agent run,
 7usable for logging, caching, metrics, or any run-scoped operations.
 8"""
 9
10import uuid
11from contextvars import ContextVar
12
13_agent_run_id: ContextVar[str | None] = ContextVar("agent_run_id", default=None)
14
15
16def get_agent_run_id() -> str | None:
17    return _agent_run_id.get()
18
19
20def set_agent_run_id(run_id: str) -> None:
21    _agent_run_id.set(run_id)
22
23
24def clear_agent_run_id() -> None:
25    _agent_run_id.set(None)
26
27
28def generate_agent_run_id() -> str:
29    return f"run_{uuid.uuid4().hex[:16]}"
def get_agent_run_id() -> str | None: 
17def get_agent_run_id() -> str | None:
18    return _agent_run_id.get()
def set_agent_run_id(run_id: str) -> None: 
21def set_agent_run_id(run_id: str) -> None:
22    _agent_run_id.set(run_id)
def clear_agent_run_id() -> None: 
25def clear_agent_run_id() -> None:
26    _agent_run_id.set(None)
def generate_agent_run_id() -> str: 
29def generate_agent_run_id() -> str:
30    return f"run_{uuid.uuid4().hex[:16]}"