Build a Webhook Delivery Service

Introduction

30 Second Summary

When you buy something online and instantly get a confirmation email, or when a GitHub push triggers a CI build, that's a webhook doing its job. Something happened, and an HTTP callback delivered the news to every system that needed to know.

In this project, you will build a production-grade webhook delivery service that accepts event subscriptions, fans out signed HTTP callbacks to matching subscribers, and retries failures with exponential backoff.

What You'll Build

A webhook delivery service API with subscription management, HMAC-signed payload delivery, a retry engine, and auto-generated interactive Swagger documentation.

By the end of this project, you'll have:

• A REST API that registers webhook subscriptions and ingests events with async fan-out to all matching subscribers.
• A delivery worker that sends HMAC-signed HTTP callbacks and classifies responses into a state machine (pending, delivered, failed, dead_letter).
• A retry engine with exponential backoff and jitter that prevents thundering herd problems and tracks every attempt in a queryable database.
• Secret Mission: Add a manual replay endpoint that lets you re-trigger failed deliveries on demand.

Are there any prerequisites?

This project assumes you're comfortable with Python basics (functions, loops, dictionaries) and have a general understanding of REST APIs and HTTP methods.

You don't need prior experience with FastAPI, async Python, or webhook systems. The project teaches all of these from scratch.

Before We Start

Before We Start...

Set Up the Project

Before you can build the webhook delivery service, you need a clean project environment with the right tools installed.

In this step, you will create a project folder, set up a Python virtual environment, install three key packages, and confirm everything runs.

In this step, get ready to:

• Create a project folder and virtual environment.
• Install FastAPI, aiosqlite, and httpx.
• Create the project file structure and verify the dev server starts.

Create a project folder and virtual environment

A virtual environment isolates your project's packages from other Python projects on your machine.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir webhook-service
cd webhook-service

• Confirm the folder exists by running:

ls

You should see the empty webhook-service folder listed on your Desktop.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type VS Code and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Navigate to your Desktop, select the webhook-service folder, then click Open.

• Open the integrated terminal in VS Code by pressing Ctrl+` (both macOS and Windows).

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, preventing version conflicts between projects.

• Create a virtual environment by running this command:

macOS

python3 -m venv venv

Windows

python -m venv venv

• Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activation command from inside the webhook-service folder where the venv directory was created.

On Windows, if you see a permissions error, run this first: Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned. Then retry the activation command.

Help me activate my Python virtual environment.

• Install the three project dependencies by running:

pip install "fastapi[standard]" aiosqlite httpx

What are these packages?

• fastapi[standard] installs FastAPI plus its dev server, auto-generated docs, and validation tools.
• aiosqlite lets you query a SQLite database without blocking the async event loop.
• httpx is an async HTTP client that your delivery worker will use to send webhooks.

The install finishes when you see a success message listing all installed packages.

Seeing a pip install error?

Make sure (venv) appears in your prompt before running pip install. Installing without the venv active puts packages in the wrong location.

If you see a network timeout, check your internet connection and retry the same command.

Help me fix my pip install error.

Create the project file structure

Your webhook service will be split across four Python modules, each with a clear responsibility.

• Create the four project files by running:

macOS

touch main.py database.py schemas.py delivery.py

Windows

New-Item main.py, database.py, schemas.py, delivery.py

• Confirm the files exist by running:

ls

You should see main.py, database.py, schemas.py, and delivery.py listed alongside the venv folder.

What is each file for?

• main.py holds the FastAPI app and all route handlers.
• database.py manages the SQLite schema and database connections.
• schemas.py defines Pydantic models for request and response validation.
• delivery.py contains the background worker, retry logic, and HMAC signing.

• In the VS Code Explorer sidebar on the left, click main.py to open it.
• Add this minimal FastAPI app to verify your server can start:

from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI()


# Health check endpoint to verify the server is running
@app.get("/")
async def root():
    return {"message": "Webhook Delivery Service"}

What does this code do?

• from fastapi import FastAPI imports the framework class that powers your entire API.
• app = FastAPI() creates the application instance. Every route you define later attaches to this object.
• @app.get("/") registers a GET endpoint at the root URL. It acts as a health check you can hit to confirm the server is alive.

• Save main.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
• Start the development server by running this in your terminal:

fastapi dev

You should see output including a line that says Uvicorn running on http://127.0.0.1:8000.

• Open your web browser and navigate to http://127.0.0.1:8000/docs.

You should see an interactive Swagger UI page titled FastAPI with your GET / endpoint listed.

What is Swagger UI?

FastAPI auto-generates interactive API documentation from your route definitions. Swagger UI lets you test every endpoint directly in the browser without writing any client code. You will use it throughout this project to verify your work.

Swagger UI not loading?

Check your terminal for error messages. A ModuleNotFoundError means the virtual environment is not activated. Stop the server with Ctrl+C, activate the venv, and run fastapi dev again.

If the browser shows a connection refused error, make sure the terminal still shows the server running. Do not close the terminal tab while the server is active.

Help me get the FastAPI dev server running.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the Swagger UI is loading in your browser.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Webhook Delivery Service"}

Verify all imports work

The server proves FastAPI is installed correctly. Now confirm the other two packages are importable too.

• Open a second terminal in VS Code by clicking the + icon in the terminal panel.
• Make sure (venv) appears in the new terminal prompt.
• Verify all three packages import without errors by running:

python -c "import fastapi; import aiosqlite; import httpx; print('All packages ready')"

You should see All packages ready printed with no error messages.

Seeing an ImportError?

If a package fails to import, make sure (venv) is active in the terminal where you run the check. The new terminal panel may not auto-activate the venv.

Run the activation command again (source venv/bin/activate on macOS or venv\Scripts\activate on Windows), then retry the import check.

Help me fix my import errors.

Your environment is ready and all packages are confirmed working. Next up, you will build the database layer that stores subscriptions, events, and delivery records.

Build the Database Schema and Subscription API

Your project skeleton is ready, but your service has no memory yet. Without a database, subscriptions vanish the moment the server restarts.

In this step, you will build the persistence layer that stores webhook subscriptions, events, and delivery records. Then you will expose your first two API endpoints so callers can register and list subscriptions.

In this step, get ready to:

• Build the database schema with three tables and an index for efficient delivery polling.
• Define Pydantic models for request and response validation.
• Create subscription API endpoints and test them in Swagger UI.

Build the database layer

Your webhook service needs to persist three kinds of records: which URLs are subscribed to which event types, which events have arrived, and the status of every delivery attempt. A single SQLite database handles all of this with zero external infrastructure.

• Open database.py in VS Code and add the full database layer by pasting this code:

import aiosqlite

DATABASE_PATH = "webhooks.db"


async def get_db():
    # Helper to open a connection with Row factory for dict-like access
    db = await aiosqlite.connect(DATABASE_PATH)
    db.row_factory = aiosqlite.Row
    return db


async def init_db():
    # Create all tables and indexes on startup
    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.executescript("""
            CREATE TABLE IF NOT EXISTS subscriptions (
                id TEXT PRIMARY KEY,
                url TEXT NOT NULL,
                secret TEXT NOT NULL,
                event_types TEXT NOT NULL,
                is_active INTEGER DEFAULT 1,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS events (
                id TEXT PRIMARY KEY,
                event_type TEXT NOT NULL,
                payload TEXT NOT NULL,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS deliveries (
                id TEXT PRIMARY KEY,
                event_id TEXT NOT NULL,
                subscription_id TEXT NOT NULL,
                url TEXT NOT NULL,
                payload TEXT NOT NULL,
                secret TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                attempt_count INTEGER DEFAULT 0,
                max_attempts INTEGER DEFAULT 5,
                next_attempt_at TEXT DEFAULT (datetime('now')),
                last_status_code INTEGER,
                last_error TEXT,
                created_at TEXT DEFAULT (datetime('now')),
                delivered_at TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id),
                FOREIGN KEY (subscription_id) REFERENCES subscriptions(id)
            );

            CREATE INDEX IF NOT EXISTS idx_deliveries_status_next
            ON deliveries(status, next_attempt_at);
        """)
        await db.commit()

What does this code do?

DATABASE_PATH sets the SQLite file name. get_db() opens a connection with aiosqlite.Row so you can access columns by name instead of index.

init_db() creates three tables. subscriptions stores webhook URLs and their signing secrets. events stores every incoming event. deliveries tracks each individual attempt to POST an event to a subscriber.

The index on (status, next_attempt_at) lets the background worker efficiently find deliveries that are ready to retry without scanning the whole table.

• Save database.py.
• You won't see a change yet. The next file (schemas.py) defines the data shapes, then main.py wires everything together so you can test in the browser.

Seeing a red underline on the import?

Make sure your virtual environment is activated. The terminal prompt should show (venv) at the beginning of the line.

help me fix my aiosqlite import error

✔️ Awesome, I've got everything!

Great. Make sure you have saved database.py before moving on.

ⓧ I'd like to double check the full code

import aiosqlite

DATABASE_PATH = "webhooks.db"


async def get_db():
    db = await aiosqlite.connect(DATABASE_PATH)
    db.row_factory = aiosqlite.Row
    return db


async def init_db():
    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.executescript("""
            CREATE TABLE IF NOT EXISTS subscriptions (
                id TEXT PRIMARY KEY,
                url TEXT NOT NULL,
                secret TEXT NOT NULL,
                event_types TEXT NOT NULL,
                is_active INTEGER DEFAULT 1,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS events (
                id TEXT PRIMARY KEY,
                event_type TEXT NOT NULL,
                payload TEXT NOT NULL,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS deliveries (
                id TEXT PRIMARY KEY,
                event_id TEXT NOT NULL,
                subscription_id TEXT NOT NULL,
                url TEXT NOT NULL,
                payload TEXT NOT NULL,
                secret TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                attempt_count INTEGER DEFAULT 0,
                max_attempts INTEGER DEFAULT 5,
                next_attempt_at TEXT DEFAULT (datetime('now')),
                last_status_code INTEGER,
                last_error TEXT,
                created_at TEXT DEFAULT (datetime('now')),
                delivered_at TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id),
                FOREIGN KEY (subscription_id) REFERENCES subscriptions(id)
            );

            CREATE INDEX IF NOT EXISTS idx_deliveries_status_next
            ON deliveries(status, next_attempt_at);
        """)
        await db.commit()

Define the Pydantic schemas

FastAPI uses Pydantic models to validate incoming request bodies and shape outgoing responses. Each model acts as a contract between your API and its callers.

• Open schemas.py and paste all of the data models:

from pydantic import BaseModel, HttpUrl


# What the caller sends when creating a subscription
class SubscriptionCreate(BaseModel):
    url: HttpUrl
    event_types: list[str]


# What the API returns for a subscription
class SubscriptionResponse(BaseModel):
    id: str
    url: str
    secret: str
    event_types: list[str]
    is_active: bool
    created_at: str


# What the caller sends when ingesting an event
class EventCreate(BaseModel):
    event_type: str
    payload: dict


# What the API returns after ingesting an event
class EventResponse(BaseModel):
    id: str
    event_type: str
    deliveries_created: int


# What the API returns for a delivery record
class DeliveryResponse(BaseModel):
    id: str
    event_id: str
    subscription_id: str
    status: str
    attempt_count: int
    max_attempts: int
    next_attempt_at: str | None
    last_status_code: int | None
    last_error: str | None
    created_at: str
    delivered_at: str | None

What does this code do?

SubscriptionCreate validates that callers provide a proper URL (HttpUrl rejects malformed URLs) and a list of event types they want to receive.

SubscriptionResponse includes the auto-generated secret used for HMAC signing. DeliveryResponse exposes delivery state so callers can inspect retry progress and failure details.

Fields marked str | None are nullable because they only get populated after a delivery attempt runs.

• Save schemas.py.
• You won't see a change yet. The next chunk wires these models into your API endpoints.

✔️ Awesome, I've got everything!

Great. Make sure you have saved schemas.py before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, HttpUrl


class SubscriptionCreate(BaseModel):
    url: HttpUrl
    event_types: list[str]


class SubscriptionResponse(BaseModel):
    id: str
    url: str
    secret: str
    event_types: list[str]
    is_active: bool
    created_at: str


class EventCreate(BaseModel):
    event_type: str
    payload: dict


class EventResponse(BaseModel):
    id: str
    event_type: str
    deliveries_created: int


class DeliveryResponse(BaseModel):
    id: str
    event_id: str
    subscription_id: str
    status: str
    attempt_count: int
    max_attempts: int
    next_attempt_at: str | None
    last_status_code: int | None
    last_error: str | None
    created_at: str
    delivered_at: str | None

Build the subscription endpoints

Now you will replace the placeholder app in main.py with the real service. The new version initializes the database on startup using a FastAPI lifespan handler.

• Open main.py and replace everything in the file with this setup code:

import asyncio
import json
import secrets
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import aiosqlite
from fastapi import FastAPI, HTTPException

from database import DATABASE_PATH, init_db
from schemas import (
    DeliveryResponse,
    EventCreate,
    EventResponse,
    SubscriptionCreate,
    SubscriptionResponse,
)


# Lifespan runs init_db() once when the server starts
@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield


app = FastAPI(
    title="Webhook Delivery Service",
    description="A reliable webhook delivery system with retry logic, HMAC signing, and delivery tracking.",
    version="1.0.0",
    lifespan=lifespan,
)

What does this code do?

The lifespan function is an async context manager that runs once when the server boots. It calls init_db() to create all tables, then yield lets the server run until shutdown.

Passing lifespan=lifespan to FastAPI() replaces the older @app.on_event pattern. The title, description, and version populate the auto-generated Swagger docs.

• Save main.py and start the development server by running this command in your terminal:

fastapi dev

• Confirm the server starts without import errors. You should see output including Uvicorn running on http://127.0.0.1:8000.

Seeing a ModuleNotFoundError?

Double-check that your virtual environment is active (look for (venv) in your terminal prompt). If you see an error about database or schemas, confirm both files are saved in the same webhook-service folder as main.py.

help me fix my FastAPI import error

The server is running, but it has no endpoints yet. Time to add the POST endpoint that creates subscriptions.

• Stop the server with Ctrl+C.
• In main.py, add the POST endpoint below the app = FastAPI(...) block:

@app.post("/subscriptions", response_model=SubscriptionResponse, status_code=201)
async def create_subscription(subscription: SubscriptionCreate):
    # Generate a unique ID and signing secret for this subscription
    sub_id = str(uuid.uuid4())
    secret = f"whsec_{secrets.token_hex(24)}"
    event_types_str = json.dumps(subscription.event_types)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    # Insert the subscription into the database
    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.execute(
            "INSERT INTO subscriptions (id, url, secret, event_types, created_at) VALUES (?, ?, ?, ?, ?)",
            (sub_id, str(subscription.url), secret, event_types_str, created_at),
        )
        await db.commit()

    return SubscriptionResponse(
        id=sub_id,
        url=str(subscription.url),
        secret=secret,
        event_types=subscription.event_types,
        is_active=True,
        created_at=created_at,
    )

What does this code do?

This endpoint receives a URL and list of event types from the caller. It generates a UUID for the subscription ID and a signing secret prefixed with whsec_ (a convention webhook services use so you can identify the key type at a glance).

The status_code=201 tells FastAPI to return HTTP 201 Created instead of the default 200. The response_model ensures the response body matches SubscriptionResponse exactly.

• Save main.py.
• You won't test this endpoint yet. The next chunk adds GET so you can test both together.

• Add the GET endpoint below the create_subscription function:

@app.get("/subscriptions", response_model=list[SubscriptionResponse])
async def list_subscriptions():
    # Fetch all active subscriptions from the database
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM subscriptions WHERE is_active = 1")
        rows = await cursor.fetchall()

    return [
        SubscriptionResponse(
            id=row["id"],
            url=row["url"],
            secret=row["secret"],
            event_types=json.loads(row["event_types"]),
            is_active=bool(row["is_active"]),
            created_at=row["created_at"],
        )
        for row in rows
    ]

What does this code do?

This endpoint queries all subscriptions where is_active = 1 and returns them as a list. The row_factory = aiosqlite.Row setting lets you access columns by name (row["url"]) instead of by index.

Since event_types is stored as a JSON string in SQLite, json.loads() converts it back into a Python list before returning it to the caller.

• Save main.py and start the server again by running:

fastapi dev

• Open your browser and go to http://127.0.0.1:8000/docs.
• You should see the Swagger UI with POST /subscriptions and GET /subscriptions listed.
• Click POST /subscriptions, then click Try it out.
• Replace the example request body with this JSON:

{
  "url": "https://httpbin.org/post",
  "event_types": ["order.created", "order.updated"]
}

• Click Execute.
• You should see a 201 response with an id, a secret starting with whsec_, and the event types you submitted.
• Now click GET /subscriptions, click Try it out, then Execute.
• You should see a list containing the subscription you just created.

Getting a 422 Validation Error on POST?

FastAPI returns 422 when the request body does not match the Pydantic model. Check that your JSON uses url (not URL) and event_types (not eventTypes). The url field must be a full URL including https://.

help me debug my 422 validation error

✔️ Awesome, I've got everything!

Great. Make sure all three files are saved before moving on.

ⓧ I'd like to double check the full code

import aiosqlite

DATABASE_PATH = "webhooks.db"


async def get_db():
    db = await aiosqlite.connect(DATABASE_PATH)
    db.row_factory = aiosqlite.Row
    return db


async def init_db():
    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.executescript("""
            CREATE TABLE IF NOT EXISTS subscriptions (
                id TEXT PRIMARY KEY,
                url TEXT NOT NULL,
                secret TEXT NOT NULL,
                event_types TEXT NOT NULL,
                is_active INTEGER DEFAULT 1,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS events (
                id TEXT PRIMARY KEY,
                event_type TEXT NOT NULL,
                payload TEXT NOT NULL,
                created_at TEXT DEFAULT (datetime('now'))
            );

            CREATE TABLE IF NOT EXISTS deliveries (
                id TEXT PRIMARY KEY,
                event_id TEXT NOT NULL,
                subscription_id TEXT NOT NULL,
                url TEXT NOT NULL,
                payload TEXT NOT NULL,
                secret TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                attempt_count INTEGER DEFAULT 0,
                max_attempts INTEGER DEFAULT 5,
                next_attempt_at TEXT DEFAULT (datetime('now')),
                last_status_code INTEGER,
                last_error TEXT,
                created_at TEXT DEFAULT (datetime('now')),
                delivered_at TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id),
                FOREIGN KEY (subscription_id) REFERENCES subscriptions(id)
            );

            CREATE INDEX IF NOT EXISTS idx_deliveries_status_next
            ON deliveries(status, next_attempt_at);
        """)
        await db.commit()

from pydantic import BaseModel, HttpUrl


class SubscriptionCreate(BaseModel):
    url: HttpUrl
    event_types: list[str]


class SubscriptionResponse(BaseModel):
    id: str
    url: str
    secret: str
    event_types: list[str]
    is_active: bool
    created_at: str


class EventCreate(BaseModel):
    event_type: str
    payload: dict


class EventResponse(BaseModel):
    id: str
    event_type: str
    deliveries_created: int


class DeliveryResponse(BaseModel):
    id: str
    event_id: str
    subscription_id: str
    status: str
    attempt_count: int
    max_attempts: int
    next_attempt_at: str | None
    last_status_code: int | None
    last_error: str | None
    created_at: str
    delivered_at: str | None

import asyncio
import json
import secrets
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import aiosqlite
from fastapi import FastAPI, HTTPException

from database import DATABASE_PATH, init_db
from schemas import (
    DeliveryResponse,
    EventCreate,
    EventResponse,
    SubscriptionCreate,
    SubscriptionResponse,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield


app = FastAPI(
    title="Webhook Delivery Service",
    description="A reliable webhook delivery system with retry logic, HMAC signing, and delivery tracking.",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/subscriptions", response_model=SubscriptionResponse, status_code=201)
async def create_subscription(subscription: SubscriptionCreate):
    sub_id = str(uuid.uuid4())
    secret = f"whsec_{secrets.token_hex(24)}"
    event_types_str = json.dumps(subscription.event_types)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.execute(
            "INSERT INTO subscriptions (id, url, secret, event_types, created_at) VALUES (?, ?, ?, ?, ?)",
            (sub_id, str(subscription.url), secret, event_types_str, created_at),
        )
        await db.commit()

    return SubscriptionResponse(
        id=sub_id,
        url=str(subscription.url),
        secret=secret,
        event_types=subscription.event_types,
        is_active=True,
        created_at=created_at,
    )


@app.get("/subscriptions", response_model=list[SubscriptionResponse])
async def list_subscriptions():
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM subscriptions WHERE is_active = 1")
        rows = await cursor.fetchall()

    return [
        SubscriptionResponse(
            id=row["id"],
            url=row["url"],
            secret=row["secret"],
            event_types=json.loads(row["event_types"]),
            is_active=bool(row["is_active"]),
            created_at=row["created_at"],
        )
        for row in rows
    ]

Your database is wired up, your models are validated, and you have a working subscription API. Next up, you will build the event ingestion endpoint that fans out deliveries to all matching subscribers.

Implement Event Ingestion and Fan-out

Your subscription API is working. Users can register URLs and event types they care about. Now you need the other half of the equation: a way to receive events and route them to the right subscribers.

In this step, you'll build the fan-out pattern. When an event arrives, your service checks every active subscription, matches by event type, and creates a pending delivery record for each match. This is the same pattern that powers webhook systems at Stripe, GitHub, and Shopify.

In this step, get ready to:

• Build a POST /events endpoint that stores events and fans out delivery records to matching subscribers.
• Add a GET /deliveries endpoint to query delivery status with optional filters.
• Test the full ingestion and fan-out flow in Swagger UI.

Build the event ingestion endpoint

The POST /events endpoint is the entry point for all events flowing into your system. It does two things in a single database transaction: stores the event, then finds every subscription whose event types match and creates a pending delivery for each one.

What is fan-out?

Fan-out means taking a single input (one event) and distributing it to multiple outputs (many delivery records). One order.completed event might fan out to three subscribers: an email service, an analytics tracker, and an inventory system.

The endpoint returns HTTP 202 (Accepted) instead of 200 (OK) because delivery happens asynchronously. The event is acknowledged immediately, but actual HTTP callbacks happen later in the background worker.

• In main.py, add the following function below your list_subscriptions endpoint:

@app.post("/events", response_model=EventResponse, status_code=202)
async def ingest_event(event: EventCreate):
    event_id = str(uuid.uuid4())
    payload_str = json.dumps(event.payload)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        # Store the incoming event
        await db.execute(
            "INSERT INTO events (id, event_type, payload, created_at) VALUES (?, ?, ?, ?)",
            (event_id, event.event_type, payload_str, created_at),
        )

        # Query all active subscriptions to find matches
        cursor = await db.execute(
            "SELECT * FROM subscriptions WHERE is_active = 1"
        )
        subscriptions = await cursor.fetchall()

        # Fan out: create a delivery record for each matching subscription
        deliveries_created = 0
        for sub in subscriptions:
            event_types = json.loads(sub["event_types"])
            if event.event_type in event_types or "*" in event_types:
                delivery_id = str(uuid.uuid4())
                await db.execute(
                    """
                    INSERT INTO deliveries (id, event_id, subscription_id, url, payload, secret, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (delivery_id, event_id, sub["id"], sub["url"], payload_str, sub["secret"], created_at),
                )
                deliveries_created += 1

        await db.commit()

    return EventResponse(
        id=event_id,
        event_type=event.event_type,
        deliveries_created=deliveries_created,
    )

What does this code do?

• status_code=202 tells FastAPI to return HTTP 202 Accepted. This signals that the event was received but delivery is asynchronous.
• The function stores the event in the events table, then loops through all active subscriptions to check for matches.
• The matching logic checks if the event type is in the subscription's list OR if the subscription uses a wildcard "*" to receive all events.
• For each match, a new delivery record is inserted with status='pending'. The background worker (built in the next step) picks these up and sends the actual HTTP requests.

• Save main.py.

Seeing an import error on startup?

If the server reports an import error for EventCreate or EventResponse, confirm that your schemas.py has both classes defined. These were created in the previous step.

Help me debug my import error in main.py

Add a deliveries query endpoint

You need a way to inspect delivery records to verify that fan-out is working. This endpoint also supports optional query filters for status and subscription ID, which is essential for debugging in production.

• In main.py, add the following function below your ingest_event endpoint:

@app.get("/deliveries", response_model=list[DeliveryResponse])
async def list_deliveries(status: str | None = None, subscription_id: str | None = None):
    # Build a dynamic query with optional filters
    query = "SELECT * FROM deliveries WHERE 1=1"
    params: list[str] = []

    if status:
        query += " AND status = ?"
        params.append(status)
    if subscription_id:
        query += " AND subscription_id = ?"
        params.append(subscription_id)

    # Return most recent deliveries first, capped at 50
    query += " ORDER BY created_at DESC LIMIT 50"

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(query, params)
        rows = await cursor.fetchall()

    return [
        DeliveryResponse(
            id=row["id"],
            event_id=row["event_id"],
            subscription_id=row["subscription_id"],
            status=row["status"],
            attempt_count=row["attempt_count"],
            max_attempts=row["max_attempts"],
            next_attempt_at=row["next_attempt_at"],
            last_status_code=row["last_status_code"],
            last_error=row["last_error"],
            created_at=row["created_at"],
            delivered_at=row["delivered_at"],
        )
        for row in rows
    ]

What does this code do?

• The WHERE 1=1 trick makes it easy to append optional filters with AND without worrying about whether a previous condition exists.
• Query parameters like ?status=pending or ?subscription_id=abc narrow results. FastAPI automatically parses these from the URL.
• Results are sorted newest-first and capped at 50 to prevent slow queries on large tables.

• Save main.py.
• Check the terminal where fastapi dev is running. It should reload without errors.

Seeing a syntax error?

Make sure the new function is placed after the closing return statement of ingest_event, not inside it. Python indentation determines scope, so the @app.get decorator must be at the top level (no indentation).

Help me fix my syntax error in main.py

Test the fan-out in Swagger UI

Time to see the full event-to-delivery pipeline in action. You'll create a subscription, fire an event, and verify that a delivery record appears.

• Switch back to your browser with Swagger UI at http://127.0.0.1:8000/docs and refresh the page.
• Expand the POST /subscriptions section and click Try it out.
• Replace the request body with this JSON and click Execute:

{
  "url": "https://httpbin.org/post",
  "event_types": ["order.completed", "order.refunded"]
}

• Confirm you see a 201 response with your subscription's id and secret.
• Now expand the POST /events section and click Try it out.
• Replace the request body with this JSON and click Execute:

{
  "event_type": "order.completed",
  "payload": {"order_id": "abc-123", "amount": 49.99}
}

• Confirm you see a 202 response with "deliveries_created": 1. This proves the fan-out matched your subscription.
• Expand the GET /deliveries section, click Try it out, then click Execute.
• You should see a delivery record with "status": "pending" and the URL, payload, and secret from your subscription.

Why is the status still pending?

The delivery record exists, but nothing has actually sent the HTTP request yet. The background delivery worker (which you'll build in the next step) is responsible for picking up pending deliveries and executing them. Right now, you're looking at the "intent to deliver" sitting in the queue.

Seeing deliveries_created: 0?

This means the fan-out found no matching subscriptions. Check that the event_type in your POST /events request matches one of the strings in your subscription's event_types array. If you previously deleted the webhooks.db file or restarted fresh, you'll need to create a new subscription first.

Help me debug why my fan-out isn't creating deliveries

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py before moving on.

ⓧ I'd like to double check the full code

Here is the full main.py at this point:

import asyncio
import json
import secrets
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import aiosqlite
from fastapi import FastAPI, HTTPException

from database import DATABASE_PATH, init_db
from schemas import (
    DeliveryResponse,
    EventCreate,
    EventResponse,
    SubscriptionCreate,
    SubscriptionResponse,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield


app = FastAPI(
    title="Webhook Delivery Service",
    description="A reliable webhook delivery system with retry logic, HMAC signing, and delivery tracking.",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/subscriptions", response_model=SubscriptionResponse, status_code=201)
async def create_subscription(subscription: SubscriptionCreate):
    sub_id = str(uuid.uuid4())
    secret = f"whsec_{secrets.token_hex(24)}"
    event_types_str = json.dumps(subscription.event_types)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.execute(
            "INSERT INTO subscriptions (id, url, secret, event_types, created_at) VALUES (?, ?, ?, ?, ?)",
            (sub_id, str(subscription.url), secret, event_types_str, created_at),
        )
        await db.commit()

    return SubscriptionResponse(
        id=sub_id,
        url=str(subscription.url),
        secret=secret,
        event_types=subscription.event_types,
        is_active=True,
        created_at=created_at,
    )


@app.get("/subscriptions", response_model=list[SubscriptionResponse])
async def list_subscriptions():
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM subscriptions WHERE is_active = 1")
        rows = await cursor.fetchall()

    return [
        SubscriptionResponse(
            id=row["id"],
            url=row["url"],
            secret=row["secret"],
            event_types=json.loads(row["event_types"]),
            is_active=bool(row["is_active"]),
            created_at=row["created_at"],
        )
        for row in rows
    ]


@app.post("/events", response_model=EventResponse, status_code=202)
async def ingest_event(event: EventCreate):
    event_id = str(uuid.uuid4())
    payload_str = json.dumps(event.payload)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        await db.execute(
            "INSERT INTO events (id, event_type, payload, created_at) VALUES (?, ?, ?, ?)",
            (event_id, event.event_type, payload_str, created_at),
        )

        cursor = await db.execute(
            "SELECT * FROM subscriptions WHERE is_active = 1"
        )
        subscriptions = await cursor.fetchall()

        deliveries_created = 0
        for sub in subscriptions:
            event_types = json.loads(sub["event_types"])
            if event.event_type in event_types or "*" in event_types:
                delivery_id = str(uuid.uuid4())
                await db.execute(
                    """
                    INSERT INTO deliveries (id, event_id, subscription_id, url, payload, secret, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (delivery_id, event_id, sub["id"], sub["url"], payload_str, sub["secret"], created_at),
                )
                deliveries_created += 1

        await db.commit()

    return EventResponse(
        id=event_id,
        event_type=event.event_type,
        deliveries_created=deliveries_created,
    )


@app.get("/deliveries", response_model=list[DeliveryResponse])
async def list_deliveries(status: str | None = None, subscription_id: str | None = None):
    query = "SELECT * FROM deliveries WHERE 1=1"
    params: list[str] = []

    if status:
        query += " AND status = ?"
        params.append(status)
    if subscription_id:
        query += " AND subscription_id = ?"
        params.append(subscription_id)

    query += " ORDER BY created_at DESC LIMIT 50"

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(query, params)
        rows = await cursor.fetchall()

    return [
        DeliveryResponse(
            id=row["id"],
            event_id=row["event_id"],
            subscription_id=row["subscription_id"],
            status=row["status"],
            attempt_count=row["attempt_count"],
            max_attempts=row["max_attempts"],
            next_attempt_at=row["next_attempt_at"],
            last_status_code=row["last_status_code"],
            last_error=row["last_error"],
            created_at=row["created_at"],
            delivered_at=row["delivered_at"],
        )
        for row in rows
    ]

Your event pipeline is wired up. Events come in, fan out to matching subscribers, and sit as pending delivery records in the database. Next up, you'll build the async delivery worker that actually sends these webhooks and signs them with HMAC for security.

Build the Async Delivery Worker with HMAC Signing

Your fan-out logic creates delivery records in the database, but nothing actually sends them yet. Right now those records just sit there with a pending status.

In this step, you will build the engine that picks up pending deliveries and fires them off as signed HTTP requests. This includes HMAC payload signing (so receivers can verify authenticity), exponential backoff with jitter (so failed deliveries retry gracefully), and a background worker loop that polls the database every 5 seconds.

In this step, get ready to:

• Implement HMAC-SHA256 signing and a backoff schedule in delivery.py.
• Build the delivery function and processing loop that classifies HTTP responses.
• Wire a background worker into FastAPI's lifespan and test the full delivery lifecycle.

Implement HMAC signing and backoff calculation

Webhook receivers need a way to verify that an incoming HTTP request actually came from your service and was not tampered with in transit. HMAC-SHA256 solves this by signing the payload with a shared secret.

• In VS Code, open delivery.py and add the following imports, constant, and signing function:

import hashlib
import hmac
import random
import time
from datetime import datetime, timedelta, timezone

import aiosqlite
import httpx

from database import DATABASE_PATH

# Delays in seconds: immediate, 30s, 5min, 30min, 2hr
BACKOFF_SCHEDULE = [0, 30, 300, 1800, 7200]


def sign_payload(payload: str, secret: str, timestamp: str) -> str:
    # Format: "{timestamp}.{payload}" signed with the subscription secret
    message = f"{timestamp}.{payload}"
    signature = hmac.new(
        secret.encode(),
        message.encode(),
        hashlib.sha256,
    ).hexdigest()
    return signature

What does this code do?

sign_payload concatenates the Unix timestamp and the JSON payload with a dot separator, then produces a hex-encoded HMAC-SHA256 digest using the subscription's secret.

The receiver can repeat the same operation with their copy of the secret. If the signatures match, the payload is authentic and unmodified.

BACKOFF_SCHEDULE defines how long to wait between retry attempts. The delays escalate from immediate to 2 hours, giving failing endpoints time to recover.

• Below sign_payload, add the backoff calculation function:

def calculate_next_attempt(attempt_count: int) -> str:
    # Pick the delay from the schedule (cap at the last value)
    if attempt_count >= len(BACKOFF_SCHEDULE):
        base_delay = BACKOFF_SCHEDULE[-1]
    else:
        base_delay = BACKOFF_SCHEDULE[attempt_count]

    # Add jitter: random 0-25% of base delay to prevent thundering herd
    jitter = random.uniform(0, base_delay * 0.25) if base_delay > 0 else 0
    delay = base_delay + jitter
    next_time = datetime.now(timezone.utc) + timedelta(seconds=delay)
    return next_time.strftime("%Y-%m-%d %H:%M:%S")

Why add jitter?

Imagine 100 deliveries all targeting the same endpoint that just went down. Without jitter, all 100 would retry at exactly the same second, potentially overwhelming the endpoint the moment it recovers.

Adding a random value between 0 and 25% of the base delay spreads retries across a window. This is called preventing the "thundering herd" problem.

• Save delivery.py.

You will not see a visible change yet. The next chunk uses these functions to actually send webhooks.

Build the delivery function and processing loop

The signing function prepares the cryptographic headers, but you still need the function that actually fires the HTTP request and a loop that processes pending deliveries from the database.

• In delivery.py, add the deliver_webhook function below calculate_next_attempt:

async def deliver_webhook(delivery: dict) -> tuple[int | None, str | None]:
    payload_str = delivery["payload"]
    timestamp = str(int(time.time()))
    signature = sign_payload(payload_str, delivery["secret"], timestamp)

    # Standard webhook headers for verification
    headers = {
        "Content-Type": "application/json",
        "X-Webhook-ID": delivery["id"],
        "X-Webhook-Timestamp": timestamp,
        "X-Webhook-Signature": f"sha256={signature}",
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.post(
                delivery["url"],
                content=payload_str,
                headers=headers,
            )
            return response.status_code, None
        except (httpx.TimeoutException, httpx.ConnectError) as e:
            return None, str(e)

What do the webhook headers mean?

• X-Webhook-ID: the delivery UUID. Receivers use this for idempotency (deduplicating repeated deliveries).
• X-Webhook-Timestamp: Unix epoch when the request was sent. Receivers can reject old timestamps to prevent replay attacks.
• X-Webhook-Signature: the HMAC digest prefixed with sha256=. Receivers verify this against their own computed signature.

If the request times out (30 seconds) or the connection fails entirely, the function returns None for the status code and the error message as a string.

• Below deliver_webhook, add the process_deliveries function that polls the database and classifies results:

async def process_deliveries() -> int:
    processed = 0
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        # Poll for deliveries that are ready to send
        cursor = await db.execute(
            """
            SELECT id, event_id, subscription_id, url, payload, secret,
                   attempt_count, max_attempts
            FROM deliveries
            WHERE status IN ('pending', 'failed')
            AND next_attempt_at <= ?
            LIMIT 10
            """,
            (now,),
        )
        deliveries = await cursor.fetchall()

        for delivery in deliveries:
            delivery_dict = dict(delivery)
            status_code, error = await deliver_webhook(delivery_dict)
            new_attempt_count = delivery_dict["attempt_count"] + 1

            if status_code and 200 <= status_code < 300:
                # Success: mark as delivered
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'delivered',
                        attempt_count = ?,
                        last_status_code = ?,
                        delivered_at = datetime('now')
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, delivery_dict["id"]),
                )
            elif status_code and 400 <= status_code < 500 and status_code not in (408, 429):
                # Client error (not timeout/rate-limit): permanent failure
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, f"HTTP {status_code}", delivery_dict["id"]),
                )
            elif new_attempt_count >= delivery_dict["max_attempts"]:
                # Max retries exhausted: permanent failure
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, error or f"Max attempts exceeded", delivery_dict["id"]),
                )
            else:
                # Transient failure: schedule retry with backoff
                next_attempt = calculate_next_attempt(new_attempt_count)
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'failed',
                        attempt_count = ?,
                        next_attempt_at = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, next_attempt, status_code, error, delivery_dict["id"]),
                )

            processed += 1

        await db.commit()
    return processed

How does the status classification work?

The delivery state machine has four states:

• pending and failed: eligible for processing. The worker picks these up when next_attempt_at is in the past.
• delivered: terminal success. A 2xx response from the receiver.
• dead_letter: terminal failure. Either a 4xx client error (the URL is fundamentally wrong) or max retries exhausted.

408 (timeout) and 429 (rate limit) are excluded from the "immediate dead_letter" rule because they are transient. Those get retried like 5xx errors.

• Save delivery.py.

You will not see a change yet. The processing loop needs a background worker to call it repeatedly.

Seeing a syntax error on save?

Make sure the f"HTTP {status_code}" strings use regular double quotes, not curly quotes copied from a PDF or web page. Also check that the elif branches are at the same indentation level as the if they belong to.

help me debug a syntax error in my process_deliveries function

Wire the worker into FastAPI and test the lifecycle

The processing loop runs once when called, but your service needs it running continuously in the background. You will create a simple polling worker and attach it to FastAPI's lifespan so it starts automatically with the server.

• At the bottom of delivery.py, add the delivery_worker function:

async def delivery_worker():
    import asyncio

    # Poll every 5 seconds for deliveries ready to send
    while True:
        try:
            await process_deliveries()
        except Exception as e:
            print(f"Delivery worker error: {e}")
        await asyncio.sleep(5)

Why import asyncio inside the function?

This avoids a circular import. The main.py file imports from delivery.py, and both files use asyncio. Importing asyncio inside the function keeps the module's top-level imports clean and focused on the delivery domain.

• Save delivery.py.

Now you need to update main.py to import and start this worker.

• In main.py, find this import block near the top:

from database import DATABASE_PATH, init_db
from schemas import (

• Add the delivery_worker import between those two lines, so the imports now look like this:

from database import DATABASE_PATH, init_db
from delivery import delivery_worker
from schemas import (

• Next, find the lifespan function:

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield

• Replace it with the following version that starts and cancels the worker:

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    # Start the background delivery worker
    worker_task = asyncio.create_task(delivery_worker())
    yield
    # Gracefully cancel on shutdown
    worker_task.cancel()
    try:
        await worker_task
    except asyncio.CancelledError:
        pass

What does this lifespan do?

Everything before yield runs at startup. asyncio.create_task schedules the worker on the event loop so it runs concurrently with incoming requests.

Everything after yield runs at shutdown. Cancelling the task and catching CancelledError ensures a clean exit without error tracebacks.

• Finally, add a GET /deliveries/{delivery_id} endpoint at the bottom of main.py to inspect individual deliveries:

@app.get("/deliveries/{delivery_id}", response_model=DeliveryResponse)
async def get_delivery(delivery_id: str):
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM deliveries WHERE id = ?", (delivery_id,))
        row = await cursor.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="Delivery not found")

    return DeliveryResponse(
        id=row["id"],
        event_id=row["event_id"],
        subscription_id=row["subscription_id"],
        status=row["status"],
        attempt_count=row["attempt_count"],
        max_attempts=row["max_attempts"],
        next_attempt_at=row["next_attempt_at"],
        last_status_code=row["last_status_code"],
        last_error=row["last_error"],
        created_at=row["created_at"],
        delivered_at=row["delivered_at"],
    )

What does this endpoint do?

This lets you look up a specific delivery by its UUID. You will use it to confirm whether the worker processed a delivery and what status it ended up with.

• Save main.py.
• If your server is already running, it will hot-reload automatically. Otherwise, start it by running:

fastapi dev

• Open http://127.0.0.1:8000/docs in your browser.
• Use the POST /subscriptions endpoint to create a subscription pointing to https://httpbin.org/post with event type order.created.
• Use the POST /events endpoint to ingest an event with type order.created and any payload (e.g. {"order_id": "abc123"}).
• Wait about 5 seconds for the worker to poll.
• Use the GET /deliveries endpoint. You should see a delivery with status set to delivered and a populated last_status_code of 200.

Delivery still showing as pending?

Check your terminal for error messages from the worker. If you see an import error, make sure you added from delivery import delivery_worker in main.py.

If the delivery shows dead_letter instead of delivered, the URL might be unreachable from your network. Try using https://httpbin.org/post which accepts any POST and returns 200.

help me debug why my delivery worker is not processing pending deliveries

✔️ Awesome, I've got everything!

Great. Double check you have saved both delivery.py and main.py.

ⓧ I'd like to double check the full code

Here is the full delivery.py:

import hashlib
import hmac
import random
import time
from datetime import datetime, timedelta, timezone

import aiosqlite
import httpx

from database import DATABASE_PATH

BACKOFF_SCHEDULE = [0, 30, 300, 1800, 7200]


def sign_payload(payload: str, secret: str, timestamp: str) -> str:
    message = f"{timestamp}.{payload}"
    signature = hmac.new(
        secret.encode(),
        message.encode(),
        hashlib.sha256,
    ).hexdigest()
    return signature


def calculate_next_attempt(attempt_count: int) -> str:
    if attempt_count >= len(BACKOFF_SCHEDULE):
        base_delay = BACKOFF_SCHEDULE[-1]
    else:
        base_delay = BACKOFF_SCHEDULE[attempt_count]

    jitter = random.uniform(0, base_delay * 0.25) if base_delay > 0 else 0
    delay = base_delay + jitter
    next_time = datetime.now(timezone.utc) + timedelta(seconds=delay)
    return next_time.strftime("%Y-%m-%d %H:%M:%S")


async def deliver_webhook(delivery: dict) -> tuple[int | None, str | None]:
    payload_str = delivery["payload"]
    timestamp = str(int(time.time()))
    signature = sign_payload(payload_str, delivery["secret"], timestamp)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-ID": delivery["id"],
        "X-Webhook-Timestamp": timestamp,
        "X-Webhook-Signature": f"sha256={signature}",
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.post(
                delivery["url"],
                content=payload_str,
                headers=headers,
            )
            return response.status_code, None
        except (httpx.TimeoutException, httpx.ConnectError) as e:
            return None, str(e)


async def process_deliveries() -> int:
    processed = 0
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        cursor = await db.execute(
            """
            SELECT id, event_id, subscription_id, url, payload, secret,
                   attempt_count, max_attempts
            FROM deliveries
            WHERE status IN ('pending', 'failed')
            AND next_attempt_at <= ?
            LIMIT 10
            """,
            (now,),
        )
        deliveries = await cursor.fetchall()

        for delivery in deliveries:
            delivery_dict = dict(delivery)
            status_code, error = await deliver_webhook(delivery_dict)
            new_attempt_count = delivery_dict["attempt_count"] + 1

            if status_code and 200 <= status_code < 300:
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'delivered',
                        attempt_count = ?,
                        last_status_code = ?,
                        delivered_at = datetime('now')
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, delivery_dict["id"]),
                )
            elif status_code and 400 <= status_code < 500 and status_code not in (408, 429):
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, f"HTTP {status_code}", delivery_dict["id"]),
                )
            elif new_attempt_count >= delivery_dict["max_attempts"]:
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, error or f"Max attempts exceeded", delivery_dict["id"]),
                )
            else:
                next_attempt = calculate_next_attempt(new_attempt_count)
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'failed',
                        attempt_count = ?,
                        next_attempt_at = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, next_attempt, status_code, error, delivery_dict["id"]),
                )

            processed += 1

        await db.commit()
    return processed


async def delivery_worker():
    import asyncio

    while True:
        try:
            await process_deliveries()
        except Exception as e:
            print(f"Delivery worker error: {e}")
        await asyncio.sleep(5)

Here is the full main.py:

import asyncio
import json
import secrets
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone

import aiosqlite
from fastapi import FastAPI, HTTPException

from database import DATABASE_PATH, init_db
from delivery import delivery_worker
from schemas import (
    DeliveryResponse,
    EventCreate,
    EventResponse,
    SubscriptionCreate,
    SubscriptionResponse,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    worker_task = asyncio.create_task(delivery_worker())
    yield
    worker_task.cancel()
    try:
        await worker_task
    except asyncio.CancelledError:
        pass


app = FastAPI(
    title="Webhook Delivery Service",
    description="A reliable webhook delivery system with retry logic, HMAC signing, and delivery tracking.",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/subscriptions", response_model=SubscriptionResponse, status_code=201)
async def create_subscription(subscription: SubscriptionCreate):
    sub_id = str(uuid.uuid4())
    secret = f"whsec_{secrets.token_hex(24)}"
    event_types_str = json.dumps(subscription.event_types)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        await db.execute(
            "INSERT INTO subscriptions (id, url, secret, event_types, created_at) VALUES (?, ?, ?, ?, ?)",
            (sub_id, str(subscription.url), secret, event_types_str, created_at),
        )
        await db.commit()

    return SubscriptionResponse(
        id=sub_id,
        url=str(subscription.url),
        secret=secret,
        event_types=subscription.event_types,
        is_active=True,
        created_at=created_at,
    )


@app.get("/subscriptions", response_model=list[SubscriptionResponse])
async def list_subscriptions():
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM subscriptions WHERE is_active = 1")
        rows = await cursor.fetchall()

    return [
        SubscriptionResponse(
            id=row["id"],
            url=row["url"],
            secret=row["secret"],
            event_types=json.loads(row["event_types"]),
            is_active=bool(row["is_active"]),
            created_at=row["created_at"],
        )
        for row in rows
    ]


@app.post("/events", response_model=EventResponse, status_code=202)
async def ingest_event(event: EventCreate):
    event_id = str(uuid.uuid4())
    payload_str = json.dumps(event.payload)
    created_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        await db.execute(
            "INSERT INTO events (id, event_type, payload, created_at) VALUES (?, ?, ?, ?)",
            (event_id, event.event_type, payload_str, created_at),
        )

        cursor = await db.execute(
            "SELECT * FROM subscriptions WHERE is_active = 1"
        )
        subscriptions = await cursor.fetchall()

        deliveries_created = 0
        for sub in subscriptions:
            event_types = json.loads(sub["event_types"])
            if event.event_type in event_types or "*" in event_types:
                delivery_id = str(uuid.uuid4())
                await db.execute(
                    """
                    INSERT INTO deliveries (id, event_id, subscription_id, url, payload, secret, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (delivery_id, event_id, sub["id"], sub["url"], payload_str, sub["secret"], created_at),
                )
                deliveries_created += 1

        await db.commit()

    return EventResponse(
        id=event_id,
        event_type=event.event_type,
        deliveries_created=deliveries_created,
    )


@app.get("/deliveries", response_model=list[DeliveryResponse])
async def list_deliveries(status: str | None = None, subscription_id: str | None = None):
    query = "SELECT * FROM deliveries WHERE 1=1"
    params: list[str] = []

    if status:
        query += " AND status = ?"
        params.append(status)
    if subscription_id:
        query += " AND subscription_id = ?"
        params.append(subscription_id)

    query += " ORDER BY created_at DESC LIMIT 50"

    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(query, params)
        rows = await cursor.fetchall()

    return [
        DeliveryResponse(
            id=row["id"],
            event_id=row["event_id"],
            subscription_id=row["subscription_id"],
            status=row["status"],
            attempt_count=row["attempt_count"],
            max_attempts=row["max_attempts"],
            next_attempt_at=row["next_attempt_at"],
            last_status_code=row["last_status_code"],
            last_error=row["last_error"],
            created_at=row["created_at"],
            delivered_at=row["delivered_at"],
        )
        for row in rows
    ]


@app.get("/deliveries/{delivery_id}", response_model=DeliveryResponse)
async def get_delivery(delivery_id: str):
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT * FROM deliveries WHERE id = ?", (delivery_id,))
        row = await cursor.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="Delivery not found")

    return DeliveryResponse(
        id=row["id"],
        event_id=row["event_id"],
        subscription_id=row["subscription_id"],
        status=row["status"],
        attempt_count=row["attempt_count"],
        max_attempts=row["max_attempts"],
        next_attempt_at=row["next_attempt_at"],
        last_status_code=row["last_status_code"],
        last_error=row["last_error"],
        created_at=row["created_at"],
        delivered_at=row["delivered_at"],
    )

Your delivery engine is alive. Pending deliveries get picked up, signed, and fired off to subscriber URLs. Next up, you will test the retry behaviour by pointing a subscription at a failing endpoint and watching the backoff schedule in action.

Add the Retry Engine and Test the Full Lifecycle

Your delivery worker can send webhooks and update their status. But how does it decide when to retry a failed delivery?

In this step, you will trace through the retry engine you built in the previous step. You will understand how exponential backoff with jitter prevents a thundering herd problem. Then you will run a full end-to-end test to watch deliveries succeed and fail in real time.

In this step, get ready to:

• Understand the exponential backoff schedule and jitter logic.
• Test a failing delivery and observe the retry state machine.
• Test a successful delivery and confirm the full lifecycle works end-to-end.

Understand the backoff schedule

When a webhook delivery fails, you do not want to immediately hammer the endpoint again. You also do not want every failed delivery retrying at the exact same moment.

• In VS Code, open delivery.py and find the BACKOFF_SCHEDULE constant near the top of the file:

BACKOFF_SCHEDULE = [0, 30, 300, 1800, 7200]

What do these numbers mean?

Each number is the base delay in seconds before the next retry attempt:

• Attempt 1: 0 seconds (immediate retry).
• Attempt 2: 30 seconds.
• Attempt 3: 300 seconds (5 minutes).
• Attempt 4: 1800 seconds (30 minutes).
• Attempt 5: 7200 seconds (2 hours).

The gaps grow exponentially. This gives a failing endpoint time to recover without overwhelming it with constant requests.

• Now find the calculate_next_attempt function below it:

def calculate_next_attempt(attempt_count: int) -> str:
    if attempt_count >= len(BACKOFF_SCHEDULE):
        base_delay = BACKOFF_SCHEDULE[-1]
    else:
        base_delay = BACKOFF_SCHEDULE[attempt_count]

    jitter = random.uniform(0, base_delay * 0.25) if base_delay > 0 else 0
    delay = base_delay + jitter
    next_time = datetime.now(timezone.utc) + timedelta(seconds=delay)
    return next_time.strftime("%Y-%m-%d %H:%M:%S")

What is jitter and why does it matter?

Jitter adds a random amount (0 to 25% of the base delay) on top of the scheduled wait. If 100 deliveries all fail at the same time, jitter spreads their retries across a window instead of firing them all at the same second.

Without jitter, a recovering server would get slammed by a synchronized wave of retries. This is called the thundering herd problem.

Review the delivery state machine

The retry logic in process_deliveries() moves each delivery through a state machine based on the HTTP response.

• Scroll down to the process_deliveries function and review the four outcome branches after each delivery attempt:

How does the state machine work?

• 2xx response: status becomes delivered. The delivery is complete.
• 4xx response (except 408/429): status becomes dead_letter immediately. A 4xx means the client is wrong, so retrying will not help.
• Max attempts exceeded: status becomes dead_letter. The service gave up after 5 tries.
• 5xx, timeout, or network error: status becomes failed and next_attempt_at is set using the backoff schedule. The worker will pick it up again later.

Why are 408 and 429 excluded from the dead_letter rule?

HTTP 408 (Request Timeout) and 429 (Too Many Requests) are transient. They indicate the server is alive but temporarily unable to handle the request. Retrying later is the correct response.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on to testing.

ⓧ I'd like to double check the full code

Here is the complete delivery.py for reference:

import hashlib
import hmac
import random
import time
from datetime import datetime, timedelta, timezone

import aiosqlite
import httpx

from database import DATABASE_PATH

BACKOFF_SCHEDULE = [0, 30, 300, 1800, 7200]


def sign_payload(payload: str, secret: str, timestamp: str) -> str:
    message = f"{timestamp}.{payload}"
    signature = hmac.new(
        secret.encode(),
        message.encode(),
        hashlib.sha256,
    ).hexdigest()
    return signature


def calculate_next_attempt(attempt_count: int) -> str:
    if attempt_count >= len(BACKOFF_SCHEDULE):
        base_delay = BACKOFF_SCHEDULE[-1]
    else:
        base_delay = BACKOFF_SCHEDULE[attempt_count]

    jitter = random.uniform(0, base_delay * 0.25) if base_delay > 0 else 0
    delay = base_delay + jitter
    next_time = datetime.now(timezone.utc) + timedelta(seconds=delay)
    return next_time.strftime("%Y-%m-%d %H:%M:%S")


async def deliver_webhook(delivery: dict) -> tuple[int | None, str | None]:
    payload_str = delivery["payload"]
    timestamp = str(int(time.time()))
    signature = sign_payload(payload_str, delivery["secret"], timestamp)

    headers = {
        "Content-Type": "application/json",
        "X-Webhook-ID": delivery["id"],
        "X-Webhook-Timestamp": timestamp,
        "X-Webhook-Signature": f"sha256={signature}",
    }

    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.post(
                delivery["url"],
                content=payload_str,
                headers=headers,
            )
            return response.status_code, None
        except (httpx.TimeoutException, httpx.ConnectError) as e:
            return None, str(e)


async def process_deliveries() -> int:
    processed = 0
    async with aiosqlite.connect(DATABASE_PATH) as db:
        db.row_factory = aiosqlite.Row
        now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        cursor = await db.execute(
            """
            SELECT id, event_id, subscription_id, url, payload, secret,
                   attempt_count, max_attempts
            FROM deliveries
            WHERE status IN ('pending', 'failed')
            AND next_attempt_at <= ?
            LIMIT 10
            """,
            (now,),
        )
        deliveries = await cursor.fetchall()

        for delivery in deliveries:
            delivery_dict = dict(delivery)
            status_code, error = await deliver_webhook(delivery_dict)
            new_attempt_count = delivery_dict["attempt_count"] + 1

            if status_code and 200 <= status_code < 300:
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'delivered',
                        attempt_count = ?,
                        last_status_code = ?,
                        delivered_at = datetime('now')
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, delivery_dict["id"]),
                )
            elif status_code and 400 <= status_code < 500 and status_code not in (408, 429):
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, f"HTTP {status_code}", delivery_dict["id"]),
                )
            elif new_attempt_count >= delivery_dict["max_attempts"]:
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'dead_letter',
                        attempt_count = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, status_code, error or f"Max attempts exceeded", delivery_dict["id"]),
                )
            else:
                next_attempt = calculate_next_attempt(new_attempt_count)
                await db.execute(
                    """
                    UPDATE deliveries
                    SET status = 'failed',
                        attempt_count = ?,
                        next_attempt_at = ?,
                        last_status_code = ?,
                        last_error = ?
                    WHERE id = ?
                    """,
                    (new_attempt_count, next_attempt, status_code, error, delivery_dict["id"]),
                )

            processed += 1

        await db.commit()
    return processed


async def delivery_worker():
    import asyncio

    while True:
        try:
            await process_deliveries()
        except Exception as e:
            print(f"Delivery worker error: {e}")
        await asyncio.sleep(5)

Test a failing delivery

Time to watch the retry engine in action. You will create a subscription pointing to a URL that does not exist, then observe the delivery failing and retrying.

• Switch back to Swagger UI at http://127.0.0.1:8000/docs in your browser.
• Expand the POST /subscriptions endpoint and click Try it out.
• Replace the request body with this JSON:

{
  "url": "http://localhost:9999/webhook",
  "event_types": ["order.created"]
}

• Click Execute.

Why localhost:9999?

Nothing is running on port 9999. Every delivery attempt to this URL will fail with a connection error, triggering the retry logic.

• Now expand POST /events and click Try it out.
• Replace the request body with:

{
  "event_type": "order.created",
  "payload": {"order_id": "12345", "amount": 99.99}
}

• Click Execute. You should see a 202 response with deliveries_created: 1.
• Wait about 10 seconds for the background worker to pick up the delivery and attempt it.
• Expand GET /deliveries and click Try it out.
• In the status query parameter field, enter failed.
• Click Execute.

You should see a delivery record with status: "failed" and attempt_count set to 1 or higher.

What should I see?

The delivery record shows the retry engine at work. attempt_count tracks how many times the worker tried. next_attempt_at shows when the next retry is scheduled. last_error describes why the delivery failed.

After the first attempt, the backoff schedule sets the next retry 30 seconds out. If you wait long enough, attempt_count will increment with each retry until it reaches 5 and becomes dead_letter.

Not seeing any deliveries?

• Make sure your dev server is still running in the terminal. If you stopped it, restart with fastapi dev.
• Check that the event_type in your event matches one of the event_types in your subscription. Both should be order.created.
• Wait at least 10 seconds. The worker polls every 5 seconds, and there is a short delay before the first attempt.

Help me debug why my deliveries aren't being processed.

Test a successful delivery and review Swagger docs

Now you will create a subscription pointing to a real endpoint that accepts POST requests. This will confirm the happy path of the state machine.

• Expand POST /subscriptions again and click Try it out.
• Replace the request body with:

{
  "url": "https://httpbin.org/post",
  "event_types": ["user.signup"]
}

• Click Execute.

What is httpbin.org?

httpbin.org is a free public service that accepts HTTP requests and echoes them back. The /post endpoint accepts any POST request and returns a 200 response. It is perfect for testing webhook delivery without running your own server.

• Expand POST /events and click Try it out.
• Replace the request body with:

{
  "event_type": "user.signup",
  "payload": {"user_id": "abc123", "email": "test@example.com"}
}

• Click Execute. You should see a 202 response with deliveries_created: 1.
• Wait about 10 seconds for the background worker to deliver it.
• Expand GET /deliveries and click Try it out.
• In the status query parameter field, enter delivered.
• Click Execute.

You should see a delivery record with status: "delivered", attempt_count: 1, and a last_status_code of 200.

What does the full state machine look like?

You have now confirmed both paths:

• Happy path: pending -> delivered (the httpbin.org test).
• Failure path: pending -> failed -> ... -> dead_letter (the localhost:9999 test).

This is the same delivery state machine that production webhook services like Stripe and GitHub use.

Delivery still showing as pending?

• If the delivery to httpbin.org is stuck on pending, your machine might not have internet access. Try opening https://httpbin.org/post in your browser to confirm it loads.
• Check your terminal for any error messages from the worker. The delivery_worker function prints exceptions to the console.

Help me figure out why my webhook delivery is stuck on pending.

• Finally, scroll to the top of the Swagger UI page at http://127.0.0.1:8000/docs.

You should see all your endpoints documented: POST /subscriptions, GET /subscriptions, POST /events, GET /deliveries, and GET /deliveries/{delivery_id}. FastAPI auto-generated this interactive documentation from your Pydantic models and route decorators.

Show this in an interview

Swagger UI at /docs is a live, interactive API reference that anyone can explore without reading your source code. It shows request schemas, response models, and lets you fire real requests. This is the kind of polish that stands out in a portfolio.

Your webhook delivery service is fully operational. You have a working state machine, retry engine with exponential backoff, HMAC signing, and auto-generated API documentation. Next up, the Secret Mission challenges you to add a manual replay endpoint for failed deliveries.

Secret mission

Add a Manual Replay Endpoint

Every production webhook service (Stripe, GitHub, Hookdeck) gives operators a way to manually re-trigger failed deliveries. Your service has a retry engine, but once a delivery hits dead_letter, it stays there forever. Can you add a replay endpoint that resets a failed delivery and gives it another chance?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• webhook-service project folder (contains all Python files, virtual environment, and SQLite database).
• FastAPI development server (running process).

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the project further.

• Your FastAPI dev server continues running and serving requests at http://127.0.0.1:8000.
• Your webhooks.db file retains all subscriptions, events, and delivery records you created during the project.
• The virtual environment stays activated in your terminal session.

Pause - I'll come back to this later

Shut down the running server process to free up your terminal. Your files stay on disk and you can restart anytime.

• In the terminal where FastAPI is running, press Ctrl+C to stop the development server.
• Deactivate the virtual environment by running:

deactivate

When you're ready to resume, reactivate the virtual environment and start the server again.

• On macOS/Linux, run:

cd ~/Desktop/webhook-service
source venv/bin/activate
fastapi dev

• On Windows, run:

cd ~\Desktop\webhook-service
venv\Scripts\activate
fastapi dev

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your code, database, and virtual environment permanently.

• In the terminal where FastAPI is running, press Ctrl+C to stop the development server.
• Deactivate the virtual environment by running:

deactivate

• On macOS/Linux, delete the entire project folder by running:

cd ~/Desktop
rm -rf webhook-service

• On Windows, delete the entire project folder by running:

cd ~\Desktop
rmdir /s /q webhook-service

This removes all four Python modules, the webhooks.db database, the venv virtual environment, and all installed packages.

Nice Work!

Nice Work!

You just built a production-grade webhook delivery service from scratch. This is the same architecture that powers systems like Stripe's event notifications and GitHub's webhook infrastructure.

You've learned how to:

• Build an event-driven delivery service that accepts subscriptions, ingests events, and fans out HMAC-signed HTTP callbacks to matching subscribers.
• Implement a retry engine with exponential backoff and jitter that tracks delivery state through a full state machine (pending, delivered, failed, dead_letter).
• Expose a queryable REST API with auto-generated interactive Swagger documentation ready to demo in an interview.
• Secret Mission: Added a manual replay endpoint that re-triggers failed or dead-letter deliveries on demand.

Ready to quiz yourself?