outworx-hooks 1.3.1

Lightweight webhook monitoring SDK for Python

outworx-hooks

Lightweight webhook monitoring SDK for Python -- know exactly when your webhooks break.

Works with FastAPI, Flask, and Django. Monitor everything from your dashboard at hooks.outworx.io.

Installation

pip install outworx-hooks

Quick Start

1. Initialize

Add your API key once at app startup, or set the OUTWORX_HOOKS_API_KEY environment variable.

import outworx_hooks

outworx_hooks.init(api_key="your-api-key")

Or via environment variable:

OUTWORX_HOOKS_API_KEY=your-api-key

2. Wrap Your Webhook Handler

FastAPI

from fastapi import FastAPI
from outworx_hooks.integrations.fastapi import OutworxHooksMiddleware
from outworx_hooks import TrackOptions

app = FastAPI()

app.add_middleware(
    OutworxHooksMiddleware,
    options=TrackOptions(provider="stripe", event_type_field="type")
)

@app.post("/webhooks/stripe")
async def handle_webhook():
    # ... process webhook
    return {"status": "ok"}

Flask

from flask import Flask
from outworx_hooks.integrations.flask import with_webhook_monitoring
from outworx_hooks import TrackOptions

app = Flask(__name__)

@app.route("/webhooks/stripe", methods=["POST"])
@with_webhook_monitoring(TrackOptions(provider="stripe", event_type_field="type"))
def handle_webhook():
    # ... process webhook
    return {"status": "ok"}

Django

Add the middleware to your MIDDLEWARE setting:

# settings.py
MIDDLEWARE = [
    ...
    'outworx_hooks.integrations.django.OutworxHooksMiddleware',
]

OUTWORX_HOOKS_OPTIONS = {
    'provider': 'stripe',
    'event_type_field': 'type',
    'capture_body': True
}

Configuration

import outworx_hooks

outworx_hooks.init(
    api_key="your-api-key",
    endpoint="https://hooks.outworx.io/api/ingest", # default
    debug=True, # log events to console
    timeout=3000, # HTTP timeout in ms
    on_error=lambda err: print(err), # error callback
)

Track Options

Each adapter accepts TrackOptions:

Option	Type	Default	Description
provider	str	--	Webhook provider name (e.g., "stripe", "shopify")
event_type_header	str	--	Header name to extract event type from
event_type_field	str	--	Body field to extract event type from (e.g., "type")
capture_body	bool	False	Capture request + response bodies (disabled by default for privacy)
capture_headers	bool	True	Capture request headers (sensitive ones redacted)
metadata	dict	--	Custom metadata attached to every event
signature_secret	str	--	Auto-verify signature (see below)
signature_verifier	callable	--	Custom verifier function
reject_invalid_signatures	bool	True	Respond with 401 when verification fails
signature_tolerance	int	300	Replay-attack window (seconds) for timestamp providers
idempotency_key	callable	--	Extract a dedup key from the request (see below)
idempotency_ttl	int	86400	TTL (seconds) for idempotency cache. Max 604800 (7d)

Signature Verification

Built-in support for verifying webhook signatures for Stripe, GitHub, Shopify, Svix / Clerk, and Slack. When you provide a signature_secret, the SDK:

1. Computes the expected HMAC-SHA256 signature
2. Compares it with the one in the request header (timing-safe, via hmac.compare_digest)
3. Rejects replay attacks (for timestamp-based providers)
4. Returns 401 Invalid webhook signature without calling your handler
5. Reports signature_valid to your Outworx dashboard for every request

FastAPI

from fastapi import FastAPI, Request
from outworx_hooks import init, TrackOptions
from outworx_hooks.integrations.fastapi import OutworxHooksMiddleware

init(api_key=os.environ["OUTWORX_HOOKS_API_KEY"])

app = FastAPI()
app.add_middleware(
    OutworxHooksMiddleware,
    options=TrackOptions(
        provider="stripe",
        signature_secret=os.environ["STRIPE_WEBHOOK_SECRET"],
    ),
)

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    # Signature has already been verified
    body = await request.json()
    return {"received": True}

Flask

from flask import Flask, request
from outworx_hooks import init, TrackOptions
from outworx_hooks.integrations.flask import with_webhook_monitoring

init(api_key=os.environ["OUTWORX_HOOKS_API_KEY"])

app = Flask(__name__)

@app.route("/webhooks/stripe", methods=["POST"])
@with_webhook_monitoring(TrackOptions(
    provider="stripe",
    signature_secret=os.environ["STRIPE_WEBHOOK_SECRET"],
))
def stripe_webhook():
    return {"received": True}

Django

# settings.py
OUTWORX_HOOKS_OPTIONS = {
    "provider": "stripe",
    "signature_secret": os.environ["STRIPE_WEBHOOK_SECRET"],
}

MIDDLEWARE = [
    "outworx_hooks.integrations.django.OutworxHooksMiddleware",
    # ...
]

Custom verifier

For providers not in the built-in list, pass your own function:

from outworx_hooks import TrackOptions

options = TrackOptions(
    provider="my-service",
    signature_verifier=lambda raw_body, headers: my_check(
        raw_body, headers.get("X-My-Signature")
    ),
)

Standalone verifiers

Import and call the per-provider verifiers directly:

from outworx_hooks.security import verify_stripe_signature

valid = verify_stripe_signature(
    raw_body=raw_body,
    header=request.headers.get("stripe-signature"),
    secret=os.environ["STRIPE_WEBHOOK_SECRET"],
)

Idempotency

Webhook providers retry deliveries when your handler times out or returns a non-2xx response — which can cause you to double-process the same event (double-charge the customer, send duplicate emails, etc.). Pass an idempotency_key function and the SDK will short-circuit retries with the cached response from the first successful delivery.

FastAPI

from outworx_hooks import init, TrackOptions
from outworx_hooks.integrations.fastapi import OutworxHooksMiddleware

init(api_key=os.environ["OUTWORX_HOOKS_API_KEY"])

app = FastAPI()
app.add_middleware(
    OutworxHooksMiddleware,
    options=TrackOptions(
        provider="stripe",
        signature_secret=os.environ["STRIPE_WEBHOOK_SECRET"],
        # Dedupe on the Stripe event ID. Retries of the same event return
        # the cached response without re-invoking the handler.
        idempotency_key=lambda req, body, headers: body.get("id") if body else None,
    ),
)

Flask

@app.route("/webhooks/stripe", methods=["POST"])
@with_webhook_monitoring(TrackOptions(
    provider="stripe",
    idempotency_key=lambda req, body, headers: body.get("id") if body else None,
))
def stripe_webhook():
    ...

Recommended keys per provider

# Stripe — event ID in the body
idempotency_key=lambda req, body, headers: body.get("id") if body else None

# GitHub — delivery ID header
idempotency_key=lambda req, body, headers: headers.get("x-github-delivery")

# Shopify — webhook ID header
idempotency_key=lambda req, body, headers: headers.get("x-shopify-webhook-id")

# Svix / Clerk — message ID header
idempotency_key=lambda req, body, headers: headers.get("svix-id")

Returning None from the function skips idempotency for that request.

Failure behavior

If the Outworx backend is unreachable when the SDK tries to check for duplicates, the check fails open — your handler runs as normal. Idempotency failures never block webhook delivery.

If your handler throws or returns 5xx, the idempotency key is not committed, so the next retry is free to re-run the handler. Stale reservations older than 30 seconds are automatically cleared.

Dashboard

View your webhook activity at hooks.outworx.io.

License

Business Source License 1.1 — see LICENSE for full text.

Free to use in production with the Outworx Hooks service. You may not use this SDK or any derivative work to offer a hosted webhook monitoring, analytics, or alerting service that competes with Outworx. The license converts to Apache 2.0 on 2030-04-17.

For commercial licensing inquiries, contact [e-mail].