FastAPI OpenAPI Callbacks 回调文档化实战：用 `callbacks` 声明外部 API 契约

FastAPI OpenAPI Callbacks 回调文档化实战：用callbacks声明外部 API 契约

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

本篇技术指南围绕 FastAPI 提供的OpenAPI 回调（OpenAPI Callbacks）能力展开，讲解如何在你开发的 API 需要反向调用由第三方开发者实现的外部 API 时，通过少量代码在/docs与/openapi.json中完整描述该外部 API 应当实现的路径、请求体与响应。读完本文，你将掌握callbacks=参数的完整用法、OpenAPI 3 路径表达式的写法，以及它如何避免回调对接时的“隐式约定”。

什么是 OpenAPI 回调

当你的 API 中某个*路径操作（path operation）*处理请求后，会向一个由外部开发者创建的外部 API再发起一次请求时，这一次反向请求就称为回调（callback）。

它的成因通常是一个双向协作流程：外部开发者编写的软件先向你的 API 发请求，你的 API 处理完毕后**“回拨”**，向外部开发者提供的外部 API 再发一次请求。因为你的服务最终要调用别人写的接口，所以你需要一种方式，把“这个外部 API 到底应该长什么样”讲清楚——它应该暴露哪个路径操作、接收什么请求体、返回什么响应，等等。

FastAPI 对此的解法非常直观：复用你自己写 API 时的那套自动文档机制，用一段只用于“声明”的代码把外部 API 的形状描述出来，然后把它挂到对应路径操作上。这些回调契约最终会出现在你自己的/docs（Swagger UI）与/openapi.json中，外部开发者据此即可正确地实现那段被回调的外部 API。

本文对应的官方文档为 docs/en/docs/advanced/openapi-callbacks.md（以及法语译本 docs/fr/docs/advanced/openapi-callbacks.md），配套示例源码位于 docs_src/openapi_callbacks/tutorial001_py310.py。

一个带回调的发票应用：业务场景

以文档中的发票（invoice）应用为例，想象你在开发一个允许创建发票的 API：

• 发票包含id、title（可选）、customer、total字段；
• 你的 API 使用者（即外部开发者）通过POST请求在你的 API 中创建一张发票；
• 随后你的 API 会（假设地）依次执行：
  1. 把发票发送给外部开发者的某个客户；
  2. 完成收款；
  3. 向 API 使用者（外部开发者）发回一条通知——这一步的实现方式，就是由你的 API主动向外部开发者提供的外部 API发送一条 POST 请求，这即是本教程讨论的“回调”。

在这个流程里有一个关键的工程痛点：通知回调发送时的请求体、目标路径、以及你期望对方返回什么，只有形成明确文档，外部开发者才能“照着实现”。下文演示的就是如何在不真正实现回调的前提下，把这段契约完整文档化。

一个普通的 FastAPI 应用（未加回调前）

先看尚未引入回调时应用的正常形态：它只有一个路径操作，接收Invoice请求体，外加一个名为callback_url的查询参数，用于承载后续回调的目标 URL。

from fastapi import APIRouter, FastAPI from pydantic import BaseModel, HttpUrl app = FastAPI() class Invoice(BaseModel): id: str title: str | None = None customer: str total: float class InvoiceEvent(BaseModel): description: str paid: bool class InvoiceEventReceived(BaseModel): ok: bool invoices_callback_router = APIRouter() @invoices_callback_router.post( "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass @app.post("/invoices/", callbacks=invoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): """ Create an invoice. This will (let's imagine) let the API user (some external developer) create an invoice. And this path operation will: * Send the invoice to the client. * Collect the money from the client. * Send a notification back to the API user (the external developer), as a callback. * At this point is that the API will somehow send a POST request to the external API with the notification of the invoice event (e.g. "payment successful"). """ # Send the invoice, collect the money, send the notification (the callback) return {"msg": "Invoice received"}

关于上面代码，有几点补充说明：

• callback_url使用了 Pydantic 的HttpUrl类型（from pydantic import BaseModel, HttpUrl）。它会在请求到达时校验 URL 的合法格式。从仓库中的 OpenAPI 快照测试可以看到，它在最终生成的 schema 中表现为format: "uri"的字符串，且带minLength: 1、maxLength: 2083约束，测试断言见 tests/test_tutorial/test_openapi_callbacks/test_tutorial001.py。
• 由于示例函数签名使用了str | None这类联合类型写法，代码需要Python 3.10 及以上，这也是该文件命名为tutorial001_py310.py的原因。
• 整个文件里唯一“新鲜”的部分，是@app.post("/invoices/", ...)装饰器上传入的callbacks=invoices_callback_router.routes参数——下面的篇幅专门解释它。

从仓库根目录运行这段示例，即可查看文档效果：

# 方式一：使用 fastapi CLI fastapi dev docs_src/openapi_callbacks/tutorial001_py310.py # 方式二：使用 uvicorn（docs_src 为可导入包） uvicorn docs_src.openapi_callbacks.tutorial001_py310:app --reload

文档化回调：关键是“声明”而非“执行”

真实的回调实现代码高度依赖你的业务逻辑，且因应用而异，通常只有一两行，例如：

callback_url = "https://example.com/api/v1/invoices/events/" httpx.post(callback_url, json={"description": "Invoice paid", "paid": True})

回调本身不过是一次普通 HTTP 请求——文档中也建议你在自行实现时使用 HTTPX、Requests 等 HTTP 客户端库。

然而，回调最值得重视的部分，是确保你的 API 使用者（外部开发者）严格按照你的 API 将要发送的数据来正确实现外部 API。因此，本教程接下来做的并不是实现回调，而是添加文档代码：声明外部 API 该接收什么请求体、该返回什么响应，使其出现在 Swagger UI（/docs）中，让外部开发者据此开发。

实现上的一个“心智模型”提示：写这段回调文档代码时，不妨假设你就是那位外部开发者，正在实现外部 API，而不是你自己的 API。这个视角会让你更容易决定参数、请求体模型、响应模型应该放在哪里。

编写回调文档代码的完整步骤

第一步：创建一个回调用的APIRouter

新建一个独立的APIRouter，用它来承载一个或多个回调路径操作：

from fastapi import APIRouter, FastAPI invoices_callback_router = APIRouter()

第二步：定义回调的路径操作

回调路径操作与普通 FastAPI 路径操作的写法几乎一致：

• 声明它应当接收的请求体，例如body: InvoiceEvent；
• 声明它应当返回的响应模型，例如response_model=InvoiceEventReceived。

@invoices_callback_router.post( "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass

它和普通路径操作相比有两处主要差异：

1. 函数体内不需要任何真实代码——你的应用永远不会执行这段逻辑，它只用于文档化外部 API。因此函数体直接写pass即可。
2. 路径本身可以包含 OpenAPI 3 的表达式（Key Expression），从而引用“原始请求”中的参数和报文片段（详见下一步）。

第三步：理解回调路径表达式{$callback_url}与{$request.body.id}

回调的路径可以包含 OpenAPI 3 规范所定义的 Key Expression，其中可以使用变量引用发送给你的 API的原始请求的各个部分。本例使用路径：

"{$callback_url}/invoices/{$request.body.id}"

它的含义是：最终回调地址 = 原始请求中callback_url参数的值，拼接/invoices/，再拼接原始请求 JSON 请求体里id字段的值。

为了验证这条路径表达式的效果，追踪一条完整的调用链（也可由仓库测试 tests/test_tutorial/test_openapi_callbacks/test_tutorial001.py 印证）：

1. 外部开发者向你的 API 发送请求：

   https://yourapi.com/invoices/?callback_url=https://www.external.org/events

   请求体 JSON 为：

   { "id": "2expen51ve", "customer": "Mr. Richie Rich", "total": "9999" }

2. 你的 API 处理发票后，在稍后的某个时间点向callback_url指向的外部 API 发起回调请求：

   https://www.external.org/events/invoices/2expen51ve

3. 回调请求体大约形如：

   { "description": "Payment celebration", "paid": true }

4. 你的 API 期望该外部 API 返回形如：

   { "ok": true }

注意回调 URL 中同时融合了两个来源：查询参数callback_url提供的基址（https://www.external.org/events），以及原始 JSON 请求体里的发票id（2expen51ve）。

第四步：将回调路由挂载到你的路径操作上

现在，把已经定义好的回调路径操作，通过callbacks=参数传给你自己的 API的路径操作装饰器：

@app.post("/invoices/", callbacks=invoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): ...

关键细节：这里传给callbacks=的并不是路由器对象invoices_callback_router本身，而是它的.routes属性（即invoices_callback_router.routes）。FastAPI 会解析这些路由条目，把它转换为回调的 OpenAPI 文档。

第五步：在/docs中查看结果

启动应用后访问 http://127.0.0.1:8000/docs，你会看到POST /invoices/接口下多出一个Callbacks章节，清晰地展示外部 API 应有的样子：

同时访问 http://127.0.0.1:8000/openapi.json，可以看到结构化的回调声明——callbacks以回调函数名invoice_notification为键，其值是“表达式路径 → PathItem”的映射：

"callbacks": { "invoice_notification": { "{$callback_url}/invoices/{$request.body.id}": { "post": { "summary": "Invoice Notification", "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/InvoiceEvent" } } } }, "responses": { "200": { "description": "Successful Response" } } } } } }

回调涉及的InvoiceEvent、InvoiceEventReceived等请求/响应模型也会一并进入components.schemas。以上完整快照可在 tests/test_tutorial/test_openapi_callbacks/test_tutorial001.py 的test_openapi_schema中核对。

源码原理：FastAPI 如何生成 callbacks 文档

为了让“文档代码不执行、仅生成文档”的机制落地，FastAPI 从参数接收到 OpenAPI 序列化有一整条清晰链路，可从本仓库源码逐段验证：

1. 参数接收与存储：fastapi/routing.py 中，APIRoute的构造参数callbacks: list[BaseRoute] | None = None会被存入route.callbacks；FastAPI/APIRouter的get、post、put、delete、patch、options、head、trace以及include_router等接口也都暴露了callbacks参数。若在APIRouter(...)或FastAPI(...)顶层传入callbacks，会作为默认值合并到该作用域内的所有路径操作上（参见 fastapi/routing.py 中_RouterIncludeContext对callbacks=[*parent_router.callbacks, *(callbacks or [])]的处理）。

2. 从路由到 OpenAPI operation：fastapi/openapi/utils.py 的get_openapi_path会检查route.callbacks，对其中每一个APIRoute回调递归调用get_openapi_path以生成独立的 PathItem，再以回调名称与表达式路径为键组装成operation["callbacks"]：

if route.callbacks: callbacks = {} for callback in route.callbacks: if isinstance(callback, routing.APIRoute): cb_path, cb_security_schemes, cb_definitions = get_openapi_path(...) callbacks[callback.name] = {callback.path: cb_path} operation["callbacks"] = callbacks

同时，fastapi/openapi/utils.py 中收集 flat models 时会调用get_fields_from_routes(api_route.callbacks)，把回调的请求体与响应模型纳入组件components.schemas，保证外部 API 的模型也拥有独立的 schema 定义。

3. 数据类型建模：fastapi/openapi/models.py 中Operation的callbacks字段类型为dict[str, dict[str, "PathItem"] | Reference] | None，恰好对应 OpenAPI 3 规范中callbacks对象的形状。

4. 不会被真实路由：从 fastapi/routing.py 的注释与 Docstring（"This is only for OpenAPI documentation, the callbacks won't be used directly"）可以看到，回调路由并不会真正加入应用的可调用路由表，只服务于文档输出。因此示例里invoice_notification函数体为pass是安全的——它永远不会被你的应用调用。

5. 测试覆盖：仓库测试文件 tests/test_tutorial/test_openapi_callbacks/test_tutorial001.py 做了三类验证：真实请求/invoices/返回{"msg": "Invoice received"}；直接调用mod.invoice_notification({})覆盖“占位函数”分支；以及test_openapi_schema用快照断言完整 OpenAPI JSON（含上文的callbacks结构、operationId如invoice_notification__callback_url__invoices___request_body_id__post、各 schema 定义）。operationId的存在意味着回调同样支持为每个路径操作生成唯一 ID，便于工具链引用。

更多使用形态与边界说明

• 多个回调：一个APIRouter里可以注册多个回调路径操作，callbacks接受的是路由列表，因此可同时文档化多条外部回调接口。
• 顶层默认回调：如果在FastAPI(...)、APIRouter(...)或include_router(...)层面传入callbacks，会应用到该作用域下（以及include_router挂载的子路由中）所有未显式覆盖的路径操作。
• 与 Webhooks 的区别：回调是绑定在某个具体路径操作上的“回拨”契约，而 OpenAPI 3.1 还提供了与之相似但不依赖具体路径操作的 webhooks 能力，可参考仓库文档 docs/en/docs/advanced/openapi-webhooks.md 以及 fastapi/applications.py 中webhooks参数的说明（"This is similar tocallbacksbut it doesn't depend on specificpath operations"，自 OpenAPI 3.1.0 / FastAPI 0.99.0 起可用）。
• 不执行、不约束运行时：回调文档只影响 OpenAPI 输出，不会在运行时拦截或校验外部 API 的真实行为；是否真正发送回调、何时发送，仍完全由你的业务代码决定。

小结

OpenAPI Callbacks 解决的是多团队 API 协作中“反向调用契约”的文档化难题。在 FastAPI 中你无需学习新的 DSL，只需三步：用APIRouter定义外部 API 的路径操作（函数体写pass）、用{$callback_url}/{$request.body.id}这类 OpenAPI 3 表达式拼接动态路径、再用callbacks=...routes把它挂到自己的路径操作装饰器上。你的/docs与/openapi.json随即会携带完整的回调契约，外部开发者拿到文档即可精确实现被回调的接口，从源头消除对接中的猜测与返工。

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi