Claude API 流式输出（SSE）实战：从打字机效果到工具调用全流程

Claude API 流式输出SSE实战从打字机效果到工具调用全流程接入 Claude API 的人多半都从messages.create()一把梭开始——发个请求等几秒整段返回。等到要做聊天框、做长文生成、做 Agent 工具调用才发现这条路走不动响应慢、用户看着白屏、长输出还会撞 524 超时。这时候必须切到流式输出。Claude API 的流式协议走的是SSEServer-Sent Events事件结构和 OpenAI 不一样做错一步前端就会白屏或乱码。本文把 Python、Node.js、cURL 三套写法跑一遍再覆盖 Tool Use 流式、断流重连、前端打字机效果和生产级容错。读完你能拿到三种语言的最小可运行代码SSE 事件类型对照表避免对着官方文档反复翻Tool Use 流式输出的处理姿势这是 Agent 项目最大的坑断流如何无损续传前端 EventSource 实现打字机效果一、为什么必须用流式三个非用不可的场景把流式当成锦上添花是最大的认知偏差。下面三个场景里不开流式直接是工程缺陷。场景 1长上下文输出2K tokensOpus 4.7 生成一段 4K tokens 的长回复从首 token 到最后一 token 普遍在 30 秒以上。非流式模式下HTTP 请求要么白屏 30 秒要么直接撞到反向代理或 CDN 的 60 秒超时上限最常见的是 524。流式模式下首 token 通常 800ms 内就到用户立刻看到内容。场景 2Agent 工具调用循环Agent 一次推理可能产生 3-5 个 tool_use 块。非流式模式下你要等整段响应回来才能开始执行工具。流式模式下第一个 tool_use 块完成时立刻就能开始调用工具第二个块继续生成整体延迟接近减半。场景 3聊天界面与 IDE 集成Cursor、Claude Code、Cherry Studio 这类客户端能给出丝滑打字机体验全靠 SSE。自己做内部 ChatBot 想达到一样的效果前端必须用 EventSource 或 fetch ReadableStream 接收。二、SSE 事件结构:先把这张表背下来Claude 的流式响应是一连串 SSE 事件,事件类型有 9 种。不理解事件结构直接写解析逻辑,必踩坑。事件类型触发时机关键字段你要做什么message_start响应开始message.id,usage.input_tokens记录 message id;初始化输入 token 计数content_block_start一个新的内容块开始index,content_block.type判断是text、thinking还是tool_usecontent_block_delta增量内容delta.type,delta.text/delta.partial_json拼接到对应 index 的内容块content_block_stop当前内容块结束index关闭该块(tool_use完整后可以触发执行)message_delta消息级元数据更新delta.stop_reason,usage.output_tokens更新输出 token 计数与停止原因message_stop整个响应结束—收尾,关闭连接ping保活心跳—忽略即可error服务端错误error.type,error.message立即终止;考虑重试关键认知:一次响应中可能有多个content_block_*序列,index用来区分。text块的增量在delta.text,tool_use块的增量在delta.partial_json(注意是字符串增量,需要拼接后再 JSON.parse)。三、最小可运行代码:三种语言3.1 Python(Anthropic SDK,推荐)官方 SDK 帮你封装好了 SSE 解析,最干净的写法是messages.stream()上下文管理器:importanthropic clientanthropic.Anthropic(api_keysk-你的密钥,base_urlhttps://gw.claudeapi.com)withclient.messages.stream(modelclaude-opus-4-7,max_tokens2048,messages[{role:user,content:写一段关于流式输出的技术博客开头}])asstream:fortextinstream.text_stream:print(text,end,flushTrue)final_messagestream.get_final_message()print(f\n\n[输入{final_message.usage.input_tokens}tokens, f输出{final_message.usage.output_tokens}tokens])stream.text_stream只迭代文本增量,过滤掉了 thinking、tool_use 等其他块,做纯聊天最方便。如果要拿到原始事件做更精细的控制(例如同时处理 thinking 和 text):withclient.messages.stream(modelclaude-opus-4-7,max_tokens2048,messages[{role:user,content:...}])asstream:foreventinstream:ifevent.typecontent_block_start:print(f\n[block #{event.index}start:{event.content_block.type}])elifevent.typecontent_block_delta:ifevent.delta.typetext_delta:print(event.delta.text,end,flushTrue)elifevent.typemessage_stop:print(\n[done])3.2 Node.js / TypeScriptimportAnthropicfromanthropic-ai/sdk;constclientnewAnthropic({apiKey:sk-你的密钥,baseURL:https://gw.claudeapi.com,});conststreamclient.messages.stream({model:claude-opus-4-7,max_tokens:2048,messages:[{role:user,content:写一段关于流式输出的技术博客开头}],});forawait(consteventofstream){if(event.typecontent_block_deltaevent.delta.typetext_delta){process.stdout.write(event.delta.text);}}constfinalawaitstream.finalMessage();console.log(\n\n[input${final.usage.input_tokens}, output${final.usage.output_tokens}]);或者用更简洁的.on(text)事件订阅:client.messages.stream({model:claude-opus-4-7,max_tokens:2048,messages:[{role:user,content:...}],}).on(text,(text)process.stdout.write(text)).on(finalMessage,(msg)console.log(\ndone:,msg.usage));3.3 cURL(原始 SSE,调试和反代必备)排查问题时直接用 cURL 看原始事件流是最快的方式:curl-Nhttps://gw.claudeapi.com/v1/messages\-Hx-api-key: sk-你的密钥\-Hanthropic-version: 2023-06-01\-Hcontent-type: application/json\-d{ model: claude-opus-4-7, max_tokens: 1024, stream: true, messages: [{role: user, content: Hello}] }注意-N关闭缓冲,否则 cURL 会卡到响应结束才一次性输出。返回会看到一串:event: message_start data: {type:message_start,message:{...}} event: content_block_start data: {type:content_block_start,index:0,...} event: content_block_delta data: {type:content_block_delta,index:0,delta:{type:text_delta,text:Hello}} ...四、Tool Use 流式:Agent 项目的高频踩坑点工具调用块的增量字段是delta.partial_json,是字符串拼接后才能解析。直接对每个 delta 单独JSON.parse必然失败。importjson tool_inputs:dict[int,str]{}# index → accumulated json stringtool_meta:dict[int,dict]{}# index → {name: ..., id: ...}withclient.messages.stream(modelclaude-opus-4-7,max_tokens2048,tools[{name:get_weather,description:查询某个城市的天气,input_schema:{type:object,properties:{city:{type:string}},required:[city]}}],messages[{role:user,content:上海现在天气怎么样?}])asstream:foreventinstream:ifevent.typecontent_block_start:blockevent.content_blockifblock.typetool_use:tool_meta[event.index]{name:block.name,id:block.id}tool_inputs[event.index]elifevent.typecontent_block_delta:ifevent.delta.typeinput_json_delta:tool_inputs[event.index]event.delta.partial_jsonelifevent.typecontent_block_stop:ifevent.indexintool_inputs:argsjson.loads(tool_inputs[event.index])metatool_meta[event.index]print(f[tool call]{meta[name]}({args}))# 这里立刻去执行工具,不必等整个响应结束关键点:content_block_start时拿到tool_use.name和tool_use.idinput_json_delta拼字符串content_block_stop时再json.loads提早 parse 是最常见的错误,会抛JSONDecodeError。五、断流重连:生产环境必须做的事国内网络抖动、移动端切换基站、反向代理重启,都会导致 SSE 半路中断。Claude API 不支持服务端续传,但可以通过记录已收到的内容做客户端侧降级。健壮模式:defstream_with_retry(messages,max_retries2):accumulatedforattemptinrange(max_retries1):try:withclient.messages.stream(modelclaude-opus-4-7,max_tokens4096,messagesmessages)asstream:fortextinstream.text_stream:accumulatedtextyieldtextreturnexcept(anthropic.APIConnectionError,anthropic.APITimeoutError)ase:ifattemptmax_retries:raise# 把已生成内容拼到上下文,让模型续写messagesmessages[{role:assistant,content:accumulated},{role:user,content:请继续输出,不要重复已经写过的内容。}]print(f\n[stream broke at{len(accumulated)}chars, retrying...])要点:只对网络层异常(APIConnectionError、APITimeoutError)重试,不要对 4xx 重试重试时把已收到的内容塞进 assistant 消息,让模型续写而不是从头来max_retries不要超过 3,否则用户会等到崩溃六、前端打字机效果:浏览器侧实现浏览器原生EventSource不支持自定义 header(API Key 没法塞),所以必须用 fetch ReadableStream。中间最好加一层后端代理鉴权,避免把 Key 暴露到前端。后端转发示例(Node.js / Express):importexpressfromexpress;importAnthropicfromanthropic-ai/sdk;constappexpress();app.use(express.json());constclientnewAnthropic({apiKey:process.env.CLAUDE_API_KEY,baseURL:https://gw.claudeapi.com,});app.post(/api/chat,async(req,res){res.setHeader(Content-Type,text/event-stream);res.setHeader(Cache-Control,no-cache);res.setHeader(Connection,keep-alive);conststreamclient.messages.stream({model:claude-sonnet-4-6,max_tokens:2048,messages:req.body.messages,});forawait(consteventofstream){if(event.typecontent_block_deltaevent.delta.typetext_delta){res.write(data:${JSON.stringify({text:event.delta.text})}\n\n);}}res.write(data: [DONE]\n\n);res.end();});前端消费:asyncfunctionchat(messages:any[],onText:(s:string)void){constrespawaitfetch(/api/chat,{method:POST,headers:{Content-Type:application/json},body:JSON.stringify({messages}),});constreaderresp.body!.getReader();constdecodernewTextDecoder();letbuffer;while(true){const{value,done}awaitreader.read();if(done)break;bufferdecoder.decode(value,{stream:true});constlinesbuffer.split(\n\n);bufferlines.pop()??;for(constlineoflines){if(!line.startsWith(data: ))continue;constdataline.slice(6);if(data[DONE])return;const{text}JSON.parse(data);onText(text);}}}注意buffer拼接逻辑——SSE 的\n\n分隔可能跨 chunk,不缓冲会丢字。七、踩坑清单坑 1:Nginx 反代默认会缓冲 SSE,前端永远收不到首 token。在反代配置里加proxy_buffering off;,必要时再加proxy_cache off;与proxy_read_timeout 600s;。Cloudflare 用户检查 “Caching” 是否被命中,必要时关掉。坑 2:tool_use块的 input 是partial_json不是text。不要在text_delta分支里找工具参数,会一直拿不到。事件结构记牢:text_delta走text块,input_json_delta走tool_use块。坑 3:streamTrue与count_tokens不兼容。想精确统计成本,就在流结束后从final_message.usage拿,或者请求前用client.messages.count_tokens()预估。坑 4:移动端 4G/5G 切换时 keepalive 会断。在message_start时记下 message id,断开后重试时把已收到的文本塞进 assistant 消息让模型续写(见第五节代码)。坑 5:Extended Thinking 与流式同时开启时事件多了一类。思考块走thinking_delta而非text_delta,前端如果直接拿text_delta拼会丢掉思考内容。明确决定要不要展示思考过程,再写解析分支。八、性能参考以下是同样的 prompt(请生成一段 2K tokens 的技术文档)在国内网络下的实测对比:模式首 token 延迟总耗时用户感知非流式—28.4s白屏 28 秒流式(Sonnet 4.6)720ms26.9s立即看到首字流式(Opus 4.7)980ms35.2s立即看到首字流式(Haiku 4.5)410ms11.8s极致顺滑非流式 ≈ 流式总耗时,但用户体验差距巨大——首 token 是体验的分水岭。小结流式输出不是可选优化,是 Claude API 进生产环境的必经之路。三个核心要点:事件结构先理解再写代码——9 种事件、text_deltavsinput_json_delta必须分清Tool Use 走 partial_json 拼接——content_block_stop时再 parse生产环境必做断流重连——只对网络异常重试,重试时让模型续写