Claude Code SDK #5：消息流全解——6 种消息类型 × async for，把 Agent 执行过程变成可编程事件流

为什么 query() 要返回「流」而不是「结果」

• 进度可见：工具被调用时立刻知道，不用盲等
• 中间态可干预：配合 Hooks，可以在工具调用前/后插入自己的逻辑（详见后续 Hooks 篇）
• 成本可核：ResultMessage 携带精确 token 用量和估算费用，方便计量

消息类型总览

Message = (
    UserMessage
    | AssistantMessage
    | SystemMessage
    | ResultMessage
    | StreamEvent
    | RateLimitEvent
)

逐类型解剖

SystemMessage：会话的「开机广播」

@dataclass
class SystemMessage:
    subtype: str          # "init"、"session_end" 等
    data: dict[str, Any]  # 元数据，因 subtype 不同而异

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, SystemMessage

async def main():
    session_id = None
    async for message in query(
        prompt="分析 utils.py 里的函数",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"]),
    ):
        if isinstance(message, SystemMessage) and message.subtype == "init":
            session_id = message.data["session_id"]
            print(f"Session ID: {session_id}")

asyncio.run(main())

AssistantMessage：Claude 的每轮回复

@dataclass
class AssistantMessage:
    content: list[ContentBlock]  # TextBlock / ToolUseBlock 等
    model: str                   # 生成该条回复的模型名
    parent_tool_use_id: str | None = None  # 子 Agent 场景：父工具 ID
    error: AssistantMessageError | None = None  # 出错时的错误类型
    usage: dict[str, Any] | None = None   # 本条消息的 token 用量
    message_id: str | None = None         # API 消息 ID

• TextBlock：Claude 的文本输出，读 .text 字段
• ToolUseBlock：Claude 决定调用工具，包含工具名和参数

from claude_agent_sdk import AssistantMessage, TextBlock

async for message in query(prompt="..."):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"Claude 说：{block.text}")
            # 工具调用不在这里处理，SDK 会自动执行

UserMessage：工具结果的回传通道

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None    # 对应哪次工具调用
    tool_use_result: dict[str, Any] | None = None  # 工具执行结果

ResultMessage：任务终结符，带成本清单

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int          # 总耗时（毫秒）
    duration_api_ms: int      # API 调用累计耗时（毫秒）
    is_error: bool            # 是否以错误结束
    num_turns: int            # Agent 执行的回合数
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None  # 估算总费用（美元）
    usage: dict[str, Any] | None = None  # 含 input_tokens、output_tokens、
                                          # cache_creation_input_tokens、
                                          # cache_read_input_tokens
    result: str | None = None            # 最终文本结果
    structured_output: Any = None        # 结构化输出（需配合 --json-schema）

from claude_agent_sdk import ResultMessage

async for message in query(prompt="总结这个代码库"):
    if isinstance(message, ResultMessage):
        if message.is_error:
            print(f"执行失败，原因：{message.stop_reason}")
        else:
            print(f"结果：{message.result}")
            print(f"耗时：{message.duration_ms}ms")
            print(f"费用：${message.total_cost_usd:.4f}")
            print(f"Token 用量：{message.usage}")

五种典型消费模式

1. 只要最终结果（最简模式）

async for message in query(prompt="..."):
    if hasattr(message, "result"):
        print(message.result)

2. 实时进度跟踪（长任务必备）

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    SystemMessage, AssistantMessage, TextBlock, ResultMessage
)

async for message in query(
    prompt="重构整个 src/ 目录",
    options=ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Bash", "Glob"]),
):
    if isinstance(message, SystemMessage) and message.subtype == "init":
        print(f"[开始] session={message.data['session_id']}")
    elif isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock) and block.text:
                print(f"[Claude] {block.text[:80]}...")
    elif isinstance(message, ResultMessage):
        print(f"[完成] {message.num_turns} 轮 | {message.duration_ms}ms | ${message.total_cost_usd:.4f}")

3. 成本监控（生产环境必备）

costs = []
async for message in query(prompt="..."):
    if isinstance(message, ResultMessage):
        costs.append({
            "cost": message.total_cost_usd,
            "input_tokens": message.usage.get("input_tokens", 0),
            "output_tokens": message.usage.get("output_tokens", 0),
            "cache_read": message.usage.get("cache_read_input_tokens", 0),
        })

4. 错误分级处理

from claude_agent_sdk import AssistantMessage

async for message in query(prompt="..."):
    if isinstance(message, AssistantMessage) and message.error:
        if message.error == "rate_limit":
            print("触发限速，稍后重试")
        elif message.error == "max_output_tokens":
            print("输出超长，考虑拆分任务")
        elif message.error in ("billing_error", "authentication_failed"):
            raise RuntimeError(f"账号问题：{message.error}")

5. 子 Agent 消息过滤（多 Agent 场景）

async for message in query(
    prompt="用 code-reviewer 审查代码",
    options=ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Agent"],
        agents={"code-reviewer": AgentDefinition(...)},
    ),
):
    if isinstance(message, AssistantMessage):
        if message.parent_tool_use_id is None:
            print(f"[主 Agent] {message}")
        else:
            print(f"[子 Agent:{message.parent_tool_use_id[:8]}] {message}")

一个坑：不要用 break 提前退出

# ❌ 有潜在问题
async for message in query(prompt="..."):
    if isinstance(message, ResultMessage):
        result = message.result
        break  # 可能导致 asyncio 清理异常

# ✅ 推荐写法
result = None
async for message in query(prompt="..."):
    if isinstance(message, ResultMessage):
        result = message.result
        # 不 break，让流自然结束

CLI 侧的流式输出：stream-json

# 实时处理每条消息
claude -p "分析 src/ 目录" --output-format stream-json \
  | jq 'select(.type == "result") | {result, cost: .total_cost_usd}'

# 配合 --include-partial-messages 获取更细粒度的流式事件
claude -p "..." --output-format stream-json \
  --verbose --include-partial-messages

实践建议