Webhook monitoring for Flask.
A single decorator wraps each webhook route. Captures latency, status, and signature validity without touching your handler logic. 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, Flask config, or a secrets manager).
# .env
OUTWORX_HOOKS_API_KEY=ow_live_...
STRIPE_WEBHOOK_SECRET=whsec_...Wrap the webhook handler
Apply the @with_webhook_monitoring decorator to any Flask route. The signature is verified before your function runs, and duplicate retries (same event ID within 24h) short-circuit with the cached response.
# app.py
import os
from flask import Flask
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"],
# Dedupe on the Stripe event ID — retries return the cached response.
idempotency_key=lambda req, body, headers: body.get("id") if body else None,
))
def stripe_webhook():
# Signature verified, duplicates filtered.
return {"received": True}One app, many webhook sources
The decorator works per-route, so each endpoint gets its own provider, secret, and idempotency strategy. Built-in HMAC-SHA256 verification covers Stripe, GitHub, Shopify, Svix / Clerk, and Slack.
# Each route can have its own provider, secret, and idempotency key.
@app.route("/webhooks/github", methods=["POST"])
@with_webhook_monitoring(TrackOptions(
provider="github",
signature_secret=os.environ["GITHUB_WEBHOOK_SECRET"],
idempotency_key=lambda req, body, headers: headers.get("x-github-delivery"),
))
def github_webhook():
# ... handle GitHub event
return {"ok": True}
@app.route("/webhooks/shopify", methods=["POST"])
@with_webhook_monitoring(TrackOptions(
provider="shopify",
signature_secret=os.environ["SHOPIFY_WEBHOOK_SECRET"],
idempotency_key=lambda req, body, headers: headers.get("x-shopify-webhook-id"),
))
def shopify_webhook():
# ... handle Shopify event
return {"ok": True}Other Python frameworks
Using FastAPI or Django? Each has its own integration pattern — same SDK, same dashboard. 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 state is wrong. Set require_processing_mark=True and call g.outworx_track.processed() when your business logic finishes:
from flask import Flask, g, 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"],
require_processing_mark=True, # ← opt in
))
def stripe_webhook():
body = request.get_json()
charge_customer(body)
g.outworx_track.processed() # ← explicit ack
return {"received": True}Full guide: Silent drop detection.