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 /products 提交 {"name":"陶瓷杯"}，得到 201，记录返回的 id 与响应头 Location: /products/{id}。description 默认空字符串，is_active 默认 false，price_cents 默认 0（免费教学样品）。
2. GET /products 得到 JSON 数组，默认最多 10 条；填写 q 过滤名称或描述，填写 is_active=true 或 is_active=false 筛选上架状态，填写 limit 限制返回条数。先筛选再限制条数，不传 is_active 时保留两种状态。
3. GET /products/{product_id} 用刚才的 ID 读取商品。
4. PUT /products/{product_id} 提交 {"name":"蓝色陶瓷杯","description":"350ml","is_active":true,"price_cents":1990}。四个可修改字段都必须出现，少一个得到 422。
5. DELETE /products/{product_id} 得到 204 和空响应体。此时不要调用前端的 response.json()。
6. 再读同一个 ID，得到 404；名称提交空字符串、纯空白或超过 120 个字符，得到 422。

名称、描述会去掉首尾空白；描述最多 16000 字符，与后续数据库 TEXT 的字符预算一致。is_active 是布尔类型，示例使用 JSON 的 true / false；Pydantic 默认还允许部分类型转换，不等同于严格模式。价格 price_cents 则明确使用严格整数分：1990 表示 19.90 元，合法范围为 0～2147483647，与后续 MySQL INT 一致；小数、字符串、布尔值和负数都会返回 422。未声明的请求体字段会返回 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 的实际内容

本实验模拟商品清单导出进度，只传输进度文本，不生成实际文件。prompt 是清单名称；/generate 和 /stream 分别演示一次返回和逐段返回。WebSocket 则只回显商品咨询文本，不实现客服回复逻辑。

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

Windows PowerShell：

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

macOS / Linux：

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

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

event: token
data: {"text": "准备导出商品清单："}

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

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 受这个演示依赖保护，/products、/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 /products	可选 q、is_active；limit 默认 10，范围 1～100	200 商品数组	查询参数
POST /products	name；可选 description、is_active、price_cents	201 商品对象；Location 指向新资源	请求体与默认值
GET /products/{product_id}	正整数 ID	200 商品对象	路径参数、404
PUT /products/{product_id}	name、description、is_active、price_cents 都必填	200 商品对象	完整替换
DELETE /products/{product_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":"catalog"}，1～200 字符	200 {"text":"准备导出商品清单：catalog。导出进度演示完成。"}	一次性响应
GET /stream	prompt 默认 catalog，1～200 字符	200 text/event-stream	有限 SSE 流
WS /ws	文本；文本 ping 是特殊约定	echo 或 pong JSON 消息	同一连接上的双向通信

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

{"error":{"code":"http_error","message":"Product 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/is_active 组合过滤与默认 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：沿用的基线依赖锁；本次实际验收环境见 VERIFICATION.md。

所有商品保存在当前 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/