# Send Webhooks with FastAPI

This guide will show you how to send webhooks with FastAPI using Hookdeck's [Outpost](https://hookdeck.com/outpost) — an open-source event dispatcher. Outpost gives you a world-class webhook and event destination infrastructure in minutes for 1/10th the cost of alternatives.

Outpost supports [Event Destinations](https://eventdestinations.org/), allowing you to deliver events to webhook endpoints, as well as to queues, and popular brokers or event gateways. These include AWS SQS, RabbitMQ, GCP Pub/Sub, Amazon EventBridge, Kafka, and more—all from a single platform.

The team behind it has drawn on their experience running [Hookdeck Event Gateway](https://hookdeck.com/) and delivering over 100 billion events to thousands of customers.

## Why use Outpost?

Sending webhooks can be trickier than you think. From managing fan‑out load, retries and backoff, to endpoint health management, and security/operational overhead. Outpost takes care of all of that so you only need to think about is how to send a webhook.

## How to Send Webhooks with FastAPI + Outpost

Start by deploying an [Outpost instance](https://hookdeck.com/docs/outpost/guides/deployment) locally, via Docker, or deploy to production. For guidance, check out the [Outpost quickstart](https://hookdeck.com/docs/outpost/quickstarts) or the [GitHub repository](https://github.com/hookdeck/outpost).

## FastAPI Support in Outpost

You can interact with Outpost using the [Python SDK](https://github.com/hookdeck/outpost/tree/main/sdks/outpost-python) or the REST API. Each SDK provides a convenient way to interact with the Outpost API, including publishing events, managing topics, and configuring destinations.

```bash
pip install fastapi outpost_sdk uvicorn python-dotenv

```

## Set Your Admin API Key

Before you can start making calls to the Outpost API you need to set an API key. The API uses bearer token authentication with a token of your choice configured through the `API_KEY` environment variable.

Create a `.env` file in your FastAPI project root:

```bash
ADMIN_API_KEY=<YOUR_BEARER_TOKEN_HERE>
OUTPOST_URL=http://localhost:3333

```

## Send Webhooks (Publish Events)

In order to send a webhook, you have to instantiate Outpost, create a tenant, and create a destination for that tenant. Then you're ready to publish events to it.

### 1. Create an Outpost instance

```python
with Outpost() as outpost

```

You can set the security parameters through the security optional parameter when initializing the SDK client instance. The selected scheme will be used by default to authenticate with the API for all operations that support it. For example:

```python
with Outpost(
    security=models.Security(
        admin_api_key="<YOUR_BEARER_TOKEN_HERE>",
    ),
) as outpost

```

### 2. Create a new tenant.

A tenant in Outpost represents a user/team/organization in your product. Use the `upsert()` method to idempotently create or update a tenant: `outpost.tenants.upsert()`

### 3. Create a destination for that tenant.

The `create()` method is used to specify a new destination for the tenant. A destination is a specific instance of a destination type. For example, a webhook destination with a particular URL.

The request body structure depends on the [destination type](https://hookdeck.com/docs/outpost/destinations/webhook). Here we're creating a destination for a webhook, but it could be SQS, RabbitMQ, etc.

```python
  res = outpost.destinations.create(destination_create={
        "tenant_id": tenant_id,
        "id": user-provided-id,
        "type": models.DestinationWebhook,
        "topics": models.TopicsEnum.WILDCARD_,
        "config": {
            "url": "https://example.com/webhook-receiver",
        },
    })

```

Note that we specify topics. A topic is a way to categorize events and is a common concept found in Pub/Sub messaging.

### 4. Publish an event for the created tenant.

You can now use the `publish()` method to send an event to the specified topic, routed to a specific destination.

```python
  res = outpost.publish(data={
        "user_id": "userid",
        "status": "active",
    }, id="evt_custom_123", tenant_id="<TENANT_ID>", destination_id="<DESTINATION_ID>", topic="topic.name", metadata={
        "source": "crm",
    })

```

`id`: Is optional but recommended. If left empty, it will be generated by hashing the topic, data, and timestamp.

`tenant_id`: The tenant ID to publish the event must match an existing tenant, otherwise it will be ignored.

`destination_id`: Optional. Used to force delivery to a specific destination regardless of the topic.

`topic`: Optional. Assumed to match ANY topic if left empty. If set, it must match one of the configured topics.

`metadata`: Arbitrary key-value mapping for event metadata.

`data`: The `data` JSON object will be sent as the webhook body. Now, every time you publish an event to this tenant, Outpost will send an HTTP POST request with the data payload to the destination URL.

Other fields:

`eligible_for_retry`: Optional, defaults to `true`. Controls whether an event should be automatically retried.

`retries`: Configuration to override the default retry behavior of the client.

## Complete Example

Combining the examples above, you should end up with a FastAPI application that looks like the following:

First, create an Outpost service (`services/outpost_client.py`):

```python
import os
from dotenv import load_dotenv
from outpost_sdk import Outpost, models

load_dotenv()

class OutpostService:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(OutpostService, cls).__new__(cls)
            cls._instance._initialize()
        return cls._instance

    def _initialize(self):
        admin_api_key = os.getenv("ADMIN_API_KEY")
        server_url = os.getenv("OUTPOST_URL", "http://localhost:3333")

        if not admin_api_key:
            raise ValueError("Please set the ADMIN_API_KEY environment variable.")

        self.client = Outpost(
            security=models.Security(admin_api_key=admin_api_key),
            server_url=f"{server_url}/api/v1",
        )

    def get_client(self):
        return self.client

```

Then, create request/response models (`models/webhook_models.py`):

```python
from typing import Optional, Dict, Any
from pydantic import BaseModel

class PublishEventRequest(BaseModel):
    tenant_id: str
    topic: str
    data: Optional[Dict[str, Any]] = None

class WebhookReceiverRequest(BaseModel):
    id: str
    data: Dict[str, Any]
    metadata: Optional[Dict[str, str]] = None
    timestamp: str

class WebhookSetupResponse(BaseModel):
    success: bool
    message: str
    data: Dict[str, Optional[str]]

class PublishEventResponse(BaseModel):
    success: bool
    message: str
    event_id: Optional[str] = None

class PortalUrlResponse(BaseModel):
    success: bool
    portal_url: Optional[str] = None

```

Then, create API routes (`routes/webhooks.py`):

```python
import uuid
import logging
from fastapi import APIRouter, HTTPException
from services.outpost_client import OutpostService
from models.webhook_models import (
    PublishEventRequest,
    PublishEventResponse,
    WebhookReceiverRequest,
    WebhookSetupResponse,
    PortalUrlResponse,
)

router = APIRouter(prefix="/webhooks", tags=["webhooks"])
outpost_service = OutpostService()
outpost = outpost_service.get_client()

logger = logging.getLogger(__name__)

@router.post("/setup", response_model=WebhookSetupResponse)
async def setup_webhook():
    """
    Create a new tenant and destination for webhook delivery.
    """
    try:
        tenant_id = f"tenant_{str(uuid.uuid4())}"
        topic = "user.created"
        destination_name = f"My Test Destination {str(uuid.uuid4())}"

        # 1. Create a tenant
        logger.info(f"Creating tenant: {tenant_id}")
        tenant = outpost.tenants.upsert(tenant_id=tenant_id)
        logger.info("Tenant created successfully")

        # 2. Create a destination for the tenant
        logger.info(f"Creating destination: {destination_name} for tenant {tenant_id}...")
        destination = outpost.destinations.create(
            tenant_id=tenant_id,
            destination_create=models.DestinationCreateWebhook(
                type=models.DestinationCreateWebhookType.WEBHOOK,
                config=models.WebhookConfig(
                    url="https://example.com/webhook-receiver"
                ),
                topics=[topic],
            ),
        )
        logger.info("Destination created successfully")

        return WebhookSetupResponse(
            success=True,
            message="Tenant and destination created successfully",
            data={
                "tenant_id": tenant.id if hasattr(tenant, 'id') else tenant_id,
                "destination_id": destination.id if hasattr(destination, 'id') else None,
            },
        )

    except Exception as e:
        logger.error(f"An error occurred: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/publish", response_model=PublishEventResponse)
async def publish_event(payload: PublishEventRequest):
    """
    Publish an event to a tenant's topic.
    """
    try:
        if not payload.tenant_id or not payload.topic:
            raise HTTPException(
                status_code=400,
                detail="tenant_id and topic are required",
            )

        event_payload = payload.data or {
            "user_id": "user_456",
            "order_id": "order_xyz",
            "timestamp": str(uuid.uuid4()),
        }

        logger.info(
            f"Publishing event to topic {payload.topic} for tenant {payload.tenant_id}..."
        )
        event = outpost.publish(
            data=event_payload,
            tenant_id=payload.tenant_id,
            topic=payload.topic,
            eligible_for_retry=True,
        )

        logger.info("Event published successfully")
        return PublishEventResponse(
            success=True,
            message="Event published successfully",
            event_id=event.id if hasattr(event, 'id') else None,
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/receiver")
async def webhook_receiver(payload: WebhookReceiverRequest):
    """
    Receive and process incoming webhooks from Outpost.
    """
    try:
        logger.info(f"Webhook received: {payload}")

        return {
            "success": True,
            "message": "Webhook received",
        }

    except Exception as e:
        logger.error(f"An error occurred: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/portal/{tenant_id}", response_model=PortalUrlResponse)
async def get_portal_url(tenant_id: str):
    """
    Get the Tenant User Portal URL for a specific tenant.
    """
    try:
        if not tenant_id:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required",
            )

        portal_url = outpost.tenants.get_portal_url(tenant_id=tenant_id)

        return PortalUrlResponse(
            success=True,
            portal_url=portal_url.url if hasattr(portal_url, 'url') else None,
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"An error occurred: {e}")
        raise HTTPException(status_code=500, detail=str(e))

```

Finally, create your main FastAPI application (`main.py`):

```python
import logging
import os
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from dotenv import load_dotenv
from routes.webhooks import router as webhooks_router

load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create FastAPI app
app = FastAPI(
    title="Outpost Webhooks API",
    description="A FastAPI application for sending webhooks with Outpost",
    version="1.0.0",
)

# Include routers
app.include_router(webhooks_router)

@app.get("/health")
async def health_check():
    """
    Health check endpoint.
    """
    return JSONResponse(
        status_code=200,
        content={"status": "ok"}
    )

@app.get("/")
async def root():
    """
    Root endpoint.
    """
    return {
        "message": "Welcome to Outpost Webhooks API",
        "docs": "/docs",
        "health": "/health",
    }

if __name__ == "__main__":
    import uvicorn

    port = int(os.getenv("PORT", 8000))
    uvicorn.run(app, host="0.0.0.0", port=port)

```

Create a `requirements.txt` file:

```text
fastapi==0.104.1
uvicorn[standard]==0.24.0
outpost-sdk==1.2.0
python-dotenv==1.0.0
pydantic==2.4.0

```

Install dependencies and run:

```bash
pip install -r requirements.txt
python main.py

```

Or using uvicorn directly:

```bash
uvicorn main:app --reload

```

### Expose Tenant User Portal

Your application can offer the user the option to configure their destination, view and retry their events using a themeable [Tenant User Portal](https://hookdeck.com/docs/outpost/features/tenant-user-portal). Note that if the portal is used, then the Outpost API needs to be exposed to the public internet (or proxy requests to the Outpost API).

You can use `get_portal_url()` to return a redirect URL containing a JWT to authenticate the user with the portal. Then redirect your users to this URL or use it in an iframe in your FastAPI templates.

#### Build your own UI

While Outpost offers a user portal, you may want to build your own UI for users to manage their destinations and view their events. The portal is built using the Outpost API with JWT authentication. You can leverage the same API to [build your own UI](https://hookdeck.com/docs/outpost/guides/building-your-own-ui).

## Help with debugging and testing webhooks

To help you and your users work with webhooks, we provide some additional tools. Like [Hookdeck Console](https://console.hookdeck.com/), a web-based inspector to test and debug incoming webhooks. As well as the [Hookdeck CLI](https://hookdeck.com/docs/cli) to forward webhook events to a local web server.

## More information

That's everything you need to start sending webhooks with Outpost using FastAPI. For more information, please check out the [Outpost documentation](https://hookdeck.com/docs/outpost/overview).

If you have any questions, we'd love to help. Please join the [Hookdeck community on Slack](https://join.slack.com/t/hookdeckdevelopers/shared_invite/zt-yw7hlyzp-EQuO3QvdiBlH9Tz2KZg5MQ).

### Related resources

* [Outpost Documentation](https://hookdeck.com/docs/outpost/overview)
* [GitHub Repository](https://github.com/hookdeck/outpost)
* [Python SDK](https://github.com/hookdeck/outpost/tree/main/sdks/outpost-python)
* [Event Destinations Initiative](https://eventdestinations.org/)
* [FastAPI Documentation](https://fastapi.tiangolo.com/)
* [Pydantic Documentation](https://docs.pydantic.dev/)