リアルタイムでレスポンスをストリーミングする
テキストとツール呼び出しがストリーミングされるときに、Agent SDK からリアルタイムレスポンスを取得します
デフォルトでは、Agent SDK は Claude がレスポンスの生成を完了した後に、完全な AssistantMessage オブジェクトを返します。テキストとツール呼び出しが生成されるときにインクリメンタルな更新を受け取るには、オプションで include_partial_messages（Python）または includePartialMessages（TypeScript）を true に設定して、部分的なメッセージストリーミングを有効にします。
ストリーミング出力を有効にする
ストリーミングを有効にするには、オプションで include_partial_messages（Python）または includePartialMessages（TypeScript）を true に設定します。これにより、SDK は到着した生の API イベントを含む StreamEvent メッセージを返すようになり、通常の AssistantMessage と ResultMessage に加えて返されます。
コードは以下の処理を実行する必要があります：
- 各メッセージのタイプをチェックして、
StreamEventを他のメッセージタイプから区別する StreamEventの場合、eventフィールドを抽出してそのタイプをチェックする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:
SDKPartialAssistantMessagewithtype: 'stream_event'
どちらも生の Claude API イベントを含み、蓄積されたテキストではありません。テキストデルタを自分で抽出して蓄積する必要があります。各タイプの構造は以下の通りです：
@dataclass
class StreamEvent:
uuid: str # このイベントの一意の識別子
session_id: str # セッション識別子
event: dict[str, Any] # 生の Claude API ストリームイベント
parent_tool_use_id: str | None # サブエージェントからの場合は親ツール ID
type SDKPartialAssistantMessage = {
type: "stream_event";
event: BetaRawMessageStreamEvent; // Anthropic SDK から
parent_tool_use_id: string | null;
uuid: UUID;
session_id: string;
ttft_ms?: number; // メッセージ開始イベントにのみ存在する、最初のトークンまでの時間（ミリ秒）
};
event フィールドには、Claude API からの生のストリーミングイベントが含まれます。一般的なイベントタイプは以下の通りです：
| イベントタイプ | 説明 |
|---|---|
message_start |
新しいメッセージの開始 |
content_block_start |
新しいコンテンツブロック（テキストまたはツール使用）の開始 |
content_block_delta |
コンテンツへのインクリメンタルな更新 |
content_block_stop |
コンテンツブロックの終了 |
message_delta |
メッセージレベルの更新（停止理由、使用量） |
message_stop |
メッセージの終了 |
メッセージフロー
部分的なメッセージが有効な場合、メッセージは以下の順序で返されます：
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 - 最終結果
部分的なメッセージが有効でない場合（Python では include_partial_messages、TypeScript では includePartialMessages）、StreamEvent を除くすべてのメッセージタイプを受け取ります。一般的なタイプには SystemMessage（セッション初期化）、AssistantMessage（完全なレスポンス）、ResultMessage（最終結果）、および会話履歴がコンパクト化されたときを示すコンパクト境界メッセージ（TypeScript では SDKCompactBoundaryMessage、Python では SystemMessage with subtype "compact_boundary"）が含まれます。
テキストレスポンスをストリーミングする
生成されるときにテキストを表示するには、delta.type が text_delta である content_block_delta イベントを探します。これらには、インクリメンタルなテキストチャンクが含まれます。以下の例は、各チャンクが到着するときに出力します：
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())
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 入力を蓄積します。3 つのイベントタイプを使用します：
content_block_start: ツール開始content_block_deltawithinput_json_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...] のようなステータスインジケータを表示します。ツールが実行されていないときはテキストが通常にストリーミングされ、ツール完了は「完了」メッセージをトリガーします。このパターンは、マルチステップエージェントタスク中に進捗を表示する必要があるチャットインターフェースに役立ちます。
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 結果は最終的な
ResultMessage.structured_outputにのみ表示され、ストリーミングデルタとしては表示されません。詳細は構造化出力を参照してください。
次のステップ
テキストとツール呼び出しをリアルタイムでストリーミングできるようになったので、これらの関連トピックを探索してください：
- インタラクティブ対ワンショットクエリ: ユースケースに応じて入力モードを選択する
- 構造化出力: エージェントから型付き JSON レスポンスを取得する
- 権限: エージェントが使用できるツールを制御する