Webhooks seem simple: receive an HTTP POST, do something. In practice, they’re a minefield of security issues, reliability problems, and edge cases. Let’s build webhooks that actually work in production.
Receiving Webhooks Basic Receiver 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 from fastapi import FastAPI, Request, HTTPException, Header from typing import Optional import hmac import hashlib app = FastAPI() WEBHOOK_SECRET = "your-webhook-secret" @app.post("/webhooks/stripe") async def stripe_webhook( request: Request, stripe_signature: Optional[str] = Header(None, alias="Stripe-Signature") ): payload = await request.body() # Verify signature if not verify_stripe_signature(payload, stripe_signature, WEBHOOK_SECRET): raise HTTPException(status_code=401, detail="Invalid signature") event = await request.json() # Process event event_type = event.get("type") if event_type == "payment_intent.succeeded": await handle_payment_success(event["data"]["object"]) elif event_type == "customer.subscription.deleted": await handle_subscription_cancelled(event["data"]["object"]) return {"received": True} def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool: """Verify Stripe webhook signature.""" if not signature: return False # Parse signature header elements = dict(item.split("=") for item in signature.split(",")) timestamp = elements.get("t") expected_sig = elements.get("v1") # Compute expected signature signed_payload = f"{timestamp}.{payload.decode()}" computed_sig = hmac.new( secret.encode(), signed_payload.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(computed_sig, expected_sig) Idempotent Processing Webhooks can be delivered multiple times. Make your handlers idempotent:
...