1. Cloud Agent 的 Web 交互层为什么绕不开 SSE 与持久化Cloud Agent 是什么简单说它把原本跑在终端里的智能体能力搬到浏览器里让多个用户同时在线对话、上传文件、触发工具调用。能做什么你可以让它在网页里读 Excel、查数据库、生成报告全程流式输出。适合谁适合正在做 AI 应用后端、想把 Agent 能力产品化的开发者。我最近在做一个 Cloud Agent 项目核心链路是浏览器发起请求 → 服务端跑 Agent 循环 → 事件流式推回前端 → 消息和结果落盘。这条链路里有两个必须解决的问题事件怎么到浏览器以及数据怎么存下来不丢。Claude Code 的输出端是终端渲染器事件通过 AsyncGenerator 直接传给 React 组件没有网络传输这一步。但 Web 场景不一样服务端和浏览器之间隔着网络需要一套流式协议。选 SSE 还是 WebSocketWebSocket 双向通信能力更强但 Cloud Agent 的交互本质是单向的服务端推送流式结果客户端只在初始请求和点停止时发信号。SSE 更轻原生支持断线重连服务端框架内置支持不需要额外库。所以选了 SSE。前端接收方式也有讲究。浏览器有 EventSource API原生处理 SSE但它不支持自定义 headers加不了 JWT token。所以用 fetch ReadableStream 手动解析 SSE 字节流。多写了点代码但认证链路不需要妥协。线缆格式很直接event: text data: {content: ...} event: tool_result data: {tool_use_id:xxx,content:...,is_error:false}定义了 10 种事件类型覆盖对话全部状态text 对应 LLM 输出文本 tokencontent_block_start 对应 LLM 开始输出 tool_usecontent_block_delta 对应 tool_use 参数流式增量content_block_stop 对应参数接收完毕tool_result 对应工具执行完成usage 对应每轮 LLM 调用结束的 token 统计done 对应 Agent 循环正常结束error 对应循环异常user 对应中断插入的消息user_question 对应 AskUserQuestion 触发。一轮 query() 调用中事件的典型时序是这样的text 先到然后是 content_block_start 表示开始调工具接着 content_block_delta 参数逐个到达content_block_stop 参数完整tool_result 执行结果返回usage 统计 token最后 done 结束。中间可能穿插多轮比如第二轮工具调用后 LLM 基于结果继续分析。事件命名有个小插曲。最初沿用了旧版本中类似消息类型的名字后来翻 Claude Code 源码发现它的事件名更简洁就统一过去了。多用户多会话的状态隔离是另一个坑。一个用户可能同时开着好几个 Session每个会话的消息、SSE 连接状态、工具执行状态都要独立。用 Zustand store 按 sessionId 分区用户从会话 A 切到 B 时A 的 SSE 流在后台继续跑到结束不会因为切走了就被强行终止。B 的消息从独立分区加载不会和 A 的数据串到一起。这个设计踩过一个坑切换会话时 SSE 事件串到了另一个会话里。排查发现是 store 没有严格按 sessionId 隔离收到 SSE 事件后直接往当前会话的数组里 push没检查事件的 sessionId 和当前显示的 sessionId 是否一致。修完之后加了分区边界检查tool_use 的 running 状态也按 sessionId 独立追踪。切会话后后台流继续跑的设计有一个副作用后台流跑完最后一轮时 tool_result 事件没有地方显示。用户切回来时看到的是完整结果看不到工具执行的中间状态。体验不够好但业务侧的对话轮数通常在 3 到 5 轮后台跑完很快。没做更复杂的恢复逻辑属于知道不完美但优先级不够的范畴。2. 用 TaoToken 统一 Key 打通 API 通道的前置准备在动手写 SSE 和持久化之前得先把 API 通道配好。Cloud Agent 要调用大模型每个模型厂商的 Key 格式、Base URL、鉴权方式都不一样如果每个都单独配代码里会散落一堆 if-else。TaoToken 的作用就是把这些统一成一个 Key、一个 Base URL代码里只认一套配置。TaoToken 是什么它是一个 API 聚合通道把不同模型厂商的接口统一成 OpenAI 兼容格式。能做什么你只需要一个 Key就能在代码里切换不同模型不用改鉴权逻辑。适合谁适合正在做多模型接入、不想维护多套鉴权代码的开发者。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 地址是 https://taotoken.net/api注意 API 地址不加 UTM 参数。前置准备分三步。第一步注册账号并创建 API Key。登录官网后进入控制台在 API Keys 页面创建一个新 Key复制保存。这个 Key 就是后面所有请求的凭证。第二步确认你要用的模型 ID。TaoToken 支持多种模型每个模型有对应的 Model ID。你可以在模型对话页面测试模型是否可用也可以在文档页面查看完整的模型列表。Model ID 的格式通常是厂商名/模型名比如 claude-sonnet-4-20250514 这种。第三步把 Base URL、Key、Model ID 三件套记下来。后面配置 Claude Code、Cline、Codex 或者自己写代码时都是围绕这三个值展开。如果你用的是 Claude Code配置方式是在 settings.json 里写{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_AUTH_TOKEN: 你的TaoToken Key, ANTHROPIC_MODEL: claude-sonnet-4-20250514 } }如果你用的是 Cline 或者 Roo Code 这类 VS Code 插件在设置里填 Base URL 为 https://taotoken.net/apiAPI Key 填你的 TaoToken KeyModel ID 填你要用的模型。如果你用的是 Codex配置在 auth.json 里{ openai_api_key: 你的TaoToken Key, openai_api_base: https://taotoken.net/api }三件套的核心逻辑是一样的Base URL 指向 TaoToken 的 API 地址Key 用 TaoToken 的 KeyModel ID 指定你要调用的模型。这样你的代码里只需要维护一套鉴权逻辑切换模型时只改 Model ID 就行。对于 Cloud Agent 项目我建议把这三件套放在环境变量里不要硬编码在代码中。比如export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_API_KEY你的Key export TAOTOKEN_MODELclaude-sonnet-4-20250514然后在代码里读取环境变量。这样本地开发和线上部署可以用不同的 Key也不会把 Key 提交到 Git 仓库。配置完成后先做一个最简单的连通性验证确认 Key 和 Base URL 没问题。用 curl 发一个请求curl -X POST https://taotoken.net/api/v1/messages \ -H Content-Type: application/json \ -H x-api-key: 你的TaoToken Key \ -H anthropic-version: 2023-06-01 \ -d { model: claude-sonnet-4-20250514, max_tokens: 100, messages: [{role: user, content: 你好}] }如果返回正常的 JSON 响应说明通道通了。如果返回 401说明 Key 有问题如果返回 404说明 Base URL 或路径有问题。这一步验证通过后再往下写 SSE 和持久化逻辑。3. 可复制的 SSE 服务端与前端配置SSE 服务端的核心是把 Agent 循环产生的事件流式推给浏览器。我用的是 Hono 框架它内置了 SSE 支持。先看服务端的路由配置import { Hono } from hono; import { streamSSE } from hono/streaming; const app new Hono(); app.get(/api/sessions/:sessionId/stream, async (c) { const sessionId c.req.param(sessionId); const authToken c.req.header(Authorization)?.replace(Bearer , ); if (!authToken) { return c.json({ error: unauthorized }, 401); } return streamSSE(c, async (stream) { const abortController new AbortController(); sessionAbortControllers.set(sessionId, abortController); try { for await (const event of query(sessionId, abortController.signal)) { await stream.writeSSE({ event: event.type, data: JSON.stringify(event.payload), }); } } catch (err) { await stream.writeSSE({ event: error, data: JSON.stringify({ message: err.message }), }); } finally { sessionAbortControllers.delete(sessionId); } }); });这段代码的关键点从 header 里取 JWT token 做鉴权用 AbortController 管理中断用 streamSSE 把事件逐个写出去。每个事件的 event 字段是事件类型data 字段是 JSON 序列化的负载。前端接收不能用 EventSource因为它不支持自定义 headers。用 fetch ReadableStream 手动解析async function connectSSE(sessionId: string, token: string) { const response await fetch(/api/sessions/${sessionId}/stream, { headers: { Authorization: Bearer ${token}, Accept: text/event-stream, }, }); const reader response.body!.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop() || ; let currentEvent ; for (const line of lines) { if (line.startsWith(event: )) { currentEvent line.slice(7); } else if (line.startsWith(data: )) { const data JSON.parse(line.slice(6)); handleEvent(currentEvent, data, sessionId); } } } }这段代码手动解析 SSE 字节流按行分割遇到 event: 开头记录事件类型遇到 data: 开头解析 JSON 并分发。buffer 用来处理跨 chunk 的半行数据。handleEvent 函数里要做 sessionId 边界检查这是之前踩坑后加的function handleEvent(eventType: string, data: any, sessionId: string) { const currentSessionId useStore.getState().currentSessionId; if (sessionId ! currentSessionId) { return; } // 按事件类型分发到对应的 store 分区 switch (eventType) { case text: useStore.getState().appendText(sessionId, data.content); break; case tool_result: useStore.getState().setToolResult(sessionId, data.tool_use_id, data); break; // ... } }断线重连的处理SSE 原生支持 Last-Event-ID服务端可以在每个事件里带上 id 字段客户端重连时自动带上 Last-Event-ID header。服务端根据这个 id 从上次断开的位置继续推。实现方式是在 writeSSE 时加 idawait stream.writeSSE({ id: String(event.sequence), event: event.type, data: JSON.stringify(event.payload), });客户端重连时fetch 请求带上 Last-Event-ID header服务端从对应的 sequence 之后继续推。如果服务端没有实现断点续传客户端重连后会从头开始推这时候需要前端做去重。数据持久化的配置分三块。消息用 JSONL 存路径是 data/messages/{sessionId}.jsonlappend-only 写入import { appendFile, readFile } from fs/promises; async function appendMessage(sessionId: string, message: any) { const path data/messages/${sessionId}.jsonl; await appendFile(path, JSON.stringify(message) \n, utf-8); } async function loadMessages(sessionId: string) { const path data/messages/${sessionId}.jsonl; const content await readFile(path, utf-8); return content.trim().split(\n).map(line JSON.parse(line)); }元数据用 SQLite 存Bun 内置了 bun:sqliteimport { Database } from bun:sqlite; const db new Database(data/app.db); db.exec(PRAGMA journal_mode WAL); db.exec( CREATE TABLE IF NOT EXISTS aac_sessions ( id TEXT PRIMARY KEY, project_id TEXT NOT NULL, name TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ) ); db.exec( CREATE TABLE IF NOT EXISTS aac_files ( id TEXT PRIMARY KEY, project_id TEXT NOT NULL, original_name TEXT NOT NULL, storage_name TEXT NOT NULL, source TEXT NOT NULL, size INTEGER NOT NULL, created_at INTEGER NOT NULL ) );文件用磁盘存路径是 data/projects/{projectId}/files/通过 aac_files 表的 source 字段区分身份user_upload 是用户上传的原始文件Agent 只能读不能改temp 是 Agent 生成的中间文件可以自由读写final_output 是用户标记的定稿Agent 不能动。Chat 路由里的持久化逻辑很保守先写盘再推 SSE。这样即使 SSE 推失败客户端断了消息已经落盘了不会出现看到一半刷新页面中间几条消息丢了的情况。async function handleChat(sessionId: string, userMessage: string) { await appendMessage(sessionId, { role: user, content: userMessage }); for await (const event of query(sessionId)) { await appendMessage(sessionId, event); await stream.writeSSE({ event: event.type, data: JSON.stringify(event.payload) }); } }4. 验证请求与成功结果从连通性到数据落库回读配置写完了得验证整条链路是通的。验证分四步API 连通性、SSE 事件流、断线重连、数据落库回读。第一步验证 TaoToken API 连通性。用 curl 发一个最简单的请求curl -X POST https://taotoken.net/api/v1/messages \ -H Content-Type: application/json \ -H x-api-key: 你的TaoToken Key \ -H anthropic-version: 2023-06-01 \ -d { model: claude-sonnet-4-20250514, max_tokens: 50, messages: [{role: user, content: 回复OK两个字母}] }成功的结果是返回一个 JSON里面 content 数组的第一项 text 字段是 OK。如果返回 401检查 Key 是否正确如果返回 404检查 Base URL 是否写成了 https://taotoken.net/api 而不是 https://taotoken.net/api/v1/messages 这种完整路径。第二步验证 SSE 事件流。启动服务端后用 curl 直接请求 SSE 端点curl -N http://localhost:3000/api/sessions/test-session/stream \ -H Authorization: Bearer 你的JWT-N 参数关闭缓冲让 curl 实时输出。成功的结果是看到一串 event: 和 data: 交替的输出比如event: text data: {content:你} event: text data: {content:好} event: usage data: {input_tokens:10,output_tokens:5} event: done data: {}如果只看到连接建立但没有事件输出检查 query() 函数是否正常产生事件如果事件输出到一半断了检查是否有未捕获的异常。第三步验证断线重连。在浏览器里打开页面发起一个对话然后在事件流还没结束时手动刷新页面。刷新后前端会重新发起 SSE 请求带上 Last-Event-ID header。成功的结果是刷新后能看到之前已经推送的消息并且后续事件继续正常推送。如果刷新后消息重复了说明服务端没有实现断点续传前端需要做去重。去重逻辑是按消息的 sequence 或 id 判断已经存在的消息不重复添加。第四步验证数据落库回读。发起一个对话等对话结束后检查三个地方消息 JSONL 文件是否存在cat data/messages/test-session.jsonl成功的结果是看到一行行的 JSON每行是一条消息或事件。SQLite 元数据是否写入sqlite3 data/app.db SELECT id, name, created_at FROM aac_sessions成功的结果是看到会话记录。文件是否落盘ls -la data/projects/test-project/files/成功的结果是看到上传的文件和 Agent 生成的文件。回读验证重启服务端然后请求加载会话消息curl http://localhost:3000/api/sessions/test-session/messages \ -H Authorization: Bearer 你的JWT成功的结果是返回之前对话的完整消息列表说明数据持久化没问题。5. 本篇常见错误排查401、local proxy failed、reading choices、OAuth配置和验证过程中会遇到一些典型报错这里逐个排查。401 Unauthorized。这个最常见原因通常是 Key 不对或 header 格式不对。检查三件事TaoToken Key 是否复制完整有没有多余空格header 名是否正确Anthropic 格式用 x-api-keyOpenAI 格式用 Authorization: BearerBase URL 是否写对TaoToken 的 API 地址是 https://taotoken.net/api不要多加或少加路径。如果用的是 Claude Code401 还可能是 settings.json 里的 ANTHROPIC_AUTH_TOKEN 没生效。检查环境变量是否被覆盖可以用 echo $ANTHROPIC_AUTH_TOKEN 确认。local proxy failed。这个报错通常出现在 Claude Code 或 Cline 这类工具里原因是工具尝试走本地代理但代理没启动。检查工具的代理配置如果不需要代理把代理设置关掉。如果用的是 TaoToken 的 Base URL不需要额外配代理。reading choices 报错。这个通常出现在 OpenAI 兼容格式的响应解析里原因是响应结构不符合预期。检查请求的 Model ID 是否正确有些模型返回的是 Anthropic 格式而不是 OpenAI 格式。如果用的是 TaoToken确认请求路径是 /v1/messagesAnthropic 格式还是 /v1/chat/completionsOpenAI 格式两者返回结构不同。OAuth 相关报错。如果用的是 Claude Code 的 OAuth 登录方式但配置了 TaoToken 的 Key可能会冲突。解决方式是明确用 Key 鉴权不要走 OAuth。在 settings.json 里只配 ANTHROPIC_AUTH_TOKEN不要配 OAuth 相关的字段。SSE 连接建立但没有事件。检查 query() 函数是否正常产生事件可以在 for await 循环里加日志。另外检查 streamSSE 的 writeSSE 是否被正确 await如果没有 await事件可能丢失。事件串到别的会话。这是之前踩过的坑原因是 store 没有按 sessionId 隔离。检查 handleEvent 函数里是否有 sessionId 边界检查收到事件后先判断 sessionId 是否等于当前显示的 sessionId不等就直接 return。数据落盘了但回读为空。检查文件路径是否正确JSONL 文件路径是 data/messages/{sessionId}.jsonlSQLite 路径是 data/app.db。另外检查文件权限服务端进程是否有读写权限。断线重连后消息重复。原因是服务端没有实现断点续传客户端重连后从头推。解决方式是服务端在每个事件里带 sequence id客户端重连时带 Last-Event-ID服务端从对应位置继续推。如果暂时不想实现前端做去重也行。6. 把统一 Key 和持久化链路固化成项目规范整条链路跑通后最后一步是把它固化成项目规范避免后面加功能时破坏现有设计。API 通道规范所有模型调用统一走 TaoToken 的 Base URL 和 Key代码里只维护一套鉴权逻辑。Model ID 放在配置文件或环境变量里切换模型时只改这一个值。新增模型时先在模型对话页面测试可用性再写进配置。SSE 事件规范所有事件类型统一定义在 types 文件里服务端和前端共用同一份类型定义。新增事件类型时先加类型定义再加服务端产生逻辑最后加前端处理逻辑。事件命名保持简洁用 text、tool_result 这种短名字不要用冗长的描述性名字。持久化规范消息走 JSONL元数据走 SQLite文件走磁盘。判断标准是数据的访问模式需要条件筛选和关联查询的走 SQLite顺序读写的大体积数据走文件系统。新增数据类型时先判断它属于哪一类再决定存储方式。会话隔离规范所有跟会话相关的状态都按 sessionId 分区收到事件后先做边界检查。新增状态时确认它是否需要按会话隔离如果需要放进对应的分区里。中断处理规范所有可能长时间运行的操作都要支持 AbortController中断时生成 synthetic tool_result 保证消息列表自洽。新增工具时确认它是否响应 abort signal如果不响应在工具执行前加检查点。这套规范的核心思路是统一入口、按访问模式选存储、按会话隔离状态、中断时保证一致性。后面加功能时按这个思路走不会出大问题。如果你还没配 TaoToken 的 Key可以去 API Keys 页面创建一个然后在接入文档里看完整的配置说明。想先测试模型是否可用可以在模型对话页面直接试。如果是长期做编码和 Agent 开发可以考虑 Coding Plan把常用模型和额度固定下来。