agent_backend/examples

API Lab · FastAPI 笔记服务

这份代码配合「API 服务开发」模块。用同一个小服务练习请求参数、REST CRUD、错误响应、依赖、异步等待、SSE 与 WebSocket。代码直接运行，不需要模型账户或模型密钥。

这是各课核心功能的参考实现。课后扩展由学生按课件添加，例如第 06 课的 offset 分页、第 13 课的 POST /stream，以及 /private/status。PromptIn、encode_sse() 与 token_events() 已提供，第 13 课可直接复用。代码不预置全部练习答案。

推荐 Python 3.11 或更新版本；本项目在 Windows、Python 3.12.6 下实际验证。依赖版本固定在 requirements.txt。数据库、Redis、任务队列、Docker 与服务器部署留到后续模块。

1. 启动服务

先解压并进入 api-lab 文件夹。后续命令都在这个文件夹执行。

Windows PowerShell：

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt
$env:DEMO_API_KEY = 'classroom-key'
python -m uvicorn main:app --reload

如果 PowerShell 不允许执行激活脚本，不需要修改系统策略，改为直接调用虚拟环境中的 Python：

.\.venv\Scripts\python.exe -m pip install -r requirements.txt
$env:DEMO_API_KEY = 'classroom-key'
.\.venv\Scripts\python.exe -m uvicorn main:app --reload

macOS / Linux：

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
export DEMO_API_KEY='classroom-key'
python -m uvicorn main:app --reload

终端保留运行，打开：

• http://127.0.0.1:8000/health：应返回 {"status":"ok"}。
• http://127.0.0.1:8000/docs：展开端点，点击 Try it out → Execute，观察 Request URL、Response code 和 Response body。
• http://127.0.0.1:8000/openapi.json：查看接口描述。

main:app 表示导入 main.py 中的 app 对象。--reload 在代码改动后重启进程，因此笔记会清空。按 Ctrl+C 停止服务。

2. 跑一遍 CRUD

在 /docs 按顺序执行：

1. POST /notes 提交 {"title":"第一篇笔记"}，得到 201，记录返回的 id 与响应头 Location: /notes/{id}。content 默认空字符串，pinned 默认 false。
2. GET /notes 得到 JSON 数组，默认最多 10 条；填写 q 过滤标题或正文，填写 pinned=true 或 pinned=false 筛选置顶状态，填写 limit 限制返回条数。先筛选再限制条数，不传 pinned 时保留两种状态。
3. GET /notes/{note_id} 用刚才的 ID 读取笔记。
4. PUT /notes/{note_id} 提交 {"title":"修改后的笔记","content":"完整替换","pinned":true}。三个可修改字段都必须出现，少一个得到 422。
5. DELETE /notes/{note_id} 得到 204 和空响应体。此时不要调用前端的 response.json()。
6. 再读同一个 ID，得到 404；标题提交空字符串、纯空白或超过 120 个字符，得到 422。

标题、正文会去掉首尾空白。pinned 是布尔类型，示例使用 JSON 的 true / false；Pydantic 默认还允许部分类型转换，不等同于严格模式。未声明的请求体字段会返回 422。

3. 浏览器联调

打开第二个终端，同样进入 api-lab，激活虚拟环境后执行：

python -m http.server 5500

打开 http://localhost:5500/web-client.html。不要双击 HTML 使用 file://。页面有三个可以直接操作的实验：

• 创建笔记：fetch() 发送 POST JSON，输出状态码与响应体。
• SSE：通过 EventSource 收到三个 token 事件和一个 done 事件；完整文本逐段显示。
• WebSocket：连接后发送文本并收到 echo，发送文本 ping 收到 pong，然后断开并重连。

打开浏览器开发者工具的 Network 面板。POST JSON 的跨域请求会触发 OPTIONS 预检；SSE 在连接内连续收到事件；WebSocket 在 Messages / Frames 中显示双向消息。不同浏览器的面板名称略有差异。

服务仅允许 http://localhost:5500 与 http://127.0.0.1:5500 这两个浏览器来源。域名、协议或端口不同就是不同来源。CORS 是浏览器的跨域读取规则，不能替代身份验证；WebSocket 也不由这段 CORS 中间件提供鉴权。

要复现第 10 课的来源不匹配，先临时把 allow_origins 收窄为只有 http://127.0.0.1:5500，再从 http://localhost:5500 打开前端；实验后恢复两个来源。

4. SSE 的实际内容

先启动 API。在另一终端执行以下命令，观察每个事件约隔 0.4 秒到达：

Windows PowerShell：

curl.exe -N 'http://127.0.0.1:8000/stream?prompt=hello'

macOS / Linux：

curl -N 'http://127.0.0.1:8000/stream?prompt=hello'

协议内容如下。每个事件末尾的空行负责划分事件边界。

event: token
data: {"text": "收到："}

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

event: token
data: {"text": "。处理完成。"}

event: done
data: {}

token 和 done 是本例约定的事件名；每个 token 的内容是演示片段，不代表模型分词器中的 token。data 用 json.dumps() 编码，所以文本里的引号和换行不会破坏 SSE 帧。

对照 POST /generate：它也模拟三个 0.4 秒处理步骤，但等约 1.2 秒后才一次返回完整 JSON。SSE 每完成一个步骤就发送一次。流正常完成或中途关闭时，token_events() 的 finally 会在服务端终端打印 stream closed。

EventSource 默认使用 GET，不能直接提交 JSON 请求体或任意自定义请求头。本例使用查询参数；需要 POST 请求体时可按课程扩展为 fetch() 与 ReadableStream。页面在收到 done、发生错误、手动停止、再次开始或离开页面时关闭旧连接，避免默认重连再次生成内容。

服务检查客户端是否断开。Cache-Control: no-cache 减少缓存干扰；X-Accel-Buffering: no 用于支持它的代理。真实线上流式到达仍取决于代理与客户端配置，不能只靠这个响应头保证。

5. 依赖与异步等待

GET /private 演示 Depends() 与 X-API-Key 请求头：

• 服务进程未设置 DEMO_API_KEY：返回 503，其余端点仍可使用。
• 设置后，漏传或传错 X-API-Key：返回 401。
• 设置为 classroom-key 后传入同样的请求头：返回 200。

classroom-key 只是课堂示例值。只有 /private 受这个演示依赖保护，/notes、/stream 和 /ws 没有身份验证。

GET /slow?delay=1 使用 await asyncio.sleep() 模拟可等待的 I/O。合法等待时间是 0～5 秒。服务启动后运行：

python concurrency_client.py

脚本同时发出四个 1 秒请求，打印每个响应与总耗时。在本机空闲状态下，总耗时通常接近 1 秒加网络与调度开销。async def 不会自动把同步文件读写、同步网络调用或 CPU 密集计算变为不阻塞的工作。

GET /upstream-demo?delay=1 通过共享的 HTTPX AsyncClient 调用本机 /slow，成功返回 {"upstream":{"delay":1,"status":"done"}}。客户端在 lifespan 中创建，并在应用关闭时释放。连接超时为 1 秒，其余默认超时为 3 秒；delay=4 会触发读取超时，接口返回 504。上游非成功状态或连接失败返回 502。

这个出站实验固定调用 http://127.0.0.1:8000，不访问公网，也不接受用户输入任意 URL。请按启动步骤使用 8000 端口；若修改服务端口，要同时修改 lifespan 中的 base_url。

6. 端点契约

方法与路径	输入	成功响应	练习目标
GET /health	无	200 {"status":"ok"}	路由与启动
GET /notes	可选 q、pinned；limit 默认 10，范围 1～100	200 笔记数组	查询参数
POST /notes	title；可选 content、pinned	201 笔记对象；Location 指向新资源	请求体与默认值
GET /notes/{note_id}	正整数 ID	200 笔记对象	路径参数、404
PUT /notes/{note_id}	title、content、pinned 都必填	200 笔记对象	完整替换
DELETE /notes/{note_id}	正整数 ID	204 空响应体	删除语义
GET /private	请求头 X-API-Key	200 {"message":"API key accepted"}	依赖与环境变量
GET /slow	delay 默认 1，范围 0～5	200 {"delay":1,"status":"done"}	异步等待
GET /upstream-demo	delay 默认 1，范围 0～5	200 {"upstream":{"delay":1,"status":"done"}}	HTTPX 生命周期、502/504
POST /generate	{"prompt":"hello"}，1～200 字符	200 {"text":"收到：hello。处理完成。"}	一次性响应
GET /stream	prompt 默认 hello，1～200 字符	200 text/event-stream	有限 SSE 流
WS /ws	文本；文本 ping 是特殊约定	echo 或 pong JSON 消息	同一连接上的双向通信

HTTP 异常与请求校验错误统一为：

{"error":{"code":"http_error","message":"Note not found","details":[]}}

校验失败时，code 为 validation_error，details 包含 loc、msg、type。这个教学处理器覆盖 HTTPException 与 RequestValidationError；未处理的程序异常、CORS 预检拒绝和 WebSocket 消息不属于该 JSON 契约。生产环境的日志、鉴权与错误追踪在后续模块补充。

7. 验收

python -m pytest -q
python -m pip check

测试覆盖 CRUD 全流程与 Location、PUT 必填字段、404/422、标题限制、q/pinned 组合过滤与默认 limit、环境变量与请求头、SSE 帧边界、中文与换行内容、WebSocket echo/ping/重连、CORS 预检和健康端点。HTTPX 上游成功、超时、HTTP 错误、连接失败使用 MockTransport 验证，不访问公网；同时检查 lifespan 关闭客户端。TestClient 的 SSE 测试检查协议格式与完整内容，不用于判断真实网络的首段到达时间。

实际验证结果记录在 VERIFICATION.md。示例的测试会重置内存笔记，适合独立测试进程，不要将测试代码作为线上操作入口。

8. 文件与使用边界

• main.py：核心功能参考实现，保留在一个文件中便于逐节查找。课程可进一步练习拆分 models、routers 和 dependencies。
• web-client.html：不需要构建工具的本地浏览器客户端。
• concurrency_client.py：四个并发请求的实测脚本。
• test_main.py：可执行验收。
• requirements.txt：已验证的固定依赖。

所有笔记保存在当前 Python 进程的内存中，重启清空、多个 worker 彼此不共享。请按示例运行单个开发进程。这里没有用户隔离、持久化、线上密钥管理、限流或任务队列；课程后续再接入数据库及部署能力。

官方参考

• FastAPI 请求体与模型：https://fastapi.tiangolo.com/tutorial/body/
• FastAPI 错误处理：https://fastapi.tiangolo.com/tutorial/handling-errors/
• FastAPI 自定义与流式响应：https://fastapi.tiangolo.com/advanced/custom-response/
• FastAPI WebSocket：https://fastapi.tiangolo.com/advanced/websockets/
• FastAPI CORS：https://fastapi.tiangolo.com/tutorial/cors/
• FastAPI 测试：https://fastapi.tiangolo.com/tutorial/testing/