Client

AxonPush and AsyncAxonPush are the two entry points. They take the same keywords and expose the same resource accessors; the async one returns coroutines and is closed with await.

Construct one

from axonpush import AxonPush

with AxonPush() as client:
  client.events.publish("boot", {"ok": True}, channel_id)

With no arguments the client reads AXONPUSH_API_KEY, AXONPUSH_TENANT_ID and AXONPUSH_BASE_URL from the environment. Pass them explicitly when you manage credentials yourself:

client = AxonPush(
    api_key="ak_...",
    tenant_id="3f1c…",          # your org UUID
    base_url="https://api.axonpush.xyz",
    environment="prod",
)

Outside a with block, close it yourself, client.close() for the sync client, await client.close() for the async one. Both are idempotent.

Keywords

All are keyword-only, and all fall back to the matching AXONPUSH_* environment variable.

The resolved configuration is available as client.settings, a frozen Settings model. client.environment and client.fail_open are shortcuts to the two fields you are most likely to assert on.

Fail-open

fail_open defaults to True: if the API is unreachable, DNS failure, refused connection, read timeout, the SDK swallows APIConnectionError and the resource call returns None instead of raising. Observability never takes down the host application.

client = AxonPush()                 # fail_open=True
result = client.events.publish(...) # None when the API is unreachable

This applies only to connection failures. A 401, a 422 or a 500 still raises; those are bugs in the call, not in the network.

Because every resource method can return None, type checkers will make you handle it. In one-shot scripts an assert is the usual answer; in library code check for None.

client = AxonPush(fail_open=False)

Suppressed and background failures are logged through the stdlib logging module under the axonpush logger, and queue drops under axonpush.publisher:

import logging
logging.getLogger("axonpush").setLevel(logging.WARNING)

A publish rejected by the server for a configuration reason, an unknown environment slug, a channel UUID that does not exist, is logged at ERROR with the server’s hint attached, rate-limited to one line per minute per error code, because that class of failure never fixes itself.

Redaction

Before a payload leaves the process the client walks it and applies three rules:

1. Keys that look like credentials, authorization, cookie, password, secret, api_key, access_token, private_key and friends, always become [REDACTED].
2. Under the default content_capture_mode="metadata_only", keys that carry model content, prompt, prompts, message, messages, completion, input, output, response, tool_arguments, tool_result, retrieval_documents, also become [REDACTED]. Set content_capture_mode="full" to keep them, or content_capture_mode="redacted" to keep the first 256 characters of each, suffixed …[REDACTED_PREVIEW].
3. Any string longer than max_content_length is truncated and suffixed with …[TRUNCATED].

redact_keys adds your own key names to rule 1:

client = AxonPush(
    content_capture_mode="full",
    redact_keys=["customer_email", "ssn"],
    max_content_length=16_384,
)

The rules apply to both payload and metadata on events.publish.

Errors

Everything the SDK raises subclasses AxonPushError.

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

Each instance carries status_code, code, hint and request_id, parsed from the backend’s error envelope. request_id falls back to the X-Request-Id response header when the body omits it, quote it in a support ticket.

RetryableError is the branch for transient failures. The transport already retries them up to max_retries (honouring Retry-After on a 429), so catching it means the retries are already spent:

import time
from axonpush import AuthenticationError, AxonPushError, RateLimitError, RetryableError

try:
    client.events.publish(...)
except AuthenticationError:
    raise                                   # a key problem - do not retry
except RateLimitError as exc:
    time.sleep(exc.retry_after or 1.0)
except RetryableError:
    ...                                     # transient - back off and try later
except AxonPushError as exc:
    log.error("axonpush rejected the call: %s (request_id=%s)", exc, exc.request_id)

Framework integrations never raise. Every callback catches its own exceptions and logs them, so a broken publish cannot break the agent it is watching.

The async client across event loops

AsyncAxonPush builds its httpx.AsyncClient lazily and rebuilds it whenever the running event loop changes. That makes it safe in the serverless and worker patterns that drive each task through a fresh asyncio.run(...): a module-level client constructed once keeps working across invocations rather than stalling on the previous loop’s closed primitives.

client = AsyncAxonPush()          # module level, constructed once

def lambda_handler(event, context):
    return asyncio.run(handle(event))   # a new loop every invocation