FastAPI webhook receiver

Receive a JSON event, verify its Stormedo signature, and print it in your terminal. The receiver returns 204 to acknowledge the delivery.

Install and run

Use Python 3.10 or newer. This example was exercised with Python 3.14.6, FastAPI 0.141.1, Uvicorn 0.52.4, and the Stormedo SDK 0.1.0a0 release candidate.

Bash

mkdir stormedo-fastapi-receiver
cd stormedo-fastapi-receiver
python -m venv .venv
source .venv/bin/activate
pip install fastapi==0.141.1 uvicorn==0.52.4 stormedo
curl --fail --output receiver.py \
  https://stormedo.com/examples/fastapi-webhook-receiver/receiver.py

Set the receiver configuration, then start it:

Bash

export STORMEDO_PROJECT_ID='prj_...'
export PUBLIC_WEBHOOK_URL='https://your-app.example/events'
uvicorn receiver:create_app --factory --host 127.0.0.1 --port 3000

Configure your HTTPS reverse proxy or tunnel to forward to 127.0.0.1:3000. The handler reads the original body bytes and verifies the signature before parsing JSON.

Obtain the actual project ID from data.project_id in a successful quickstart send. Configure the complete public URL in both the receiver and publisher, including any query string. Receiving needs the project ID, not the publisher API key.

Complete receiver

Download receiver.py.

receiver.py

import json
import os
from urllib.parse import urlsplit

from fastapi import FastAPI, Request, Response
from stormedo import AsyncDeliveryVerifier, DeliveryVerificationError


def create_app(*, verifier=None):
    project_id = os.environ["STORMEDO_PROJECT_ID"]
    public_url = os.environ["PUBLIC_WEBHOOK_URL"]
    destination = urlsplit(public_url)
    if destination.scheme != "https" or not destination.hostname or destination.username or destination.password or destination.fragment:
        raise ValueError("PUBLIC_WEBHOOK_URL must be the public HTTPS destination URL.")
    verifier = verifier or AsyncDeliveryVerifier(project_id)

    app = FastAPI(redirect_slashes=False)

    @app.post(destination.path or "/")
    async def receive(request: Request):
        def single_header(name):
            values = [value for key, value in request.scope["headers"] if key.lower() == name]
            return values[0].decode("latin-1") if len(values) == 1 and values[0] else None

        signature = single_header(b"stormedo-signature")
        request_id = single_header(b"stormedo-request-id")
        if not signature or not request_id:
            return Response(status_code=400)
        if request.headers.get("content-encoding", "identity") != "identity":
            return Response(status_code=415)
        raw_body = bytearray()
        async for chunk in request.stream():
            if len(raw_body) + len(chunk) > 256 * 1024:
                return Response(status_code=413)
            raw_body.extend(chunk)
        try:
            delivery = await verifier.verify_delivery(
                signature=signature, request_id=request_id,
                expected_url=public_url, method=request.method, raw_body=raw_body,
            )
        except DeliveryVerificationError as error:
            print("Delivery verification failed:", error.code)
            return Response(status_code=503 if error.code == "jwks_unavailable" else 401)
        try:
            event = json.loads(raw_body)
        except (ValueError, UnicodeDecodeError):
            return Response(status_code=400)
        print("Received event:", delivery.request_id, event)
        return Response(status_code=204)

    return app

Send a signed event

From another terminal, set STORMEDO_TOKEN to the project API key and PUBLIC_WEBHOOK_URL to the receiver’s public HTTPS URL. This command uses curl 7.87 or newer for --url-query:

Bash

curl --request POST 'https://api.stormedo.com/send' \
  --url-query "url=$PUBLIC_WEBHOOK_URL" \
  --url-query 'retry=3' \
  --header "Authorization: Bearer $STORMEDO_TOKEN" \
  --header 'Idempotency-Key: order-123-created' \
  --header 'Content-Type: application/json' \
  --data '{"type":"order.created","order_id":"ord_123"}'

Stormedo signs the outbound attempt automatically. Expect API 202, a Received event: line in the receiver’s terminal, and destination 204 in request history.

Before adding business logic

This example logs each verified event. Stormedo provides at-least-once delivery, so retries can print the same event again. Signature verification does not prevent duplicates.

Before replacing the log with an action such as updating an order or sending an email, validate the event fields and add receiver idempotency. Return 2xx only after your work or a durable queue handoff has committed.

The receiver rejects missing or duplicate signing headers and bodies over 256 KiB. Invalid signatures return 401; a temporary public-key fetch failure returns 503 so Stormedo can retry if attempts remain. Keep the raw body intact until verification finishes.

See Receive deliveries for configuration and response handling, or Troubleshoot delivery for failed attempts.