实时流式输出响应

在 Agent SDK 里启用部分消息流：StreamEvent 参考、消息流转顺序、流式工具调用、构建流式 UI 示例与已知限制。

默认情况下，Agent SDK 在 Claude 生成完每个非空内容块（如文本块或工具调用）之后，为它产出一个完整的 AssistantMessage。要在文本和工具调用生成过程中接收增量更新，请启用部分消息流。本页讲输出流式（实时接收 token）；输入模式（你如何发送消息）见「向智能体发送消息」。你也可以通过 CLI 用 Agent SDK 流式传输响应。

启用流式输出

要启用流式，在选项里把 include_partial_messages（Python）或 includePartialMessages（TypeScript）设为 true。这会让 SDK 在通常的 AssistantMessage 和 ResultMessage 之外，还在原始 API 事件到达时产出携带它们的 StreamEvent 消息。你的代码随后需要：

检查每条消息的类型，把 StreamEvent 与其他消息类型区分开
对 StreamEvent，提取 event 字段并检查它的 type
找 delta.type 为 text_delta 的 content_block_delta 事件，它们包含实际的文本块

下面的例子启用流式并在文本块到达时打印它们。注意嵌套的类型检查：先检查 StreamEvent，再检查 content_block_delta，再检查 text_delta：

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="List the files in my project", 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())

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "List the files in my project",
  options: {
    includePartialMessages: true,
    allowedTools: ["Bash", "Read"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;
    if (event.type === "content_block_delta") {
      if (event.delta.type === "text_delta") {
        process.stdout.write(event.delta.text);
      }
    }
  }
}

StreamEvent 参考

启用部分消息时，你收到的是被包在对象里的原始 Claude API 流式事件。该类型在每个 SDK 里名字不同：Python——StreamEvent（从 claude_agent_sdk.types 导入）；TypeScript——带 type: 'stream_event' 的 SDKPartialAssistantMessage。两者都包含原始的 Claude API 事件，而不是累积的文本，你需要自己提取并累积文本增量。

parent_tool_use_id 字段在 Python 里总是 None、在 TypeScript 里总是 null：流事件只为主会话发出，来自子智能体的 token 级增量不会被转发；要把输出归属到某个子智能体，用带 parent_tool_use_id 的完整消息（见「检测子智能体调用」）。Claude Code 把 user_message_uuid 设在轮次的第一个非 ping 流事件上，并在轮次所应答的消息改变时再次设置（条件见 user_message_uuid）；Python 的 StreamEvent 不暴露这个字段。

event 字段包含来自 Claude API 的原始流事件。常见事件类型：

事件类型	说明
`message_start`	新消息开始
`content_block_start`	新内容块（文本或工具使用）开始
`content_block_delta`	内容的增量更新
`content_block_stop`	内容块结束
`message_delta`	消息级更新（停止原因、用量）
`message_stop`	消息结束

消息流转

Claude Code 在每个非空内容块完成时发出一个 AssistantMessage，所以带一个文本块和一个工具调用的响应会产出两个 AssistantMessage 对象。每个只携带它自己的内容块，两者共享同一个消息 ID（在 TypeScript 里读作 message.message.id，在 Python 里读作 message.message_id）。启用部分消息时，每个 AssistantMessage 都在该块的 content_block_stop 事件之前到达，你按这个顺序收到消息：

StreamEvent (message_start)
StreamEvent (content_block_start) - text block
StreamEvent (content_block_delta) - text chunks...
AssistantMessage - complete text block
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use block
StreamEvent (content_block_delta) - tool input chunks...
AssistantMessage - complete tool_use block
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
... tool executes ...
... more streaming events for next turn ...
ResultMessage - final result

没有启用部分消息时，你收到除 StreamEvent 之外的所有消息类型，常见的有 SystemMessage（会话初始化）、AssistantMessage（完整内容块）、ResultMessage（最终结果），以及指示对话历史何时被压缩的压缩边界消息（TypeScript 里是 SDKCompactBoundaryMessage；Python 里是 subtype 为 "compact_boundary" 的 SystemMessage）。

流式工具调用

工具调用也是增量流式的：你可以跟踪工具何时开始、在它生成时接收它的输入，并看到它何时完成。下面的例子跟踪当前被调用的工具，并在 JSON 输入流式到达时累积它，用到三种事件类型：content_block_start——工具开始；带 input_json_delta 的 content_block_delta——输入块到达；content_block_stop——工具调用完成。

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


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

    # 跟踪当前工具并累积它的输入 JSON
    current_tool = None
    tool_input = ""

    async for message in query(prompt="Read the README.md file", 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"Starting tool: {current_tool}")

            elif event_type == "content_block_delta":
                delta = event.get("delta", {})
                if delta.get("type") == "input_json_delta":
                    # 在 JSON 输入流式到达时累积它
                    chunk = delta.get("partial_json", "")
                    tool_input += chunk
                    print(f"  Input chunk: {chunk}")

            elif event_type == "content_block_stop":
                # 工具调用完成——显示最终输入
                if current_tool:
                    print(f"Tool {current_tool} called with: {tool_input}")
                    current_tool = None


asyncio.run(stream_tool_calls())

import { query } from "@anthropic-ai/claude-agent-sdk";

// 跟踪当前工具并累积它的输入 JSON
let currentTool: string | null = null;
let toolInput = "";

for await (const message of query({
  prompt: "Read the README.md file",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      // 新的工具调用开始
      if (event.content_block.type === "tool_use") {
        currentTool = event.content_block.name;
        toolInput = "";
        console.log(`Starting tool: ${currentTool}`);
      }
    } else if (event.type === "content_block_delta") {
      if (event.delta.type === "input_json_delta") {
        // 在 JSON 输入流式到达时累积它
        const chunk = event.delta.partial_json;
        toolInput += chunk;
        console.log(`  Input chunk: ${chunk}`);
      }
    } else if (event.type === "content_block_stop") {
      // 工具调用完成——显示最终输入
      if (currentTool) {
        console.log(`Tool ${currentTool} called with: ${toolInput}`);
        currentTool = null;
      }
    }
  }
}

构建流式 UI

这个例子把文本流式和工具流式组合成一个连贯的 UI。它跟踪智能体当前是否在执行工具（用 in_tool 标志），在工具运行时显示 [Using Read...] 这样的状态指示；不在工具里时文本正常流式输出，工具完成触发 "done" 消息。这种模式适用于需要在多步骤智能体任务期间显示进度的聊天界面。

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


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

    # 跟踪我们当前是否在工具调用里
    in_tool = False

    async for message in query(
        prompt="Find all TODO comments in the codebase", 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[Using {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:
                    sys.stdout.write(delta.get("text", ""))
                    sys.stdout.flush()

            elif event_type == "content_block_stop":
                if in_tool:
                    # 工具调用结束
                    print(" done", flush=True)
                    in_tool = False

        elif isinstance(message, ResultMessage):
            # 智能体完成了所有工作
            print(f"\n\n--- Complete ---")


asyncio.run(streaming_ui())

import { query } from "@anthropic-ai/claude-agent-sdk";

// 跟踪我们当前是否在工具调用里
let inTool = false;

for await (const message of query({
  prompt: "Find all TODO comments in the codebase",
  options: {
    includePartialMessages: true,
    allowedTools: ["Read", "Bash", "Grep"]
  }
})) {
  if (message.type === "stream_event") {
    const event = message.event;

    if (event.type === "content_block_start") {
      if (event.content_block.type === "tool_use") {
        // 工具调用开始——显示状态指示
        process.stdout.write(`\n[Using ${event.content_block.name}...]`);
        inTool = true;
      }
    } else if (event.type === "content_block_delta") {
      // 只在不执行工具时才流式输出文本
      if (event.delta.type === "text_delta" && !inTool) {
        process.stdout.write(event.delta.text);
      }
    } else if (event.type === "content_block_stop") {
      if (inTool) {
        // 工具调用结束
        console.log(" done");
        inTool = false;
      }
    }
  } else if (message.type === "result") {
    // 智能体完成了所有工作
    console.log("\n\n--- Complete ---");
  }
}

已知限制

结构化输出：启用部分消息时，JSON 以工具调用的未经验证的 input_json_delta 块流式传输，只有经过验证的结果才到达最终的 ResultMessage.structured_output（细节见「结构化输出」）。

下一步

既然你能实时流式传输文本和工具调用，可以探索这些相关主题：交互式与一次性查询（为你的用例选择输入模式）；结构化输出（从智能体得到类型化的 JSON 响应）；权限（控制智能体能使用哪些工具）。