企业官网建设流程全解析

如何在 Prefect Cloud 创建 webhook 接收事件回调

【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect

如果你的外部系统——CI 任务、cron 脚本、模型更新流水线——需要通知 Prefect Cloud，让它据此触发自动化，Prefect Cloud 的 webhook 就是这条集成路径：每个 webhook 暴露一个唯一的 URL 端点，接收外部系统的 HTTP 请求，并通过你定义的 Jinja2 模板把请求转换为工作区里的 Prefect 事件，事件随后可以作为自动化（automation）的触发器。本文的目标是：在 Prefect Cloud UI 中创建一个 webhook，用一次真实的 HTTP 请求调用它，在 Event Feed 中确认事件已生成，并基于该事件创建一个自动化。

前提条件

• 可以访问 Prefect Cloud，并能进入当前工作区的Webhooks页面。
• 外部发送方（你的脚本或服务）能够对 Prefect Cloud 发起 HTTP 请求。
• 如果你的账户启用了账号级的 webhook 认证强制策略（见下文“为 webhook 配置 API key 认证”），创建 webhook 时必须关联一个 service account。

在 UI 中创建 Webhook

1. 进入 Prefect Cloud 的Webhooks页面，点击Create Webhook。
2. 为 webhook 填写名称，并编写一个 Jinja2 模板。模板决定入站 HTTP 请求如何被转换成一个 Prefect 事件。
3. 保存后，Prefect Cloud 会为该 webhook 生成一个唯一 URL。端点形如https://api.prefect.cloud/hooks/下的随机不透明 URL（例如https://api.prefect.cloud/hooks/AERylZ_uewzpDx-8fcweHQ，此为文档示例值）。该 URL 由 Prefect Cloud 分配，不能通过 API 自行设定；随时可以轮换（rotate）URL，而不会丢失已有关联配置。

模板必须渲染出合法 JSON

模板渲染后的结果必须是能解析的合法 JSON（文档原文以json.loads举例），并且至少包含：

• event：事件名称；
• resource["prefect.resource.id"]：资源 ID。

这两项是所有 Prefect 事件的要求。其余缺失字段（如occurred、id）由 Prefect Cloud 自动填充默认值，模板中不必写。如果模板没有输出payload字段，Prefect Cloud 会默认填入一组调试信息，包含 HTTP method、headers 和 body。

模板中可用的请求上下文

Jinja2 模板上下文包含入站 HTTP 请求的三部分内容：

• method：大写后的 HTTP 方法字符串，如GET、POST；
• headers：请求头构成的大小写不敏感字典，可按任意大小写取值（例如{{ headers['Content-Length'] }}、{{ headers['content-length'] }}等价）。注意Authorization头会被移除，不会进入模板；
• body：请求体，按Content-Type做尽力解析。application/json会被解析为 JSON 对象；application/x-www-form-urlencoded会被解析为扁平的键值对字典，body['friendly_name']与body.friendly_name等价。属性式访问只适用于合法的 Python 标识符，带连字符的键要写成body['friendly-name']。其他 Content-Type 会先尝试按 JSON 解析，解析失败时 body 以 Pythonstr形式提供给模板。

模板支持全部内建 Jinja2 语法块和过滤器，另含jinja2-humanize-extensions包提供的过滤器。缺值时可以用default过滤器兜底，例如{{ body.friendly_name|default(body.model) }}。

一个动态模板示例

以下示例从请求 body 中读取model_id、friendly_name、run_count三个字段（与后文的 curl 调用对应）：

{ "event": "model-update", "resource": { "prefect.resource.id": "product.models.{{ body.model_id}}", "prefect.resource.name": "{{ body.friendly_name }}", "run_count": "{{body.run_count}}" } }

可选：完全静态的模板

如果 webhook 只针对单一资源、单一事件，模板可以完全静态。适用于对方团队只是在 cron 脚本末尾加一行curl的场景——静态模板不依赖请求 body，因此GET、HEAD、DELETE这类无 body 的请求也可以调用它：

{ "event": "model.refreshed", "resource": { "prefect.resource.id": "product.models.recommendations", "prefect.resource.name": "Recommendations [Products]", "producing-team": "Data Science" } }

调用方式就是一条不带 body 的请求（URL 替换为你自己的 webhook 端点，以下为文档示例值）：

curl https://api.prefect.cloud/hooks/AERylZ_uewzpDx-8fcweHQ

每次命中该 webhook 都会在工作区生成一个同名的 Prefect 事件。

调用 Webhook 端点

HTTP 方法与是否携带 body 的对应关系（来自 Webhooks 概念文档）：

• 模板为静态、或模板不依赖请求 body 时，可用GET、HEAD、DELETE（headers 仍可用于模板）；
• 模板需要从 body 取值时，使用POST、PUT、PATCH。

webhook 端点对外非常“安静”：成功只返回204 No Content（个别场景除外，见下文认证部分），任何解析请求出错则返回400 Bad Request。因此不要指望从响应体获得更多信息——是否真正生效要看 Event Feed。

对上文的动态模板，用任意 HTTP 客户端发送POST即可，body 带上模板要读取的字段（YOUR_UNIQUE_WEBHOOK_ID替换为你的 webhook ID）：

curl -X POST https://api.prefect.cloud/hooks/YOUR_UNIQUE_WEBHOOK_ID \ -d "model_id=my_model_123" \ -d "friendly_name=My Awesome Model" \ -d "run_count=15"

验证事件是否生成

1. 调用 webhook 后，进入 Prefect Cloud 的Event Feed，应能看到一条与本次调用对应的新事件。

2. 点击该事件进入事件详情页，核对事件名和资源标签是否与模板预期一致。

这一步就是成功的判断依据：Event Feed 中出现了模板渲染出的事件。如果在调用后看不到事件，或事件数据不符合预期，按下一节的排障顺序检查。

从事件创建 Automation

在事件详情页（在 Event Feed 中点击事件即可进入）点击Automate按钮，系统会基于这条事件预填一个自动化触发器。点击Next定义该自动化要执行的动作，例如运行一个 deployment 或发送通知。这样就完成了从外部 HTTP 请求到 Prefect 自动化的整条链路。

事件接收异常的排查

调用后 Event Feed 中没有预期事件、或事件数据不对时，文档给出的排查项：

• 查看 Event Feed：先看是否存在与该 webhook 相关的任何事件，即使与你的预期不完全一致。
• 查找prefect-cloud.webhook.failed事件：Prefect Cloud 处理 webhook 出错（例如模板不合法、请求格式错误）时会生成prefect-cloud.webhook.failed事件，其中包含收到的 HTTP method、headers、body，以及模板实际渲染出的内容，是定位模板问题的主要线索。
• 核对请求本身：复查发给 webhook 的 URL、HTTP 方法、headers 和 body。
• 复查模板：确认 Jinja2 表达式访问的确实是请求中你打算使用的部分，如body.field_name、headers['Header-Name']。

为 webhook 配置 API key 认证

在 Prefect Cloud Pro 或 Enterprise 层级账户中，可以为 webhook 关联一个 service account，使端点使用该 service account 的 API key 进行认证。启用后，外部系统在调用时需要携带：

Authorization: Bearer <your-api-key>

<your-api-key>替换为该 service account 的 API key。如果账户启用了账号级的强制认证设置，所有未关联 service account 的已有 webhook 会被自动禁用，新 webhook 必须在创建时就关联 service account。

可选：通过 CLI 以编程方式创建

除了在 UI 创建，文档说明也可以用prefect cloud webhook命令组管理 webhook，例如：

prefect cloud webhook create your-webhook-name \ --description "Receives webhooks from your system" \ --template '{ "event": "your.event.name", "resource": { "prefect.resource.id": "your.resource.id" } }'

其中 webhook 名称、描述和模板按你的集成替换。ls、get、toggle、rotate等其他子命令可用prefect cloud webhook --help查看。文档还提到可通过 Prefect Cloud API 或 Terraform Provider 管理 webhook。

参考文档

• 本文主路径来源：创建 webhook 的 how-to 文档
• 端点规则、模板上下文、认证与失败事件的详细说明：Webhooks 概念文档
• 事件机制：Events 概念文档
• 自动化定义：Automations 概念文档 与 创建自动化

【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect