Traces & spans
Start traces and add spans, alerts, groups, messages and metadata with the Recount Python SDK: start_trace, add_span and init.
recount.init(options=None)
Optionally configure the SDK. The SDK reads RECOUNT_API_KEY from the environment automatically, so init is only needed to override defaults.
import recount
recount.init({
"apiKey": "your-api-key", # overrides RECOUNT_API_KEY env var
"apiUrl": "https://...", # override the ingest endpoint
"environment": "production", # overrides RECOUNT_ENVIRONMENT env var
"debug": True,
})If no API key is found, the SDK disables itself. All calls become no-ops with no errors thrown.
Options
| Option | Type | Description |
|---|---|---|
apiKey | str | Your Recount API key |
apiUrl | str | Override the default ingest URL |
environment | str | Tag all traces with this environment (e.g. 'production'). Reads RECOUNT_ENVIRONMENT env var |
debug | bool | Log SDK operations to stdout. Reads RECOUNT_DEBUG env var |
recount.start_trace(name, fn)
Wraps an async function in a trace. The trace is sent to Recount when fn resolves or raises.
async def assess_claim():
# work happens here
...
await recount.start_trace("Assess claim", assess_claim)name must be 5–500 characters.
Pass options as the second argument, with the callback as the third:
async def trace_fn():
# work happens here
...
await recount.start_trace(
"Insurance claim - CLM-2024-03941",
{
"metadata": {"claimId": "CLM-2024-03941", "policyNumber": "POL-8821-AU"},
"retentionPeriod": "30d",
},
trace_fn,
)Options
| Option | Type | Description |
|---|---|---|
metadata | dict[str, str | int | float | bool | None] | Filterable key-value metadata |
environment | str | Override the environment for this trace |
retentionPeriod | RetentionPeriod | How long to keep the trace. Default: '30d' |
Retention periods
RetentionPeriod = Literal["1d", "7d", "30d", "90d", "1y"]recount.named(name, fn)
Sets the span name for the next auto-instrumented call inside fn. Integrations like wrap_anthropic, wrap_openai, and wrap_collection pick up this name instead of their default (model name, operation name, etc.).
result = 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' instead of 'claude-sonnet-4-6'The name applies to exactly one span - the first auto-instrumented call that runs inside fn. Subsequent calls revert to their default naming.
| Parameter | Type | Description |
|---|---|---|
name | str | Span name to use for the next auto-instrumented call |
fn | Callable[[], Awaitable[T]] | Callback containing the instrumented call |
Returns the result of fn.
recount.add_span(type_, name, ...)
Add a span to the currently active trace. Must be called inside a start_trace callback.
There are two forms: static and callback.
Static form
Pass an options dict to set input, output, durationMs, and other fields manually:
recount.add_span("db", "Policy lookup", {
"metadata": {"table": "policies", "queryType": "sql"},
"input": {
"query": "SELECT * FROM policies WHERE number = $1",
"params": ["POL-8821-AU"],
},
"output": {"rows": 1},
})Callback form
Pass an async callback and Recount handles timing and output automatically. The span's durationMs is set from how long the callback takes, and its output is set from the callback's return value.
async def load_policy():
return await db.policies.find_one({"number": "POL-8821-AU"})
# ^ becomes the span output automatically
policy = await recount.add_span("json", "Load policy", load_policy)Pass options as the third argument to include input, redact, or metadata:
async def load_policy():
return await db.policies.find_one({"number": "POL-8821-AU"})
await recount.add_span(
"json",
"Load policy",
{"input": {"policyNumber": "POL-8821-AU"}, "redact": ["holder.name"]},
load_policy,
)Empty span dropping
Spans with no input and no output are silently dropped at ingest time (except message spans, which carry their text separately). Groups that produce no surviving children are also dropped. If all spans are dropped, the trace is not sent.
This means you can safely call add_span inside conditional logic; spans that contribute nothing are discarded automatically.
| Span type | Drop condition |
|---|---|
json, table, chart, http, llm, image, db | No input and no output |
group | All child spans were dropped |
message | Never dropped |
Set debug: True in recount.init() (or RECOUNT_DEBUG=1) to log dropped spans to stdout.
Overloads
| Signature | Description |
|---|---|
add_span(type_, name, options=None) | Static - set all fields manually |
add_span(type_, name, fn) | Callback - auto-timing, auto-output |
add_span(type_, name, opts, fn) | Callback with options |
Span types
| Type | Use for |
|---|---|
message | Plain text output |
json | Structured JSON data |
table | Tabular data (use recount.table()) |
chart | Charts (use recount.chart()) |
http | HTTP requests (usually via recount.fetch()) |
llm | LLM calls, see LLM spans |
image | Images rendered inline, see Image spans |
db | Database queries, see Database spans |
group | Group child spans together, see Groups |
Options
| Option | Type | Description |
|---|---|---|
description | str | Short description shown below the span header in the dashboard (max 1,000 chars) |
input | Any | Input to this step |
output | Any | Output from this step |
durationMs | int | How long this step took |
metadata | DbSpanMetadata | When type_="db", set collection, table, or queryType |
redact | list[str] | Fields to redact from input/output. See Redaction |
public | bool | Mark this span as public (visible to end users in shared views) |
recount.add_alert(label, level, options=None)
Flag the most recently added span with a warn or error badge. Call it right after the add_span (or add_message) it applies to - there's no built-in threshold, so decide with a plain if:
recount.add_span("json", "Query time", {"output": {"durationMs": 5240}})
if duration_ms > 1000:
recount.add_alert("Query duration (ms)", "warn", {
"checks": [{"value": duration_ms}],
})See Alerts for the full API, including multiple checks and usage inside groups.
recount.add_group(name, builder, options=None)
Group related spans together under a named, collapsible section. See Groups for full details.
def build_group(g):
g.span("table", "Current plan", {"output": recount.table_from_object(current_plan)})
g.span("table", "Recommended plan", {"output": recount.table_from_object(best_plan)})
recount.add_group("Plan comparison", build_group)recount.add_message(name, text=None, opts=None)
Shorthand for adding a message span. Useful for log-style checkpoints.
recount.add_message("Validation passed", "All policy checks cleared", {"level": "info"})
recount.add_message(
"High-value claim",
"Claim exceeds $10,000 - flagged for review",
{"level": "warn"},
)
# text is optional - name alone is enough
recount.add_message("Cache warmed")Options
| Option | Type | Description |
|---|---|---|
description | str | Short description shown below the span header in the dashboard (max 1,000 chars) |
level | "info" | "warn" | "error" | Severity. Default: "info" |
public | bool | Mark this span as public |
recount.add_metadata(data)
Attach additional metadata to the active trace. Useful when metadata is only known partway through execution.
async def trace_fn():
application = await load_application(id)
# Attach identifiers once we have them
recount.add_metadata({
"applicationId": application["id"],
"customerId": application["customerId"],
})
# ... rest of logic
await recount.start_trace("Process application", trace_fn)Metadata values must be strings, numbers, booleans, or None. Metadata is merged with any metadata set on start_trace.
| Parameter | Type | Description |
|---|---|---|
data | dict[str, str | int | float | bool | None] | Key-value pairs to merge into trace metadata |