Claude Agent SDK Session Manager 实战用契约驱动的 Agent Team 工作流构建全栈会话管理应用【免费下载链接】context-engineering-introContext engineering is the new vibe coding - its the way to actually make AI coding assistants work. Claude Code is the best for this so thats 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 契约、以及如何保证重载页面后前端渲染出的气泡数与数据库行数一一对应文本累积。二、技术栈选型计划对技术栈做了明确约定各层职责清晰层技术前端React、TypeScript、Vite、Tailwind、shadcn/ui后端Python、FastAPI、sse-starlette数据库SQLite aiosqliteAgent SDKclaude-agent-sdk选型理由可以从依赖清单反推sse-starlette用于把 SDK 的异步事件流转为 HTTP Server-Sent Eventsaiosqlite提供异步 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构建 Schema、CRUD 函数、Pydantic 模型把函数签名和模型定义发送给 leadLead 验证后转发给后端 AgentPhase 2后端 Agent在收到数据库契约之后构建 FastAPI 应用、路由、SSE 流、SDK 客户端把完整 API 契约发送给 lead必须包含精确的 endpoint URL并标注尾部斜杠精确的请求/响应 JSON 结构精确的 SSE 事件格式含所有事件类型成功与错误场景的状态码Lead 验证后转发给前端 AgentPhase 3前端 Agent在收到 API 契约之后构建 React 应用、组件、API 客户端严格对齐已验证的 API 契约禁止猜测 endpoint URL 或响应结构——一律使用收到的契约Phase 4Lead 验证契约比对contract diff——对比后端真实 endpoint 与前端 fetch 调用启动前后端两个服务运行 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 而非散文描述。EndpointsMethodEndpoint (exact)Request BodyResponseGET/health—{status: ok}POST/api/sessions/{title: ..., system_prompt?: ..., working_directory?: ..., model?: ...}SessionResponse(200)GET/api/sessions/—SessionResponse[](200)GET/api/sessions/{id}—{session: SessionResponse, messages: MessageResponse[]}(200) or 404POST/api/sessions/{id}/chat{message: ...}SSE streamDELETE/api/sessions/{id}—204 No Content注意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(promptuser_message, optionsClaudeAgentOptions(resumesession_id)): if msg.type text: accumulated_text msg.content # Accumulate, dont 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_namemsg.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(..., optionsClaudeAgentOptions(resumesession_id))这一调用形态——resume参数把 SDK 的上下文恢复能力与自建数据库的历史展示能力缝合在一起。捕获 Session ID新会话时从 init 消息中取出 session_idif 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 中孤儿化跨切面关注点反模式的正面解法ConcernOwnerCoordinates WithDetailText chunk accumulationBackendFrontendBackend accumulates streamed text chunks into ONE DB row. Frontend renders one bubble per DB row on reload.URL trailing slashesBackendFrontendFastAPI router uses trailing slashes on collection endpoints (/api/sessions/). Frontend fetch URLs must match exactly.Response envelopeBackendFrontendGET session returns{session: {...}, messages: [...]}, NOT a flat object. Frontend must destructure.UI accessibilityFrontendLead (for E2E testing)All interactive elements needaria-labelattributes. Delete buttons must be clickable (notopacity-0without focus fallback).SSE event formatBackendFrontendExact JSON shapes for each event type documented in API Contract. Both sides must match.最后一行的UI 可访问性值得单独强调aria-label不只关乎无障碍更关乎可自动化测试——后续 E2E 环节的 agent-browser CLI 正是靠ref定位可交互元素opacity-0的隐藏按钮在自动化工具眼里等同于不存在。九、前端组件设计与样式约定组件职责划分SessionSidebar按last_accessed排序的会话列表新建会话按钮打开 NewSessionDialog点击会话加载每个会话带删除按钮aria-labelDelete sessionhover 与 focus 时都必须可见兼顾可访问性与自动化NewSessionDialog标题输入必填系统提示词 textarea可选placeholder 给示例工作目录输入可选默认 .创建按钮ChatView底部消息输入框发送按钮流式期间禁用展示会话标题与元数据MessageList可滚动容器新消息自动滚动到底连续 assistant 消息分组MessageBlock按message_type渲染text聊天气泡user右侧蓝色assistant左侧灰色tool_useToolUseCard 组件tool_result内联结果过长可折叠thinking斜体、弱化、默认折叠ToolUseCard图标 工具名头部折叠态一行摘要展开态格式化的 JSON 输入与输出is_errortrue时的错误态样式样式约定shadcn/ui 作为组件底座Tailwind 做自定义样式支持暗色模式响应式移动端侧边栏折叠用户消息右对齐、蓝色背景助手消息左对齐、灰色背景工具卡片细边框、展开/折叠动画十、依赖清单后端fastapi0.109.0 uvicorn[standard]0.27.0 sse-starlette1.8.0 aiosqlite0.19.0 claude-agent-sdk0.1.0 pydantic2.0.0 python-dotenv1.0.0前端react react-dom typescript vite tailwindcss shadcn/ui lucide-react这些版本约束对应了仓库 CLAUDE.md 中使用 pydantic 做数据校验、FastAPI 构建 API的全局约定python-dotenv也与仓库 CLAUDE.md 中用 python_dotenv 管理环境变量的规则呼应。十一、验收标准新建会话用户用标题 可选系统提示词创建会话 → 会话被保存聊天用户发消息 → 响应实时流式返回 → 工具使用内联可见恢复用户点击历史会话 → 完整消息历史加载 → 可继续对话删除用户删除会话 → 会话与消息一并移除错误处理网络/SDK 错误被优雅展示响应式桌面端与移动端均可正常使用十二、分层验证从单层到全栈计划把验证拆成各 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 需自验不需要后端侧边栏以空状态渲染New Session 对话框能打开与关闭组件渲染无控制台错误暗色模式切换正常如已实现不同宽度下的响应式布局端到端验证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 22? 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 22 # 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 - its the way to actually make AI coding assistants work. Claude Code is the best for this so thats 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考