第 2 章：理解消息系统

和 Claude 对话就像收快递 —— 你得知道包裹里装的是什么，才能正确处理。

上一章我们用 query() 和 Claude 说了第一句话，也拿到了回复。但那些 message 到底是什么东西？里面又装了些啥？这一章我们把这个"黑箱"彻底拆开看看。

2.1 Claude 返回的"包裹"里有什么？

想象你网购了一箱东西。快递包裹（Message）到了，你拆开一看，里面可能有好几件商品（ContentBlock）。有的是衣服（文字回复），有的是说明书（思考过程），有的是提货单（工具调用请求）。

用代码说话：

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, ResultMessage,
    TextBlock, ToolUseBlock, ThinkingBlock
)

async def main():
    options = ClaudeAgentOptions(
        permission_mode="bypassPermissions",  # 自动批准工具使用
        max_turns=1,                          # 只让 Claude 回复一轮
    )

    async for message in query(prompt="你好，请做一下自我介绍", options=options):
        # 每个 message 就是一个"包裹"
        print(f"收到包裹类型: {type(message).__name__}")

        # 如果是 Claude 的回复，拆开看看里面有什么
        if isinstance(message, AssistantMessage):
            print(f"  里面有 {len(message.content)} 件物品:")
            for i, block in enumerate(message.content):
                print(f"  第 {i+1} 件: {type(block).__name__}")

asyncio.run(main())

运行后你会看到类似这样的输出：

收到包裹类型: AssistantMessage
  里面有 1 件物品:
  第 1 件: TextBlock
收到包裹类型: ResultMessage

两个核心概念：

概念	比喻	说明
Message	快递包裹	一次完整的消息，有 5 种类型
ContentBlock	包裹里的物品	消息的具体内容，有 4 种类型

一个 AssistantMessage（Claude 的回复）里可能同时包含多个 ContentBlock。比如 Claude 一边回答你的问题（TextBlock），一边决定调用某个工具（ToolUseBlock），这两个就会出现在同一个消息的 content 列表里。

2.2 五种消息类型

query() 返回的消息流里，你会遇到五种类型的消息。日常开发中最常打交道的是前两种。

2.2.1 AssistantMessage —— Claude 的回复（最常用）

这是你最常见到的消息类型。每当 Claude 说话、思考、或者想用工具，都会以 AssistantMessage 的形式发给你。

@dataclass
class AssistantMessage:
    content: list[ContentBlock]    # 内容块列表（文字、思考、工具调用等）
    model: str                     # 使用的模型，如 "claude-sonnet-4-5-20250514"
    parent_tool_use_id: str | None # 如果是工具调用的后续回复，这里会有对应的 ID
    error: str | None              # 如果出错了，这里会有错误类型

用法：

if isinstance(message, AssistantMessage):
    print(f"模型: {message.model}")
    for block in message.content:
        if isinstance(block, TextBlock):
            print(f"Claude 说: {block.text}")

2.2.2 ResultMessage —— 对话结束信号

这是每次对话的"收据"，告诉你这次对话花了多少钱、用了多长时间。它永远是消息流中的最后一条消息。

@dataclass
class ResultMessage:
    subtype: str                       # 结果子类型
    duration_ms: int                   # 总耗时（毫秒）
    duration_api_ms: int               # API 调用耗时（毫秒）
    is_error: bool                     # 是否出错
    num_turns: int                     # 对话轮数
    session_id: str                    # 会话 ID
    total_cost_usd: float | None       # 花了多少美元
    usage: dict[str, Any] | None       # Token 用量详情
    result: str | None                 # 最终结果文本

这就好比你去餐厅吃完饭，服务员递来的账单：

if isinstance(message, ResultMessage):
    print(f"耗时: {message.duration_ms / 1000:.1f} 秒")
    print(f"费用: ${message.total_cost_usd:.4f}" if message.total_cost_usd else "费用: 免费（订阅用户）")
    print(f"对话轮数: {message.num_turns}")
    print(f"是否出错: {message.is_error}")
    print(f"会话 ID: {message.session_id}")

2.2.3 UserMessage —— 你发的消息

@dataclass
class UserMessage:
    content: str | list[ContentBlock]  # 消息内容
    uuid: str | None                   # 消息唯一标识
    parent_tool_use_id: str | None     # 关联的工具调用 ID

在 query() 模式下，你通常看不到 UserMessage，因为你的输入直接传给了 Claude，不会回显给你。

但在 ClaudeSDKClient（多轮对话模式）里，当 Claude 调用工具后，SDK 会自动把工具执行结果封装成 UserMessage 发回给 Claude。这时你就能在消息流里看到它了。

2.2.4 SystemMessage —— 系统级通知

@dataclass
class SystemMessage:
    subtype: str               # 通知子类型（如 "init" 等）
    data: dict[str, Any]       # 通知携带的数据

系统消息是 SDK 内部的通知，比如初始化完成、状态变更等。大多数情况下你可以直接忽略它。如果你好奇，可以打印出来看看：

if isinstance(message, SystemMessage):
    print(f"系统通知: {message.subtype}")
    print(f"通知数据: {message.data}")

2.2.5 StreamEvent —— 流式事件（高级）

@dataclass
class StreamEvent:
    uuid: str                          # 事件唯一标识
    session_id: str                    # 会话 ID
    event: dict[str, Any]              # 原始的 Anthropic API 流式事件
    parent_tool_use_id: str | None     # 关联的工具调用 ID

StreamEvent 是部分消息的实时更新 —— 就像你看直播打字，一个字一个字蹦出来。需要在 ClaudeAgentOptions 里设置 include_partial_messages=True 才能收到。初学阶段先跳过。

消息类型速查表

类型	出现频率	说明	你需要关注吗？
`AssistantMessage`	每次都有	Claude 的回复，包含文字、工具调用等	必须关注
`ResultMessage`	每次最后一条	对话结束的"收据"	建议关注
`UserMessage`	ClaudeSDKClient 模式	你发的消息或工具执行结果	看场景
`SystemMessage`	偶尔	系统级通知	通常忽略
`StreamEvent`	需手动开启	流式部分更新	高级用法

2.3 四种内容块

AssistantMessage 的 content 字段是一个列表，里面装的就是 ContentBlock。一共有四种。

2.3.1 TextBlock —— 最常见，Claude 的文字回复

@dataclass
class TextBlock:
    text: str  # Claude 说的话

这是你最常遇到的内容块，没什么花头，就是一段文字：

if isinstance(block, TextBlock):
    print(block.text)

2.3.2 ThinkingBlock —— Claude 的"内心独白"

@dataclass
class ThinkingBlock:
    thinking: str   # 思考过程的文字
    signature: str  # 签名（用于验证）

有时候我们希望 Claude 在回答之前先"想一想"，特别是处理复杂数学题、逻辑推理的时候。开启扩展思考（Extended Thinking）后，你就能看到 ThinkingBlock —— 这就像是 Claude 的内心独白，让你知道它是怎么一步步推理出答案的。

什么时候会出现？ 只有显式开启扩展思考配置后才会出现，默认不会有。后面 2.5 节会详细讲怎么开启。

2.3.3 ToolUseBlock —— Claude 想用某个工具

@dataclass
class ToolUseBlock:
    id: str                    # 这次工具调用的唯一 ID
    name: str                  # 工具名称，如 "Bash"、"Read"、"Write"
    input: dict[str, Any]      # 工具参数

Claude 不只会说话，它还会"动手"。当 Claude 决定要执行命令、读文件、写文件时，它会发出一个 ToolUseBlock，相当于跟你说："我想用这个工具，参数是这些。"

if isinstance(block, ToolUseBlock):
    print(f"Claude 想使用工具: {block.name}")
    print(f"参数: {block.input}")
    # 例如：
    # Claude 想使用工具: Bash
    # 参数: {"command": "echo 'hello world'"}

常见的内置工具名：

工具名	用途
`Bash`	执行终端命令
`Read`	读取文件内容
`Write`	写入文件
`Edit`	编辑文件（精确替换）
`Glob`	按模式搜索文件
`Grep`	在文件中搜索内容

2.3.4 ToolResultBlock —— 工具执行的结果

@dataclass
class ToolResultBlock:
    tool_use_id: str                                  # 对应哪次工具调用
    content: str | list[dict[str, Any]] | None        # 执行结果
    is_error: bool | None                             # 是否出错

工具执行完后，结果会以 ToolResultBlock 的形式出现。tool_use_id 跟 ToolUseBlock 的 id 对应，这样你就知道这是哪次调用的结果。

if isinstance(block, ToolResultBlock):
    print(f"工具调用 {block.tool_use_id} 的结果:")
    print(f"  内容: {block.content}")
    print(f"  是否出错: {block.is_error}")

内容块速查表

类型	比喻	出现条件
`TextBlock`	Claude 说的话	几乎每次都有
`ThinkingBlock`	Claude 的内心独白	需要开启扩展思考
`ToolUseBlock`	"我想用这个工具"	Claude 需要执行操作时
`ToolResultBlock`	工具执行的回执	工具执行完毕后

2.4 实战：写一个消息检查器

光看不练假把式。来写一个"消息检查器"，让 Claude 执行一个会用到工具的任务，然后把每种消息和内容块的详细信息都打印出来。

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, UserMessage, SystemMessage, ResultMessage,
    TextBlock, ThinkingBlock, ToolUseBlock, ToolResultBlock,
)

async def main():
    options = ClaudeAgentOptions(
        permission_mode="bypassPermissions",  # 自动批准所有工具使用
        max_turns=2,                          # 允许 2 轮（调用工具 + 返回结果）
    )

    # 让 Claude 执行一个简单的命令，这样我们能看到工具调用相关的消息
    prompt = "请用 Bash 工具运行 echo 'Hello from Claude!'，然后告诉我结果。"

    print("=" * 60)
    print("消息检查器 - 观察 Claude 返回的每一条消息")
    print("=" * 60)

    message_count = 0
    async for message in query(prompt=prompt, options=options):
        message_count += 1
        print(f"\n--- 消息 #{message_count} ---")
        print(f"类型: {type(message).__name__}")

        if isinstance(message, AssistantMessage):
            print(f"模型: {message.model}")
            if message.error:
                print(f"错误: {message.error}")
            print(f"内容块数量: {len(message.content)}")

            for i, block in enumerate(message.content):
                print(f"\n  [内容块 {i+1}] {type(block).__name__}")

                if isinstance(block, TextBlock):
                    # 文字太长就截断显示
                    text = block.text[:200] + "..." if len(block.text) > 200 else block.text
                    print(f"    文本: {text}")

                elif isinstance(block, ThinkingBlock):
                    thinking = block.thinking[:200] + "..." if len(block.thinking) > 200 else block.thinking
                    print(f"    思考过程: {thinking}")

                elif isinstance(block, ToolUseBlock):
                    print(f"    工具名: {block.name}")
                    print(f"    调用ID: {block.id}")
                    print(f"    参数: {block.input}")

                elif isinstance(block, ToolResultBlock):
                    print(f"    对应调用ID: {block.tool_use_id}")
                    print(f"    是否出错: {block.is_error}")
                    content = str(block.content)
                    content = content[:200] + "..." if len(content) > 200 else content
                    print(f"    结果: {content}")

        elif isinstance(message, UserMessage):
            content = str(message.content)
            content = content[:200] + "..." if len(content) > 200 else content
            print(f"内容: {content}")
            if message.parent_tool_use_id:
                print(f"关联工具调用ID: {message.parent_tool_use_id}")

        elif isinstance(message, SystemMessage):
            print(f"子类型: {message.subtype}")
            print(f"数据: {message.data}")

        elif isinstance(message, ResultMessage):
            print(f"是否出错: {message.is_error}")
            print(f"对话轮数: {message.num_turns}")
            print(f"总耗时: {message.duration_ms / 1000:.1f} 秒")
            print(f"API 耗时: {message.duration_api_ms / 1000:.1f} 秒")
            if message.total_cost_usd is not None:
                print(f"费用: ${message.total_cost_usd:.4f}")
            else:
                print("费用: 无（订阅用户）")
            print(f"会话ID: {message.session_id}")
            if message.result:
                result = message.result[:200] + "..." if len(message.result) > 200 else message.result
                print(f"结果: {result}")

    print(f"\n{'=' * 60}")
    print(f"总共收到 {message_count} 条消息")

asyncio.run(main())

完整代码见 examples/message_inspector.py

运行后你大概会看到这样的消息序列：

--- 消息 #1 ---
类型: AssistantMessage
模型: claude-sonnet-4-5-20250514
内容块数量: 2

  [内容块 1] TextBlock
    文本: 我来帮你运行这个命令。

  [内容块 2] ToolUseBlock
    工具名: Bash
    调用ID: toolu_01XYZ...
    参数: {'command': "echo 'Hello from Claude!'"}

--- 消息 #2 ---
类型: AssistantMessage
模型: claude-sonnet-4-5-20250514
内容块数量: 1

  [内容块 1] TextBlock
    文本: 命令已成功运行，输出结果是：Hello from Claude!

--- 消息 #3 ---
类型: ResultMessage
是否出错: False
对话轮数: 2
总耗时: 3.2 秒
...

注意观察消息序列的规律： 1. Claude 先说一句话（TextBlock）+ 发起工具调用（ToolUseBlock） 2. SDK 自动执行工具，把结果反馈给 Claude 3. Claude 根据工具结果给出最终回答（TextBlock） 4. 最后收到 ResultMessage，对话结束

2.5 扩展思考模式

有些问题需要 Claude "深思熟虑" —— 比如复杂的数学比较、逻辑推理、代码分析。这时候可以开启扩展思考，让 Claude 先在"脑子里"想清楚，再给你答案。

怎么开启？

有两种方式：

方式一：精确控制思考预算

options = ClaudeAgentOptions(
    thinking={"type": "enabled", "budget_tokens": 10000},  # 最多用 10000 个 token 来思考
    max_turns=1,
)

方式二：用 effort 参数（更简单）

options = ClaudeAgentOptions(
    effort="high",    # 可选 "low"、"medium"、"high"、"max"
    max_turns=1,
)

effort 更直观 —— 就像告诉 Claude "这道题你多想想"。thinking 则给你精确控制思考预算的能力。

看到 ThinkingBlock 时怎么处理？

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ThinkingBlock,
)

async def main():
    options = ClaudeAgentOptions(
        thinking={"type": "enabled", "budget_tokens": 10000},
        max_turns=1,
    )

    async for message in query(prompt="9.11 和 9.8 哪个大？请仔细想想。", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ThinkingBlock):
                    print("Claude 的思考过程:")
                    print("-" * 40)
                    print(block.thinking)
                    print("-" * 40)
                    print()
                elif isinstance(block, TextBlock):
                    print(f"Claude 的最终回答: {block.text}")

asyncio.run(main())

完整代码见 examples/thinking_example.py

运行后你会看到类似这样的输出：

Claude 的思考过程:
----------------------------------------
用户问的是 9.11 和 9.8 哪个大。
这是在比较两个小数。
9.11 = 9 + 0.11
9.8 = 9 + 0.8
0.8 > 0.11
所以 9.8 > 9.11
----------------------------------------

Claude 的最终回答: 9.8 比 9.11 大。虽然 11 比 8 大，但在小数中...

小贴士：很多人（包括某些 AI）会被 "9.11 vs 9.8" 这个问题绊倒，因为直觉上 11 > 8。但开启扩展思考后，Claude 会像做数学题一样一步步推理，大大降低出错概率。

2.6 小结

这一章我们彻底搞清楚了 Claude 消息系统的结构：

Message 是包裹：有 5 种类型（AssistantMessage、ResultMessage、UserMessage、SystemMessage、StreamEvent）
ContentBlock 是包裹里的物品：有 4 种类型（TextBlock、ThinkingBlock、ToolUseBlock、ToolResultBlock）
最常用的组合：AssistantMessage + TextBlock —— Claude 的文字回复
想看 Claude 怎么想的：开启扩展思考，观察 ThinkingBlock
想知道花了多少钱：看最后的 ResultMessage

掌握了消息系统，你就能准确地从 Claude 的回复中提取出你想要的信息。下一章我们来聊聊 ClaudeAgentOptions —— 也就是怎么"指挥"Claude 按你的意思来工作。