Telemetry

pip install "axonpush[otel]"

axonpush.telemetry is the OpenTelemetry-native path. It reuses your application’s own TracerProvider, attaches a BatchSpanProcessor, and ships spans as real OTLP over HTTP to POST {base}/v1/traces. The spans follow the GenAI semantic conventions (gen_ai.operation.name, gen_ai.request.model, gen_ai.usage.*), so they are portable to any OTLP backend, not just axonpush, and there is no proprietary event format to translate.

Quickstart

Set AXONPUSH_BASE_URL, AXONPUSH_API_KEY and AXONPUSH_CHANNEL_ID in the environment, then configure once at startup, wrap each model call in a GenAI span, and record usage when the response returns.

from axonpush.telemetry import (
    configure_telemetry,
    genai_span,
    record_genai_response,
    record_genai_content,
)

handle = configure_telemetry(
    service_name="my-agent",
    environment="prod",
)
tracer = handle.tracer()

with genai_span(
    tracer,
    operation="chat",
    request_model="gpt-4o",
    system="openai",
    agent_name="research-agent",
) as span:
    response = call_model(...)

    record_genai_response(
        span,
        response_model="gpt-4o",
        input_tokens=1200,
        output_tokens=350,
        reasoning_tokens=64,
        cache_read_tokens=900,
        cache_write_tokens=300,
    )
    record_genai_content(span, prompt=prompt, completion=text, handle=handle)

handle.flush()  # serverless: call at the end of each invocation

configure_telemetry resolves base_url, api_key and channel_id from the argument first, then the matching AXONPUSH_* variable. The exporter posts to {base}/v1/traces with X-API-Key for auth and X-Axonpush-Channel for routing. Every keyword is optional; a zero-argument call works once the environment is set.

Reuse the app’s TracerProvider

configure_telemetry never replaces a provider you already own:

• Pass tracer_provider= and it attaches to that.
• If the global provider is already a real SDK TracerProvider, it attaches to it.
• Otherwise it creates one with a resource describing the service (service.name, deployment.environment.name, service.version) and installs it as the global provider.

So an app that already has OpenTelemetry keeps its own instrumentation and just gains the axonpush exporter. Attaching for the same (endpoint, channel) twice is a no-op, so calling configure_telemetry from more than one module is safe.

Spanning a GenAI call

with genai_span(
    tracer,
    operation="chat",
    request_model="gpt-4o",
    system="openai",
    agent_name="research-agent",
) as span:
    ...

genai_span is a context manager that starts a CLIENT span named "{operation} {request_model}" (override with name=). It sets gen_ai.operation.name and gen_ai.request.model, and when supplied gen_ai.system / gen_ai.provider.name from system, and gen_ai.agent.name from agent_name.

record_genai_response records the response and usage attributes, including the cache-token counts that map to the semconv gen_ai.usage.cache_read_input_tokens and gen_ai.usage.cache_write_input_tokens:

record_genai_response(
    span,
    response_model="gpt-4o",
    finish_reasons=["stop"],
    input_tokens=1200,
    output_tokens=350,
    reasoning_tokens=64,
    cache_read_tokens=900,
    cache_write_tokens=300,
)

Content capture and redaction

Prompts and completions are off by default. record_genai_content emits them as span events (gen_ai.content.prompt / gen_ai.content.completion) rather than attributes, so large payloads do not inflate the span, and it is gated by the capture policy:

Credential-shaped keys are always stripped regardless of mode, and any keys you name in redact_keys are removed too. Set the policy once on the handle:

handle = configure_telemetry(
    service_name="my-agent",
    content_capture="redacted",
    redact_keys=["email", "ssn"],
)

record_genai_content(span, prompt=prompt, completion=text, handle=handle)

Or pass the policy per call with content_capture= and redact_keys= instead of handle=.

Flush on shutdown

The BatchSpanProcessor exports in the background. Drain it on a graceful exit, and on serverless, where the container is frozen between invocations, so the atexit hook is unreliable, call flush() at the end of each invocation. configure_telemetry logs a hint when it detects a serverless host.

handle.flush(timeout_ms=2000)  # returns True if it drained in time
handle.shutdown()              # flush and stop the processor; idempotent