Hub Python Library documentation
Webhooks
Webhooks
Webhooks 是 MLOps 相关功能的基础。它们允许你监听特定仓库的新变更,或监听你关注的所有用户/组织下的所有仓库。本指南将首先介绍如何以编程方式管理 Webhooks。然后,我们将了解如何利用 huggingface_hub 创建一个监听 Webhooks 的服务器,并将其部署到 Space 中。
本指南假定您已熟悉 Huggingface Hub 上 webhook 的概念。如需进一步了解 webhook 本身,请先阅读此指南。
管理 Webhooks
huggingface_hub 允许您以编程方式管理 Webhook。您可以列出已有的 Webhook,创建新的 Webhook,并对其进行更新、启用、禁用或删除。本节将指导您使用 Hugging Face Hub 的 API 函数完成这些操作。
创建 Webhook
要创建新的 webhook,请使用 create_webhook(),并指定应发送 payload 的 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"
)Webhook 还可以触发在 Hugging face 基础设施上运行一个 Job,而不是将 payload 发送到 URL。 在这种情况下,您需要传入源 Job 的 ID。
from huggingface_hub import create_webhook
# Example: Creating a webhook that triggers a Job
webhook = create_webhook(
job_id=job_id,
watched=[{"type": "user", "name": "your-username"}, {"type": "org", "name": "your-org-name"}],
domains=["repo", "discussion"],
secret="your-secret"
)Webhook 会触发 Job,并将 webhook 负载放入环境变量 WEBHOOK_PAYLOAD 中。
源 Job 的密钥不会被复制到 webhook 中。如果触发的 Job 需要密钥(例如 HF_TOKEN),请通过 secrets={"HF_TOKEN": "hf_***"} 传入。使用 update_webhook() 时,必须将 job_id 与 secrets 一起传入。列出的密钥会替换已存储的值,空字符串会删除某个密钥,而其他所有密钥保持不变。
有关 Hugging Face Jobs、可用硬件(CPU、GPU)和 UV 脚本的更多信息,请参阅 Jobs 文档。
列出 Webhooks
要查看所有已配置的 webhook,可以使用 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() 来实现,之后可以通过 enable_webhook() 重新启用该 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")从 CLI 管理 Webhooks
上述所有操作也可以通过 hf 命令行界面使用。这在脚本编写和智能体工作流中尤为有用。
# List all webhooks
>>> hf webhooks ls
# Get details about a specific webhook
>>> hf webhooks info <webhook_id>
# Create a webhook that pings a URL
>>> hf webhooks create --url https://example.com/hook --watch model:bert-base-uncased --domain repo
# Create a webhook that triggers a Job
>>> hf webhooks create --job-id <job_id> --watch user:julien-c
# Create a webhook that triggers a Job with your HF token as a secret
>>> hf webhooks create --job-id <job_id> --watch bucket:my-org/my-bucket --secrets HF_TOKEN
# Update an existing webhook
>>> hf webhooks update <webhook_id> --url https://new-url.com/hook
# Enable / disable a webhook
>>> hf webhooks enable <webhook_id>
>>> hf webhooks disable <webhook_id>
# Delete a webhook (with confirmation prompt)
>>> hf webhooks delete <webhook_id>--watch 选项使用 type:name 格式,其中 type 为 model、dataset、space、org 或 user 之一。可以重复使用以监视多个项目。要查看完整的选项列表,请运行 hf webhooks <command> --help。更多详细信息,请参阅 CLI 指南。
Webhooks 服务器
本指南部分将使用的基础类是 WebhooksServer()。它是一个用于轻松配置服务器的类,该服务器可以接收来自 Huggingface Hub 的 webhooks。该服务器基于 Gradio 应用构建。它提供了一个 UI,用于向您或您的用户展示操作说明,并提供一个 API 来监听 webhooks。
要查看 webhook 服务器的运行示例,请查看 Spaces CI Bot。它是一个 Space,当某个 Space 上打开 PR 时,会启动临时环境。
这是一个实验性功能。这意味着我们仍在改进 API。未来可能会在不另行通知的情况下引入破坏性变更。请务必在您的依赖项中固定
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,你将看到一个包含已注册 webhook 说明的落地页。
- Gradio 应用底层是一个 FastAPI 服务器。其中新增了一个 POST 路由
/webhooks/trigger_training。 该路由用于监听 webhook,并在触发时运行trigger_training函数。FastAPI 会自动解析 payload,并将其作为 WebhookPayload 对象传递给该函数。这是一个pydantic对象,包含触发该 webhook 的事件的所有信息。 - Gradio 应用还打开了一个隧道,用于接收来自互联网的请求。这是有趣的部分:你可以在 https://huggingface.co/settings/webhooks 上配置一个 Webhook,将其指向你的本地机器。这对于调试你的 webhook 服务器并在部署到 Space 之前快速迭代非常有用。
- 最后,日志还提示你当前服务器未通过密钥进行保护。这在本地调试时没有问题,但后续需要留意。
默认情况下,服务器会在脚本末尾启动。如果在 notebook 中运行,可以通过调用
decorated_function.launch()手动启动服务器。由于使用的是唯一的服务器,即使有多个端点,也只需启动一次服务器。
配置 Webhook
现在您已有一个正在运行的 webhook 服务器,接下来需要配置一个 Webhook 以开始接收消息。
前往 https://huggingface.co/settings/webhooks,点击“添加新的 webhook”并配置您的 Webhook。设置您想要监控的目标仓库以及 Webhook URL,此处为 https://1fadb0f52d8bf825fc.gradio.live/webhooks/trigger_training。

就是这样!现在你可以通过更新目标仓库(例如推送一个提交)来触发该 webhook。查看你的 Webhook 的 Activity 选项卡,以查看已触发的事件。既然你已经搭建好了可用的环境,就可以进行测试并快速迭代。如果你修改了代码并重启服务器,你的公共 URL 可能会发生变化。如有需要,请记得在 Hub 上更新 webhook 配置。
部署到 Space
现在你已经拥有了一个可用的 webhook 服务器,目标是将其部署到 Space 中。前往 https://huggingface.co/new-space 创建一个 Space。为其命名,选择 Gradio SDK,然后点击“Create Space”。将你的代码上传到 Space 中,文件名为 app.py。你的 Space 将自动启动!有关 Spaces 的更多详细信息,请参阅此 指南。
您的 webhook 服务器现在已运行在一个公开的 Space 上。在大多数情况下,您希望使用密钥对其进行保护。前往您的 Space 设置 > “Repository secrets” 部分 > “Add a secret”。将 WEBHOOK_SECRET 环境变量设置为您选择的值。返回 Webhooks settings 并在 webhook 配置中设置该密钥。现在,您的服务器将只接受带有正确密钥的请求。
就是这样!您的 Space 现在已准备好接收来自 Hub 的 webhook。请注意,如果您在免费的 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 来实现,它会覆盖默认的界面。例如,您可以为用户添加说明,或添加一个表单以手动触发 webhooks。在创建 WebhooksServer() 时,您可以使用 add_webhook() 装饰器来注册新的 webhooks。
以下是一个完整的示例:
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.launch()- 我们使用 Gradio blocks 定义了一个自定义 UI。该 UI 将显示在服务器的着陆页上。
- 我们创建一个带有自定义 UI 和密钥的 WebhooksServer() 对象。密钥是可选的,可以通过
WEBHOOK_SECRET环境变量进行设置。 - 我们注册了一个具有显式名称的 webhook。这将在
/webhooks/say_hello处创建一个端点。 - 我们注册了一个具有隐式名称的 webhook。这将在
/webhooks/goodbye处创建一个端点。 - 我们启动服务器。这一步是可选的,因为脚本结束时会自动启动服务器。