Webhook monitoring for FastAPI.
ASGI middleware that wraps every webhook route — captures latency, status, and signature validity without touching your handler logic. Async-safe with under 3ms of overhead per request.
Add the package
pip install outworx-hooks
# or
poetry add outworx-hooks
# or
uv add outworx-hooksSet environment variables
Grab your API key from the dashboard and load it however you normally would (.env, Pydantic settings, Doppler, etc).
# .env
OUTWORX_HOOKS_API_KEY=ow_live_...
STRIPE_WEBHOOK_SECRET=whsec_...App-wide configuration
Register OutworxHooksMiddleware once. The signature is verified before your route handler runs, and duplicate retries (same event ID within 24h) short-circuit with the cached response.
# app/main.py
import os
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"],
# Dedupe on the Stripe event ID — retries return the cached response.
idempotency_key=lambda req, body, headers: body.get("id") if body else None,
),
)
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
# Signature is already verified, duplicates filtered.
body = await request.json()
return {"received": True}Per-route configuration
When one app handles webhooks from multiple providers, the @with_webhook_monitoring decorator gives you per-route control over provider, secret, and idempotency keys. Built-in HMAC-SHA256 verification covers Stripe, GitHub, Shopify, Svix / Clerk, and Slack.
# Per-route provider config — useful when one app handles multiple providers.
from outworx_hooks import TrackOptions
from outworx_hooks.integrations.fastapi import with_webhook_monitoring
@app.post("/webhooks/github")
@with_webhook_monitoring(TrackOptions(
provider="github",
signature_secret=os.environ["GITHUB_WEBHOOK_SECRET"],
idempotency_key=lambda req, body, headers: headers.get("x-github-delivery"),
))
async def github_webhook(request: Request):
payload = await request.json()
# ... handle GitHub event
return {"ok": True}Other Python frameworks
On a different stack? Flask uses a @with_webhook_monitoring decorator; Django uses OutworxHooksMiddleware with OUTWORX_HOOKS_OPTIONS. The full TrackOptions reference lives in the main SDK docs.
Catch silent drops
When your handler returns 200 but never actually processes the event, everything looks fine — until a customer notices state is wrong. Set require_processing_mark=True and call processed() when your business logic finishes:
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"],
require_processing_mark=True, # ← opt in
),
)
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
body = await request.json()
charge_customer(body)
request.state.outworx_track.processed() # ← explicit ack
return {"received": True}Full guide: Silent drop detection.