Hub Python Library documentation

Webhooks

Hugging Face's logo
Join the Hugging Face community

and get access to the augmented documentation experience

to get started

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 服务器!让我们详细拆解一下具体发生了什么:

  1. 通过用 webhook_endpoint() 装饰一个函数,后台已创建了一个 WebhooksServer() 对象。 如你所见,该服务器是一个运行在 http://127.0.0.1:7860 上的 Gradio 应用。如果你在浏览器中打开此 URL,你将看到一个包含已注册 webhook 说明的落地页。
  2. Gradio 应用底层是一个 FastAPI 服务器。其中新增了一个 POST 路由 /webhooks/trigger_training。 该路由用于监听 webhook,并在触发时运行 trigger_training 函数。FastAPI 会自动解析 payload,并将其作为 WebhookPayload 对象传递给该函数。这是一个 pydantic 对象,包含触发该 webhook 的事件的所有信息。
  3. Gradio 应用还打开了一个隧道,用于接收来自互联网的请求。这是有趣的部分:你可以在 https://huggingface.co/settings/webhooks 上配置一个 Webhook,将其指向你的本地机器。这对于调试你的 webhook 服务器并在部署到 Space 之前快速迭代非常有用。
  4. 最后,日志还提示你当前服务器未通过密钥进行保护。这在本地调试时没有问题,但后续需要留意。

默认情况下,服务器会在脚本末尾启动。如果在 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()
  1. 我们使用 Gradio blocks 定义了一个自定义 UI。该 UI 将显示在服务器的着陆页上。
  2. 我们创建一个带有自定义 UI 和密钥的 WebhooksServer() 对象。密钥是可选的,可以通过 WEBHOOK_SECRET 环境变量进行设置。
  3. 我们注册了一个具有显式名称的 webhook。这将在 /webhooks/say_hello 处创建一个端点。
  4. 我们注册了一个具有隐式名称的 webhook。这将在 /webhooks/goodbye 处创建一个端点。
  5. 我们启动服务器。这一步是可选的,因为脚本结束时会自动启动服务器。
Update on GitHub