Webhooks

A webhook endpoint is attached to one channel. When an event is published there and passes the endpoint’s event-type filter, the backend POSTs it to your URL and records the attempt.

Register an endpoint

from axonpush import AxonPush

with AxonPush() as client:
    endpoint = client.webhooks.create_endpoint(
        url="https://your-server.example/axonpush",
        channel_id=channel_id,
        event_types=["agent.error"],
        description="Page on agent errors",
    )

    print(endpoint.endpoint_id, endpoint.url, endpoint.active)

Every argument is keyword-only.

Keyword	Type	Required	What it does
`url`	`str`	yes	Where axonpush POSTs.
`channel_id`	`str`	yes	Source channel UUID.
`event_types`	`list[str]`	no	Only these types fire. Omit to receive every event on the channel.
`secret`	`str`	no	Signing secret. The server generates one when omitted.
`description`	`str`	no	Free-form note, shown in the dashboard.

The signing secret is shown once

The create response, and only the create response, carries raw_secret. Store it at that moment; later reads expose signing_secret_prefix and has_secret, never the value.

endpoint = client.webhooks.create_endpoint(url=..., channel_id=...)

if endpoint.raw_secret:
    save_to_vault(endpoint.raw_secret)     # your only chance

WebhookEndpointCreateResponseDto is a different shape from the endpoint records list_endpoints() returns, it is the only one with raw_secret on it. If you skip storing it, delete the endpoint and create a new one.

List and delete

for ep in client.webhooks.list_endpoints(channel_id):
    print(ep.endpoint_id, ep.url, ep.active, ep.event_types)

client.webhooks.delete_endpoint(endpoint_id)

list_endpoints takes the channel UUID; delete_endpoint takes the endpoint UUID. Endpoint records carry endpoint_id, id, url, channel_id, org_id, active, event_types, description, has_secret, signing_secret_prefix, created_at and updated_at.

The field is active, not is_active. The runtime was always sending active; the schema was corrected to match in v0.0.14.

Inspect deliveries

for d in client.webhooks.deliveries(endpoint_id):
    print(d.status, int(d.attempts), d.status_code, d.error)

Field	Type	Meaning
`delivery_id`	`str`	This attempt record.
`endpoint_id`	`str`	Endpoint it was sent to.
`event_id`	`str`	Event that triggered it.
`status`	`DeliveryStatus`	`pending`, `success`, `failed` or `retrying`.
`attempts`	`float`	Attempts so far.
`status_code`	`float`	HTTP status your server returned.
`error`	`str`	Transport or application error, when one occurred.
`response_body`	`str`	What your server replied.
`last_attempt_at` / `next_attempt_at`	`datetime`	Retry scheduling.
`created_at`	`datetime`	When the delivery was queued.

DeliveryStatus is a str enum, so d.status == "failed" works:

from axonpush import DeliveryStatus

failed = [d for d in deliveries if d.status is DeliveryStatus.FAILED]

Return a 2xx to acknowledge. Anything else is retried on a backoff, and the attempts show up in this log.

Async

async with AsyncAxonPush() as client:
    endpoint = await client.webhooks.create_endpoint(
        url="https://your-server.example/axonpush",
        channel_id=channel_id,
        event_types=["agent.error"],
    )
    deliveries = await client.webhooks.deliveries(endpoint.endpoint_id)

The request axonpush sends

The delivery body and signature header are defined by the backend, not the SDK, see the webhook API reference for the exact contract your handler has to satisfy.