axonpush

Python SDK for axonpush, observability and event infrastructure for AI agent systems.

Publish, trace, and deliver agent events. Drop-in integrations for LangChain, LangGraph Deep Agents, OpenAI Agents SDK, Anthropic, CrewAI, and the Python observability stack (stdlib logging, Loguru, structlog, OpenTelemetry, Sentry).

v1.0.0 is a breaking release. All IDs are str UUIDs; models live under a flat axonpush.models namespace; the realtime feature (MQTT / SSE / WebSocket / connect_realtime) has been removed. See CHANGELOG.md for the migration guide.

Install

pip install axonpush                  # or: uv add axonpush
pip install axonpush[langchain]       # LangChain / LangGraph
pip install axonpush[deepagents]      # LangChain Deep Agents
pip install axonpush[openai-agents]   # OpenAI Agents SDK
pip install axonpush[anthropic]       # Anthropic
pip install axonpush[crewai]          # CrewAI
pip install axonpush[loguru]          # Loguru sink
pip install axonpush[structlog]       # structlog processor
pip install axonpush[otel]            # OpenTelemetry SpanExporter
pip install axonpush[rq]              # Redis Queue durable backend
pip install axonpush[all]             # everything above

Installing the package also puts axonpush-eval on your PATH: the release gate that replays a dataset revision against a candidate in CI and exits non-zero when it regresses. Same flags and exit codes as the TypeScript and .NET builds. See the CLI reference.

Quick start

from axonpush import AxonPush, EventType

# Reads AXONPUSH_API_KEY / AXONPUSH_TENANT_ID / AXONPUSH_BASE_URL from env
# when the kwargs are omitted.
with AxonPush() as client:
    event = client.events.publish(
        "web_search",
        {"query": "AI agent frameworks"},
        channel_id="…channel uuid…",
        agent_id="researcher",
        event_type=EventType.AGENT_TOOL_CALL_START,
    )
    # event.event_id is server-assigned; event.queued is True within ~1 ms.

    listing = client.events.list(channel_id="…channel uuid…", limit=20)
    for ev in listing.data:
        print(ev.event_type, ev.identifier)

Async

import asyncio
from axonpush import AsyncAxonPush

async def main():
    async with AsyncAxonPush() as client:
        await client.events.publish(
            "web_search",
            {"query": "AI agents"},
            channel_id="…channel uuid…",
            agent_id="researcher",
            event_type="agent.tool_call.start",
        )

asyncio.run(main())

Zero-instrumentation: the LLM gateway

The fastest self-serve path to full observability is the axonpush gateway: no SDK, callback handler, or framework wrapper. Point an existing OpenAI (or Anthropic) client's base_url at the gateway and add the x-axonpush-api-key default header. Every call, tool call, cost, token count, and latency is captured, and any moderation / govern / spend policy runs inline, with zero code changes. It works alongside the framework integrations below.

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.axonpush.xyz/gw/openai",
    default_headers={"x-axonpush-api-key": os.environ["AXONPUSH_API_KEY"]},
    # api_key still reads OPENAI_API_KEY; the gateway forwards it upstream.
)

The path segment (/gw/openai or /gw/anthropic) selects the wire shape. Add x-axonpush-target: openrouter|anthropic|groq|together|vercel|openai to route to a different upstream provider, with that provider's key in the standard Authorization: Bearer <provider_key> header.

Configuration

Every kwarg falls back to an AXONPUSH_… env var; constructor kwargs win.

AxonPush(
    api_key="ak_…",        # AXONPUSH_API_KEY
    tenant_id="…",         # AXONPUSH_TENANT_ID
    base_url="https://api.axonpush.xyz",  # AXONPUSH_BASE_URL
    environment="prod",    # AXONPUSH_ENVIRONMENT
    timeout=30.0,          # AXONPUSH_TIMEOUT
    max_retries=3,         # AXONPUSH_MAX_RETRIES
    fail_open=False,       # AXONPUSH_FAIL_OPEN
)

fail_open=True swallows APIConnectionError and returns None from every resource call, useful when axonpush observability must never break the host application.

Authentication

The SDK supports two ingest credentials:

# Server-side: API key
client = AxonPush(api_key="ak_…", tenant_id="…")

# Untrusted / browser context: public ingest token
client = AxonPush(api_key="pt_…", tenant_id="…")

Both are passed via the api_key argument (or AXONPUSH_API_KEY); the SDK sends ak_ values on X-API-Key and pt_ values on X-Public-Token.

Resources

The client exposes Stripe-style resource accessors:

Every resource has an Async sibling on AsyncAxonPush, and is importable from axonpush.resources.

events.list() and events.search() return an EventListResponseDto with .data (list) and .meta (cursor + count).

Errors

from axonpush import (
    AxonPushError,            # base
    APIConnectionError,       # network / DNS / read timeout
    AuthenticationError,      # 401
    ForbiddenError,           # 403
    NotFoundError,            # 404
    ValidationError,          # 422 / code='validation_error'
    RateLimitError,           # 429, carries .retry_after
    ServerError,              # 5xx
    RetryableError,           # mixin: APIConnectionError, RateLimitError, ServerError
)

Every exception carries request_id, status_code, code, hint parsed from the backend's { code, message, hint, requestId } envelope. Anything that subclasses RetryableError is safe to retry; the SDK's transport already retries them up to max_retries with exponential backoff.

Integrations

All log/span integrations emit OpenTelemetry-shaped payloads (severityNumber, severityText, body, attributes, resource) so events line up with anything else you ship to an OTel-compatible backend.

Publishing modes

Every integration accepts a mode parameter:

from redis import Redis
from axonpush.integrations.langchain import AxonPushCallbackHandler

handler = AxonPushCallbackHandler(
    client, channel_id="…",
    mode="rq",
    rq_options={"redis_conn": Redis(), "queue_name": "axonpush"},
)

Then run rq worker axonpush somewhere.

Tracing

Group related events with a shared trace_id. The SDK auto-creates one per call, but you'll usually want to pin a trace to a logical request:

from axonpush import get_or_create_trace

trace = get_or_create_trace()
client.events.publish(
    "step.start", {"step": 1},
    channel_id="…",
    trace_id=trace.trace_id,
    span_id=trace.next_span_id(),
)

detail = client.traces_v2.detail(trace.trace_id)
summary = detail.summary
print(summary.event_count, summary.duration_ms, summary.tool_call_count)

get_or_create_trace() reads the active context (set via with TraceContext(...):) when one exists, so framework integrations propagate the trace automatically.

OpenTelemetry-native telemetry

axonpush.telemetry ships GenAI spans as real OTLP over HTTP to ${base}/v1/traces, routed by the X-Axonpush-Channel header and authed by X-API-Key. Spans follow the OpenTelemetry GenAI semantic conventions (gen_ai.*), so they are portable to any OTLP backend, not just axonpush. This is the recommended way to send traces.

pip install axonpush[otel]   # pulls opentelemetry-exporter-otlp-proto-http

configure_telemetry() reuses the application's own TracerProvider when one is already set globally (it never replaces it) and only creates one when the app does not own OpenTelemetry. Config resolves from arguments, then the matching AXONPUSH_BASE_URL / AXONPUSH_API_KEY / AXONPUSH_CHANNEL_ID env var.

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

# Reuses or creates a TracerProvider, attaches a batch OTLP/HTTP exporter,
# and stamps a Resource with service.name / deployment.environment.name / service.version.
handle = configure_telemetry(
    service_name="my-agent",
    environment="prod",
    service_version="1.4.0",
    content_capture="metadata_only",  # or "redacted" / "full"
)
tracer = handle.tracer()

with genai_span(tracer, operation="chat", request_model="gpt-4o", system="openai") as span:
    # ... call the model ...
    record_genai_response(
        span,
        response_model="gpt-4o",
        input_tokens=12,
        output_tokens=48,
        reasoning_tokens=0,
        cache_read_tokens=0,
        cache_write_tokens=0,
    )
    record_genai_content(span, prompt="…", completion="…", handle=handle)

handle.flush()  # or handle.shutdown() on clean exit

Prompt and completion land as span events (gen_ai.content.prompt / gen_ai.content.completion), gated by content_capture: metadata_only drops content, redacted keeps short previews, full keeps it. Credential-shaped keys are always stripped regardless of mode.

On serverless hosts (AWS Lambda and friends) the batch processor's atexit flush is unreliable because the container is frozen between invocations. configure_telemetry() logs a note when it detects one; call handle.flush() at the end of each invocation.

def handler(event, context):
    with genai_span(handle.tracer(), operation="chat", request_model="gpt-4o") as span:
        ...
    handle.flush()  # block until buffered spans ship

Which should I use

The OTel-native path above is the recommended way to send traces. The legacy per-framework event-model exporter (axonpush.integrations.otel.AxonPushSpanExporter, which maps calls to /event payloads) still works and is kept for compatibility, but it is being superseded by OTel-native. Frame it as the compatibility path: keep it if you already depend on channel fan-out on the events plane, otherwise reach for axonpush.telemetry. Traces from both land in the same dashboard (/v2/traces); the server reconciles them through the OTLP normalizer, so you can migrate call sites incrementally without a gap in your traces.

Examples

examples/ contains runnable recipes: quickstart, tracing, webhooks, async, error handling, plus one example per integration. Each reads AXONPUSH_API_KEY / AXONPUSH_TENANT_ID from your environment. See examples/README.md for the full table.

Advanced

For internal contracts, the resource ownership matrix, the OpenAPI codegen layer, and exception handling internals, see SHARED-CONTRACT.md.

License

MIT