ankusa

The client SDK for Ankusa deployments: one PyPI package meant to bundle everything a non-Elixir consumer needs to talk to an Ankusa deployment. Today that's the claim-check gateway client and a webhook-receiving header helper; more clients (ingest, admin) land here as they're built.

Install

Not published yet. Until the first release, depend on it as a local path, the same way the Elixir packages in this monorepo depend on ankusa core before their first Hex release, and packages/sdk-typescript depends on itself via file:.

With uv:

[project]
dependencies = ["ankusa"]

[tool.uv.sources]
ankusa = { path = "../../packages/sdk-python" }

Or with plain pip:

pip install -e ../../packages/sdk-python

Claim-check client

Redeem a claim-check ref, verify the bytes it returns against the message's sha256, and classify failures into dead-letter vs. retry, without holding any object-store credentials. Conforms to the framework's own contract, priv/openapi/claim_check.v1.yaml: the spec is the source of truth, this package conforms to it, not the reverse.

import os

from ankusa import ClaimCheckClient, ClaimCheckError

claim_check = ClaimCheckClient(os.environ.get("CLAIM_CHECK_URL", "http://localhost:4001"))

# A queue message that carries a claim also carries its sha256:
#   "claim":  "urn:ankusa:claim:v1:<tenant>:<claim_id>"  (claim_id: uppercase ULID)
#   "sha256": 64-char lowercase hex of the claim's bytes
def resolve_body(message: dict) -> bytes:
    try:
        return claim_check.redeem(message["claim"], message["sha256"])
    except ClaimCheckError as err:
        if not err.retryable:
            # bad ref/sha256, 404, or an integrity mismatch: dead-letter, don't requeue
            raise
        # gateway unreachable or 5xx: safe to retry
        raise

redeem() does three things GET /v1/claims/... alone doesn't:

1. Parses the ref into its tenant id, claim id, and gateway path (GET /v1/claims/{tenant_id}/{claim_id}) (parse_claim_ref, also exported standalone).
2. Fetches the bytes.
3. Verifies them against the message's sha256 (the gateway itself does not check this, see "Redeem a claim" in docs/claim-check.md) before ever returning them to you.

Every failure is a ClaimCheckError subclass with a retryable attribute, so a consumer needs exactly one bit to decide dead-letter vs. retry:

health() hits GET /health for a liveness probe.

ClaimCheckClient owns an httpx.Client; use it as a context manager (with ClaimCheckClient(...) as claim_check:) or call .close() yourself when done.

Webhook receiver helper

Every receiver of Ankusa's HTTP sink needs the same handful of headers off each request; parse_headers replaces the hand-rolled self.headers.get(...) calls with one call and a typed result:

from ankusa import MissingHookIdError, parse_headers

def do_POST(self):
    try:
        hook = parse_headers(self.headers)
    except MissingHookIdError:
        self.send_response(400)
        self.end_headers()
        return

    body = self.rfile.read(int(self.headers.get("content-length", "0")))
    # hook.id, hook.source, hook.seq, hook.tenant, hook.content_type
    ...

HookHeaders.id is what a receiver dedupes on: delivery is at-least-once (see "HTTP handoff" in docs/integrations.md), so the same hook can arrive twice after a retry. Header lookup is always case-insensitive, regardless of whether the mapping passed in already is.

examples/quickstart/worker.py uses this.

Routes client

Manage route definitions and the global IP rules on the route-management listener (routes.admin.port, default 4003) — the routes tag of priv/openapi/admin.v1.yaml.

import os

from ankusa import RoutesClient

routes = RoutesClient(os.environ.get("ROUTES_URL", "http://localhost:4003"))

routes.create_route({"id": "stripe", "path": "/webhooks/stripe"})
routes.get_ip_rules()   # {"default": "allow", "rules": []}
routes.test_route({"method": "POST", "path": "/webhooks/stripe", "ip": "203.0.113.7"})

Methods: health(), list_routes(), create_route(), get_route(), replace_route(), update_route(), delete_route(), get_ip_rules(), put_ip_rules(), test_route(). Failures are RoutesError subclasses with a retryable attribute: RouteNotFoundError (404), RoutesRejectedError (400/409, carrying code, field, message, conflicting_id, max_routes), and RoutesUnavailableError (5xx/unreachable, retryable).

Admin client

The operator API on admin.port (default 4002): health, Prometheus metrics, the redacted config, the DLQ, and the quarantine list — the operations, dlq, and quarantine tags of admin.v1.yaml.

import os

from ankusa import AdminClient

admin = AdminClient(os.environ.get("ADMIN_URL", "http://localhost:4002"))

admin.health()                          # {"status": "ok", "instance": ..., "roles": [...]}
admin.list_dead_letters(limit=10)
admin.replay_dead_letters({"source_id": "demo"})
admin.list_quarantined()

Methods: health(), metrics() (Prometheus text), config(), list_dead_letters(), replay_dead_letters(), list_quarantined(). Failures are AdminError subclasses: RoleNotEnabledError (409, carrying role), AdminRejectedError (400, carrying code), and AdminUnavailableError (5xx/unreachable, retryable).

Layout

src/ankusa/
  __init__.py           # umbrella barrel: re-exports every client this package bundles
  py.typed
  webhook.py             # x-ankusa-* header parsing for HTTP-sink receivers
  claim_check/           # the claim-check gateway client
    __init__.py          # barrel for this client
    client.py
    ref.py
    errors.py
  routes/                # the route-management client (routes.admin.port)
    __init__.py
    client.py
    errors.py
  admin/                 # the operator client (admin.port)
    __init__.py
    client.py
    errors.py
tests/
  test_client.py
  test_ref.py
  test_webhook.py
  test_routes.py
  test_admin.py

A future client (say, an ingest helper) gets its own src/ankusa/<name>/ directory with the same shape, re-exported from src/ankusa/__init__.py.

Develop

uv sync
uv run pytest
uv run mypy