axonpush
Python SDK

Traces

Give related events a shared trace id, then read the trace back, summary, events, and org-wide stats.

A trace is a set of events sharing one trace_id. The SDK creates one for you on the first publish, but for anything multi-step you want to pin the trace yourself so every step lands under the same id.

Tracing GenAI model calls? The recommended path is OpenTelemetry-native telemetry: reuse your own TracerProvider and emit standard gen_ai.* spans over OTLP instead of event-model traces.

Pin a trace to a unit of work

from axonpush import AxonPush, EventType, get_or_create_trace

with AxonPush() as client:
    trace = get_or_create_trace()

    client.events.publish(
        "research.start",
        {"goal": "Find recent papers on transformer architectures"},
        channel_id,
        agent_id="research-agent",
        trace_id=trace.trace_id,
        span_id=trace.next_span_id(),
        event_type=EventType.AGENT_START,
    )

    client.events.publish(
        "web_search",
        {"query": "transformer architecture papers 2026"},
        channel_id,
        agent_id="research-agent",
        trace_id=trace.trace_id,
        span_id=trace.next_span_id(),
        event_type=EventType.AGENT_TOOL_CALL_START,
        metadata={"tool_name": "web_search"},
    )

get_or_create_trace() returns the trace already active on the current context, or creates and installs one. Because it is stored in a contextvars.ContextVar, each asyncio task, and each thread that copies its parent’s context, sees its own value, so concurrent requests do not braid their traces together.

Pass an id to adopt one that came from elsewhere, such as a request header:

trace = get_or_create_trace("3f1c9d84-6a2b-4c7e-9f10-2b8d5e6a1c47")

That installs the id as the current context and returns it. current_trace() reads the active context without creating one, returning None when there is none.

TraceContext is a plain dataclass, not a context manager, with TraceContext(...) raises. To scope a trace, call get_or_create_trace(id) at the top of the unit of work; the context var keeps it isolated per task.

TraceContext

MemberReturnsWhat it is
trace_idstrA UUID4 string, generated on construction unless supplied.
next_span_id()strA fresh 16-character lowercase hex span id.
w3c_trace_id()strThe trace id as a 32-hex W3C id, hashed if it is not already hex.
traceparent(span_id=None)strA sampled W3C traceparent header value.

You rarely build spans by hand. The transport already injects X-Axonpush-Trace-Id, X-Axonpush-Span-Id and traceparent on every request while a trace is active, and the framework integrations map their own run ids onto span_id and parent_event_id so a nested chain lands as a tree.

Read a trace back

Reading is the v2 trace surface, reached at client.traces_v2 in Python and client.tracesV2 in TypeScript. It models spans as spans, which is what you want for a waterfall.

detail = client.traces_v2.detail(trace.trace_id)
summary = detail.summary

print(summary.event_count, summary.duration_ms, summary.tool_call_count)
print(summary.agents, summary.models)

detail() returns the trace’s summary (a TraceSummaryV2) alongside its span hierarchy. Ingest is eventually consistent, a detail requested in the same millisecond as the last publish may not see it yet.

TraceSummaryV2

FieldTypeMeaning
trace_idstrThe trace id.
event_countfloatEvents in the trace.
agentslist[str]Distinct agent ids that contributed.
models / providerslist[str]Distinct models and providers seen.
toolslist[str]Distinct tools called.
start_time / end_timedatetimeFirst and last event.
duration_msfloatWall time in milliseconds.
error_countfloatError events.
tool_call_countfloatTool-call events.
handoff_countfloatAgent handoffs.
statusstrTrace status, when the backend derived one.
input_tokens / output_tokens / total_tokensfloatToken counts, when known.
cost_usdfloatAttributed cost, when known.

The counts come back as float because the backend serialises them as JSON numbers; wrap in int() for display.

The events in a trace

for ev in client.traces_v2.events(trace.trace_id).data:
    print(ev.span_id, ev.identifier, ev.event_type)

Returns the full stored event records, in order. For the span tree instead of raw events, use client.traces_v2.spans(trace.trace_id).

List and search traces

page = client.traces_v2.list({"limit": 20, "environment": "prod"})

for t in page.data:
    print(t.trace_id, int(t.event_count), f"{t.duration_ms:.0f}ms")

if page.meta.has_more:
    ...  # pass page.meta.cursor on the next call

list() takes the query as a mapping. Alongside limit, cursor, environment and app_id it accepts the facet filters, model, provider, tool, service, release, status, min_cost_usd / max_cost_usd, min_duration_ms / max_duration_ms, and so on. Each item is a TraceSummaryV2; meta carries cursor, has_more, limit and total.

To discover which filters are worth applying, ask the API what actually appears in the window:

facets = client.traces_v2.facets({"environment": "prod"})
keys = client.traces_v2.attribute_keys({"environment": "prod"})

Dashboard stats

stats = client.traces_v2.stats(environment="prod")

print(stats.total_traces, stats.traces_today)
print(stats.total_events, stats.events_today)
print(stats.error_count, stats.error_rate, stats.avg_trace_duration)

stats() (backed by GET /v2/traces/stats) takes app_id and environment. stats.events_by_hour is a bucketed series for the last window, suitable for plotting straight onto a chart.

Async

Every method has an async sibling:

async with AsyncAxonPush() as client:
    detail = await client.traces_v2.detail(trace_id)
    events = await client.traces_v2.events(trace_id)