Claude Agent SDK 流式响应开发

简介

流式响应（Stream），指模型不再等全部内容生成完毕一次性返回，而是分段实时推送文本块。

体验：用户可以实时看到逐字输出，减少等待感，适合网页、控制台实时展示
两种流式类型：纯文本流式、工具调用流式（tool_use 会在流中分段输出）
适用场景：Web 对话界面、CLI 实时打印、长报告生成、实时代码输出

重点：流式不改变 Agent 工具调用的完整逻辑，只是数据返回方式变成分片；工具调用依然需要等完整 tool_use 块接收完成后，再执行工具并回传 tool_result。

前置准备

Python3.10+ / Node.js 18+
安装 SDK

# Python
pip install anthropic
# TS
npm install @anthropic-ai/sdk

环境变量配置 ANTHROPIC_API_KEY

核心原理

发起请求时开启 stream=True（Python） / stream: true（TS）
服务端持续推送事件块（text_delta、tool_use 增量块等）
程序循环读取事件流，实时处理文本增量
如果流中包含工具调用：先收集完整 tool_use 信息，流结束后执行工具函数，再发起新一轮带 tool_result 的流式请求

实战1：Python 基础文本流式输出

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def stream_text_response():
    stream = client.messages.create(
        model="claude-3-5-sonnet-latest",
        max_tokens=1024,
        temperature=0.3,
        system="你是一名前端开发助手，回答简洁清晰",
        messages=[{"role": "user", "content": "简单介绍 TailwindCSS"}],
        stream=True  # 开启流式
    )

    print("开始流式输出：")
    for event in stream:
        # 文本增量事件
        if event.type == "content_block_delta":
            delta = event.delta
            if delta.type == "text_delta":
                # 实时打印增量文本
                print(delta.text, end="", flush=True)

if __name__ == "__main__":
    stream_text_response()

实战2：Python 流式 + 工具调用（重点）

流式下工具调用不会一次性返回完整 tool_use，需要拼接增量数据，等收集完整后执行工具

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

tools = [
    {
        "name": "get_current_time",
        "description": "获取当前系统时间",
        "input_schema": {"type": "object", "properties": {}, "required": []}
    }
]

def get_current_time():
    from datetime import datetime
    return {"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

def stream_with_tool():
    messages = [{"role": "user", "content": "现在几点了？"}]
    stream = client.messages.create(
        model="claude-3-5-sonnet-latest",
        max_tokens=512,
        tools=tools,
        messages=messages,
        stream=True
    )

    # 缓存tool_use信息
    tool_use_id = None
    tool_name = None
    tool_input_json = ""

    for event in stream:
        if event.type == "content_block_start":
            if event.content_block.type == "tool_use":
                tool_use_id = event.content_block.id
                tool_name = event.content_block.name
        elif event.type == "content_block_delta":
            delta = event.delta
            if delta.type == "input_json_delta":
                # 拼接增量JSON字符串
                tool_input_json += delta.partial_json

    # 流读取完毕，判断是否需要调用工具
    if tool_use_id and tool_name:
        import json
        tool_input = json.loads(tool_input_json)
        if tool_name == "get_current_time":
            tool_result = get_current_time()
        # 追加tool_result，再次流式请求
        messages.append({
            "role": "user",
            "content": [{"type": "tool_result", "tool_use_id": tool_use_id, "content": str(tool_result)}]
        })
        print("\n工具调用完成，继续流式返回回答：")
        stream2 = client.messages.create(
            model="claude-3-5-sonnet-latest",
            max_tokens=512,
            messages=messages,
            tools=tools,
            stream=True
        )
        for event in stream2:
            if event.type == "content_block_delta" and event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)

if __name__ == "__main__":
    stream_with_tool()

实战3：TypeScript / Node.js 流式示例

import Anthropic from "@anthropic-ai/sdk";const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
});
async function streamResponse() {
  const stream = await anthropic.messages.create({
    model: "claude-3-5-sonnet-latest",
    max_tokens: 1024,
    temperature: 0.3,
    system: "前端开发助手",
    messages: [{ role: "user", content: "介绍一下Nuxt 4" }],
    stream: true,
  });
  console.log("流式输出：");
  for await (const event of stream) {
    if (event.type === "content_block_delta") {
      const delta = event.delta;
      if (delta.type === "text_delta") {
        process.stdout.write(delta.text);
      }
    }
  }
}
streamResponse();

流式事件类型说明

事件类型	说明
`message_start`	消息会话开始
`content_block_start`	内容块开始（文本块 / tool_use块）
`content_block_delta`	增量数据，`text_delta`文本 / `input_json_delta`工具参数JSON片段
`content_block_stop`	当前内容块结束
`message_delta`	消息级别的状态更新（token使用量）
`message_stop`	整条消息流结束

Web端流式传输方案（后端SSE）

如果做网页对话，后端收到SDK stream，包装成 SSE（Server-Sent Events） 推送给前端：

后端接收用户提问，开启Anthropic stream
捕获 text_delta，包装为SSE data: xxx 发送前端
前端JS监听 EventSource，实时追加文字
遇到工具调用块，后端暂停流、执行工具，拿到结果后继续发起下一轮流式请求

提示：不要直接把Anthropic API暴露给前端，必须后端做代理，防止API Key泄露。

调优与最佳实践

文本场景优先流式：长文章、代码生成、聊天对话，用户感知更好；短回答可以直接非流式一次性返回，减少开销。
工具调用注意分片JSON：input_json_delta 是分段JSON碎片，不能边读边解析，必须拼接完整后再JSON.parse。
超时处理：流式长连接容易断连，增加心跳检测、断线重试机制。
Token统计：message_delta 事件中可以获取输出token消耗，用于计费、限流。
输出过滤：在增量文本推送给用户前，可做敏感词过滤。
并发控制：流式请求同样占用API配额，多用户场景增加并发限制。

常见问题

Q：流式模式下，可以实时拿到工具参数直接执行吗？

A：不行。工具参数是分段返回的不完整JSON片段，必须等 content_block_stop 收到，拼接完整JSON后才能解析执行。

Q：流式会省token费用吗？

A：不会。流式只是传输方式不同，token计费规则和普通请求完全一致。

Q：流中断怎么办？

A：可以记录当前会话 messages 完整历史，重新发起请求，传入完整消息上下文继续生成。

小结

流式响应核心：开启 stream 参数，循环读取事件增量；文本增量可实时展示；工具调用需要收集完整tool_use块再执行工具。
适合Web聊天、实时代码生成场景；工具+流式组合开发复杂度更高，注意JSON碎片拼接、断连处理。

简介