Claude Code 第五课：挂上 FastAPI，让 Agent 变成团队服务

生命不息，折腾不止。第四课你的 Agent 已经是「程序里的一个函数」了，这一课把它推到公网后面：一个 POST 进来，token 一个个流回去，同事打开网页就能用。

第四课我们用 claude-agent-sdk 把 Claude Code 塞进了 Python 函数里，run_agent() 一调，它就帮你读文件、改代码、跑命令。但那个函数只能你自己在自己机器上跑——想给团队用，总不能让每个人 ssh 进来敲命令吧？

这一课我们干的活很实在：用 FastAPI 把这个函数包成一个 HTTP 服务，支持流式输出（SSE，边生成边往浏览器推）、并发闸门（别把机器跑爆）、多轮会话（同一个任务接着聊），再配上 Nginx 和 systemd，正正经经上线。看完你手里就有一个「团队共用的 Agent 后端」了。

一、先想清楚：为什么不能拿 `claude -p` 硬凑

有人会说，无头模式不就能用了吗？claude -p "..." 塞进 subprocess，读 stdout 不就完事。真要这么干，很快会撞三堵墙：

没有结构。stdout 里是给人看的文本，你想区分「这是正文」「这是它正在调工具」「这是最后结果」，只能靠正则猜；
流式是假的。子进程的输出是攒一堆才给你，curl 在那一动不动等三十秒，用户以为你服务挂了；
没法管。谁在调、跑了多久、花了多少钱、同时跑了几个人——全都不知道，机器一卡就只能重启。

而这三点，claude-agent-sdk 恰好都给了：消息是类型化的对象（AssistantMessage、StreamEvent、ResultMessage），流式有 include_partial_messages 开关，每次执行结束 ResultMessage 里带着轮数、耗时、花费。我们要做的只是把这堆对象翻译成 HTTP 上跑得动的东西。

二、架构：五件套，缺一味就出问题

先把全局摆出来，省得后面代码看着散：

组件	干什么	少它会怎样
FastAPI	收请求、发响应	没有入口
`StreamingResponse` + SSE	把 token 实时推给前端	用户干等，以为卡死
`asyncio.Semaphore`	限制同时跑几个 Agent	几个人同时点，机器 OOM
`max_turns` + 超时	熔断，防死循环	一个 bug 烧掉一天额度
`session_id` + `resume`	多轮续聊	每句话都从零开始

有个认知必须先纠正：SDK 的一次会话，背后是一个真实的 Claude Code 子进程，不是纯协程。它要占内存、要跑 node、要读磁盘上的会话文件。所以「并发数」不是随便写个 50，得按你机器的真实家底来定——这个后面细说。

三、最小可用版：一个文件跑通流式接口

先装依赖（Python 要 3.10 以上，SDK 本身是 0.2 系列，写作时最新 0.2.174，版本以 PyPI 为准）：

1	pip install "claude-agent-sdk" fastapi "uvicorn[standard]"

然后 app.py，这段是完整可跑的：

import asyncio
import json
import os

from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, ToolUseBlock, ResultMessage,
)
from claude_agent_sdk.types import StreamEvent

app = FastAPI()

# 同时最多几个 Agent 在跑，按机器内存定
SEM = asyncio.Semaphore(int(os.getenv("AGENT_CONCURRENCY", "3")))
API_TOKEN = os.getenv("AGENT_TOKEN", "")


class AskReq(BaseModel):
    prompt: str
    session_id: str | None = None


def sse(obj: dict) -> str:
    """SSE 的线格式：data: <一行 JSON>，再加一个空行。"""
    return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"


async def run_agent(req: AskReq):
    options = ClaudeAgentOptions(
        system_prompt="你是团队内部的代码助手，回答用中文，先给结论再给细节。",
        allowed_tools=["Read", "Grep", "Glob"],       # 自动放行这几个
        disallowed_tools=["Bash", "Write", "Edit"],   # 明确禁掉危险的
        max_turns=8,                                  # 硬熔断
        cwd="/srv/agent-workspace",                   # 只在这个目录里活动
        include_partial_messages=True,                # 打开 token 级流式
        resume=req.session_id,                        # None 就是新会话
    )

    async for msg in query(prompt=req.prompt, options=options):
        # 1) token 级增量：正文一个字一个字往外冒
        if isinstance(msg, StreamEvent):
            ev = msg.event
            if ev.get("type") == "content_block_delta":
                delta = ev.get("delta", {})
                if delta.get("type") == "text_delta":
                    yield sse({"type": "text", "text": delta.get("text", "")})

        # 2) 完整块：能看到它在调哪个工具
        elif isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, ToolUseBlock):
                    yield sse({"type": "tool", "name": block.name})

        # 3) 一轮结束：带上会话号和花费
        elif isinstance(msg, ResultMessage):
            yield sse({
                "type": "done",
                "session_id": getattr(msg, "session_id", None),
                "cost_usd": getattr(msg, "total_cost_usd", None),
            })


@app.post("/agent/stream")
async def agent_stream(req: AskReq, authorization: str = Header(default="")):
    if API_TOKEN and authorization != f"Bearer {API_TOKEN}":
        raise HTTPException(401, "token 不对")

    # 排队最多等 5 秒，等不到就明确告诉调用方「忙」
    try:
        await asyncio.wait_for(SEM.acquire(), timeout=5)
    except asyncio.TimeoutError:
        raise HTTPException(503, "Agent 全在忙，稍后再试")

    async def gen():
        try:
            async for chunk in run_agent(req):
                yield chunk
        except Exception as e:                                  # 兜底，别让连接裸断
            yield sse({"type": "error", "message": str(e)})
        finally:
            SEM.release()

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",   # 告诉 Nginx 别缓冲，关键！
        },
    )


@app.get("/health")
async def health():
    return {"ok": True}

启动：

ANTHROPIC_BASE_URL="https://ai.aklibk.com" \
ANTHROPIC_AUTH_TOKEN="sk-xxxx" \
AGENT_TOKEN="team-secret" \
uvicorn app:app --host 127.0.0.1 --port 8000

测试别用普通 curl，要加 -N 关掉 curl 自己的缓冲，不然你看不出流式效果：

curl -N -X POST http://127.0.0.1:8000/agent/stream \
  -H "Authorization: Bearer team-secret" \
  -H "Content-Type: application/json" \
  -d '{"prompt":"读一下当前目录的 README，用三句话总结"}'

看到 data: {"type":"text","text":"这"} 这种一行行滚出来，就成了。

几个容易看漏的点：

StreamEvent 是原始 API 事件，不是拼好的文本。所以必须先判 message.event.get("type") == "content_block_delta"，再判 delta.get("type") == "text_delta"，两层判断一个都不能省；
开启流式后，完整块仍然会来。顺序是：一堆 StreamEvent → 一个 AssistantMessage（这块内容完整了）→ 下一块。所以别把两边都往正文里灌，否则前端会看到重复文字。上面代码里我只在流式事件里拼正文，AssistantMessage 只用来报工具名，就是这个原因；
X-Accel-Buffering: no 是给 Nginx 看的，没有它，前端会等到全部生成完才「啪」一下出现全文，白折腾。

四、并发、超时、熔断：生产环境的三道闸

本地一个人用，怎么写都行；挂到团队后面，这三件事不做，第一天就出事。

4.1 并发：Semaphore 只是第一层

asyncio.Semaphore(3) 的意思是「同时最多 3 个请求在真跑，第 4 个排队」。这个数字怎么定？跑起来之后 ps aux | grep claude 看一眼真实内存占用，再拿机器内存除以它，留三成余量。别拍脑袋写 20。

还有个更隐蔽的坑：uvicorn --workers 4 会起 4 个进程，每个进程一个独立的 Semaphore，你以为限了 3，实际能跑 12 个。所以要么保持 --workers 1（Agent 场景足够，瓶颈在模型不在 Web 层），要么把计数挪到 Redis 做全局锁。这一条踩过的人不少。

4.2 超时：卡总时长，别对生成器套 `wait_for`

max_turns 管的是「轮数」，可一轮里模型自己思考多久它不管。想卡总时长，最简单的做法是在生成器里记开始时间：

import time

async def gen():
    start = time.monotonic()
    try:
        async for chunk in run_agent(req):
            if time.monotonic() - start > 600:      # 10 分钟封顶
                yield sse({"type": "error", "message": "超时，已中断"})
                break
            yield chunk
    finally:
        SEM.release()

为什么不直接 asyncio.wait_for(run_agent(...), 600)？因为对异步生成器套超时，取消会打在 anyio 的取消作用域上，清理顺序一乱就容易出「Task was destroyed but it is pending」这类鬼故事。用时间戳判断，最土但最稳。

4.3 熔断：`max_turns` 和 `disallowed_tools` 要一起上

max_turns=8：转 8 轮还没完就强制收工，防 self-loop；
disallowed_tools=["Bash", "Write", "Edit"]：这才是真正禁工具的开关。很多人以为 allowed_tools 是白名单，没列的就用不了——不是。按官方说法，allowed_tools 只是「自动批准」清单，没列进去的工具依然可用，只是会走权限询问；真要禁，得用 disallowed_tools。
公网服务绝对不要开 bypassPermissions。那等于把 shell 递给陌生人，配上 Bash 工具就是一台肉鸡。

五、多轮会话：`session_id` + `resume`

流式跑通后，第二个需求马上来：「能不能接着上一句聊？」

可以，但别想着在服务里存 client 对象——重启就没了。正确姿势是把会话号交给客户端保管：

第一次请求，session_id 传空，SDK 开一个新会话；
结束时我们已经在 done 事件里回了 session_id；
下一次请求把它带上，服务端 resume=req.session_id，上下文就接上了。

想「在原来的基础上试另一条路」而不污染原会话，用 fork_session=True：

# 接着原会话继续（默认行为）
options = ClaudeAgentOptions(resume=session_id)
# 复制一份出来另开分支，原会话不受影响
options = ClaudeAgentOptions(resume=session_id, fork_session=True)

这里有个上线才会发现的坑：会话数据是存在本机磁盘上的（Claude Code 自己的会话目录）。你要是把服务部署成两个副本做负载均衡，请求 A 打到副本一、请求 B 打到副本二，resume 直接报找不到会话。要么把会话目录挂成共享存储，要么按 session_id 做一致性哈希，要么老老实实单副本。没有共享存储就别吹多副本。

六、上线：systemd + Nginx，SSE 重点在「别缓冲」

6.1 systemd 托管

[Unit]
Description=Claude Agent HTTP API
After=network.target

[Service]
WorkingDirectory=/srv/agent-api
Environment="ANTHROPIC_BASE_URL=https://ai.aklibk.com"
Environment="ANTHROPIC_AUTH_TOKEN=sk-xxxx"
Environment="AGENT_TOKEN=team-secret"
Environment="AGENT_CONCURRENCY=3"
ExecStart=/srv/agent-api/.venv/bin/uvicorn app:app --host 127.0.0.1 --port 8000 --workers 1
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target

systemctl daemon-reload && systemctl enable --now agent-api，然后 journalctl -u agent-api -f 盯着日志。

6.2 Nginx 反代（这一段是成败关键）

location /agent/ {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;

    proxy_buffering off;          # 不缓冲，来一点发一点
    proxy_cache off;
    proxy_read_timeout 3600s;     # 长任务别 60 秒被掐
    chunked_transfer_encoding on;
}

proxy_buffering on 是 Nginx 的默认值，它会攒够一个 buffer 再往下发。SSE 场景下这就等于把流式废掉——很多人「本地流式、线上一坨」，九成是这个原因。改完记得 nginx -t && systemctl reload nginx。

6.3 接哪家模型：只认两个环境变量

注意上面 unit 文件里那两行 ANTHROPIC_*。SDK 驱动的是 Claude Code，所以接哪个模型跟 CLI 完全一样，靠环境变量，代码一行不用改。

Claude 这类国外模型，国内直连官方要处理网络和绑卡，走中转站 ai.aklibk.com 就是另一条路：国内直连、免绑卡、人民币按量付费，把 ANTHROPIC_BASE_URL 和 ANTHROPIC_AUTH_TOKEN 一设，SDK 内部那个子进程就会把请求打过去，剩下的体验跟官方一致。团队内部用，这套最省事。

七、前端怎么接：`fetch` 读流，别用 `EventSource`

这里有个新手必踩的坑：浏览器的 EventSource 只支持 GET，而我们要 POST 一个 prompt 过去，所以用它没戏。正确做法是 fetch + ReadableStream 手动切 SSE：

const res = await fetch("/agent/stream", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "Authorization": "Bearer team-secret",
  },
  body: JSON.stringify({ prompt: "帮我看看这个报错", session_id: lastSessionId }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = "";

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  buf += decoder.decode(value, { stream: true });

  // SSE 以空行分隔事件，最后一段可能不完整，留在 buf 里
  const parts = buf.split("\n\n");
  buf = parts.pop();
  for (const part of parts) {
    if (!part.startsWith("data: ")) continue;
    const ev = JSON.parse(part.slice(6));
    if (ev.type === "text") {
      out.textContent += ev.text;           // 一个字一个字长出来
    } else if (ev.type === "tool") {
      status.textContent = `正在用 ${ev.name}…`;
    } else if (ev.type === "done") {
      lastSessionId = ev.session_id;
      status.textContent = `完成，花费 $${ev.cost_usd ?? "?"}`;
    } else if (ev.type === "error") {
      status.textContent = `出错：${ev.message}`;
    }
  }
}

buf.split("\n\n") 然后 pop() 这一手是精髓：网络分片可能刚好把一个事件切成两半，剩下的半截必须留到下一轮，不然 JSON.parse 随机报错。

八、避坑清单

按踩坑概率从高到低排：

线上不流式：99% 是 Nginx proxy_buffering 没关，或者没设 X-Accel-Buffering: no；
allowed_tools 当成禁用名单：它只是免确认清单，要禁用必须 disallowed_tools；
--workers 多开把并发限流失效：每个进程一个 Semaphore，数字乘起来才是真实并发；
正文重复两遍：StreamEvent 和 AssistantMessage 都往正文里写，实际是同一段内容；
公网开 bypassPermissions + Bash：等于把服务器送人，务必用 disallowed_tools 兜底；
receive_response() 里 break：多轮客户端提前跳出会留清理问题，用标志位让它自然走完；
多副本 + 本地会话目录：resume 找不到会话，先确认会话目录是不是共享的；
忘了 cwd：不指定工作目录，Agent 会在服务进程的当前目录里翻，权限和路径都可能不对；
用老版本 SDK：0.2.164 之前的版本，流被切断时可能不发 message_stop，前端会永远「加载中」，升级能解决。

九、小结

到这一课，你的 Claude Code 技能树是这样长起来的：

第一课：装好、跑通、会跟它对话；
第二课：学会拆 Subagents，让一队并行干活；
第三课：无头模式，claude -p 进 CI，无人值守；
第四课：Agent SDK，把 Agent 变成一个 Python 函数；
第五课：FastAPI + SSE + 并发闸门 + 会话恢复，把它变成一个团队能用的服务。

从「我一个人在终端里敲」到「同事打开网页就能用」，中间隔的不是什么高深技术，就是流式怎么推、并发怎么限、会话怎么续这三件脏活。做完这三件，你的 Agent 才算真正上线。

生命不息，折腾不止。第五课先到这儿。下一篇（Claude Code 第六课）咱们给这个服务装「门禁和仪表盘」：用 Hooks 在工具真正执行前拦下危险命令、把每次调用记成可查的审计日志，再按用户统计 token 和花费——让它从「能跑」变成「敢给老板看」。到时候见。