默认情况下 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 中
与其他概念的关联
- streaming-vs-single-mode - 输入模式(流式输入 vs 单消息输入)
- agent-loop - StreamEvent 在消息流中的位置
- structured-outputs - 结构化输出与流式的兼容性
实践要点
- StreamEvent 包含原始 API 事件,需要自己积累文本增量,不会自动拼接
- 不需要实时输出的场景(CI 管道、后台任务)不要启用,避免额外开销
- TypeScript 中 StreamEvent 类型为
SDKPartialAssistantMessage,type === "stream_event"