Build a Python webhook receiver with FastAPI

A Python webhook receiver lets another service tell your application that something happened. A report is ready, a file arrived, or a job changed state. The sender makes an HTTP POST request to your webhook endpoint, and your application decides whether to accept the event.

This guide builds a FastAPI receiver for a synthetic report.ready event. It verifies a signature, checks the payload, and writes a receipt to a local SQLite inbox before responding. Repeated delivery of the same event returns the existing receipt. The example does not generate a report, download a file, send email, or submit a Celery task.

Validation status, November 11, 2026: Thirty local application tests passed on Windows with Python 3.12.14. A separate loopback HTTP check exercised Uvicorn, the sender command, stored rows, signature rejection, conflicting IDs, and a duplicate after restarting the server process. The restart happened after completed requests. Real-provider delivery, public HTTPS, proxy behavior, body-read timeout and disconnect paths, Linux, containers, queues, interrupted writes, power or disk loss, load, and Hivenet deployment remain untested.

What a webhook receiver needs to establish

A webhook uses an API endpoint, but the event source initiates the request. That differs from polling an API repeatedly to ask whether something has changed. Receiving a request is only the first boundary: you also need to establish who signed it, whether its content is valid, and what your success response promises.

Here, HTTP 202 means the inbox transaction committed. It does not mean the report was processed. This distinction matters when the sender times out and tries again: the original request might already have committed even if its response never arrived.

Use a normal @api.post() route for incoming HTTP requests. FastAPI's app.webhooks feature describes outgoing notifications in OpenAPI documentation; it does not implement this receiver. See the FastAPI webhook reference.

Download the local example

Download the Python webhook sample and tests. Extract it into a new private directory. It contains the complete receiver, sender, pinned top-level dependencies, tests, and README. There are no deployment credentials or databases in the archive.

The recorded checks used FastAPI 0.162.0, Starlette 1.10.0, Pydantic 2.14.1, Uvicorn 0.54.0, HTTPX 0.28.1, and pytest 9.2.0. These are the tested versions, not a claim that they are the newest. The test environment already contained these packages; a clean installation on another machine was not exercised. Recheck compatibility before updating dependencies.

The commands below use PowerShell and Python 3.12. Create a virtual environment and install the required packages:

python -m venv .venv
.venv/Scripts/python.exe -m pip install -r requirements.txt

For an occasional exercise, your own machine is enough. A continually available endpoint needs a hosting and operating decision after the request-handling behavior is understood.

Define the event and signing contract

The webhook payload contains exactly two fields:

{"type":"report.ready","report_id":"report-1"}

The type must be report.ready. A report ID contains one to 64 ASCII letters, digits, underscores, or hyphens. Extra fields are rejected. The receiver treats the report ID as an identifier, not a URL or command. It makes no outbound request based on webhook data.

Three demonstration headers accompany Content-Type: application/json: X-Demo-Event-Id, X-Demo-Timestamp, and X-Demo-Signature. The timestamp is a ten-digit Unix time in seconds. The signature is a lowercase hexadecimal HMAC-SHA256 digest over these exact bytes:

v1\nTIMESTAMP\nEVENT_ID\nRAW_BODY

Each \n represents one newline byte. The event ID follows the same character and length limits as the report ID. Those restrictions keep the signed fields unambiguous. A retry retains the event ID and the exact payload bytes, but creates a fresh timestamp and signature. Reformatting otherwise equivalent JSON changes the body digest and is treated as different content.

This is a teaching protocol. GitHub and Stripe use their own header formats and verification rules. Follow the actual provider's SDK or documented algorithm when integrating it; do not rename these headers and assume compatibility. GitHub's validation guide specifies its HMAC format, while Stripe's webhook guide covers raw-body verification, duplicate events, and delivery ordering.

Verify and record the incoming event

The complete receiver is in app.py in the download. Its signature helper preserves the original request body and uses Python's hmac implementation:

def signature(secret: bytes, timestamp: str, event_id: str, body: bytes) -> str:
    message = b"v1\n" + timestamp.encode("ascii") + b"\n" + event_id.encode("ascii") + b"\n" + body
    return hmac.new(secret, message, hashlib.sha256).hexdigest()

The route uses hmac.compare_digest() to compare the supplied and expected digest. Python documents that comparison in its HMAC reference. It checks a five-minute timestamp window before accepting an attempt, including attempts using an already recorded event ID. Clock accuracy is therefore part of the contract.

Before JSON parsing, the route counts streamed bytes and rejects bodies larger than 65,536 bytes. It does not rely on Content-Length. It also sets a ten-second body-read deadline, rejects compressed requests, and rejects missing or repeated required headers. The recorded tests exercised the byte limit without Content-Length; the timeout and disconnect paths remain untested. Public request limits and rate limiting require separate proxy and server checks.

Reading raw bytes bypasses FastAPI's usual automatic body-model handling, so the code explicitly validates the JSON with Pydantic after verifying its signature. Validation errors return a short fixed message instead of echoing the payload. See FastAPI's direct-request guidance and Starlette's request-body interfaces.

The inbox table stores the event ID, a SHA-256 body digest, the original body, and its receipt time. The event ID is the primary key. This transaction serializes the check and insert, then commits before returning:

def record(path, event_id, body):
    digest = hashlib.sha256(body).hexdigest()
    connection = sqlite3.connect(path, timeout=1)
    try:
        with connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT digest FROM inbox WHERE event_id = ?", (event_id,)
            ).fetchone()
            if row:
                if row[0] != digest:
                    raise HTTPException(409, "Event ID has different content")
                duplicate = True
            else:
                connection.execute(
                    "INSERT INTO inbox VALUES (?, ?, ?, ?)",
                    (event_id, digest, body, int(time.time())),
                )
                duplicate = False
        # The transaction context has committed before this return.
        return {"event_id": event_id, "state": "recorded", "duplicate": duplicate}
    finally:
        connection.close()

The connection context commits on a successful exit and rolls back on an exception; the connection is closed separately. Python's SQLite documentation explains these transaction and connection responsibilities. The route runs this synchronous database work in a thread pool and converts SQLite errors into HTTP 503.

Eight concurrent duplicate requests produced one stored event in the local test. That is a functional check of this small inbox, not a throughput measurement. SQLite still serializes writers, and the one-second lock wait can fail under contention.

Start the local server and send a webhook

In the extracted directory, generate a new local secret without printing it. This command refuses to overwrite an existing file:

.venv/Scripts/python.exe -c "from pathlib import Path; import secrets; Path('webhook-secret.txt').open('x', encoding='ascii').write(secrets.token_hex(32))"
$env:WEBHOOK_SECRET_FILE = (Resolve-Path webhook-secret.txt).Path
$env:WEBHOOK_DB = Join-Path (Get-Location) 'inbox.sqlite3'
.venv/Scripts/python.exe -m uvicorn app:create_app --factory --host 127.0.0.1 --port 8765 --no-access-log --no-proxy-headers

Keep the directory accessible only to the intended local user. The secret and database are excluded by the sample's .gitignore, but that does not protect them from other local users or backups. Do not commit, log, or share them. The app fails startup if the secret format is invalid or the inbox cannot be initialized.

The server listens only on loopback. In a second PowerShell terminal in the same directory, send the same event twice:

$env:WEBHOOK_SECRET_FILE = (Resolve-Path webhook-secret.txt).Path
.venv/Scripts/python.exe send_event.py evt-1 report-1
.venv/Scripts/python.exe send_event.py evt-1 report-1

The sender signs the exact JSON bytes it transmits, uses a request timeout, and does not follow redirects. It is fixed to the local lab address. The first successful response has state: recorded and duplicate: false. The second has duplicate: true; both return HTTP 202. A 202 and the state label describe receipt only.

Changing the second command to evt-1 report-2 produces HTTP 409 because that event ID already belongs to different content. The sender exits with an error for unsuccessful HTTP responses. It does not implement automatic retries.

Inspect the inbox and test failure behavior

Stop the server with Ctrl+C when finished. Retain the same database if you want to repeat the event after restarting. The sender generates a new timestamp, so the retained ID can be recognized without accepting an old signed request indefinitely.

The download includes 30 application tests and a separate HTTP checker. Run them with no other service using port 8765:

.venv/Scripts/python.exe -m pip install pytest==9.2.0
.venv/Scripts/python.exe -m pytest test_webhook.py -q
.venv/Scripts/python.exe check_http.py

The application tests cover signatures, modified body bytes and signed headers, missing and repeated headers, timestamp limits, invalid JSON and schemas, oversized bodies, duplicates, conflicting IDs, concurrent inserts, a locked database, and a simulated commit failure. They also reopen the database through a new application instance. Starlette issued a deprecation warning about its HTTPX test-client integration; the tests passed with the pinned versions.

The separate checker creates a temporary secret and inbox, starts Uvicorn on 127.0.0.1, sends actual HTTP requests, runs the sender command, and reads the recorded rows. It stops the server process after completed requests, starts another process with the same database, and verifies the duplicate. It observed two stored events and then removed the temporary test files. This does not test a crash during commit or a failed storage device.

Use the response to identify the boundary that failed: 400 for missing or malformed event headers, 401 for invalid signatures or timestamps, 413 for excess bytes, 415 for unsupported content, 422 for invalid event data, 409 for conflicting IDs, and 503 for an unavailable inbox. A sender that loses the response cannot assume that nothing was saved. Retry according to the actual provider's policy, retaining the stable event identity.

Keep acceptance separate from background processing

The inbox retains every accepted sample event and has no cleanup process. That makes repeat detection easy to inspect, but storage will grow. A real service needs a retention policy consistent with its provider's retry window and its own recovery requirements. Deleting old IDs changes what future duplicate checks can establish. A shared secret also authenticates only this one demonstration source; multiple providers or tenants need scoped secrets and identities.

Do not put consequential work in an in-process callback and assume the 202 response makes it recoverable. The FastAPI background tasks guide explains the completion and interruption limits of that approach.

For separate workers, the Celery with Redis guide provides a small queue exercise. Connecting it to this receiver needs an explicit handoff: a process can stop after committing the inbox but before publishing a task, or publish a task and fail before recording that fact. An outbox or dispatcher can make pending work discoverable, but it still needs retries, reconciliation, and idempotent downstream effects. None of that integration is implemented or tested here.

Choose a hosting path after the local checks

A CPU VM can host a continuously available receiver when you need control over the process, local storage, and adjacent workers. The FastAPI deployment guide covers the surrounding Docker and HTTPS procedure. Its shared API-key example does not replace provider-specific webhook verification.

A managed event receiver or serverless endpoint may fit better when you want less host maintenance or intermittent traffic. Compare execution limits, storage behavior, delivery handling, operating work, and cost for the actual workload. This small example establishes no capacity or savings claim.

Before exposing a service, separately verify the provider's signing format, registered HTTPS URL, retry schedule, event ordering, secret rotation, proxy limits, and recovery behavior. A committed SQLite record is not a backup or a high-availability design. Loss of the VM or disk needs a separate recovery plan. If considering Hivenet, discuss the CPU workload with its traffic, storage, and access requirements and confirm the available deployment path.