The capture surface is small: one decorator for the agent, one for tools, and helpers for model calls. Everything is a no-op when the gate is off, so you can leave it in place.

## @capture.agent

Wrap your agent entry point. Inside the decorated call, the SDK builds a span tree and records every tool and model call made beneath it, then writes one capture file.

from evalshift import capture, record_model_call

@capture.agent(suite="support_agent")
def handle_ticket(query: str) -> str:
    answer = call_model(query)
    record_model_call(
        model_id="claude-opus-4-8",
        input=query,
        output=answer,
        input_tokens=812,
        output_tokens=140,
        cost_usd=0.012,
        latency_ms=940,
    )
    return answer

record_model_call records an already-complete call. Its keyword args: model_id, input, output, input_tokens, output_tokens, cost_usd, and optional latency_ms. It is a no-op outside an active session.

There is no separate “final output” field on the agent span — persist the agent’s answer as the last model call’s output.

## @capture.tool

Decorate tools to record a span per call, with arguments and result. Nested tools nest in the tree automatically. Use it bare or with an explicit name:

@capture.tool                 # name defaults to the function name
def search_orders(user_id: str): ...

@capture.tool(name="refund")  # or name it explicitly
def issue_refund(order_id: str): ...

## Async, streaming & concurrency

@capture.agent and @capture.tool auto-detect coroutine functions and wrap them in an async wrapper. Context propagates across await and into asyncio.gather child tasks, so concurrent tool calls land in the tree with the right parentage and a dense, collision-free ordering.

import asyncio

@capture.agent(suite="support_agent")          # async is auto-detected
async def handle_ticket(query: str) -> str:
    # concurrent tools get correct parentage + ordering
    orders, profile = await asyncio.gather(
        search_orders(user_id),
        load_profile(user_id),
    )
    return await summarize(orders, profile)

### Streaming model calls

For token streams, open a recorder with capture.model_call(...), accumulate chunks with add_text, optionally record usage with set_usage, and exactly one model-call span is recorded when the block exits. It works as a sync with or an async with. It must run inside an active agent — a @capture.agent call or an agent session — a bare capture.model_call outside one is an inert no-op.

@capture.agent(suite="support_agent")
def answer(query: str) -> str:
    with capture.model_call(model_id="claude-opus-4-8", input=query) as rec:
        for chunk in stream:
            rec.add_text(chunk.text)
        rec.set_usage(input_tokens=812, output_tokens=140, cost_usd=0.012)
    return "done"
# exactly one model_call span recorded on exit (works with 'async with' too)

## Context-manager form

When a decorator doesn’t fit, capture.agent_session(...) (and capture.agent_session_async(...)) capture an inline block instead. They yield None when capture is disabled. Always pass agent_input — any JSON-able value; it is identity only, hashed into the envelope’s input_hash (the dedup key) and never stored raw. For a chat agent, pass the full messages list.

from evalshift import capture

with capture.agent_session(suite="support_agent", agent_input=messages) as session:
    ...  # session is None when the gate is off

The session’s scope is dynamic (contextvar-based), not lexical: once an agent wrapper or session is active, @capture.tool calls and model-call recorders attach to it from any function called inside it, however deep — recording calls don’t need to sit lexically inside the decorated function or with block. Sessions are also the recommended primitive for multi-turn conversations: one with block per turn, with a fresh turn_index.

## Failures are the best telemetry

When your agent raises, the SDK records an error event, still writes the partial capture, then re-raises the original exception unchanged. Your control flow is never altered.

See Redaction to control what those payloads contain, and Config, sinks & hygiene for where files land.

## Known limit

• +Tools run on a raw threading.Thread (rather than asyncio.to_thread / run_in_executor) start a fresh context and are silently uncaptured. Use the asyncio helpers so context copies into the worker.