默认情况下 SDK 在 Claude 完成每轮响应后才 yield 完整的 AssistantMessage。通过启用 include_partial_messages,可以在生成过程中实时接收 StreamEvent 消息,实现打字机效果或实时工具调用状态显示。

核心机制

启用方式:设置 include_partial_messages=True(Python)或 includePartialMessages: true(TypeScript)。

消息流顺序(启用后):

StreamEvent(message_start)
StreamEvent(content_block_start) - 文本块开始
StreamEvent(content_block_delta) - 文本增量...
StreamEvent(content_block_stop)
StreamEvent(content_block_start) - 工具调用块开始
StreamEvent(content_block_delta) - 工具参数增量...
StreamEvent(content_block_stop)
StreamEvent(message_delta)
StreamEvent(message_stop)
AssistantMessage - 完整消息
... 工具执行 ...
ResultMessage - 最终结果

提取文本增量:检查 content_block_delta 事件中 delta.type == "text_delta" 的内容。

提取工具调用增量:检查 content_block_delta 事件中 delta.type == "input_json_delta" 的 partial_json 字段。

关键细节

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
 
async for message in query(
    prompt="List the files in my project",
    options=ClaudeAgentOptions(include_partial_messages=True),
):
    if isinstance(message, StreamEvent):
        event = message.event
        if event.get("type") == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                print(delta.get("text", ""), end="", flush=True)

构建流式 UI(追踪工具调用状态):

in_tool = False
async for message in query(prompt="...", options=options):
    if isinstance(message, StreamEvent):
        event = message.event
        if event.get("type") == "content_block_start":
            if event.get("content_block", {}).get("type") == "tool_use":
                print(f"\n[Using {event['content_block']['name']}...]", end="")
                in_tool = True
        elif event.get("type") == "content_block_delta":
            if not in_tool and event.get("delta", {}).get("type") == "text_delta":
                print(event["delta"]["text"], end="", flush=True)
        elif event.get("type") == "content_block_stop" and in_tool:
            print(" done")
            in_tool = False

已知限制:

  • 启用 max_thinking_tokens 时不会 yield StreamEvent(只有完整消息)
  • 结构化输出(structured_output)不以流式增量形式出现,只在最终 ResultMessage 中

与其他概念的关联

实践要点

  • StreamEvent 包含原始 API 事件,需要自己积累文本增量,不会自动拼接
  • 不需要实时输出的场景(CI 管道、后台任务)不要启用,避免额外开销
  • TypeScript 中 StreamEvent 类型为 SDKPartialAssistantMessage,type === "stream_event"