FastAPIPython
Python FastAPI Webhook Guide: Async Handlers & HMAC Signature Verification
Build high-performance asynchronous webhook receivers in Python FastAPI. Read raw request bytes and verify HMAC-SHA256 signatures with background tasks.
Direct Answer / Quick Implementation Guide
In FastAPI, extract raw request bytes using `await request.body()` instead of Pydantic models for signature validation. Verify the signature with Python’s `hmac.compare_digest()` before dispatching processing to `BackgroundTasks`.
Essential Implementation Takeaways
- •Call `await request.body()` to get the raw `bytes` object.
- •Use `hmac.compare_digest` for timing-safe signature comparison.
- •Use FastAPI `BackgroundTasks` to offload work and return 200 OK in < 10ms.
- •Parse verified bytes with `json.loads(raw_body)`.
Common Gotchas in FastAPI & How to Fix Them
Using Pydantic model parameter in route function
Fix:Use `request: Request` parameter and call `await request.body()` directly to preserve exact byte formatting.
Complete Production Boilerplate
Python# main.py (FastAPI Webhook Handler)
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
import hmac
import hashlib
import json
app = FastAPI()
WEBHOOK_SECRET = "whsec_your_secret_key"
def process_webhook_payload(payload: dict):
# Long-running background database or sync logic
print(f"Processing event: {payload.get('type')}")
@app.post("/api/webhook")
async def webhook_listener(request: Request, background_tasks: BackgroundTasks):
raw_body = await request.body()
signature_header = request.headers.get("stripe-signature")
if not signature_header:
raise HTTPException(status_code=400, detail="Missing signature header")
# Compute HMAC-SHA256
expected_sig = hmac.new(
WEBHOOK_SECRET.encode("utf-8"),
raw_body,
hashlib.sha256
).hexdigest()
# Timing-safe signature check
if not hmac.compare_digest(signature_header, expected_sig):
raise HTTPException(status_code=401, detail="Signature verification failed")
payload = json.loads(raw_body.decode("utf-8"))
# Enqueue background task and respond immediately
background_tasks.add_task(process_webhook_payload, payload)
return {"status": "success"}Step-by-Step Setup Guide
1
Import `Request`, `HTTPException`, and `BackgroundTasks` from `fastapi`.
2
Read raw bytes via `raw_body = await request.body()`.
3
Extract signature from `request.headers`.
4
Compute HMAC and compare with `hmac.compare_digest()`.
5
Schedule long-running work via `background_tasks.add_task()` and return 200.
How to Test Locally on Your Machine
- Run `uvicorn main:app --port 8000 --reload`.
- Use SafeWebhook Replay Drawer to send captured events to `http://localhost:8000/api/webhook`.
- Check FastAPI terminal output for background task execution logs.
Frequently Asked Questions: FastAPI Webhooks
Why should I use BackgroundTasks in FastAPI webhooks?
`BackgroundTasks` runs your processing function after the HTTP 200 response is dispatched to the client, preventing webhook gateway timeouts.
Test Your FastAPI Webhook Handler Live
Capture payloads, simulate errors, and replay directly to your localhost server.