LLM spans

Trace Anthropic and OpenAI calls in Python automatically - prompts, responses, models and token counts, streaming included.

Anthropic - `wrap_anthropic`

Wrap your async Anthropic client once and every messages.create() call is traced automatically. Import from recount.integrations.anthropic:

from anthropic import AsyncAnthropic
from recount.integrations.anthropic import wrap_anthropic

anthropic = wrap_anthropic(AsyncAnthropic())

Pass redact to strip sensitive fields from all spans created by this client:

anthropic = wrap_anthropic(AsyncAnthropic(), redact=["user.ssn"])

Then use it exactly as before:

async def trace_fn():
    # llm span recorded automatically - messages, model, response, tokens
    message = await anthropic.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )

    result = parse_response(message.content[0].text)


await recount.start_trace("Assess claim", trace_fn)

Each span captures the provider, model, messages (and system prompt if set), response text, token counts, and stop reason.

Naming spans

By default the span name is the model identifier (e.g. claude-sonnet-4-6). Wrap the call with recount.named() to give it a meaningful name:

message = await recount.named(
    "Classify intent",
    lambda: anthropic.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    ),
)
# -> llm span named 'Classify intent'

Streaming

Pass stream=True as normal - the wrapped client returns an async generator that yields events as they arrive, and records the span once the stream finishes:

stream = await anthropic.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=messages,
    stream=True,
)

text = ""
async for event in stream:
    if event.type == "content_block_delta" and event.delta.type == "text_delta":
        text += event.delta.text
# span recorded here, after the stream is consumed

Install

pip install "recount-sdk[anthropic]"

AsyncAnthropic is required - the wrapper awaits create() internally, so a synchronous Anthropic client will not work.

OpenAI - `wrap_openai`

Same pattern for OpenAI. Import from recount.integrations.openai:

from openai import AsyncOpenAI
from recount.integrations.openai import wrap_openai

openai = wrap_openai(AsyncOpenAI())

Pass redact to strip sensitive fields from all spans created by this client:

openai = wrap_openai(AsyncOpenAI(), redact=["user.ssn"])

Every chat.completions.create() call is traced automatically:

async def trace_fn():
    # llm span recorded automatically
    completion = await openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )


await recount.start_trace("Generate summary", trace_fn)

Naming spans

Same as Anthropic - wrap with recount.named() to override the default model-name span:

completion = await recount.named(
    "Summarise claim",
    lambda: openai.chat.completions.create(model="gpt-4o", messages=messages),
)
# -> llm span named 'Summarise claim'

Streaming

Pass stream_options={"include_usage": True} to capture token counts in the stream:

stream = await openai.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    stream=True,
    stream_options={"include_usage": True},
)

text = ""
async for chunk in stream:
    delta = chunk.choices[0].delta.content if chunk.choices else None
    if delta:
        text += delta
# span recorded here

Install

pip install "recount-sdk[openai]"

AsyncOpenAI is required - the wrapper awaits create() internally, so a synchronous OpenAI client will not work.

Manual spans

For other providers or when you need full control, use add_span with type_="llm" directly.

recount.add_span("llm", "Classify intent", {
    "input": {
        "messages": [
            {"role": "system", "content": "You are a claims assessor."},
            {"role": "user", "content": "Is this claim fraudulent? ..."},
        ],
        "provider": "anthropic",
        "model": "claude-sonnet-4-6",
    },
    "output": {"response": "No indicators of fraud detected."},
    "durationMs": 340,
})

Plain prompt

If you're not using a chat-style API, pass prompt instead of messages:

recount.add_span("llm", "Summarise claim", {
    "input": {
        "prompt": "Summarise the following claim in one sentence: ...",
        "provider": "openai",
        "model": "gpt-4o",
    },
    "output": {"response": "Rear-end collision claim for $7,590 damage."},
    "durationMs": 820,
})

Callback form

Use the callback form to auto-set output and timing:

async def call_llm():
    async with httpx.AsyncClient() as client:
        res = await client.post(
            "https://api.yourprovider.com/generate",
            json={"messages": messages},
        )
        return {"response": res.json()["text"]}


result = await recount.add_span(
    "llm",
    "Classify intent",
    {"input": {"messages": messages, "provider": "anthropic", "model": "claude-sonnet-4-6"}},
    call_llm,
)

Input fields

Field	Type	Description
`messages`	`list[LlmMessage]`	Chat messages (role + content pairs)
`prompt`	`str`	Plain string prompt (alternative to messages)
`system`	`str`	System prompt
`provider`	`str`	Provider name, e.g. `'anthropic'`
`model`	`str`	Model identifier, e.g. `'gpt-4o'`

Output fields

Field	Type	Description
`response`	`str \| dict`	The model's response
`inputTokens`	`int`	Prompt token count
`outputTokens`	`int`	Completion token count
`stopReason`	`str`	Why the model stopped generating

input and output are plain dicts - the SDK doesn't export dedicated LlmSpanInput/LlmSpanOutput types, so pass the fields above directly.

On this page