What Is a Webhook? Building a Solid Webhook Setup with FastAPI

Installing a doorbell instead of checking the door every minute

A webhook is an HTTP request that one system sends to another automatically when an event happens. In short, it is the promise “I’ll let you know when something happens.”

Imagine you are waiting for a parcel. You have two options: open the door every five minutes to see whether the courier has arrived, or install a doorbell and wait. The first is polling, the second is a webhook.

With polling, the client keeps asking: “Has the payment gone through?” Most of the time the answer is “no”. That wears the server out for nothing, and you only learn about the event at the next query. With a webhook, the moment the event happens the other side sends a POST to a URL you chose. No wasted traffic, no delay.

	Polling	Webhook
Who starts it?	The receiver keeps asking	The sender notifies when an event happens
Delay	Up to the polling interval	Almost instant
Wasted requests	Many	None
What the receiver needs	A cron job or scheduler	A public HTTPS endpoint
Difficulty	Simple but wasteful	Needs security, retries and ordering

That is also why webhooks are sometimes called a “reverse API”: normally you call the API; here the API calls you.

How does a webhook work?

The flow has four steps, and both sides have responsibilities.

Registration: The receiver gives the sending system a URL (e.g. https://api.yoursite.com/webhooks/stripe) and chooses which events it wants to listen to. A shared secret key is usually generated at this point.
Event: Something happens on the sender’s side: a payment is received, a commit is pushed, an SMS arrives.
Delivery: The sender POSTs the event as JSON to that URL. The request header carries the body signed with the secret (an HMAC).
Acknowledgement: The receiver verifies the signature and quickly returns 2xx. If no 2xx comes back or the request times out, the sender tries again later.

The receiver’s golden rule: acknowledge first, process later. Do not do long-running work (sending email, building a report) inside the HTTP request. Record the event, return 200, and finish the work in the background.

Real-world examples

Service	Typical event	What is it for?
Stripe	`invoice.paid`, `customer.subscription.deleted`	Keeping subscription status in sync with your own database
GitHub	`push`, `pull_request`	Triggering CI/CD, posting a Slack notification
Twilio / WhatsApp	incoming message	Replying to the user automatically
Shopify	`orders/create`	Passing the order to the warehouse or accounting

If you work with Stripe, it is worth knowing that after a payment the user’s browser may never reach the “success” page. They may close the tab, or their connection may drop. That is why the single source of truth for subscription status is the webhook, not the redirect.

Let’s write a webhook receiver with FastAPI

Instead of a single-file example, let’s build a small but tidy structure you can carry into a real project. Scenario: a payment service sends us payment.succeeded and payment.failed events.

webhook-receiver/
├── app/
│   ├── main.py            # FastAPI application
│   ├── config.py          # secret and settings (from .env)
│   ├── security.py        # HMAC signature verification
│   ├── schemas.py         # Pydantic event model
│   ├── store.py           # idempotency (never process the same event twice)
│   ├── handlers.py        # event type -> function mapping
│   └── routers/
│       └── webhooks.py    # endpoint
└── sender.py              # sending side (next section)

Setup:

pip install fastapi uvicorn pydantic-settings httpx
uvicorn app.main:app --reload

1. Settings: `config.py`

Never hard-code the secret. pydantic-settings reads the WEBHOOK_SECRET environment variable automatically.

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    webhook_secret: str = "whsec_change_me"
    signature_tolerance_seconds: int = 300  # reject requests older than 5 minutes


settings = Settings()

2. Signature verification: `security.py`

This is the most important part of the post. Your webhook URL is open to the internet; without a signature check, anyone can send you a fake “payment received” event.

We use a scheme similar to the one Stripe uses: X-Webhook-Signature: t=<timestamp>,v1=<signature>. The signature is the HMAC-SHA256 hash of the text timestamp.body.

import hashlib
import hmac
import time

from fastapi import HTTPException, Request

from app.config import settings


def compute_signature(secret: str, timestamp: str, body: bytes) -> str:
    signed_payload = f"{timestamp}.".encode() + body
    return hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()


async def verify_signature(request: Request) -> bytes:
    """Header format: X-Webhook-Signature: t=1731590000,v1=<hex>"""
    header = request.headers.get("X-Webhook-Signature")
    if not header:
        raise HTTPException(status_code=401, detail="Missing signature")

    try:
        parts = dict(item.split("=", 1) for item in header.split(","))
        timestamp, received_sig = parts["t"], parts["v1"]
    except (ValueError, KeyError):
        raise HTTPException(status_code=401, detail="Malformed signature")

    # Against replay attacks: do not accept old requests
    if abs(time.time() - int(timestamp)) > settings.signature_tolerance_seconds:
        raise HTTPException(status_code=401, detail="Request too old")

    body = await request.body()  # RAW body: sign it before turning it into JSON
    expected_sig = compute_signature(settings.webhook_secret, timestamp, body)

    # compare_digest instead of ==: constant-time comparison against timing attacks
    if not hmac.compare_digest(expected_sig, received_sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

    return body

There are three details here, and all three are often skipped in real projects:

The raw body (request.body()): If you compute the signature after parsing the JSON and turning it back into a string, whitespace or key order changes and the signature never matches.
hmac.compare_digest: A plain == stops at the first differing character. An attacker can measure response times and guess the signature character by character.
Timestamp check: Someone who captures a valid request may try to resend it hours later (a replay). Because the timestamp is inside the signature, it cannot be changed.

3. Event model: `schemas.py`

from datetime import datetime
from typing import Any

from pydantic import BaseModel


class WebhookEvent(BaseModel):
    id: str              # the event's unique id -> idempotency key
    type: str            # e.g. "payment.succeeded"
    created_at: datetime
    data: dict[str, Any]

4. Idempotency: `store.py`

Almost every system that sends webhooks works “at-least-once”: it sends an event at least once, and sometimes more than once. If your 200 gets lost on the network, the same event arrives again. To make sure a customer does not get two invoice emails, we record the event ID.

"""In memory for the demo. In production use Redis (SET NX + TTL)
or a UNIQUE(event_id) constraint in Postgres."""

_processed: set[str] = set()


def mark_if_new(event_id: str) -> bool:
    """Returns True and marks the event if it is new."""
    if event_id in _processed:
        return False
    _processed.add(event_id)
    return True

5. Event routing: `handlers.py`

To avoid writing a giant if/elif block as event types multiply, we set up a registry with a small decorator. Adding a new event is as easy as writing a new function.

import logging
from collections.abc import Callable

from app.schemas import WebhookEvent

logger = logging.getLogger("webhooks")

HANDLERS: dict[str, Callable[[WebhookEvent], None]] = {}


def on(event_type: str):
    """Decorator that binds a function to a given event type."""
    def decorator(func: Callable[[WebhookEvent], None]):
        HANDLERS[event_type] = func
        return func
    return decorator


@on("payment.succeeded")
def handle_payment_succeeded(event: WebhookEvent) -> None:
    logger.info("Payment received: %s", event.data.get("amount"))
    # activate the subscription, send the invoice email...


@on("payment.failed")
def handle_payment_failed(event: WebhookEvent) -> None:
    logger.warning("Payment failed: %s", event.data.get("customer_id"))


def dispatch(event: WebhookEvent) -> None:
    handler = HANDLERS.get(event.type)
    if handler is None:
        logger.info("Skipped unknown event type: %s", event.type)
        return
    handler(event)

For an unknown event type we do not raise an error, we just log it. If the sender adds a new event type tomorrow, your system should not blow up.

6. Endpoint: `routers/webhooks.py`

This is where all the pieces meet. Because the signature check is a FastAPI dependency, it runs before the function body is entered.

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from pydantic import ValidationError

from app.handlers import dispatch
from app.schemas import WebhookEvent
from app.security import verify_signature
from app.store import mark_if_new

router = APIRouter(prefix="/webhooks", tags=["webhooks"])


@router.post("/payments", status_code=200)
async def receive_payment_webhook(
    background_tasks: BackgroundTasks,
    body: bytes = Depends(verify_signature),
):
    try:
        event = WebhookEvent.model_validate_json(body)
    except ValidationError:
        raise HTTPException(status_code=400, detail="Invalid payload")

    # If the same event arrives a second time, still return 200 but don't process it
    if not mark_if_new(event.id):
        return {"status": "duplicate"}

    # Acknowledge first, process later
    background_tasks.add_task(dispatch, event)
    return {"status": "accepted"}

Note: we return 200, not an error, for a repeated event. If we returned 409, the sender would treat it as a failure and keep retrying.

7. Application: `main.py`

import logging

from fastapi import FastAPI

from app.routers import webhooks

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="Webhook Receiver")
app.include_router(webhooks.router)


@app.get("/health")
def health():
    return {"ok": True}

Let’s test it

The small test below checks all four scenarios. It uses the sign function from the next section.

import json, time, uuid
from fastapi.testclient import TestClient
from app.main import app
from app.config import settings
from sender import sign

client = TestClient(app)

def post(event, sig=None):
    body = json.dumps(event).encode()
    return client.post("/webhooks/payments", content=body,
        headers={"X-Webhook-Signature": sig or sign(settings.webhook_secret, body)})

ev = {"id": "evt_" + uuid.uuid4().hex, "type": "payment.succeeded",
      "created_at": "2026-11-14T12:00:00Z", "data": {"amount": 4900}}

print(post(ev).json())                                       # {'status': 'accepted'}
print(post(ev).json())                                       # {'status': 'duplicate'}
print(post(ev, sig=f"t={int(time.time())},v1=fake").json())  # {'detail': 'Invalid signature'}
print(post(ev, sig="t=1000,v1=fake").json())                 # {'detail': 'Request too old'}

Production note: BackgroundTasks runs in the same process. If the server restarts before the work is done, the work is lost and the event stays marked as “processed”. In a serious system, first write the event to the database with status='pending', hand the work to a queue such as Celery, ARQ or Dramatiq, and set status='done' once it succeeds.

The other side of the coin: sending webhooks

When your own SaaS needs to send webhooks to its customers, the job is reversed. Now signing, and retrying when the other side’s server is down, are your responsibility.

import hashlib
import hmac
import json
import time

import httpx

RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


def sign(secret: str, body: bytes) -> str:
    timestamp = str(int(time.time()))
    sig = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()
    return f"t={timestamp},v1={sig}"


def send_webhook(url: str, secret: str, event: dict, max_attempts: int = 5) -> bool:
    body = json.dumps(event, separators=(",", ":")).encode()

    for attempt in range(1, max_attempts + 1):
        headers = {
            "Content-Type": "application/json",
            "X-Webhook-Signature": sign(secret, body),  # a fresh timestamp on every attempt
            "X-Webhook-Id": event["id"],
        }
        try:
            response = httpx.post(url, content=body, headers=headers, timeout=10)
            if 200 <= response.status_code < 300:
                return True
            if response.status_code not in RETRYABLE_STATUS:
                return False  # errors like 400/401 do not go away when retried
        except httpx.TransportError:
            pass  # connection error, timeout -> retry

        if attempt < max_attempts:
            time.sleep(2 ** attempt)  # 2, 4, 8, 16 s: exponential backoff

    return False  # write the event to a dead-letter table here

There are a few decisions worth noting in this code:

Exponential backoff: If the other server is already struggling, sending a request every second only chokes it further. We double the wait on every attempt.
A new signature on every attempt: The timestamp is part of the signature. If you reuse an old signature, you will hit the receiver’s 5-minute tolerance.
We do not retry every error: A 401 is a signature error, and it will still be a 401 an hour later. Only temporary errors (5xx, 429, timeouts) are worth retrying.
Dead-letter: When all attempts are used up, do not lose the event. Write it to a separate table and give the customer a “resend” button in the dashboard.

In a real system, instead of waiting in a loop with time.sleep, you put each attempt on a queue as a delayed job. Stripe retries failed webhooks on a schedule spread over three days. The loop above is kept simple to show the logic.

Extras worth knowing

There is no ordering guarantee

Events may not arrive in the order they were sent. A subscription.updated event can arrive before subscription.created. The fix: look at the event’s created_at field and ignore it if it is older than the record you have. Or treat the event only as a “something changed” signal and fetch the current state from the API again. With Stripe, the second approach is very common.

Thin vs fat payloads

Fat payload: All of the event’s data arrives in the body. It is fast, but the data can go stale and it carries sensitive information.
Thin payload: Only {"type": "order.updated", "id": "ord_123"} arrives, and you fetch the details from the API. You always work with current data, at the cost of an extra request.

Secret rotation

When you need to change the secret (a leak, an employee leaving, and so on), allow a transition period. For a while the sender sends two signatures at once (v1=old,v1=new), and the receiver accepts the request if either one matches.

Security checklist

The endpoint is reachable only over HTTPS
The HMAC signature is verified over the raw body with compare_digest
There is a timestamp tolerance (replay protection)
The secret lives in an environment variable, not in the code or the repo
There is an upper limit on payload size (e.g. 1 MB)
On the sending side: customer-supplied URLs are blocked from reaching internal addresses such as localhost or 169.254.169.254 (SSRF protection)

The last item is critical if you are building a product that sends webhooks. If a user enters your cloud provider’s metadata address as their URL, your server may read its own secrets out to them.

Tools that help during development

Tool	What for?
webhook.site	An instant temporary URL to inspect incoming requests
ngrok / Cloudflare Tunnel	Exposing the FastAPI app on your localhost to the internet
Stripe CLI (`stripe listen --forward-to localhost:8000/...`)	Forwarding Stripe events straight to your local machine and triggering them
Svix, Hookdeck	Webhook infrastructure services that provide retries, logs, signatures and the rest out of the box

Common mistakes

Verifying the signature over parsed JSON.
Doing 30 seconds of work inside the endpoint and hitting the sender’s timeout. Result: the same event arrives again and the work is done twice.
Returning 4xx for a repeated event and getting stuck in an endless retry loop.
Not logging every incoming request. When something goes wrong, the answer to “did the event ever arrive?” lives only in the logs.

Final word

The essence of a webhook fits in one sentence: “when something happens, I’ll send you an HTTP request.” But a reliable webhook system is built from four habits woven around that sentence: verify the signature, acknowledge quickly, never process the same event twice, never lose a failed one.

Once you have those four in place, you can handle events from Stripe, GitHub or your own service with the same confidence. You can start by copying the structure above and adding your own events to handlers.py.

This English version was translated from the Turkish original with Claude.