philiprehberger-webhook-relay-client 0.1.0

Python SDK + HMAC verifier for the Webhook Relay API

philiprehberger-webhook-relay-client

Python SDK + HMAC verifier for the Webhook Relay API. Python 3.10+, zero runtime dependencies, PEP 561 typed.

Installation

pip install philiprehberger-webhook-relay-client

Verify an incoming webhook (receiver side)

from philiprehberger_webhook_relay_client import verify_signature

def webhook_handler(request):
    raw = request.body  # bytes — DO NOT json.loads + re-dumps
    if not verify_signature(
        secret=os.environ["WEBHOOK_SECRET"],
        body=raw,
        header=request.headers.get("X-Webhook-Signature"),
    ):
        return Response("Bad signature", status=400)

    event = json.loads(raw)
    # ... handle event

The body must be the exact bytes received. json.dumps(json.loads(...)) will reorder keys or change whitespace and break the signature.

The format matches Stripe and Svix (t=<ts>,v1=<hex> over "{ts}.{body}" with HMAC-SHA256), so the same verifier accepts signatures from any sender using that convention.

Send an event (sender side)

from philiprehberger_webhook_relay_client import WebhookRelayClient

relay = WebhookRelayClient(api_key=os.environ["WEBHOOK_RELAY_KEY"])

event = relay.ingest(
    event_type="order.created",
    payload={"order_id": 42},
    idempotency_key="order-42-created",
)
print(event["id"], event["deliveries_summary"])

The client raises WebhookRelayError on 4xx/5xx with the RFC 7807 problem payload preserved:

from philiprehberger_webhook_relay_client import WebhookRelayError

try:
    relay.ingest(event_type="", payload={})
except WebhookRelayError as err:
    print(err.status, err.title, err.detail)

Subscriptions, deliveries, and the rest

sub = relay.create_subscription(
    url="https://my-app.example.com/webhooks",
    name="orders inbound",
    event_filter="order.*",
)
print(sub["signing_secret"])    # store this now — shown only once

page = relay.list_deliveries(status="failed")

relay.pause_subscription(sub["id"])
relay.resume_subscription(sub["id"])
rotated = relay.rotate_subscription_secret(sub["id"])

For endpoints the typed surface doesn't cover (manual retry, dead-letters, webhook test probe) drop down to request():

relay.request("POST", f"/v1/deliveries/{delivery_id}/retry")

Compatible senders

Use sign_body to stand up a compatible sender for testing, or to verify your verifier matches the wire format:

from philiprehberger_webhook_relay_client import sign_body

signed = sign_body("whsec_shared", raw_body)
requests.post(receiver_url, data=raw_body, headers={"X-Webhook-Signature": signed.header})

Pointing at a different host

relay = WebhookRelayClient(
    api_key=api_key,
    base_url="https://relay.staging.internal",
)

Links

• API: https://api.webhook-relay.dcsuniverse.com
• Docs: https://webhook-relay.dcsuniverse.com
• OpenAPI spec: https://webhook-relay.dcsuniverse.com/openapi.yaml
• Source: https://github.com/philiprehberger/py-webhook-relay-client

License

MIT