第 12 期：FastAPI 包一层——流式、会话、最简鉴权

Part 1 十一期都是命令行：一次 uv run python -m ep11.main ...，进程跑完退出。给别人用的服务不能这样——别人不会开终端敲命令，他们要的是一个一直开着、随时能发消息进来的服务。这一期把第 11 期那个客服 agent 包进 FastAPI，补三样命令行版本没有、服务必须有的能力：流式（前端不用等一整轮跑完才看到反应）、会话（服务常驻，图和连接只建一次，不是每个请求重开一次）、 最简鉴权（挡住随便什么人都能调）。

敲进去

第 12 期的代码在 code/ep12/，graph.py/tools.py/state.py 跟第 11 期一字不改，新增一个 app.py。

会话：常驻进程，只建一次连接

CLI 版本每次调用都是一个新进程：开数据库文件、连 MCP 服务器、编译图，用完就退出。服务不是这样——用 FastAPI 的 lifespan，这些事只在进程启动时做一次，编译好的图存在 app.state 上，每个请求来了直接复用：

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with AsyncSqliteSaver.from_conn_string(str(CHECKPOINT_DB)) as saver, \
            AsyncSqliteStore.from_conn_string(str(MEMORY_DB)) as store:
        await store.setup()
        mcp_tools = await load_mcp_tools()
        app.state.graph = build_graph(saver, store, TOOLS + mcp_tools)
        yield


app = FastAPI(lifespan=lifespan)

从 CLI 脚本换成常驻服务，这是最容易漏掉的一步——第 7 期连 MCP 服务器要起子进程、握手，那一套代价一次性的，摊到进程的整个生命周期里；如果每个请求都重新走一遍，一个客服 agent 服务会话开得越多，启动那几秒的延迟就被越多用户看在眼里。

最简鉴权：一把共享密钥

bearer = HTTPBearer()

def check_auth(creds: HTTPAuthorizationCredentials = Depends(bearer)) -> None:
    if creds.credentials != settings.API_KEY:
        raise HTTPException(status_code=401, detail="密钥不对")

API_KEY 是这个服务自己的密钥，跟 MODEL_API_KEY（打给模型端点的那把）是两回事——一把管“谁能调这个服务“，一把管“这个服务拿什么身份去调模型“，第 11 期就已经分清楚的“这是两把钥匙“的道理，这一期多了一把。Depends(check_auth) 挂在每个业务路由上，/health 不挂，运维探活不需要密钥。

流式：把内部结构翻译成对外的几种事件

def _sse(event: dict) -> str:
    return f"data: {json.dumps(event, ensure_ascii=False)}\n\n"


async def sse_events(graph, run_input, config: dict) -> AsyncIterator[str]:
    async for update in graph.astream(run_input, config=config, stream_mode="updates"):
        if "__interrupt__" in update:
            (info,) = update["__interrupt__"]
            yield _sse({"type": "interrupt", "payload": info.value})
            continue
        for node, changed in update.items():
            parts = changed if isinstance(changed, list) else [changed]
            for part in parts:
                for msg in part.get("messages", []):
                    if node == "agent" and msg.tool_calls:
                        for call in msg.tool_calls:
                            yield _sse({"type": "tool_call", "name": call["name"], "args": call["args"]})
                    elif node == "agent":
                        yield _sse({"type": "answer", "content": msg.content})
    yield _sse({"type": "done"})

stream_mode="updates" 从第 3 期起就在用，CLI 版本拿它来 print；这里换成拿它来拼 SSE（Server-Sent Events）事件。翻译这一步是故意的：update 的原始结构（节点名、Command 有没有被用、messages 键的形状）是图内部的实现细节，第 9 期那次因为一个工具返回 Command就让消费端的形状假设崩掉的教训还在——对外只暴露 tool_call/answer/interrupt/done 四种事件，图内部随便怎么重构，这份对外协议不用跟着变。

两个路由，一个正常问、一个走审批之后续着问：

@app.post("/chat", dependencies=[Depends(check_auth)])
async def chat(req: ChatRequest, request: Request) -> StreamingResponse:
    graph = request.app.state.graph
    config = build_run_config(req.thread_id, req.user_id)
    state = {"messages": [HumanMessage(req.message)]}
    return StreamingResponse(sse_events(graph, state, config), media_type="text/event-stream")


@app.post("/chat/resume", dependencies=[Depends(check_auth)])
async def resume(req: ResumeRequest, request: Request) -> StreamingResponse:
    graph = request.app.state.graph
    config = build_run_config(req.thread_id, req.user_id)
    return StreamingResponse(
        sse_events(graph, Command(resume=req.decision), config), media_type="text/event-stream"
    )

跑起来

cd code
export MODEL_BASE_URL=https://api.deepseek.com/v1   # 或你的网关
export MODEL_API_KEY=sk-xxxx
export MODEL_NAME=deepseek-v4-flash
export API_KEY=sk-test-123
uv run uvicorn ep12.app:app --port 8000

另开一个终端：

curl -N -X POST http://localhost:8000/chat \
  -H "Authorization: Bearer sk-test-123" -H "Content-Type: application/json" \
  -d '{"user_id":"wang","thread_id":"api1","message":"订单 KL-778 能不能改期"}'

你应该看到什么

没带密钥，直接拒绝

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:8000/chat \
    -H "Content-Type: application/json" -d '{"user_id":"wang","thread_id":"api1","message":"hi"}'
401

密钥错的也是 401，不区分“没带“和“带错了“——都不该告诉调用方到底差在哪。

流式：工具调用先到，答案后到

$ curl -N -X POST http://localhost:8000/chat \
    -H "Authorization: Bearer sk-test-123" -H "Content-Type: application/json" \
    -d '{"user_id":"wang","thread_id":"api1","message":"订单 KL-778 能不能改期"}'
data: {"type": "tool_call", "name": "get_order", "args": {"order_id": "KL-778"}}

data: {"type": "tool_call", "name": "get_policy", "args": {"product_id": "SKU-1001", "topic": "reschedule"}}

data: {"type": "tool_call", "name": "get_current_time", "args": {"timezone": "Asia/Shanghai"}}

data: {"type": "answer", "content": "可以改期。您的东京迪士尼一日票（KL-778）出行日是 9月7日，距出行还有4天，属于「出行日前3天可免费改期一次」的范围。需要我帮您办理改期吗？"}

data: {"type": "done"}

这次模型一口气并行发了三个工具调用（get_order、get_policy、 get_current_time），SSE 把它们按到达顺序原样吐出去——前端能在答案出来之前，先给用户看“正在查订单、正在查政策“这类进度提示，这是流式相对于命令行版本“整轮跑完一次性打印“的真实差别。

会话：第二次请求记得第一次问的是什么

$ curl -N -X POST http://localhost:8000/chat \
    -H "Authorization: Bearer sk-test-123" -H "Content-Type: application/json" \
    -d '{"user_id":"wang","thread_id":"api1","message":"刚才说的是哪个订单来着？"}'
data: {"type": "answer", "content": "您刚才问的是订单 **KL-778**（东京迪士尼一日票，出行日期 2026-09-07，2 张）。这个订单可以免费改期一次，需要帮您操作吗？"}

data: {"type": "done"}

同一个 thread_id，两次独立的 HTTP 请求，AsyncSqliteSaver 让第二次请求接得上第一次的历史——服务重启也不丢，checkpointer 落的是磁盘文件，不是进程内存。

需要审批的操作：先拿到 interrupt 事件，批准后再续上

$ curl -N -X POST http://localhost:8000/chat \
    -H "Authorization: Bearer sk-test-123" -H "Content-Type: application/json" \
    -d '{"user_id":"chen","thread_id":"api2","message":"帮我取消订单 KL-901"}'
data: {"type": "tool_call", "name": "cancel_order", "args": {"order_id": "KL-901"}}

data: {"type": "interrupt", "payload": {"action": "cancel_order", "order_id": "KL-901", "customer": "陈先生", "product": "东京迪士尼一日票"}}

data: {"type": "done"}

$ curl -N -X POST http://localhost:8000/chat/resume \
    -H "Authorization: Bearer sk-test-123" -H "Content-Type: application/json" \
    -d '{"user_id":"chen","thread_id":"api2","decision":"approve"}'
data: {"type": "answer", "content": "订单 KL-901 已提交取消，正在走人工审批流程。审批通过后订单才会正式取消，如被拒绝则订单不受影响。请问还有什么可以帮您？"}

data: {"type": "done"}

前端看到 interrupt 事件就该停下来，弹一个“批准/拒绝“的界面，而不是傻等下一个事件——这次调用不会再有下一个事件了，直到有人调 /chat/resume。

两个客人同时问，互不干扰

$ (curl -sN ... -d '{"user_id":"li","thread_id":"api3", ...}' &
   curl -sN ... -d '{"user_id":"wang","thread_id":"api4", ...}' &
   wait)
data: {"type": "tool_call", "name": "get_order", "args": {"order_id": "KL-778"}}
data: {"type": "tool_call", "name": "get_order", "args": {"order_id": "KL-315"}}
data: {"type": "answer", "content": "订单 KL-778 的出行日期是 2026-09-07 ..."}
data: {"type": "done"}
data: {"type": "tool_call", "name": "get_policy", ...}
data: {"type": "answer", "content": "订单 KL-315（首尔往返机场大巴票 ...）可以改期 ..."}
data: {"type": "done"}

两条请求的 tool_call 事件交错到达（一条的第一个事件，紧跟着是另一条的第一个事件），证明它们真的在并发处理，不是排队一个处理完再处理下一个—— thread_id 不同，AsyncSqliteSaver 天然按 thread 隔离，同一个图实例服务两个互不知道对方存在的客人。

发生了什么

从 CLI 到服务，图和工具一行没改，改的是“谁来管连接的生命周期“。 build_graph、TOOLS、load_mcp_tools 全部从第 11 期原样搬过来。 CLI 版本里连接的生命周期跟一次调用绑定（async with ... as saver 包住整个 main_async）；服务版本里连接的生命周期跟进程绑定（包住整个 lifespan）。这不是重构，是同一套资源管理模式换了一个绑定的时间尺度。

对外事件协议，是故意跟图的内部结构脱钩的一层。 tool_call/ answer/interrupt/done 这四种事件名，前端只需要认这四个，不需要知道背后是哪个节点、Command 用没用上。第 9 期那次 stream_mode= "updates" 形状崩过一次消费端代码的教训，这一期用一层专门的翻译函数正面回应：内部结构可以继续因为加新工具、加新节点而变，只要 sse_events 这层翻译跟着更新，对外协议不用跟着抖一下。

最简鉴权挡的是“完全没有门槛“，不是真正的多租户隔离。 一把共享密钥，能调用的人共享同一个 API_KEY，服务分不出是谁在调——这一期的定位是“补上从零到一“，不是“补到生产级“。真实生产要的是每个调用方一把独立的 key、能单独吊销、能分别记账——这本书没有做这一步，第 14 期上线时用的仍是这一把共享密钥，那一期的常见问题里说了为什么以及该换成什么。

流式不是“更快“，是“更早开始显示“。 拿到最终答案的总时间没有变， get_order/get_policy/get_current_time 该等的网络往返一次没少；流式改变的是用户从“发出请求“到“屏幕上第一次出现反应“之间那段空白—— CLI 版本这段空白是一整轮的时长，服务版本这段空白是第一个工具调用返回的时长。

常见问题

为什么鉴权失败统一返回 401，不区分密钥缺失和密钥错误？ 区分了等于告诉一个乱猜密钥的人“你现在的问题是没带，不是猜错“，反而给了信息。统一成一种失败，是最简单也最安全的做法。

/health 为什么不用鉴权？ 探活是给负载均衡器、容器编排系统用的，它们不该知道业务密钥；/health 只回答“进程活着“，不碰任何业务数据，暴露它的代价接近于零。

SSE 和 WebSocket 该选哪个？ 这一期的场景是“客户端问一句、服务端流式答一句“，单向流最省事的选择是 SSE，浏览器原生 EventSource 就能消费，不用维护一个双向连接的状态机。要做“服务端主动推消息“（比如定时唤醒，对应上册练习 26）才用得上 WebSocket 那一套，这本书暂时不需要。

并发请求会不会把 SQLite 写坏？ 不会，aiosqlite 内部对写操作排了队；thread_id 不同的两条请求各自只碰自己那一份历史，即使排队也几乎感觉不到——上一节两个客人同时问的真机结果就是证据。真到高并发的量级，第 13 期换成 Postgres 之后会有真正的连接池。

加分练习

给 /chat 加一个请求超时（比如 60 秒没出 done 事件就断开），想清楚断开之后 checkpoint 里会停在哪个状态，下一次请求能不能接得上。
把 API_KEY 换成“多个 key 各自映射一个 user_id“的映射表，调用方不用在请求体里传 user_id，从鉴权那一步直接确定身份—— 想一想这跟现在“谁都能在请求体里填任何 user_id“比，安全边界差在哪。
用 httpx.AsyncClient 写一个小压测脚本，同时开 10 个不同 thread_id 的会话，量一下服务端处理这些并发请求时 CPU/内存的真实开销，跟“一个 CLI 进程只服务一个人“比一比。
给 interrupt 事件加一个超时——如果十分钟没人调 /chat/resume，自动按拒绝处理。想清楚这条规则该写在哪一层：app.py 里，还是图本身。

Keyboard shortcuts

笨办法学 Agent · 用 LangGraph 上线