Hub Python Library documentation

Webhooks

You are viewing main version, which requires installation from source. If you'd like regular pip install, checkout the latest stable version (v2.4.0).
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 --watch model:bert-base-uncased --domain repo

# 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