Webhook Receiver in Python: Flask & FastAPI

Python's Flask and FastAPI both make it easy to receive webhooks — but you need to be careful about how you access the raw request body for signature verification.

Flask Implementation

from flask import Flask, request, jsonify
import hmac
import hashlib
import os
import threading

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]

def verify_signature(raw_body: bytes, sig_header: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), raw_body, hashlib.sha256
    ).hexdigest()
    provided = sig_header.replace("sha256=", "")
    return hmac.compare_digest(expected, provided)

processed_events = set()

@app.post("/webhooks/github")
def github_webhook():
    raw = request.get_data()
    sig = request.headers.get("X-Hub-Signature-256", "")
    if not sig or not verify_signature(raw, sig):
        return jsonify(error="Invalid signature"), 401

    import json
    event = json.loads(raw)
    event_id = request.headers.get("X-GitHub-Delivery")

    response = jsonify(ok=True)

    def process():
        if event_id in processed_events:
            return
        processed_events.add(event_id)
        print(f"Processing {event_id}: {event.get('action')}")

    threading.Thread(target=process, daemon=True).start()
    return response

FastAPI Implementation

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
import hmac
import hashlib
import os

app = FastAPI()
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]

def verify_signature(raw_body: bytes, sig_header: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), raw_body, hashlib.sha256
    ).hexdigest()
    provided = sig_header.replace("sha256=", "")
    return hmac.compare_digest(expected, provided)

processed_events: set = set()

@app.post("/webhooks/github")
async def github_webhook(request: Request, background_tasks: BackgroundTasks):
    raw = await request.body()
    sig = request.headers.get("x-hub-signature-256", "")
    if not sig or not verify_signature(raw, sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

    import json
    event = json.loads(raw)
    event_id = request.headers.get("x-github-delivery", "")
    background_tasks.add_task(process_event, event_id, event)
    return {"ok": True}

async def process_event(event_id: str, event: dict):
    if event_id in processed_events:
        return
    processed_events.add(event_id)
    print(f"Processing {event_id}")

Key Differences: Flask vs FastAPI

Production Notes

• Use Redis for persistent idempotency across restarts
• Use hmac.compare_digest() — never string equality
• Always call request.get_data() or await request.body() before JSON parsing
• For Stripe, use stripe.Webhook.construct_event()

Use WebhookWhisper to get a public HTTPS endpoint for local development.