Appearance
Agent SDK 默认在 Claude 生成完一次响应后一次性返回 AssistantMessage。要实时看到逐段生成的文本和工具调用，需在选项中启用 include_partial_messages（Python）或 includePartialMessages（TypeScript）。启用后 SDK 会额外产出 StreamEvent，你需要从事件类型中筛选 content_block_delta 和 text_delta 来打印文本，或通过 content_block_start/input_json_delta 跟踪工具调用。注意：显式设置了 max_thinking_tokens 或使用结构化输出时，流式事件不会触发。
Claude Code Agent SDK 流式输出配置与实时响应
启用流式输出
设置 include_partial_messages=True（Python）或 includePartialMessages: true（TypeScript）即可启用。启用后，除了原有的 AssistantMessage 和 ResultMessage，SDK 还会产出 StreamEvent 消息，封装了原始 API 事件。
你的代码需要：
- 检查每条消息的类型，区分
StreamEvent和其他消息 - 对
StreamEvent提取event字段，检查其type - 寻找
content_block_delta事件中delta.type为text_delta的片段，这就是文本增量
下面是启用流式输出并实时打印文本块的例子：
python
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())typescript
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：
SDKPartialAssistantMessage，其type为'stream_event'
两者都包含原始 API 事件，而非累积的文本。你需要自行提取并累积文本增量。下面是两种结构的定义：
python
@dataclass
class StreamEvent:
uuid: str # 事件唯一标识
session_id: str # 会话标识
event: dict[str, Any] # 原始 Claude API 流事件
parent_tool_use_id: str | None # 子代理的父工具 ID（如果有）typescript
type SDKPartialAssistantMessage = {
type: "stream_event";
event: BetaRawMessageStreamEvent; // 来自 Anthropic SDK
parent_tool_use_id: string | null;
uuid: UUID;
session_id: string;
};event 字段是原始流事件，符合 Claude API 流事件格式。常见事件类型：
| 事件类型 | 描述 |
|---|---|
message_start | 新消息开始 |
content_block_start | 新的内容块开始（文本或工具调用） |
content_block_delta | 内容的增量更新 |
content_block_stop | 内容块结束 |
message_delta | 消息层级的更新（停止原因、用量） |
message_stop | 消息结束 |
消息流动顺序
开启流式输出后，消息按以下顺序到达：
text
StreamEvent (message_start)
StreamEvent (content_block_start) - 文本块
StreamEvent (content_block_delta) - 文本片段...
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_use 块
StreamEvent (content_block_delta) - 工具输入片段...
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
AssistantMessage - 完整消息，包含所有内容
... 工具执行 ...
... 下一轮流事件 ...
ResultMessage - 最终结果如果未启用流式输出（include_partial_messages/includePartialMessages 为 false），你将收到除了 StreamEvent 之外的所有消息类型，包括 SystemMessage（会话初始化）、AssistantMessage（完整响应）、ResultMessage（最终结果），以及一个紧凑的边界消息（表示会话历史被压缩，TypeScript 中为 SDKCompactBoundaryMessage，Python 中为 SystemMessage 且 subtype 为 "compact_boundary"）。
实时输出文本
要显示实时生成的文本，寻找 content_block_delta 事件且 delta.type 为 text_delta，这些事件包含增量文本片段。下面的例子打印每个片段：
python
from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent
import asyncio
async def stream_text():
options = ClaudeAgentOptions(include_partial_messages=True)
async for message in query(prompt="Explain how databases work", 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)
print()
asyncio.run(stream_text())typescript
import { query } from "@anthropic-ai/claude-agent-sdk";
for await (const message of query({
prompt: "Explain how databases work",
options: { includePartialMessages: true }
})) {
if (message.type === "stream_event") {
const event = message.event;
if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
process.stdout.write(event.delta.text);
}
}
}
console.log();实时输出工具调用
工具调用也是增量流式的。你可以跟踪工具开始、接收 JSON 输入片段、以及工具完成。下面的例子使用了三种事件：
content_block_start：工具开始content_block_delta且delta.type为input_json_delta：输入片段到达content_block_stop：工具调用完成
python
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"],
)
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":
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())typescript
import { query } from "@anthropic-ai/claude-agent-sdk";
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") {
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 标志跟踪 agent 是否正在执行工具，以便在工具运行时显示状态（如 [Using Read...]），工具完成时显示 "done"。不在工具调用阶段时正常流式输出文本。这种模式适用于需要展示多步 agent 任务进度的聊天界面。
python
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())typescript
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 ---");
}
}已知限制
部分 SDK 功能与流式输出不兼容：
- 扩展思考（Extended thinking）：当你显式设置了
max_thinking_tokens（Python）或maxThinkingTokens（TypeScript），StreamEvent消息不会触发。你只会收到每次回答后的完整消息。注意：SDK 默认未启用思考，因此只要你不启用，流式输出就正常工作。 - 结构化输出（Structured output）：JSON 结果只在最终的
ResultMessage.structured_output中出现，不会以流式增量形式提供。详见 结构化输出。
下一步
你已经可以实时流式输出文本和工具调用，接下来可以探索：
- 交互模式 vs 一次性查询：根据你的用例选择输入模式
- 结构化输出：从 agent 获取类型化的 JSON 响应
- 权限控制：控制 agent 可以使用哪些工具
常见问题
为什么我启用了 include_partial_messages 但收不到 StreamEvent？
检查你是否显式设置了 max_thinking_tokens（Python）或 maxThinkingTokens（TypeScript）。如果设置了扩展思考，流式事件会被禁用，你只会收到完整消息。另外，确认选项对象是否正确传入了 query 调用。
如何同时流式显示文本和工具调用状态？
参考 构建流式 UI 部分，使用 in_tool 标志区分。当触发 content_block_start 且内容块类型为 tool_use 时设置标志为 true，在 content_block_stop 时设置回 false。只在不执行工具时打印文本增量。
结构化输出模式下能收到流式事件吗？
不能。结构化输出的 JSON 结果只出现在最终的 ResultMessage.structured_output 中，不会以 text_delta 或 input_json_delta 形式流式输出。如果你需要流式显示，可以考虑不启用结构化输出，而是自行解析 Agent 的最终回答。