Build a Webhook Receiver with FastAPI

Build a Webhook Receiver with FastAPI

Imagine you’re waiting for a critical notification from GitHub, Stripe, or Slack, but your server is too busy processing user requests to handle it immediately. The webhook arrives, gets dropped, and you lose track of a payment or a code commit. That’s the nightmare every developer faces when building webhook receivers without a solid strategy. But with FastAPI, you can build a robust, secure, and asynchronous webhook receiver that handles incoming events like a pro—without breaking your app’s performance.

Webhooks are the backbone of modern integrations. They let external services push data to your application in real time. Yet, implementing them correctly is tricky: you need to verify signatures, handle retries, avoid duplicate processing, and respond quickly to keep the provider happy. Let’s dive into how to build a production-ready webhook receiver using FastAPI that you can deploy today.

Why FastAPI is Perfect for Webhooks

FastAPI shines in webhook scenarios because it’s built on async Python, supports automatic request validation with Pydantic, and includes background task handling out of the box. Unlike traditional frameworks, FastAPI lets you:

• Receive the raw request body without parsing it prematurely
• Verify cryptographic signatures before processing
• Offload heavy logic to background tasks or job queues
• Return a 200 OK response instantly to prevent provider retries

This aligns perfectly with the recommended webhook flow: receive → verify → persist minimal data → ACK → hand off processing [2].

Setting Up Your Project

Start by creating a clean project structure:

mkdir webhook-receiver
cd webhook-receiver
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install fastapi uvicorn python-jose[cryptography]

We’ll use python-jose for HMAC signature verification, a common requirement for providers like Stripe and Svix [7].

The Core Endpoint: Raw Body, Signature Verification, and Fast ACK

The most critical part of a webhook receiver is the endpoint. You must retrieve the raw body first, verify the signature, and return an ACK before doing any heavy work [2].

Here’s a complete, working example:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.background import BackgroundTasks
import hmac
import hashlib
import json
import time

app = FastAPI()

# Replace with your actual signing secret (e.g., from Stripe or Svix UI)
SIGNING_SECRET = "whsec_your_secret_here"

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    # Extract version and sig (e.g., "v1,abc123..." -> "abc123...")
    if "," in signature:
        version, sig = signature.split(",", 1)
    else:
        sig = signature

    # Compute expected signature
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(expected, sig)

def process_webhook_later(payload: dict, event_id: str):
    """Heavy business logic goes here (e.g., DB updates, notifications)."""
    print(f"[BACKGROUND] Processing event {event_id}: {payload}")
    # Simulate DB write or API call
    time.sleep(2)
    print(f"[BACKGROUND] Event {event_id} processed successfully!")

@app.post("/webhook")
async def webhook_endpoint(request: Request):
    # 1. Get raw body (critical for signature verification)
    raw_body = await request.body()
    headers = request.headers

    # 2. Extract signature and timestamp
    signature = headers.get("Webhook-Signature")
    timestamp = headers.get("Webhook-Timestamp")

    if not signature or not timestamp:
        return JSONResponse(status_code=400, content={"error": "Missing signature or timestamp"})

    # 3. Verify timestamp (prevent replay attacks)
    current_time = int(time.time())
    if abs(current_time - int(timestamp)) > 300:  # 5-minute tolerance
        return JSONResponse(status_code=400, content={"error": "Timestamp too old"})

    # 4. Verify signature
    if not verify_signature(raw_body, signature, SIGNING_SECRET):
        return JSONResponse(status_code=401, content={"error": "Invalid signature"})

    # 5. Parse payload (safe now, after verification)
    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError:
        return JSONResponse(status_code=400, content={"error": "Invalid JSON"})

    # 6. Check for duplicate event (idempotency)
    event_id = payload.get("id") or payload.get("event_id")
    if not event_id:
        return JSONResponse(status_code=400, content={"error": "Missing event ID"})

    # TODO: In production, check DB/cache for duplicate event_id
    # if is_duplicate(event_id): return JSONResponse(200, {"status": "already_processed"})

    # 7. Store raw payload (for debugging/retries) - minimal persistence
    # TODO: Save to DB: {"event_id": event_id, "payload": raw_body, "received_at": now}

    # 8. Hand off heavy processing to background task
    background = BackgroundTasks()
    background.add_task(process_webhook_later, payload, event_id)

    # 9. Return ACK immediately
    return JSONResponse(status_code=200, content={"status": "received", "event_id": event_id}, background=background)

Run it with:

uvicorn main:app --reload

Making It Public with ngrok

Your local server isn’t accessible to external services. Use ngrok to expose it:

ngrok http 8000

Copy the generated URL (e.g., https://abc123.ngrok.io) and configure your webhook provider (GitHub, Stripe, etc.) to send events to /webhook on that URL [1][10].

Security Best Practices You Can’t Skip

Webhooks are a prime attack surface. Here’s what to enforce:

• Signature verification: Always use the raw body; parsing it first breaks HMAC [7].
• Timestamp validation: Reject events older than 5 minutes to prevent replay attacks [7].
• Idempotency: Track event IDs to avoid processing duplicates [2][10].
• Constant-time comparison: Use hmac.compare_digest() to prevent timing attacks [7].
• Logging: Log every request and response for debugging failed deliveries [10].

Testing Without a Real Provider

You don’t need Stripe or GitHub to test. Use curl:

curl -X POST \
  -H "Content-Type: application/json" \
  -H "Webhook-Signature: v1,abc123def456" \
  -H "Webhook-Timestamp: 1720940000" \
  -d '{"id":"evt_123","type":"payment.completed","amount":99.99}' \
  http://127.0.0.1:8000/webhook

For production-like testing, generate a real signature using your secret and the exact payload bytes [7].

What to Do Next

You now have a secure, async webhook receiver that:

• Verifies signatures correctly
• Responds instantly to prevent retries
• Offloads heavy work to background tasks
• Supports idempotency and debugging

Your action today:

1. Replace SIGNING_SECRET with your provider’s secret.
2. Add DB logic to check for duplicate event_id.
3. Connect your background task to a real queue (e.g., Celery, Redis) for scalability.
4. Deploy to a cloud provider (Render, AWS, or Cloud Run) and update your ngrok URL.

Webhooks are powerful—but only when built right. With FastAPI, you get the tools to build them safely, fast, and without the usual pitfalls. Now go integrate that payment, notification, or CI/CD trigger you’ve been waiting on.

Drop a comment if you’re using Stripe, GitHub, or another provider—I’d love to see your implementation! And if this helped, share it with a teammate who’s wrestling with webhook retries.

---

If you found this helpful, consider buying me a coffee ☕ — it keeps these articles coming!

Also check out my AI tools collection: AI 次元世界 — free AI tools for developers.

---

💡 Related: **Content Creator Ultimate Bundle (Save 33%)* — $29.99*