Webhooks

Webhooks are a foundation for MLOps-related features. They allow you to listen for new changes on specific repos or to all repos belonging to particular users/organizations you’re interested in following. This guide will first explain how to manage webhooks programmatically. Then we’ll see how to leverage huggingface_hub to create a server listening to webhooks and deploy it to a Space.

This guide assumes you are familiar with the concept of webhooks on the Huggingface Hub. To learn more about webhooks themselves, you should read this guide first.

管理 Webhooks

huggingface_hub 允许你以编程方式管理你的 webhooks。你可以列出你现有的 webhooks，创建新的，以及更新、启用、禁用或删除它们。本节将指导你使用 Hugging Face Hub 的 API 函数完成这些步骤。

创建 Webhook

要创建一个新的 webhook，请使用 create_webhook() 并指定应将有效负载发送到的 URL，应监视哪些事件，并可选择设置域和密钥以确保安全。

from huggingface_hub import create_webhook

# Example: Creating a webhook
webhook = create_webhook(
    url="https://webhook.site/your-custom-url",
    watched=[{"type": "user", "name": "your-username"}, {"type": "org", "name": "your-org-name"}],
    domains=["repo", "discussion"],
    secret="your-secret"
)

列出 Webhooks

要查看你已配置的所有 webhooks，你可以使用 list_webhooks() 列出它们。这对于查看它们的 ID、URL 和状态很有用。

from huggingface_hub import list_webhooks

# Example: Listing all webhooks
webhooks = list_webhooks()
for webhook in webhooks:
    print(webhook)

更新 Webhook

如果你需要更改现有 webhook 的配置，例如 URL 或其监视的事件，你可以使用 update_webhook() 更新它。

from huggingface_hub import update_webhook

# Example: Updating a webhook
updated_webhook = update_webhook(
    webhook_id="your-webhook-id",
    url="https://new.webhook.site/url",
    watched=[{"type": "user", "name": "new-username"}],
    domains=["repo"]
)

启用和禁用 Webhooks

你可能想要暂时禁用 webhook 而不删除它。这可以使用 disable_webhook() 完成，并且 webhook 稍后可以使用 enable_webhook() 重新启用。

from huggingface_hub import enable_webhook, disable_webhook

# Example: Enabling a webhook
enabled_webhook = enable_webhook("your-webhook-id")
print("Enabled:", enabled_webhook)

# Example: Disabling a webhook
disabled_webhook = disable_webhook("your-webhook-id")
print("Disabled:", disabled_webhook)

删除 Webhook

当不再需要 webhook 时，可以使用 delete_webhook() 永久删除它。

from huggingface_hub import delete_webhook

# Example: Deleting a webhook
delete_webhook("your-webhook-id")

Webhooks 服务器

在本指南部分中我们将使用的基类是 WebhooksServer()。它是一个用于轻松配置可以从 Huggingface Hub 接收 webhooks 的服务器的类。该服务器基于 Gradio 应用程序。它有一个 UI 来显示给你或你的用户的说明，以及一个 API 来监听 webhooks。

要查看 webhook 服务器的运行示例，请查看 Spaces CI Bot。它是一个 Space，当在 Space 上打开 PR 时，它会启动临时环境。

这是一个实验性功能。这意味着我们仍在努力改进 API。未来可能会在没有事先通知的情况下引入重大更改。请务必在你的 requirements 中固定 huggingface_hub 的版本。

创建端点

实现 webhook 端点就像装饰一个函数一样简单。让我们看第一个例子来解释主要概念

# app.py
from huggingface_hub import webhook_endpoint, WebhookPayload

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

将此代码段保存在名为 'app.py' 的文件中，并使用 'python app.py' 运行它。你应该看到类似这样的消息

Webhook secret is not defined. This means your webhook endpoints will be open to everyone.
To add a secret, set `WEBHOOK_SECRET` as environment variable or pass it at initialization:
        `app = WebhooksServer(webhook_secret='my_secret', ...)`
For more details about webhook secrets, please refer to https://huggingface.co/docs/hub/webhooks#webhook-secret.
Running on local URL:  http://127.0.0.1:7860
Running on public URL: https://1fadb0f52d8bf825fc.gradio.live

This share link expires in 72 hours. For free permanent hosting and GPU upgrades (NEW!), check out Spaces: https://huggingface.co/spaces

Webhooks are correctly setup and ready to use:
  - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training
Go to https://huggingface.co/settings/webhooks to setup your webhooks.

做得好！你刚刚启动了一个 webhook 服务器！让我们分解一下到底发生了什么

通过使用 webhook_endpoint() 装饰一个函数，后台已创建了一个 WebhooksServer() 对象。正如你所看到的，此服务器是一个在 http://127.0.0.1:7860 上运行的 Gradio 应用程序。如果你在浏览器中打开此 URL，你将看到一个着陆页，其中包含有关已注册 webhooks 的说明。
Gradio 应用程序是底层的一个 FastAPI 服务器。已向其添加了一个新的 POST 路由 /webhooks/trigger_training。这是将监听 webhooks 并在触发时运行 trigger_training 函数的路由。FastAPI 将自动解析有效负载，并将其作为 WebhookPayload 对象传递给该函数。这是一个 pydantic 对象，其中包含有关触发 webhook 的事件的所有信息。
Gradio 应用程序还打开了一个隧道以接收来自互联网的请求。这是有趣的部分：你可以在 https://huggingface.co/settings/webhooks 上配置一个指向你的本地计算机的 Webhook。这对于调试你的 webhook 服务器并在将其部署到 Space 之前快速迭代很有用。
最后，日志还告诉你，你的服务器当前未通过密钥保护。对于本地调试来说，这没有问题，但对于稍后要记住这一点。

默认情况下，服务器在脚本末尾启动。如果你在 notebook 中运行它，你可以通过调用 decorated_function.run() 手动启动服务器。由于使用了唯一的服务器，即使你有多个端点，也只需启动服务器一次。

配置 Webhook

现在你已经运行了一个 webhook 服务器，你想要配置一个 Webhook 以开始接收消息。转到 https://huggingface.co/settings/webhooks，单击“添加新 webhook”并配置你的 Webhook。设置你要监视的目标仓库和 Webhook URL，此处为 https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training。

就是这样！你现在可以通过更新目标仓库（例如，推送提交）来触发该 webhook。检查 Webhook 的“活动”选项卡以查看已触发的事件。现在你已经有了一个工作设置，你可以对其进行测试并快速迭代。如果你修改了代码并重新启动服务器，你的公共 URL 可能会更改。如果需要，请确保更新 Hub 上的 webhook 配置。

部署到 Space

现在你已经有了一个工作的 webhook 服务器，目标是将其部署到 Space。转到 https://huggingface.co/new-space 创建一个 Space。给它一个名称，选择 Gradio SDK，然后单击“创建 Space”。将你的代码以上传到 Space 中的一个名为 app.py 的文件中。你的 Space 将自动启动！有关 Spaces 的更多详细信息，请参阅此指南。

你的 webhook 服务器现在在一个公共 Space 上运行。在大多数情况下，你将需要使用密钥来保护它。转到你的 Space 设置 > “仓库密钥”部分 > “添加密钥”。将 WEBHOOK_SECRET 环境变量设置为你选择的值。返回到 Webhooks 设置并在 webhook 配置中设置密钥。现在，只有具有正确密钥的请求才会被你的服务器接受。

就是这样！你的 Space 现在已准备好接收来自 Hub 的 webhooks。请记住，如果你在免费的“cpu-basic”硬件上运行 Space，它将在不活动 48 小时后关闭。如果你需要永久 Space，你应该考虑设置为升级的硬件。

高级用法

上面的指南解释了设置 WebhooksServer() 的最快方法。在本节中，我们将了解如何进一步自定义它。

多个端点

你可以在同一服务器上注册多个端点。例如，你可能想要一个端点来触发训练作业，另一个端点来触发模型评估。你可以通过添加多个 @webhook_endpoint 装饰器来做到这一点

# app.py
from huggingface_hub import webhook_endpoint, WebhookPayload

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

@webhook_endpoint
async def trigger_evaluation(payload: WebhookPayload) -> None:
    if payload.repo.type == "model" and payload.event.action == "update":
        # Trigger an evaluation job if a model is updated
        ...

这将创建两个端点

(...)
Webhooks are correctly setup and ready to use:
  - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training
  - POST https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_evaluation

自定义服务器

为了获得更高的灵活性，您也可以直接创建一个 WebhooksServer() 对象。如果您想自定义服务器的着陆页，这将非常有用。您可以通过传递一个 Gradio UI 来覆盖默认的着陆页。例如，您可以为用户添加说明，或者添加一个表单来手动触发 Webhook。在创建 WebhooksServer() 时，您可以使用 add_webhook() 装饰器注册新的 Webhook。

这是一个完整的示例

import gradio as gr
from fastapi import Request
from huggingface_hub import WebhooksServer, WebhookPayload

# 1. Define  UI
with gr.Blocks() as ui:
    ...

# 2. Create WebhooksServer with custom UI and secret
app = WebhooksServer(ui=ui, webhook_secret="my_secret_key")

# 3. Register webhook with explicit name
@app.add_webhook("/say_hello")
async def hello(payload: WebhookPayload):
    return {"message": "hello"}

# 4. Register webhook with implicit name
@app.add_webhook
async def goodbye(payload: WebhookPayload):
    return {"message": "goodbye"}

# 5. Start server (optional)
app.run()

我们使用 Gradio blocks 定义了一个自定义 UI。此 UI 将显示在服务器的着陆页上。
我们创建了一个带有自定义 UI 和密钥的 WebhooksServer() 对象。密钥是可选的，可以使用 WEBHOOK_SECRET 环境变量进行设置。
我们注册了一个带有显式名称的 Webhook。这将在 /webhooks/say_hello 创建一个端点。
我们注册了一个带有隐式名称的 Webhook。这将在 /webhooks/goodbye 创建一个端点。
我们启动服务器。这是可选的，因为您的服务器将在脚本结束时自动启动。

< > 在 GitHub 上更新

Hub Python Library