Claude学院

Claude API 流式输出 Python 实现教程（SSE 详解）

用 anthropic 官方 Python SDK 实现 Claude 流式输出，核心只有一行：把 client.messages.create() 换成 with client.messages.stream(...) as stream，然后遍历 stream.text_stream 逐字拿到输出。底层走的是 SSE（Server-Sent Events，服务器推送事件）协议，服务端把一条回复拆成一连串事件流式下发，前端拿到一个就渲染一个，于是有了「打字机」效果。下面从最小示例讲到事件类型、思考块、工具调用和错误处理。

为什么要用流式输出

非流式请求（messages.create）会一直等到 Claude 把整段话生成完才返回，长回复要等十几秒甚至更久，用户体验差，还容易触发 SDK 的 HTTP 超时。流式输出有两个实打实的好处：

  • 首字延迟低：模型生成第一个 token 就开始往回推，用户几乎立刻看到内容在动。
  • 避免超时：当 max_tokens 设得较大（比如超过 16000）时，官方建议必须用流式，否则非流式调用容易在 SDK 层超时。Claude Opus 4.8 / Sonnet 4.6 输出上限较高，长文生成尤其需要流式。

如果你还没跑通第一个 Claude 请求，建议先看 Claude API Python 示例代码：10 分钟跑通第一个程序 和 Anthropic Python SDK 安装与配置完整教程，把环境和密钥准备好再回来。

第一步：安装与最小示例

安装官方 SDK，并把 API Key 写进环境变量 ANTHROPIC_API_KEY（SDK 会自动读取，不要硬编码在代码里）：

pip install -U anthropic
export ANTHROPIC_API_KEY="你的密钥"

最简单的流式输出，用 text_stream 直接拿纯文本增量，一行一行打印出来：

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=2000,
    messages=[{"role": "user", "content": "用三句话介绍一下 SSE 协议"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()
    final = stream.get_final_message()  # 流结束后拿到完整 Message 对象
    print("\n用量：", final.usage)

关键点：stream() 必须配合 with 上下文管理器使用，退出时会自动关闭连接；text_stream 是一个生成器，每次 yield 一段文本增量；get_final_message() 在流读完后返回拼好的完整 Message，里面有 stop_reason 和 usage 等字段。如果你不关心中间事件、只要最终结果，这套写法就够了。

SSE 事件类型详解

SSE 把一条回复拆成多种事件。当你需要更精细的控制（比如区分思考块和正文、统计 token、处理工具调用）时，就要遍历 stream 本身拿到原始事件。下面是 Messages API 流式响应里最常见的事件，按出现顺序排列：

事件类型含义与处理时机
message_start消息开始，携带 message 元信息（模型、初始 usage）。整段回复的第一个事件。
content_block_start一个内容块开始。content_block.type 可能是 text、thinking、tool_use 等。
content_block_delta内容块的增量。文本是 text_delta，思考是 thinking_delta，工具参数是 input_json_delta。逐字渲染就靠它。
content_block_stop当前内容块结束。
message_delta消息级别的增量，通常携带 stop_reason 和累计 usage.output_tokens。
message_stop整条消息结束。最后一个事件。

用 for event in stream 遍历并按 event.type 分发处理：

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=2000,
    messages=[{"role": "user", "content": "解释一下什么是事件循环"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_delta":
            # 这里能拿到 stop_reason，比如 end_turn / max_tokens / tool_use
            if event.delta.stop_reason:
                print(f"\n[结束原因] {event.delta.stop_reason}")
        elif event.type == "message_stop":
            print("\n[消息结束]")

SDK 已经帮你把原始 SSE 字节解析成了结构化的 Python 对象，你不需要自己处理 data: 前缀和换行符。除非你坚持用 requests/httpx 裸调 HTTP，否则不用碰 SSE 的原始格式。想了解 Messages API 的全部参数，可以参考 Claude Messages API 全部参数说明（含代码示例）。

流式处理思考块（thinking）

Claude Opus 4.8 支持自适应思考（adaptive thinking）。开启后，模型会先输出 thinking 类型的内容块，再输出正文。如果你想在界面上区分「正在思考」和「正式回答」，就要按内容块类型分流处理。注意 Opus 4.8 默认 display 为 omitted（思考块文本为空，看起来像一段长时间停顿），要显示思考摘要需显式设置 display="summarized"：

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=4000,
    thinking={"type": "adaptive", "display": "summarized"},
    messages=[{"role": "user", "content": "一个长方形周长 20，面积最大是多少？"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            if event.delta.type == "thinking_delta":
                print(event.delta.thinking, end="", flush=True)  # 思考摘要
            elif event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)      # 最终回答

提示：thinking={"type": "enabled", "budget_tokens": N} 这种固定预算写法在 Opus 4.8 上会返回 400 错误，已经改用自适应思考；同时 temperature、top_p 等采样参数在该模型上也不再支持。模型怎么选可以看 Claude 模型怎么选？Opus / Sonnet / Haiku 选型指南。

流式处理工具调用

当 Claude 决定调用工具时，会输出 tool_use 内容块，参数以 input_json_delta 形式分片流出。这种场景下不要手动拼 JSON 字符串去解析增量，直接用 get_final_message() 拿到拼好的完整工具调用，再判断 stop_reason 是否为 tool_use：

with client.messages.stream(
    model="claude-opus-4-8",
    max_tokens=2000,
    tools=[{
        "name": "get_weather",
        "description": "查询某城市的天气",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    }],
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

if final.stop_reason == "tool_use":
    for block in final.content:
        if block.type == "tool_use":
            print("\n要调用工具:", block.name, "参数:", block.input)

工具调用的完整闭环（执行工具、把 tool_result 回传给模型继续对话）属于另一个话题，详见 Claude Tool Use 工具调用怎么用？完整代码实战。

错误处理与重试

流式请求同样会遇到限流和服务端错误。anthropic SDK 提供了类型化异常，用 isinstance 判断而不是去匹配错误字符串。SDK 默认会对 429 和 5xx 自动重试（max_retries=2），你也可以自定义：

import anthropic

client = anthropic.Anthropic(max_retries=3)

try:
    with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=2000,
        messages=[{"role": "user", "content": "你好"}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
except anthropic.RateLimitError:
    print("触发限流，请稍后重试（429）")
except anthropic.APIStatusError as e:
    print(f"接口报错 {e.status_code}: {e.type}")

遇到 429 限流的多种处理方案见 Claude API 429 限流报错怎么办？3 种重试方案；如果是 401 认证失败，参考 Claude API 报错 401 怎么解决？认证失败排查指南。

常见问题

流式输出和非流式相比，价格会更贵吗？

不会。流式和非流式按相同的 token 计费方式结算，流式只是改变了数据返回的方式，不改变计费规则。最终 token 用量可以在 get_final_message().usage 里查看。具体每个模型的单价和计费方式以 Anthropic 官网为准。

必须用 with 上下文管理器吗？能不能用异步？

推荐用 with，它能保证连接被正确关闭。异步场景把 Anthropic() 换成 AsyncAnthropic()，并用 async with client.messages.stream(...) 配合 async for 遍历 text_stream 即可，事件类型和处理逻辑完全一致。

Node.js 怎么实现同样的流式输出？

用 @anthropic-ai/sdk 的 client.messages.stream()，监听 text 事件或用 for await 遍历，再用 finalMessage() 拿完整结果，思路和 Python 一致。完整示例见 Claude API Node.js 调用示例：从安装到流式输出。

← Claude API 开发