| Tutorial
5 min read ellipse November 29, 2022

Sending Webhooks With FastAPI

Webhooks are messages ( or payload ) sent from an application after the execution of an operation. They are also used to communicate between a chain of services; for example, a payment provider emits webhook events to an e-commerce application’s endpoint after a payment operation. Convoy facilitates publishing webhook events from your application to your clients by serving as a reliable egress.

In this article, we will build a Todo API in FastAPI and use Convoy to publish webhook events for each operation on our Todo items; create, update & delete.


To follow along you would need the following

  1. A Convoy Cloud account.
  2. An Outgoing Project ID & API Key.

For the sake of brevity, we created an additional resource to help with creating user endpoints, usually, users will supply this information to you via your dashboard. [2] We have also left out other aspects of the code not necessary for this guide.

API Spec

Our API looks like this:

  • Endpoint
    GET    /endpoint
    GET    /endpoint/:id
    POST   /endpoint
    PUT    /endpoint/:id
    DELETE /endpoint/:id
  • Todo
    GET    /todo
    GET    /todo/:id
    POST   /todo
    PUT    /todo/:id
    DELETE /todo/:id

Every time we create, update and delete a todo item, we would generate the following events — todo.created, todo.updated, and todo.deleted respectively.

Let’s Build Our API

  1. Project Setup

    $ mkdir fastapi-todo-convoy && cd fastapi-todo-convoy
    $ python3 -m venv venv
    $ touch .env {config,api,events}.py

    Activate virtual environment

    $ source venv/bin/activate
    $ pip install fastapi uvicorn pydantic[dotenv]
    $ pip freeze > requirements.txt

    Install Convoy:

    $ pip install git+ssh://[email protected]/frain-dev/convoy-python

    Configure your environment variables in a .env file:

  2. Define events & configuration

    events = {
        "ping": {
            "event": "ping",
            "description": "Webhook test from application."
        "created": {
            "event": "todo.created",
            "description": "Todo created successfully"
        "retrieved": {
            "event": "todo.retrieved",
            "description": "Todo retrieved successfully"
        "updated": {
            "event": "todo.updated",
            "description": "Todo updated successfully"
        "deleted": {
            "event": "todo.deleted",
            "description": "Todo deleted successfully"
        "failed": {
            "event": "todo.failure",
            "description": "Todo not found."


    from pydantic import BaseSettings
    from typing import Optional
    class Settings(BaseSettings):
        CONVOY_API_KEY: Optional[str] = None
        CONVOY_PROJECT_ID: Optional[str] = None
        class Config:
            env_file = ".env"
  3. Endpoints API

    Start by adding the imports and a Convoy instance:

    from convoy import Convoy
    from fastapi import FastAPI, Depends
    from events import events
    from config import Settings
    app = FastAPI()
    config = Settings()
    todos = []
    convoy = Convoy({"api_key": config.CONVOY_API_KEY, "project_id": config.CONVOY_PROJECT_ID})

    Add the code for the endpoints API:

    api.py | Endpoints API
    @app.post("/endpoint", tags=["endpoint"])
    async def create_endpoint(endpoint_body: dict):
        (endpoint, result) = convoy.endpoint.create({}, endpoint_body)
        return {"Endpoint ID": endpoint["data"]["uid"]}
    @app.get("/endpoint", tags=["endpoint"])
    async def get_endpoint():
        (endpoint, result) = convoy.endpoint.all({})
        return endpoint
    @app.get("/endpoint/{endpoint_id}", tags=["endpoint"])
    async def get_endpoint(endpoint_id: str):
        (endpoint, result) = convoy.endpoint.find(endpoint_id, {})
        return endpoint
    @app.put("/endpoint/{endpoint_id}", tags=["endpoint"])
    async def update_endpoint(endpoint_id: str, endpoint_body: dict):
        (endpoint, result) = convoy.endpoint.update(endpoint_id, {}, endpoint_body)
        return endpoint
    @app.delete("/endpoint/{endpoint_id}", tags=["endpoint"])
    async def delete_endpoint(endpoint_id: str):
        (endpoint, result) = convoy.endpoint.delete(endpoint_id, {}, "")
        return endpoint

    Add the function that publishes webhooks:

    def send_webhook_event(event_type: str, endpoint: str):
        event = {
            "endpoint_id": endpoint,
            "event_type": event_type,
            "data": events[event_type]
        (res, err) = convoy.event.create({}, event)
        return res
  4. Todos API

    api.py | Todos API
    @app.get("/todo", tags=["todos"])
    async def get_todos(endpoint: str) -> dict:
        send_webhook_event("retrieved", endpoint)
        return { "data": todos }
    @app.get("/todo/{id}", tags=["todos"])
    async def get_todo(id: int, endpoint: str) -> dict:
        for todo in todos:
            if todo["id"] == id:
                send_webhook_event("retrieved", endpoint)
                return todo
        send_webhook_event("failed", endpoint)
        return { "error": "Todo not found" }
    @app.post("/todo", tags=["todos"])
    async def add_todo(todo: dict, endpoint: str) -> dict:
        send_webhook_event("created", endpoint)
        return {
            "data": { "Todo added." }
    @app.put("/todo/{id}", tags=["todos"])
    async def update_todo(id: int, body: dict, endpoint: str) -> dict:
        for todo in todos:
            if int(todo["id"]) == id:
                todo["item"] = body["item"]
                return {
                    "data": f"Todo with id {id} has been updated."
        send_webhook_event("failed", endpoint)
        return {
            "data": f"Todo with id {id} not found."
    @app.delete("/todo/{id}", tags=["todos"])
    async def delete_todo(id: int, endpoint: str) -> dict:
        for todo in todos:
            if int(todo["id"]) == id:
                send_webhook_event("deleted", endpoint)
                return {
                    "data": f"Todo with id {id} has been removed."
        send_webhook_event("failed", endpoint)
        return {
            "data": f"Todo with id {id} not found."

Publish Webhook Events

It’s time to publish your first webhook!

  1. To begin, we start our FastAPI app

    $ uvicorn api:app --host --port 8080 --reload
  2. Second, we create an endpoint with the cURL command below:

    curl -X 'POST' \
    '' \
    -H 'accept: application/json' \
    -H 'Content-Type: application/json' \
    -d ' {
            "name": "Todo endpoint",
            "description": "Endpoint for todo API.",
            "http_timeout": "10s",
            "url": "https://webhook.site/40984a4e-7b36-41fb-a234-9c2006bac8b5"

    The API returns a successful response:

    {"Endpoint ID":"da6c42b5-2a51-478c-ad5e-53097c0f61cb"}
  3. Finally, we create a Todo item, that in turn generates the webhook item. Let's use the cURL command below:

    curl -X 'POST' \
    '' \
    -H 'accept: application/json' \
    -H 'Content-Type: application/json' \
    -d '{
    "id": 1,
    "item": "Setup a convoy webhook project and publish my first webhook"

    The API returns a successful response:

    {"data":["Todo added."]}

Let’s see our event deliveries dashboard.

Deliveries dashboard

Let’s also see our webhooks endpoint

Webhooks.site endpoint page


  1. In production environments, Endpoints should be scoped to each user/business/customer or whatever makes sense in your case because at the point of generating webhooks
  2. Users can supply their endpoints through multiple means — your dashboard, the portal link
  3. In this article, we publish webhooks in our controllers, in an ideal production environment, you should publish them from your workers.


Convoy provides the ability the send webhooks to one endpoint as well as multiple endpoints. In this article, you learned how to create send webhooks from a FastAPI API. We hope you enjoyed reading this, and that you get to try it out and give us some feedback on the slack community!

More Posts

post image

Introducing Convoy App Portal

One of the essential characteristics of a reliable webhooks infrastructure is visibility. We're enabling you to extend that visibility to your customers. Unlike the Convoy Dashboard, App Portal is for just one application; giving your customers a good and extensive..

post image

Introducing Convoy

After weeks of work, I'm pleased to announce our new tool — built to send webhooks efficiently. While building out our initial API on third-party monitoring, every user we spoke to wanted asynchronous events — Do you provide webhooks to notify us about downtime?