《AI Agent 开发入门》17.2 HTTP 服务化

17.2 HTTP 服务化

17.1 把 minagent 收敛成了一个稳定的库——但库有个天生的局限：它只能被同一个进程里的 Python 代码调用。浏览器调不动它，TypeScript 前端调不动它，别的语言写的服务也调不动它。而真实的 Agent 产品里，界面往往在浏览器、推理在服务器，中间隔着一条网络。要跨过这条边界，minagent 得从「一个库」再进一步，变成「一个服务」。

本节把 minagent 推进到「HTTP 服务」：用 FastAPI 把引擎暴露成 POST /run（一次性返回）与 POST /stream（SSE 流式），把 7.1 的 Event 映射成 SSE 的 event: / data: 行，真起一个本地服务并用 curl 打通两个端点。

17.2.1 为什么值得跨出进程

先想清楚「服务化」到底解决了什么。同一个进程里，调用方直接 agent.run(...) 就完了，为什么要绕一圈 HTTP？答案在边界二字上。进程内调用要求调用方和被调用方跑在同一台机器、同一种语言、同一个生命周期里——这是很强的约束。一旦变成 HTTP 服务，这三条约束同时松开：

代价也随之而来：多了一层网络（延迟、序列化、错误传播），以及一个必须处理的新问题——「跑很久的 Agent 怎么把过程实时告诉调用方」。这正是 SSE 要解决的，也是本节的重点。

minagent 的这次「出门」，其实复用了一件早就备好的东西：7.1 那条 Event 事件流。agent.stream() 已经能把「token / 工具调用 / 步骤」逐条吐出来，服务化要做的只是把这条流接到 HTTP 上。换句话说，7.1 建的是底座，本节建的是出口。

17.2.2 依赖与最小骨架

先声明口径：FastAPI 与 uvicorn 是可选依赖。minagent 的核心（消息 / 工具 / 循环 / 事件）仍然只用标准库；只有当你要把它当服务跑时，才需要装这两样。本节实测版本是 fastapi 0.143.0 与 uvicorn 0.54.0：

python -m pip install --index-url https://mirrors.ustc.edu.cn/pypi/simple fastapi uvicorn

服务层的全部代码放进一个新模块 minagent/service.py。它的骨架极短：一个请求模型、一个「造 Agent」的工厂、一个把 Event 编成 SSE 帧的函数，然后两个路由。

17.2.3 请求模型：让 FastAPI 替你校验

入参用一个 pydantic 模型描述，这是 FastAPI 的惯用法：

# minagent/service.py
from pydantic import BaseModel


class RunRequest(BaseModel):
    input: str
    session_id: str | None = None

input 是必填的字符串，session_id 是可选（默认 None）。就这么一个类，FastAPI 会自动替你做好几件事：解析 JSON 请求体、校验字段类型、缺字段时报 422、并把它写进 OpenAPI 文档。这比自己 request.json() 然后手工 if "input" not in body 稳得多——校验逻辑声明式地写在类型里，而不是散在 if 分支里。

17.2.4 两个端点：一次性与流式

Agent 的调用有两种典型形态，对应两个端点：

/run 直接把 RunResult 压成 JSON 返回：

from typing import Any

from fastapi import FastAPI

from .agent import Agent
from .model import FakeModel
from .tools import Tool, ToolRegistry
from .types import ToolCall


def build_agent() -> Agent:
    reg = ToolRegistry()
    reg.add(Tool("add", "两数相加",
                 {"type": "object",
                  "properties": {"a": {"type": "number"}, "b": {"type": "number"}},
                  "required": ["a", "b"]},
                 lambda a, b: a + b))
    model = FakeModel([
        [ToolCall("c1", "add", {"a": 2, "b": 3})],
        "2 + 3 = 5。",
    ])
    return Agent(model, reg, system="你是计算助手", max_steps=4)


app = FastAPI(title="minagent service", version="0.17.0")


@app.post("/run")
def run_once(req: RunRequest) -> dict[str, Any]:
    result = build_agent().run(req.input)
    return {
        "final": result.final,
        "steps": [{"index": s.index, "kind": s.kind, "detail": s.detail}
                  for s in result.steps],
    }

注意返回的是普通 dict，FastAPI 会把它序列化成 JSON。这里刻意没直接返回 RunResult——它是 dataclass，Message 里还嵌着 ToolCall，直接返回要靠自定义编码器；手工挑出 final 与 steps 两个字段，既简单又明确了「对外的数据形状」。服务边界处，显式地定义输出形状，比依赖框架的魔法编码更可靠。

17.2.5 把 Event 映射成 SSE 帧

SSE（Server-Sent Events）是一种极简的服务器推送协议：响应头是 text/event-stream，body 是一串用空行分隔的帧，每帧由若干 字段: 值 行组成。最常用的两个字段是 event:（事件名）和 data:（载荷）。把 7.1 的六种 Event 一一对上，映射关系是：

映射本身是一行字符串拼接：

import json

from .events import Event


def sse(event: Event) -> str:
    payload = json.dumps(event.data, ensure_ascii=False)
    return f"event: {event.type}\ndata: {payload}\n\n"

两个细节决定它能不能用。第一，结尾必须是两个换行 \n\n——SSE 靠空行切帧，少一个换行，两帧就粘成一条，解析器读到的数据全乱。第二，data: 的载荷里不能有裸换行，因为换行会被 SSE 当成帧的边界。好在 json.dumps 默认不换行，天然满足；如果哪天载荷里真有换行（比如多行文本），就得拆成多行 data:，每行一个前缀——这也是为什么用 JSON 当载荷比塞裸文本省心。

ensure_ascii=False 是为了中文：默认 ensure_ascii=True 会把中文转成 \uXXXX 转义，虽然合法但没法读。设成 False，data: 里就是原样的中文，调试时肉眼可看。

17.2.6 StreamingResponse：让生成器直接变成响应体

流式端点只有四行，核心是 StreamingResponse：

from typing import Iterator

from fastapi.responses import StreamingResponse


@app.post("/stream")
def stream(req: RunRequest) -> StreamingResponse:
    def gen() -> Iterator[str]:
        for ev in build_agent().stream(req.input):
            yield sse(ev)

    return StreamingResponse(gen(), media_type="text/event-stream")

StreamingResponse 接受一个可迭代对象（这里是个生成器），把它产出的每一段当作响应体的一片，产一片、发一片，而不是等全部算完再发。这正好接上 7.1 那条 agent.stream()：Agent 每产出一个 Event，sse() 把它编成帧，StreamingResponse 立刻把这帧推出去。整条链路上没有任何一处「等全部攒好」，首字节延迟被压到「第一个事件产生」为止。

media_type="text/event-stream" 是 SSE 的约定类型，浏览器见到它就知道这是事件流，EventSource 会自动解析。顺带说明一个实现细节：这里 gen 是个同步生成器，Starlette 会用线程池把它包起来异步消费（iterate_in_threadpool），所以它不会阻塞事件循环。要做成纯异步，把 gen 写成 async def + async for 即可——第 14 章的 asyncio 基础已经够用，本节先用同步版保持简单。

17.2.7 真跑：起服务，curl 打通两端点

代码写完了，真起一个服务验证。用 uvicorn 拉起 service.py 里的 app，绑到高位端口 127.0.0.1:8123 避免和别的服务打架：

python -m uvicorn minagent.service:app --host 127.0.0.1 --port 8123

启动日志（真实输出）：

INFO:     Started server process [6872]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8123 (Press CTRL+C to quit)

先打 /run，一次性拿结果：

curl -s -X POST http://127.0.0.1:8123/run \
  -H 'Content-Type: application/json' \
  -d '{"input":"2+3 等于多少？"}'

真实响应：

{"final":"2 + 3 = 5。","steps":[{"index":0,"kind":"tool_call","detail":"add({'a': 2, 'b': 3}) -> 5"},{"index":1,"kind":"final","detail":"2 + 3 = 5。"}]}

final 是最终答复，steps 是轨迹——一次工具调用加一次收敛，和 17.1 里本地调用拿到的 RunResult 完全一致。服务化没有改变语义，只是换了个传输通道。

再打 /stream。这里 curl 要加 -N（关闭缓冲），否则它会攒够一批才显示，看不到「逐条到达」：

curl -s -N -X POST http://127.0.0.1:8123/stream \
  -H 'Content-Type: application/json' \
  -d '{"input":"2+3 等于多少？"}'

真实的 SSE 事件流（原样）：

event: tool_call
data: {"id": "c1", "name": "add", "arguments": {"a": 2, "b": 3}}

event: tool_result
data: {"id": "c1", "name": "add", "result": "5"}

event: step
data: {"index": 0, "kind": "tool_call"}

event: token
data: {"text": "2"}

event: token
data: {"text": " "}

event: token
data: {"text": "+"}

event: token
data: {"text": " "}

event: token
data: {"text": "3"}

event: token
data: {"text": " "}

event: token
data: {"text": "="}

event: token
data: {"text": " "}

event: token
data: {"text": "5"}

event: token
data: {"text": "。"}

event: step
data: {"index": 1, "kind": "final"}

event: final
data: {"content": "2 + 3 = 5。"}

这条流把 7.1 的整个事件序列原封不动搬到了网络上：先是 tool_call（模型要调 add），接着 tool_result（工具返回 5），一个 step 收尾；第二轮没有工具调用，直接吐了 11 个 token，最后一个 step 和一条 final 收敛。每一帧都以空行结尾，event: 和 data: 成对出现——这正是 17.2.5 那个映射的落地。

对照 /run 与 /stream 会发现：前者是「把整条流折叠成最后一个结果」，后者是「把整条流原样展开」。它们是同一份数据的两种投影，区别只在调用方想要「结果」还是「过程」。请求完，按 Ctrl+C 关掉服务即可——本地验证不需要它常驻。

17.2.8 OpenAPI 自动文档的意义

FastAPI 还白送了一份东西：OpenAPI 文档。启动服务后访问 /docs 是交互式的 Swagger UI，/redoc 是阅读版，/openapi.json 是机器可读的 schema。本节服务的 schema 实测是：

title: minagent service
version: 0.17.0
paths: ['/run', '/stream']
RunRequest props: ['input', 'session_id']

这份文档的价值有三层。第一，它是活的契约。 因为 schema 直接从 RunRequest 的类型标注和路由签名里生成，代码改一处、文档跟着变，永远不会「文档说的和代码做的对不上」。第二，它省掉前后端对齐成本。 前端不用等你写接口文档，直接读 /openapi.json，甚至能自动生成 TS 客户端代码——这正是下一节 17.3 手写客户端时那份「协议」的机器版。第三，它内建了调试入口。 /docs 页面上能直接填参数、点按钮发请求，联调时省掉一堆 curl。

不过要如实说明一个局限：/stream 这个 SSE 端点，OpenAPI 只能描述到「它接受 RunRequest、返回 text/event-stream」，描述不了流里会出现哪些 event: 类型、每种 data: 长什么样。SSE 是流式文本，不在 OpenAPI 的 schema 表达力之内。所以那六种事件的契约，得靠人和文档来维护（比如本节 17.2.5 那张映射表）——这也是为什么 17.3 的 TS 客户端必须手写解析器，而不能像调用普通接口那样靠代码生成。自动文档覆盖得了「请求长什么样」，覆盖不了「流里会有什么」。

17.2.9 真实模型下会怎样

真实模型下，这套服务的形状完全不变：路由还是那两个，SSE 帧还是那些帧，前端还是那样消费。变的只有 build_agent() 里造的是哪个 Model——把 FakeModel 换成第 2 章那个真实供应商适配器（可选）即可，Agent 与 StreamingResponse 一行都不用动。这正是把 Model 抽成协议、把服务层架在 Event 流之上的回报。

但真实场景会给服务层加三层新的复杂度。其一，流会更碎、更慢。 真实 token 一个一个来，中间可能有几百毫秒的间隔；这时需要一个心跳——定期发一个注释帧（:\n\n）或空 data:，防止中间的代理或负载均衡把「长时间无数据」的连接掐掉。其二，错误会从网络来。 客户端可能在流跑到一半时断线，服务端要能感知并停止生成（否则 Agent 还在烧 token 却没人收）；第 14 章的超时与取消正是为此准备的。其三，会话要真的跨请求。 本节每个请求都 build_agent() 造一个全新 Agent，历史不留；真实产品要按 session_id 找回 Session（7.2 的 Session 就是为此而生），把多轮对话接起来。

一句话：服务的骨架（两个端点 + 一条 SSE 映射）与真假模型无关，真实化只是往这个骨架里补心跳、补取消、补会话持久化。骨架先立住，加固是后面的事。

17.2.10 小结

• 服务化松开三条进程内约束：同语言、同机器、同生命周期；代价是网络与「过程如何实时回传」。
• POST /run 一次性返回，POST /stream 用 StreamingResponse 边产边发。
• Event → SSE 的映射就是 event: {type} + data: {json} + 空行；JSON 载荷天然规避了裸换行问题。
• StreamingResponse 接受生成器，接上 agent.stream() 后全程无「攒好再发」。
• OpenAPI 自动生成请求侧契约，但描述不了 SSE 流里的事件形状——那部分要靠手写协议与手写客户端。
• 真实化要补心跳、断线取消、session_id 会话持久化。

到这里，minagent 已经能被任意语言通过 HTTP 调用了。下一节 17.3 TS 落点一：同一协议的 TS 客户端 就用 TypeScript 写一个消费这条 SSE 流的客户端，证明「同一事件协议、两种语言」真的能协作。想回看公开面是怎么钉住的，翻 17.1 库的对外接口 。

阅读导航：上一节：17.1 库的对外接口 · 下一节：17.3 TS 落点一：同一协议的 TS 客户端 。