# FastAPI webhook receiver

Source: https://stormedo.com/examples/fastapi-webhook-receiver/

Run a FastAPI endpoint that verifies Stormedo signatures, logs the received JSON event, and acknowledges delivery.

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](https://stormedo.com/docs/get-started/first-request/). 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](https://stormedo.com/examples/fastapi-webhook-receiver/receiver.py).

```python
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](https://stormedo.com/docs/api-reference/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](https://stormedo.com/docs/concepts/idempotency/#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](https://stormedo.com/docs/how-to/receive-deliveries/) for configuration and response handling, or [Troubleshoot delivery](https://stormedo.com/docs/how-to/troubleshoot-delivery/) for failed attempts.