Attribution functions

Four entry points write attribution onto the current scope. All of them push the same AgentContext onto a stack that is isolated per thread and per asyncio task, so agents awaited together under asyncio.gather cannot steal each other's attribution.

For when to reach for which, see Agents and Tags and sessions.

agent()

capsera.agent(
    name: str,
    team: str | None = None,
    task_type: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

Returns a decorator. Wraps sync and async functions alike — coroutine functions are detected and awaited correctly — and preserves the wrapped function's name, docstring, and signature.

@capsera.agent("researcher", team="core", task_type="research")
def research(question: str) -> str: ...


@capsera.agent("classifier")
async def classify(text: str) -> str: ...

Scopes nest and the innermost value wins per field. The scope is popped in a finally, so an exception inside the function does not leak attribution into whatever runs next.

agent() has no session_id parameter. Use tag() or set_session().

tag()

with capsera.tag(
    agent: str,
    team: str | None = None,
    task_type: str | None = None,
    session_id: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
): ...

A context manager with the same fields as agent(), plus session_id. Same nesting rules, same cleanup on exception.

with capsera.tag(agent="summarizer", customer_id="acme", session_id="conv-42"):
    response = client.messages.create(...)

Reach for it when the unit of work is narrower than a function, or when the value is only known at runtime.

set_session()

capsera.set_session(session_id: str) -> None

Sets an ambient session ID for the current thread, and in an async server the current task, that lands on every subsequent event without wrapping any call. Intended for middleware.

clear_session()

capsera.clear_session() -> None

Removes the ambient session ID. Call it in a finally — a pooled worker thread outlives the request, and a session ID left behind attaches to the next request's spend.

@app.middleware("http")
async def attach_session(request, call_next):
    capsera.set_session(request.headers.get("x-session-id", "anon"))
    try:
        return await call_next(request)
    finally:
        capsera.clear_session()

How a value resolves

Each field is resolved independently, first match winning. "Explicit argument" applies to record(); an intercepted call has no arguments of its own, so it starts at the innermost scope.

An unattributed call is still recorded, under unknown, because an unattributable cost is still a cost.