Building a Reliable Webhook Receiver API with FastAPI and Handling Retries

[gitwithu]

Select Topic Area

Question

Body

Hi everyone!

I'm working on integrating webhooks into my application using FastAPI and wanted to start a discussion around best practices, especially for building robust and scalable webhook receivers.

Here’s what I’d love to explore and get feedback on:

Designing the Webhook Endpoint

-How to structure a clean /webhook endpoint in FastAPI?
-Should we validate incoming webhook payloads with HMAC signatures (e.g., for Stripe, GitHub)?
-What’s the best way to avoid blocking responses if the processing takes time?

Looking forward to hearing your thoughts, code snippets, or battle-tested solutions!
This topic can benefit anyone building webhook-driven systems like payment gateways, messaging bots, or deployment triggers.

Thanks!

[KrishPatel1010]

Hey! 👋 Great discussion topic—webhooks are critical but often tricky to get right in production. I’ve worked on a few webhook-driven systems using FastAPI, so here are some best practices that have helped me:

✅ 1. Webhook Endpoint Design
Keep it simple and versioned:

@app.post("/webhooks/v1/stripe")
async def handle_stripe_webhook(request: Request):
payload = await request.body()
# Process here...

Avoid JSON parsing too early if you need to validate raw payloads (for HMAC signature checks).

🔐 2. Payload Validation with HMAC
Absolutely yes—especially for services like Stripe, GitHub, SendGrid, etc.

Example using Stripe’s signature validation:

import stripe

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
payload = await request.body()
sig_header = request.headers.get("stripe-signature")
try:
event = stripe.Webhook.construct_event(
payload, sig_header, STRIPE_WEBHOOK_SECRET
)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid payload")
except stripe.error.SignatureVerificationError:
raise HTTPException(status_code=400, detail="Invalid signature")
# Handle event
return {"status": "success"}

You can do similar checks manually using hmac for other services:

import hmac
import hashlib

def verify_signature(secret, payload, header_signature):
computed = hmac.new(
key=secret.encode(), msg=payload, digestmod=hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, header_signature)

⚙️ 3. Non-blocking Processing
Don’t block the HTTP response—offload to:

Background tasks:

from fastapi import BackgroundTasks

@app.post("/webhook")
async def webhook(request: Request, background_tasks: BackgroundTasks):
payload = await request.json()
background_tasks.add_task(process_payload, payload)
return {"status": "received"}

Celery or a task queue: Use when things scale and require retries, visibility, etc.

Always acknowledge quickly (200 OK) unless the service requires a specific retry mechanism.

[AnMoreNight]

Hello.

Clean /webhook Endpoint (FastAPI)

==========================
from fastapi import FastAPI, Request, BackgroundTasks
import hmac, hashlib

app = FastAPI()

SECRET = b"your_webhook_secret"

def verify_signature(payload: bytes, signature: str) -> bool:
expected = hmac.new(SECRET, payload, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)

@app.post("/webhook")
async def webhook(request: Request, background_tasks: BackgroundTasks):
body = await request.body()
signature = request.headers.get("X-Signature")

if not verify_signature(body, signature):
    return {"status": "invalid signature"}

background_tasks.add_task(process_webhook, body)
return {"status": "accepted"}

def process_webhook(payload: bytes):
# Your async-safe processing logic here
pass

2. Best Practices
   -Use HMAC (e.g., SHA256) to validate payloads — especially important with Stripe, GitHub, etc.
   -Use BackgroundTasks to avoid blocking the webhook sender.
   -Return a 2xx quickly to prevent retries.
   -Log and store raw payloads for debugging/auditing if possible.

Good luck!

[gitwithu]

Thank you for your help!

[ebndev]

Hi @gitwithu & @MomotaroHero, thanks for being a part of the GitHub Community!

While we appreciate you wanting to contribute to Discussions, please note: We made the decision to disable the ability to earn Achievements in our Community in order to discourage users from participating in coordinated or inauthentic activity like rapid questions and answers in order to earn badges. You can learn more about this decision in our announcement post here Achievements will no longer be available in the Community.

As a result, we'll be unmarking the answer and locking this, and other related posts. 

Any future violations may result in a temporary or indefinite block from the Community.

Thanks for understanding and we hope to see you participate in other discussions in the future!