The WebhooksServer() class lets you create an instance of a Gradio app that can receive Huggingface webhooks. These webhooks can be registered using the add_webhook() decorator. Webhook endpoints are added to the app as a POST endpoint to the FastAPI router. Once all the webhooks are registered, the launch method has to be called to start the app.

It is recommended to accept WebhookPayload as the first argument of the webhook function. It is a Pydantic model that contains all the information about the webhook event. The data will be parsed automatically for you.

Check out the webhooks guide for a step-by-step tutorial on how to setup your WebhooksServer and deploy it on a Space.

import gradio as gr
from huggingface_hub import WebhooksServer, WebhookPayload

with gr.Blocks() as ui:
    ...

app = WebhooksServer(ui=ui, webhook_secret="my_secret_key")

@app.add_webhook("/say_hello")
async def hello(payload: WebhookPayload):
    return {"message": "hello"}

app.launch()

Decorator to start a WebhooksServer() and register the decorated function as a webhook endpoint.

This is a helper to get started quickly. If you need more flexibility (custom landing page or webhook secret), you can use WebhooksServer() directly. You can register multiple webhook endpoints (to the same server) by using this decorator multiple times.

Check out the webhooks guide for a step-by-step tutorial on how to setup your server and deploy it on a Space.

Examples: The default usage is to register a function as a webhook endpoint. The function name will be used as the path. The server will be started automatically at exit (i.e. at the end of the script).

from huggingface_hub import webhook_endpoint, WebhookPayload

@webhook_endpoint
async def trigger_training(payload: WebhookPayload):
    if payload.repo.type == "dataset" and payload.event.action == "update":
        # Trigger a training job if a dataset is updated
        ...

# Server is automatically started at the end of the script.

Advanced usage: register a function as a webhook endpoint and start the server manually. This is useful if you are running it in a notebook.

from huggingface_hub import webhook_endpoint, WebhookPayload

@webhook_endpoint
async def trigger_training(payload: WebhookPayload):
    if payload.repo.type == "dataset" and payload.event.action == "update":
        # Trigger a training job if a dataset is updated
        ...

# Start the server manually
trigger_training.launch()