webhookd-sdk (Python)

Official Python SDK for NimbusNexus Webhooks — publish events, manage your endpoints / keys / deliveries, and verify the webhooks you receive.

pip install webhookd-sdk

Verify an incoming webhook (subscribers)

When webhookd delivers a webhook it signs the body with your endpoint's signing secret. Always verify the signature before trusting the payload — it proves the request really came from webhookd and wasn't tampered with or replayed.

from webhookd_sdk import verify

# In your webhook handler — pass the RAW request body bytes (do not re-serialize the JSON):
ok = verify(
    secret=ENDPOINT_SIGNING_SECRET,
    raw_body=request.body,
    signature=request.headers["X-Webhook-Signature"],
    timestamp=request.headers["X-Webhook-Timestamp"],
)
if not ok:
    return Response(status_code=400)  # forged, tampered, or outside the 300s replay window

Publish an event (producers)

from webhookd_sdk import Client, WebhookdAPIError

with Client("https://webhooks.example.com", api_key="whsk_…") as wh:
    try:
        event = wh.publish(
            "order.created",
            {"order_id": "ord_123", "total": 4200},
            idempotency_key="order-123",   # makes the publish safe to retry
        )
        print(event.event_uid, event.deliveries_created)
    except WebhookdAPIError as e:
        print(e.status_code, e.code, e.message)   # the stable {error:{code,message}} envelope

Transient failures (connection errors, 429, 5xx) are retried with backoff (a 429 honours Retry-After); other 4xx raise WebhookdAPIError.

Manage endpoints, keys & deliveries (operators)

The same Client wraps the control-plane API — register receivers, mint keys, and drain the dead-letter queue from code (needs an admin-scoped key). Management methods return the raw JSON as dicts (snake_case, exactly as the API sends); list methods return a page — {"items": [...], "next_offset": int | None}; delete/revoke return None (a 204).

from webhookd_sdk import Client

wh = Client("https://webhooks.example.com", api_key="whsk_admin_…")

# --- Endpoints ----------------------------------------------------------------
# Create a receiver — its signing secret is in the response exactly once, so persist it now.
ep = wh.create_endpoint(
    "https://your-app.example/webhooks",
    subscriptions=[{"match_kind": "prefix", "pattern": "order."}],
    description="orders service",
)
endpoint_id, signing_secret = ep["id"], ep["secret"]

wh.list_endpoints(environment="prod")        # {"items": [...], "next_offset": ...}
wh.get_endpoint(endpoint_id)

# PATCH — send only the keys you want to change (omitted = unchanged, None = cleared):
wh.update_endpoint(endpoint_id, {"max_attempts": 10, "status": "disabled"})

wh.rotate_endpoint_secret(endpoint_id)       # returns the new secret, once
wh.enable_endpoint(endpoint_id)              # recover an auto-disabled endpoint
wh.delete_endpoint(endpoint_id)              # -> None (204)

# --- API keys -----------------------------------------------------------------
key = wh.create_api_key("ci-publisher", scope="publish", expires_in_days=90)
print(key["key"])                            # shown once
wh.revoke_api_key(key["id"])                 # -> None (204)

# --- Deliveries / dead-letter recovery ----------------------------------------
for d in wh.list_deliveries(status="dead")["items"]:
    wh.redeliver(d["id"])

Develop

pip install -e '.[dev]'
pytest && ruff check . && mypy webhookd_sdk