How to Receive Webhooks in Python (Flask, FastAPI, Django)

Ask three Python developers how to receive a webhook and you'll get three correct answers that look almost identical. The differences are small and they all live in one place: how you get at the raw request body.

Get that wrong and your signature verification fails forever, with a payload that looks perfectly fine in the logs. So let's do all three properly.

Flask

import hmac, hashlib, os
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ['WEBHOOK_SECRET'].encode()

@app.post('/webhooks')
def receive():
    raw = request.get_data()            # bytes, untouched
    signature = request.headers.get('X-Signature', '')

    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        abort(401)

    payload = request.get_json()        # only now
    enqueue(payload)
    return '', 200

request.get_data() hands back the original bytes. request.json doesn't — it parses, and once it's parsed the bytes you needed are gone.

FastAPI

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

app = FastAPI()
SECRET = os.environ['WEBHOOK_SECRET'].encode()

@app.post('/webhooks')
async def receive(request: Request):
    raw = await request.body()
    signature = request.headers.get('x-signature', '')

    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401)

    payload = await request.json()
    return {'ok': True}

FastAPI makes this one genuinely tempting to get wrong, because declaring a Pydantic model as your parameter is the idiomatic thing to do everywhere else in the framework. Do it here and FastAPI parses the body before you ever see it. Take the Request, read the bytes, parse it yourself.

Django

import hmac, hashlib, os, json
from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt

SECRET = os.environ['WEBHOOK_SECRET'].encode()

@csrf_exempt
def receive(request):
    raw = request.body
    signature = request.headers.get('X-Signature', '')

    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        return HttpResponseForbidden()

    payload = json.loads(raw)
    return HttpResponse(status=200)

If you forget @csrf_exempt, every single delivery comes back 403 and the provider's dashboard fills with failures while your code looks completely reasonable. Stripe isn't a browser. It has no CSRF token and never will.

On compare_digest

if signature == expected:                       # don't
if hmac.compare_digest(signature, expected):    # do

A plain == stops at the first byte that differs, so how long it takes tells an attacker how much of the signature they guessed correctly. Grind that long enough and you forge a valid one. compare_digest always takes the same time. It's the same number of characters to type and it removes the whole attack.

Ack First, Work Later

Providers generally wait a handful of seconds. This is a trap:

@app.post('/webhooks')
def receive():
    payload = request.get_json()
    sync_to_warehouse(payload)     # 20 seconds
    return '', 200

Twenty seconds in, the sender has long since recorded a failed delivery and queued a retry — so the sync runs twice. Push it to Celery or RQ and return immediately:

    process_event.delay(payload)
    return '', 200

Spawning a thread instead is a tempting shortcut and it half-works. The catch is that a thread dies with the process, so anything in flight during a deploy or a restart just evaporates, silently. A queue survives that.

Since retries are guaranteed, dedupe on the provider's event ID:

if ProcessedEvent.objects.filter(event_id=payload['id']).exists():
    return HttpResponse(status=200)

Return 200, not an error. The delivery worked; you'd just seen it before. Anything in the 5xx range asks for yet another copy. Idempotent webhook handling goes deeper on the transactional side of this.

The Part Your Code Can't Fix

None of the above helps when Gunicorn is restarting, or a migration is holding a lock, or the box is out of memory. The provider gets a 5xx and reacts according to its own policy — Stripe retries for three days, some services retry twice and give up, a few don't retry at all. You don't get a vote.

And you have no copy of what was sent, which is the thing you'll actually want in six weeks when someone asks why an order never got fulfilled.

Bluejay Relay sits in front of the app and takes that on. Every event gets captured and stored with its headers, then delivered to your endpoint with retries and backoff you control. Your Flask app being down for a deploy stops being an event-loss scenario and becomes a delivery backlog that drains on its own. And when you find the handler bug you shipped last Tuesday, you filter the log to that window and replay it.

The Python code stays exactly as it is above. It just stops carrying the reliability burden by itself.

---

Give your Python app a webhook buffer with retries and replay. Start free with Bluejay Relay.