FastAPI OpenAPI Webhooks 全解析：用文档化的出站事件通知反向调用用户系统

导读

本文基于 FastAPI 仓库中的 docs/es/docs/advanced/openapi-webhooks.md（英文版见 docs/en/docs/advanced/openapi-webhooks.md），系统讲解 Webhooks 与 OpenAPI 的交叉主题：如何在 FastAPI 应用中把"你的应用将主动调用用户系统"这一出站事件契约，以规范化的方式写进 OpenAPI 文档（3.1.0+ 引入的顶层 webhooks 字段），从而让 API 用户轻松实现接收端、甚至自动生成接收代码。读完本文你将掌握：webhook 与普通 path operation 的本质区别、app.webhooks 的声明语法与事件标识符语义、生成文档与 openapi.json 中 webhook 结构的对应关系，以及相关源码与测试层面的实现细节。

Webhook 是什么：请求方向的翻转

大多数时候，你的用户把请求发送给你的 API，你返回响应。但在某些场景下，情况恰好相反——你的 API（或应用）会带着某些数据，主动向用户的系统发出请求，通常是为了通知某种事件。这种由你的应用发起的、朝向用户系统的 HTTP 回调，通常被称为 webhook。

典型应用包括：

• 用户在你的平台注册了新订阅（如 new-subscription 事件）；
• 某个异步任务完成，需要通知用户系统拉取结果；
• 需要把第三方产生的状态变化实时推送给订阅方。

关键差异在于：常规流程是"用户 → 你的 API"，而 webhook 流程是"你的 API → 用户的 API"。这意味着 webhook 本质上描述的是你即将发出的请求的契约，而非你对外提供的接口。

Webhook 的标准流程

原文把 webhook 的完整协作流程拆成三个角色分工：

1. 你（服务提供方）负责定义：将在代码中定义要发送的消息，即 请求体（request body）；
2. 你还要定义触发时机：在哪些时刻（事件发生时）你的应用会发出这些请求或事件；
3. 你的用户负责提供接收地址：用户通过某种方式（例如在你提供的某个网页面板中）注册一个 URL，告诉你的应用"把 webhook 发到这个地址"。

需要特别强调的是，注册 URL 的逻辑以及真正发送请求的代码，全部由你在自己的代码里实现——OpenAPI 文档化只解决"契约描述"问题，FastAPI 不会替你维护用户注册表，也不会替你发出任何请求。发送时你完全可以按需选用 httpx、requests 等任意 HTTP 客户端库。

用 FastAPI + OpenAPI 文档化 Webhooks 的价值

借助 FastAPI 和 OpenAPI，你可以为每个 webhook 声明三类信息：

• 名称（name）：webhook 的事件标识；
• HTTP 操作类型：你的应用可能发出的请求方法，例如 POST、PUT 等；
• 请求体（body）：你的应用发送请求时会携带的数据结构。

把这些契约写进 OpenAPI 文档后，你的用户实现"接收端 API"就会容易得多——他们甚至能基于文档自动生成一部分接收端代码，因为文档已完整描述了"你将会用什么方法、以什么数据结构调用他们"。

版本前提：OpenAPI 中 webhooks 字段自 OpenAPI 3.1.0 起可用，FastAPI 自 0.99.0 起开始支持。

一个带 Webhooks 的最小应用

定义一个带 webhook 的 FastAPI 应用极其直观：创建 FastAPI 实例后，使用 webhooks 属性，以与声明 path operations 完全相同的方式（例如 @app.webhooks.post()）声明 webhook。仓库中的完整示例位于 docs_src/openapi_webhooks/tutorial001_py310.py：

from datetime import datetime

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Subscription(BaseModel):
    username: str
    monthly_fee: float
    start_date: datetime


@app.webhooks.post("new-subscription")
def new_subscription(body: Subscription):
    """
    When a new user subscribes to your service we'll send you a POST request with this
    data to the URL that you register for the event `new-subscription` in the dashboard.
    """


@app.get("/users/")
def read_users():
    return ["Rick", "Morty"]

逐段解读这个例子：

• Subscription 是出站请求体模型：三个字段 username（str）、monthly_fee（float）、start_date（datetime），完整描述了"新订阅"事件将要发送给用户系统的数据载荷。
• @app.webhooks.post("new-subscription") 声明事件：post 表示你的应用会对用户系统发起 POST 请求。webhook 声明支持与 path operation 相同的 HTTP 方法（POST、PUT 等），因此可以声明你的应用可能发出的每一种操作类型。
• 函数签名中的 body: Subscription：与普通 path operation 声明请求体完全一致，Pydantic 会据此生成对应的 OpenAPI schema（#/components/schemas/Subscription）。
• 函数 docstring 是给用户看的契约说明：它会被带入生成的 OpenAPI 文档，作为该事件的描述，帮助用户理解"在什么时机、会收到什么"。
• 函数体为空是正常的：webhook 声明只服务于文档，FastAPI 并不会把它注册为你的应用上可访问的真实接口，因此这个函数体不会被当作请求处理器去执行。

关键语义：传入的文本不是 path，而是事件标识符

原文用一个重要细节提醒读者：在 webhook 声明里，你传入的第一个字符串并不是 URL 路径。

比如 @app.webhooks.post("new-subscription") 中的 new-subscription，只是这个 webhook 的标识符（webhook 名称/事件名称），而不是类似 /items/ 那样的路径。原因很清晰：真正接收 webhook 请求的 URL 路径，应当由你的用户在别处（例如你的面板）自行定义，而你的应用只需按"事件名称"把请求发往用户注册的地址即可。换句话说，webhook 的"路径"对服务提供方是未知的、由接收方决定的，所以在 OpenAPI 文档中它只能以一个事件键名而非 URL 路径的形式出现。

app.webhooks 本质上就是一个 APIRouter

从源码来看，这个说法有据可查。在 fastapi/applications.py 的 FastAPI.__init__ 中，webhooks 是一个可注入的参数，其默认实现为：

self.webhooks: Annotated[APIRouter, ...] = webhooks or routing.APIRouter()

对应的构造函数参数文档也明确写道："Add OpenAPI webhooks. This is similar to callbacks but it doesn't depend on specific path operations." 也就是说：

• app.webhooks 与你做多文件项目结构时用的 APIRouter 是同一类型，因此你完全可以先创建一个 APIRouter，在其中集中声明所有 webhook，再通过 FastAPI(webhooks=my_router) 注入；
• 它与 callbacks（回调）的区别在于：callbacks 必须依附于某条具体的 path operation（用于声明"调用此接口后我可能回调你"），而 webhooks 不依赖任何 path operation，是全局独立的出站事件声明。

由于 self.webhooks 是独立路由器，其 routes 不会混入应用对外服务的 self.router.routes。可以看到 openapi() 方法在生成 schema 时是这样传递的（fastapi/applications.py）：

self.openapi_schema = get_openapi(
    ...
    routes=self.routes,
    webhooks=self.webhooks.routes,
    ...
)

这也印证了前文结论：webhook 路由只流向 OpenAPI 文档生成器，并不会在你的应用上产生真实可访问的端点。

查看文档效果

按常规方式启动应用（例如 uvicorn 或 fastapi dev 加载该示例模块），然后访问 http://127.0.0.1:8000/docs。

你会看到自动交互式文档中既有普通的 path operations（本例为 GET /users/），又出现了独立的 Webhooks 区块：

图片出自官方文档原图：原文档 docs/es/docs/advanced/openapi-webhooks.md 中内嵌的 /img/tutorial/openapi-webhooks/image01.png，实际文件位于仓库的 docs/en/docs/img/tutorial/openapi-webhooks/image01.png（各语言文档共享同一份图片资源）。

OpenAPI Schema 中的 webhooks 结构

Webhooks 不只是"好看"，它们会以正式字段进入 OpenAPI schema（可通过 http://127.0.0.1:8000/openapi.json 查看）。仓库测试 tests/test_tutorial/test_openapi_webhooks/test_tutorial001.py 对生成的 schema 做了精确快照断言，核心结构如下（OpenAPI 3.1.0 的顶层 webhooks 键，而非 paths）：

{
  "openapi": "3.1.0",
  "paths": {
    "/users/": {
      "get": {
        "summary": "Read Users",
        "operationId": "read_users_users__get",
        "responses": { "200": { "description": "Successful Response" } }
      }
    }
  },
  "webhooks": {
    "new-subscription": {
      "post": {
        "summary": "New Subscription",
        "description": "When a new user subscribes to your service we'll send you a POST request with this data to the URL that you register for the event `new-subscription` in the dashboard.",
        "operationId": "new_subscriptionnew_subscription_post",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": { "$ref": "#/components/schemas/Subscription" }
            }
          },
          "required": true
        },
        "responses": {
          "200": { "description": "Successful Response" },
          "422": { "description": "Validation Error", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/HTTPValidationError" } } } }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "Subscription": {
        "properties": {
          "username": { "type": "string", "title": "Username" },
          "monthly_fee": { "type": "number", "title": "Monthly Fee" },
          "start_date": { "type": "string", "format": "date-time", "title": "Start Date" }
        },
        "type": "object",
        "required": ["username", "monthly_fee", "start_date"],
        "title": "Subscription"
      }
    }
  }
}

从中可以观察出几个由实现佐证的细节：

• 事件名称即键名：webhooks 对象以 new-subscription（你传入的标识符原样）作为键，方法名 post 作为子键；
• 请求体与校验错误一并声明：requestBody 通过 $ref 引用 components.schemas.Subscription 并标记 required: true；同时自动附带 422 校验错误响应，与普通 path operation 的生成逻辑完全一致；
• 模型注册无差别：Subscription 模型（以及 HTTPValidationError、ValidationError）与普通接口共用 components.schemas 命名空间，用户端拿到文档后即可据 Subscription schema 构造接收端的数据模型。

从源码看 webhook schema 的生成路径

在 fastapi/openapi/utils.py 的 get_openapi() 中，webhook 与普通路由走的是同一套处理管线，只是输出落点不同：

• 普通路由的处理结果累积到 paths（paths.setdefault(...)，见 utils.py）；
• webhook 路由则经由 routing.iter_route_contexts(...) 逐个解析，结果累积到独立的 webhook_paths，并在最后写入顶层键 output["webhooks"]（见 utils.py）：

  for webhook_context in routing.iter_route_contexts(webhooks or []):
      api_webhook = _get_api_route_for_openapi(webhook_context)
      if api_webhook is not None:
          ...
          webhook_paths.setdefault(api_webhook.path_format, {}).update(path)
  ...
  if webhook_paths:
      output["webhooks"] = webhook_paths
  

• 关键点在于，用于生成 schema 的 all_fields 把普通路由与 webhook 路由合并后统一提取字段与模型（见 utils.py），这保证了 webhook 的请求体模型会被完整纳入 components.schemas，同时文档只输出到独立的 webhooks 字段，不会污染真实的 paths。

也就是说，OpenAPI 规范把"出站回调契约"与"入站公开接口"在结构上做了隔离——这正是 FastAPI 借助顶层 webhooks 键所实现的。

Webhooks 的扩展：与安全机制组合使用

webhook 与普通 path operation 一样，可以附加安全依赖。尽管原文档未展开此话题，仓库测试 tests/test_webhooks_security.py 给出了可直接参照的用法——为 webhook 函数参数注入 Security(bearer_scheme)：

from typing import Annotated

from fastapi import FastAPI, Security
from fastapi.security import HTTPBearer

app = FastAPI()
bearer_scheme = HTTPBearer()


@app.webhooks.post("new-subscription")
def new_subscription(
    body: Subscription, token: Annotated[str, Security(bearer_scheme)]
):
    """..."""

在该测试的 schema 快照中，webhook 操作的 security: [{"HTTPBearer": []}] 会被写入 webhooks["new-subscription"]["post"]，同时 HTTPBearer 进入 components.securitySchemes。这说明：你可以向用户精确文档化"你调用他们的 webhook 时会如何携带凭证"（例如请求头里的 Bearer 签名），便于接收方校验请求确系你方发出——这是真实生产 webhook 的安全基线。

各角色的落地职责总结

职责	归属方	是否由 FastAPI 自动完成
定义事件名称与 HTTP 方法	服务提供方（你）	✅ 声明即生成到 OpenAPI 文档
定义出站请求体 schema	服务提供方（你）	✅ Pydantic 模型自动转换
维护"事件 → 用户 URL"注册表	服务提供方（你）	❌ 需自己编写（如面板 + 数据库）
真正发送 HTTP 请求	服务提供方（你）	❌ 需在触发事件处自己调用 HTTP 客户端
提供接收 webhook 的 URL 并处理请求	用户（接收方）	❌ 用户基于你的文档实现

当文档中同时声明了事件契约（示例代码）后，你的用户只要对照 webhooks 区块中的请求体 schema 与 docstring 说明，就能低摩擦地实现并校验自己的接收端点，甚至可借助 OpenAPI 生态的代码生成工具自动产出接收端骨架代码。这正是"把 webhook 文档化"在生产协作中的最大价值。