Claude Code SDK #11：流式输出全解——一个参数打开 Agent 运行时的实时可观测性

开启流式输出只需一行

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio

async def stream_response():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Bash", "Read"],
    )
    async for message in query(prompt="列出项目里的文件", options=options):
        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)

asyncio.run(stream_response())

StreamEvent 的结构

@dataclass
class StreamEvent:
    uuid: str               # 事件唯一 ID
    session_id: str         # 会话 ID
    event: dict[str, Any]  # 原始 Claude API 流式事件
    parent_tool_use_id: str | None  # 来自子 Agent 时的父工具 ID

三层嵌套检查的逻辑

1. 先检查 message.type 是 StreamEvent（区分于 AssistantMessage / ResultMessage）
2. 再检查 event["type"] == "content_block_delta"（过滤掉 start/stop/message 类事件）
3. 最后检查 delta["type"] == "text_delta"（区分文本片段与工具输入片段）

消息流的完整顺序

StreamEvent (message_start)
StreamEvent (content_block_start)  ← 文本块开始
StreamEvent (content_block_delta)  ← 文字逐字到达...
StreamEvent (content_block_stop)
StreamEvent (content_block_start)  ← 工具调用块开始
StreamEvent (content_block_delta)  ← 工具 JSON 输入逐步到达...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage                   ← 完整消息
... 工具执行 ...
... 下一轮流式事件 ...
ResultMessage                      ← 最终结果

流式工具调用追踪

async def stream_tool_calls():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash"],
    )
    current_tool = None
    tool_input = ""

async for message in query(prompt="读取 README.md 文件", options=options):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    current_tool = content_block.get("name")
                    tool_input = ""
                    print(f"开始调用工具: {current_tool}")

elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk

elif event_type == "content_block_stop":
                if current_tool:
                    print(f"工具 {current_tool} 调用完成，参数: {tool_input}")
                    current_tool = None

构建流式 UI

async def streaming_ui():
    options = ClaudeAgentOptions(
        include_partial_messages=True,
        allowed_tools=["Read", "Bash", "Grep"],
    )
    in_tool = False

async for message in query(
        prompt="找出代码库里所有的 TODO 注释", options=options
    ):
        if isinstance(message, StreamEvent):
            event = message.event
            event_type = event.get("type")

if event_type == "content_block_start":
                content_block = event.get("content_block", {})
                if content_block.get("type") == "tool_use":
                    tool_name = content_block.get("name")
                    print(f"\n[正在使用 {tool_name}...]", end="", flush=True)
                    in_tool = True

elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "text_delta" and not in_tool:
                    # 只在工具未运行时输出文本
                    import sys
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

elif event_type == "content_block_stop":
                if in_tool:
                    print(" 完成", flush=True)
                    in_tool = False

elif isinstance(message, ResultMessage):
            print("\n\n--- 全部完成 ---")

已知限制：结构化输出不流式

五条实践建议