Claude Agent SDK Session Manager 实战：用契约驱动的 Agent Team 工作流构建全栈会话管理应用

Claude Agent SDK Session Manager 实战：用契约驱动的 Agent Team 工作流构建全栈会话管理应用

【免费下载链接】context-engineering-introContext engineering is the new vibe coding - it's the way to actually make AI coding assistants work. Claude Code is the best for this so that's what this repo is centered around, but you can apply this strategy with any AI coding assistant!项目地址: https://gitcode.com/gh_mirrors/co/context-engineering-intro

本文基于本仓库 use-cases/build-with-agent-team/example-plan/session-manager-plan.md 展开。该文档是仓库中「Build with Agent Team」技能（README.md、SKILL.md）配套的示例计划：它演示了如何把一份"要做什么"的产品构想，拆解成一份多个 Agent 可以并行开工、按契约对齐、逐层验收的完整工程计划。读完本文，你将掌握：如何用 FastAPI + SSE 封装 claude-agent-sdk 的流式能力、如何用"双存储 + 文本累积"模式解决会话恢复时的历史消息读取问题、如何把一份计划文档转化为数据库 → 后端 → 前端的契约链，并用 Agent 团队并行落地一套可复现的验证流程。

一、问题背景：为什么需要自建会话存储

计划文档开门见山地指出了要解决的核心痛点：

Claude Agent SDK 不提供拉取历史消息的 API。恢复会话时，Claude 会在内部记住上下文，但你无法以编程方式取回过去的消息用于 UI 展示。

也就是说，resume能力让 Claude 能"接得上话"，但它是一个黑盒——你不能把历史对话拿出来渲染到界面上。因此计划给出的解决方案是：

把消息存进自己的数据库，同时复用 SDK 的session_id做上下文恢复。

这个"双存储"思路是整个 Session Manager 的架构基石：数据库负责"可查询、可展示"的历史，SDK 负责"可续聊"的上下文。它同时引出了本文后续要展开的三大关键技术决策：消息按什么结构落库（数据库 Schema）、后端如何流式转发 SDK 事件（SSE 契约）、以及如何保证重载页面后前端渲染出的气泡数与数据库行数一一对应（文本累积）。

二、技术栈选型

计划对技术栈做了明确约定，各层职责清晰：

选型理由可以从依赖清单反推：sse-starlette用于把 SDK 的异步事件流转为 HTTP Server-Sent Events；aiosqlite提供异步 SQLite 访问，避免阻塞 FastAPI 的事件循环；前端用 shadcn/ui + Tailwind 快速搭出可访问的聊天界面。值得注意的是，计划刻意选择了轻量的 SQLite 而非重型数据库——会话消息的读写模式简单（按 session_id 聚合读写），单机开发/演示场景下 SQLite 完全够用，这符合"以最小成本验证方案"的工程取舍。

三、项目结构：按 Agent 边界切分目录

计划给出的目录结构本身就是"多 Agent 并行"的直接体现——每个目录都对应一个 Agent 的专属所有权：

agent-session-manager/ ├── backend/ │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI app entry │ │ ├── database.py # SQLite connection + queries │ │ ├── models.py # Pydantic models │ │ ├── routes/ │ │ │ ├── sessions.py # Session CRUD │ │ │ └── chat.py # Chat with SSE streaming │ │ └── sdk_client.py # Claude Agent SDK wrapper │ ├── requirements.txt │ └── pyproject.toml ├── frontend/ │ ├── src/ │ │ ├── components/ │ │ │ ├── SessionSidebar.tsx │ │ │ ├── ChatView.tsx │ │ │ ├── MessageList.tsx │ │ │ ├── MessageBlock.tsx │ │ │ ├── ToolUseCard.tsx │ │ │ └── NewSessionDialog.tsx │ │ ├── lib/ │ │ │ ├── api.ts │ │ │ └── types.ts │ │ ├── App.tsx │ │ └── main.tsx │ ├── package.json │ ├── vite.config.ts │ └── tailwind.config.js └── README.md

这种结构与仓库 SKILL.md 中"为每个 Agent 定义 Ownership / Does NOT touch"的原则完全一致：数据库 Agent 只碰database.py与models.py，后端 Agent 只碰routes/与sdk_client.py，前端 Agent 只碰frontend/src/，从文件系统层面就杜绝了并行开发时的文件冲突。

四、Agent 构建顺序：契约优先（Contract-First）

计划为 Agent 团队定义了严格的契约优先时序，这是整个方案能并行而不跑偏的关键。仓库 SKILL.md 第 4 步专门强调：并行 Agent 如果没有事先约定的契约，必然在 endpoint URL、响应结构、尾部斜杠、存储语义上分叉。

Phase 1：数据库 Agent

1. 构建 Schema、CRUD 函数、Pydantic 模型
2. 把函数签名和模型定义发送给 lead
3. Lead 验证后转发给后端 Agent

Phase 2：后端 Agent（在收到数据库契约之后）

1. 构建 FastAPI 应用、路由、SSE 流、SDK 客户端
2. 把完整 API 契约发送给 lead，必须包含：
   • 精确的 endpoint URL（并标注尾部斜杠）
   • 精确的请求/响应 JSON 结构
   • 精确的 SSE 事件格式（含所有事件类型）
   • 成功与错误场景的状态码
3. Lead 验证后转发给前端 Agent

Phase 3：前端 Agent（在收到 API 契约之后）

1. 构建 React 应用、组件、API 客户端，严格对齐已验证的 API 契约
2. 禁止猜测 endpoint URL 或响应结构——一律使用收到的契约

Phase 4：Lead 验证

1. 契约比对（contract diff）——对比后端真实 endpoint 与前端 fetch 调用
2. 启动前后端两个服务
3. 运行 E2E 浏览器测试

这套时序的精髓在于：契约先于实现存在，实现全部并行进行。SKILL.md 中明确列出了两种反面模式：无契约并行派生（各写各的，集成时 URL 与响应结构全对不上）和全串行派生（一个等一个，失去并行意义）。而这里的 Phase 1→2→3 是"契约链"的传递顺序，不是"串行开发"——数据库契约一旦敲定，后端和前端即可同时开工。

五、数据库 Schema：把对话结构化为可查询数据

Sessions 表

CREATE TABLE sessions ( id TEXT PRIMARY KEY, -- Claude SDK session_id title TEXT NOT NULL, system_prompt TEXT, -- Optional custom system prompt working_directory TEXT, model TEXT DEFAULT 'claude-sonnet-4-20250514', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

关键设计点：id直接用Claude SDK 的 session_id作为主键，这样"数据库记录"与"SDK 上下文"天然一一对应；last_accessed用于侧边栏按最近访问排序；model字段允许为不同会话指定不同模型。

Messages 表

CREATE TABLE messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, -- 'user', 'assistant', 'system' content TEXT, message_type TEXT NOT NULL, -- 'text', 'tool_use', 'tool_result', 'thinking' tool_name TEXT, tool_input TEXT, -- JSON tool_output TEXT, -- JSON is_error BOOLEAN DEFAULT FALSE, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); CREATE INDEX idx_messages_session ON messages(session_id);

message_type字段是这套设计的核心：它把一条 SDK 事件流拆成text / tool_use / tool_result / thinking四种可独立渲染的"消息形态"，前端MessageBlock才能按类型渲染气泡、工具卡片、折叠的思考过程。tool_input/tool_output存 JSON 字符串，配合is_error标记工具执行失败状态。外键ON DELETE CASCADE保证删除会话时消息级联清除，idx_messages_session索引支撑按会话聚合查询。

六、API 契约：唯一的集成真相源

计划用醒目的字体声明：这是权威 API 契约，前后端必须严格遵循，lead 在集成前负责核验对齐。这正是 SKILL.md 中"契约质量检查清单"的落地产物——URL 要精确到斜杠、响应结构要给出 JSON 而非散文描述。

Endpoints

注意：POST 和 GET 列表接口使用尾部斜杠（/api/sessions/）；GET by ID、DELETE 和 chat 不使用尾部斜杠。

这条"斜杠约定"看似琐碎，却是多 Agent 并行最容易踩的坑之一——FastAPI 路由trailing_slash的默认行为会导致"带斜杠与不带斜杠"被当成不同路径，前端若写错就会收获 307/404。SKILL.md 把它列为跨切面关注点并要求"双方必须精确匹配"，就是因为它反复成为集成失败的典型原因。

响应结构

SessionResponse:

{"id": "uuid", "title": "string", "system_prompt": "string|null", "working_directory": "string|null", "model": "string", "created_at": "ISO8601", "last_accessed": "ISO8601"}

MessageResponse:

{"id": 1, "session_id": "uuid", "role": "user|assistant", "content": "string|null", "message_type": "text|thinking|tool_use|tool_result", "tool_name": "string|null", "tool_input": "string|null", "tool_output": "string|null", "is_error": false, "timestamp": "ISO8601"}

GET /api/sessions/{id} 返回的是嵌套对象（不是扁平结构）：

{ "session": { SessionResponse }, "messages": [ MessageResponse, ... ] }

前端必须把这个嵌套结构解构为扁平化的SessionWithMessages作为内部状态。

"响应信封（扁平 vs 嵌套）"同样是 SKILL.md 点名的跨切面关注点——后端返回嵌套对象时，前端如果按扁平结构取值，session.title会直接变成undefined。把信封形态写进契约、并在前端解构，就从源头消除了这类集成 bug。

七、SDK 集成：双存储 + 文本累积

鉴权：无需 Mock

计划明确写道：

不需要任何 Mock。我们已经通过全局 CLI 认证完成 Anthropic 鉴权，Claude Agent SDK 会自动使用这些凭据。不要 mock SDK 或伪造响应——直接对真实 API 测试。

这是该计划的一个鲜明立场：所有验证都跑在真实 API 上，避免"mock 全绿、联调爆炸"的经典翻车路径。

关键模式：双存储 + 文本累积

这是整份计划技术含量最高的一节，值得完整保留：

SDK 以小片段（chunks）流式输出文本。不要把每个 chunk 存成一行数据库记录——那会让前端加载历史时渲染出 N 个独立气泡。正确做法是：累积文本 chunk，每条完整的文本响应只存一行。

async def chat(session_id: str, user_message: str): # 1. Store user message in OUR database await db.add_message(session_id, "user", user_message, "text") # 2. Accumulate text chunks, store other types immediately accumulated_text = "" async for msg in query(prompt=user_message, options=ClaudeAgentOptions(resume=session_id)): if msg.type == "text": accumulated_text += msg.content # Accumulate, don't store yet elif msg.type == "tool_use": # Flush accumulated text before tool use if accumulated_text: await db.add_message(session_id, "assistant", accumulated_text, "text") accumulated_text = "" await db.add_message(session_id, "assistant", None, "tool_use", tool_name=msg.tool_name, ...) elif msg.type == "done": # Flush remaining accumulated text if accumulated_text: await db.add_message(session_id, "assistant", accumulated_text, "text") accumulated_text = "" # 3. Yield EVERY event to frontend via SSE (streaming feel) yield msg

这个模式的妙处在于一石二鸟：

• 对前端：yield msg把每个事件实时转发给 SSE，用户看到的是逐字流出的打字机效果；
• 对数据库：只有遇到tool_use或done时才"冲刷"累积的文本，保证一条完整回答只对应一行记录，重载后前端按行渲染，气泡数与直觉一致。

计划特意强调了query(..., options=ClaudeAgentOptions(resume=session_id))这一调用形态——resume参数把 SDK 的上下文恢复能力与自建数据库的历史展示能力缝合在一起。

捕获 Session ID

新会话时，从 init 消息中取出 session_id：

if isinstance(message, SystemMessage) and message.subtype == "init": session_id = message.data.get("session_id")

拿到 session_id 后才能落库创建 sessions 记录，也才能让后续轮次通过resume继续对话。

SSE 事件类型

计划为前后端约定了完整的流事件联合类型：

type StreamMessage = | { type: 'text'; content: string } | { type: 'thinking'; content: string } | { type: 'tool_use'; tool_name: string; tool_input: object } | { type: 'tool_result'; content: string; is_error: boolean } | { type: 'session_init'; session_id: string } | { type: 'done'; session_id: string; total_cost_usd: number; duration_ms: number }

注意done事件携带total_cost_usd和duration_ms——这让前端可以展示每次回答的成本与耗时，也验证了 SDK 事件流的元数据价值。这组 TypeScript 联合类型与后端消息表的message_type枚举一一对应，形成"数据库字段 ↔ 前端类型"的双向约束。

八、跨切面关注点：显式分配，避免"三不管"

计划专门用一张表列出"跨越多个 Agent、必须在构建时显式指派"的行为——这是 SKILL.md 中"孤儿化跨切面关注点"反模式的正面解法：

最后一行的"UI 可访问性"值得单独强调：aria-label不只关乎无障碍，更关乎可自动化测试——后续 E2E 环节的 agent-browser CLI 正是靠@<ref>定位可交互元素，opacity-0的隐藏按钮在自动化工具眼里等同于不存在。

九、前端组件设计与样式约定

组件职责划分

SessionSidebar

• 按last_accessed排序的会话列表
• 新建会话按钮打开 NewSessionDialog
• 点击会话加载
• 每个会话带删除按钮，aria-label="Delete session"（hover 与 focus 时都必须可见，兼顾可访问性与自动化）

NewSessionDialog

• 标题输入（必填）
• 系统提示词 textarea（可选，placeholder 给示例）
• 工作目录输入（可选，默认 "."）
• 创建按钮

ChatView

• 底部消息输入框
• 发送按钮（流式期间禁用）
• 展示会话标题与元数据

MessageList

• 可滚动容器
• 新消息自动滚动到底
• 连续 assistant 消息分组

MessageBlock按message_type渲染：

• text：聊天气泡（user=右侧蓝色，assistant=左侧灰色）
• tool_use：ToolUseCard 组件
• tool_result：内联结果（过长可折叠）
• thinking：斜体、弱化、默认折叠

ToolUseCard

• 图标 + 工具名头部
• 折叠态：一行摘要
• 展开态：格式化的 JSON 输入与输出
• is_error=true时的错误态样式

样式约定

• shadcn/ui 作为组件底座
• Tailwind 做自定义样式
• 支持暗色模式
• 响应式：移动端侧边栏折叠
• 用户消息：右对齐、蓝色背景
• 助手消息：左对齐、灰色背景
• 工具卡片：细边框、展开/折叠动画

十、依赖清单

后端

fastapi>=0.109.0 uvicorn[standard]>=0.27.0 sse-starlette>=1.8.0 aiosqlite>=0.19.0 claude-agent-sdk>=0.1.0 pydantic>=2.0.0 python-dotenv>=1.0.0

前端

react react-dom typescript vite tailwindcss @shadcn/ui lucide-react

这些版本约束对应了仓库 CLAUDE.md 中"使用 pydantic 做数据校验、FastAPI 构建 API"的全局约定；python-dotenv也与仓库 CLAUDE.md 中"用 python_dotenv 管理环境变量"的规则呼应。

十一、验收标准

1. 新建会话：用户用标题 + 可选系统提示词创建会话 → 会话被保存
2. 聊天：用户发消息 → 响应实时流式返回 → 工具使用内联可见
3. 恢复：用户点击历史会话 → 完整消息历史加载 → 可继续对话
4. 删除：用户删除会话 → 会话与消息一并移除
5. 错误处理：网络/SDK 错误被优雅展示
6. 响应式：桌面端与移动端均可正常使用

十二、分层验证：从单层到全栈

计划把验证拆成"各 Agent 自验各自领域 + lead 做端到端验证"两层，对应 SKILL.md 第 7 步的定义，也与仓库 PRP 基础模板 中"Validation Loop"的理念一脉相承——先跑语法/单测，再跑集成，最后人工确认。

数据库验证（在backend/下执行）

# 1. Schema creation python -c " import asyncio from app.database import init_db, get_db asyncio.run(init_db()) print('✓ Schema created') " # 2. CRUD operations python -c " import asyncio from app.database import * async def test(): await init_db() # Create session session_id = 'test-123' await create_session(session_id, 'Test Session', 'You are helpful', '.') print('✓ Session created') # Add messages await add_message(session_id, 'user', 'Hello', 'text') await add_message(session_id, 'assistant', 'Hi there!', 'text') print('✓ Messages added') # Fetch session with messages session = await get_session_with_messages(session_id) assert session['title'] == 'Test Session' assert len(session['messages']) == 2 print('✓ Session fetch works') # List sessions sessions = await list_sessions() assert any(s['id'] == session_id for s in sessions) print('✓ Session list works') # Delete cascade await delete_session(session_id) session = await get_session_with_messages(session_id) assert session is None print('✓ Delete cascade works') asyncio.run(test()) "

这段脚本覆盖了 Schema 创建、增删改查、级联删除四个关键断言，其中get_session_with_messages正是"嵌套响应"的数据来源。

后端验证（在backend/下执行）

# 1. Start the server uvicorn app.main:app --reload & sleep 2 # 2. Health check curl -s http://localhost:8000/health | grep -q "ok" && echo "✓ Server running" # 3. Create session SESSION=$(curl -s -X POST http://localhost:8000/api/sessions \ -H "Content-Type: application/json" \ -d '{"title": "Test", "system_prompt": "Be concise"}' | jq -r '.id') echo "✓ Created session: $SESSION" # 4. List sessions curl -s http://localhost:8000/api/sessions | jq -e '.[] | select(.title == "Test")' && echo "✓ Session in list" # 5. Get session curl -s http://localhost:8000/api/sessions/$SESSION | jq -e '.title' && echo "✓ Session fetch works" # 6. SSE streaming (send message, capture first event) timeout 30 curl -s -N -X POST "http://localhost:8000/api/sessions/$SESSION/chat" \ -H "Content-Type: application/json" \ -d '{"message": "Say hello in one word"}' | head -5 echo "✓ SSE streaming works" # 7. Verify message persisted curl -s http://localhost:8000/api/sessions/$SESSION | jq -e '.messages | length > 0' && echo "✓ Messages persisted" # 8. Delete session curl -s -X DELETE http://localhost:8000/api/sessions/$SESSION && echo "✓ Session deleted" # 9. Confirm deletion curl -s http://localhost:8000/api/sessions/$SESSION | jq -e '. == null' && echo "✓ Deletion confirmed"

需要留意：契约明确规定集合端点使用尾部斜杠（POST /api/sessions/、GET /api/sessions/），而上述部分 curl 命令未带斜杠——在实际实现中，应以 API 契约为准统一斜杠约定，避免 FastAPI 把带斜杠与不带斜杠的路由判定为不同路径而返回重定向或 404。这正是计划反复强调"斜杠必须精确匹配"的用意所在。

前端验证（在frontend/下执行）

# 1. Dependencies install npm install && echo "✓ Dependencies installed" # 2. TypeScript compiles npx tsc --noEmit && echo "✓ TypeScript valid" # 3. Build succeeds npm run build && echo "✓ Build successful" # 4. Dev server starts npm run dev & sleep 3

随后可用Vercel Agent Browser CLI独立验证 UI（无需后端）：

# Install if needed npm install -g agent-browser agent-browser install # Downloads Chromium # Validate static UI elements agent-browser open "http://localhost:5173" agent-browser snapshot -i # Get interactive elements # Verify: sidebar exists, 'New Session' button visible, chat area renders agent-browser screenshot validation.png

前端 Agent 需自验（不需要后端）：

1. 侧边栏以空状态渲染
2. New Session 对话框能打开与关闭
3. 组件渲染无控制台错误
4. 暗色模式切换正常（如已实现）
5. 不同宽度下的响应式布局

端到端验证（Lead Agent）

所有 Agent 报告完成后，lead 同时启动两个服务并跑 E2E 流程（依旧强调：不 mock，直接用真实 API）：

# Start both servers cd backend && uvicorn app.main:app --port 8000 & cd frontend && npm run dev & sleep 5 # Install agent-browser if needed npm install -g agent-browser agent-browser install

完整 E2E 流程共 12 步：

# 1. CREATE SESSION agent-browser open "http://localhost:5173" agent-browser snapshot -i agent-browser click @<new-session-button-ref> agent-browser fill @<title-input-ref> "E2E Test" agent-browser fill @<system-prompt-ref> "Be extremely brief" agent-browser click @<create-button-ref> # 2. VERIFY SESSION appears in sidebar agent-browser snapshot -i # Confirm "E2E Test" visible in sidebar # 3. SEND MESSAGE agent-browser fill @<chat-input-ref> "What is 2+2?" agent-browser click @<send-button-ref> # 4. VERIFY RESPONSE streams in (wait for completion) agent-browser snapshot -i # Confirm response shows "4" or similar # 5. TOOL USE - trigger a tool agent-browser fill @<chat-input-ref> "Use a tool to list files in the current directory" agent-browser click @<send-button-ref> # 6. VERIFY TOOL card appears agent-browser snapshot -i # Confirm tool use card with Bash/Read tool visible # 7. REFRESH page agent-browser open "http://localhost:5173" # 8. VERIFY PERSISTENCE - session still in sidebar with history agent-browser snapshot -i agent-browser click @<e2e-test-session-ref> # Confirm message history loaded # 9. CONTINUE CHAT agent-browser fill @<chat-input-ref> "What did I ask you first?" agent-browser click @<send-button-ref> # 10. VERIFY CONTEXT - Claude remembers agent-browser snapshot -i # Confirm response mentions 2+2 # 11. DELETE session agent-browser click @<delete-button-ref> # 12. VERIFY DELETE agent-browser snapshot -i # Confirm "E2E Test" removed from sidebar agent-browser screenshot e2e-final.png

成功标准：

• 12 步全部通过
• 终端无服务端错误（检查 uvicorn 输出）
• SSE 流式正常（响应增量出现，而非一次性全部返回）
• 页面刷新后会话持久化正常
• 工具使用卡片用真实工具输出正确渲染（非 mock）

注意第 8 步"刷新后加载历史"和第 10 步"Claude 记得上下文"，正是对本文第一节核心设计（自建 DB 管展示 + SDK resume 管上下文）的端到端验收：历史能加载出来，说明数据库存储与读取链路正确；Claude 能记得之前的提问，说明 session_id 复用链路正确。两条链路缺一不可。

十三、这份计划在本仓库中的定位

需要说明的是，session-manager-plan.md并非可执行代码，而是本仓库use-cases/build-with-agent-team场景下的示例计划文档——它演示了 README.md 中描述的工作流：把一份计划交给/build-with-agent-team [plan-path] [num-agents]技能，技能会读取计划、拆解出数据库/后端/前端等角色、在 tmux 分屏中并行 spawn Agent，并按 SKILL.md 的契约流程协调构建。

这份计划的示范价值在于它完整展示了"一份好的计划文档长什么样"：它把问题（SDK 无历史 API）、方案（双存储）、边界（契约链、斜杠约定、响应信封）、协作规则（跨切面关注点所有权）、验收（6 条标准 + 4 层验证命令）全部写死成文档，让多个互不见面的 Agent 能各自独立开发却在集成时严丝合缝。无论你是否使用 Agent 团队，这套"计划 → 契约 → 并行实现 → 分层验证"的方法论，都适用于任何需要多角色协作的 AI 辅助开发项目。

【免费下载链接】context-engineering-introContext engineering is the new vibe coding - it's the way to actually make AI coding assistants work. Claude Code is the best for this so that's what this repo is centered around, but you can apply this strategy with any AI coding assistant!项目地址: https://gitcode.com/gh_mirrors/co/context-engineering-intro