こんにちは。よっしーです(^^)
背景
この連載では、Claude Codeの公式ドキュメントを1ページずつ読み解いていきます。公式ドキュメントは情報が網羅されている分、「結局どの機能を、どんな場面で使えばいいのか」は自分で考える必要があり、読むのに意外と時間がかかります。そこで、私が実務で使うために読み込んだ内容を「使う場面→実例コード」の順に整理して残していくことにしました。専門家の解説というより、一次情報を読んだ記録です。推測や動作を確認していない部分には、その都度そう書きます。
公式ドキュメント：https://code.claude.com/docs/ja/agent-sdk/streaming-output
1. 一言でいうと
Agent SDK は既定では、Claude が1つの内容（テキストやツール呼び出し）を作り終えてから、完成した AssistantMessage を返します。includePartialMessages（Python は include_partial_messages）を true にすると、それに加えて、Claude API の生のストリーミングイベントが StreamEvent として届くようになり、文字が生成されるそばから画面に流したり、ツールの入力が組み立てられていく様子を追ったりできます。ただし届くのは生のイベントなので、テキストのつなぎ合わせなどは自分で行います。
第91回の「ストリーミング入力とシングルメッセージ入力」は、エージェントにメッセージを送る側の話でした。今回は受け取る側の話で、名前は似ていますが別の機能です。原文も冒頭でこの区別を明記しています。
2. どういう場面で役立つか
シーン1：チャット画面で応答を少しずつ表示する
ユーザーが待っている画面では、応答が完成するまで何も出ないより、文字が順に出てくるほうが待ち時間を短く感じます。テキストの差分（text_delta）を受け取るたびに画面に足していけば、Web のチャットでよく見る表示になります。
シーン2：いま何をしているかを見せる
エージェントがツールを何度も使う作業では、黙って時間が過ぎると利用者は不安になります。ツールの呼び出しが始まった時点で「[Using Read...]」のような状態表示を出せば、進み具合が伝わります。原文の最後の例がこれで、「複数の手順を踏むエージェントの作業中に進捗を見せたいチャット画面に役立つ」と書かれています。
シーン3：途中経過をログやデバッグに使う
どのツールにどんな入力が組み立てられているかを、確定前から見られます。生成の途中で止まった場合に、どこまで進んでいたかを調べる手がかりにもなります。
不要・向かないケース
- 最後の結果だけが要るバッチ処理：完成したメッセージや結果メッセージで足ります。ストリームイベントの分岐を書く手間が増えるだけです。
- 構造化出力の JSON を途中から使いたい：途中で流れてくる JSON は検証されていません。検証済みの値は最後の
structured_outputにしか届きません（3-6）。 - サブエージェントの文字を流したい：ストリームイベントはメインのセッションの分しか来ません（3-2）。
3. コードと仕組みの解説
原文の fenced コードブロックは7本です（テキストのストリーミング、ツール呼び出しのストリーミング、ストリーミング UI の、Python と TypeScript の対訳3組6本と、メッセージが届く順序を示すテキスト1本）。どれも内容が違うので全数を引用します。コードは原文から機械的に抜き出したもので、入れ子に由来する先頭インデントを外した以外は変えていません。順序図の中の ... は、原文が「繰り返し」「途中省略」の意味で書いているものです。原文が Python を先に並べているので、その順で載せます。
3-0. バージョン要件
| 変化 | バージョン |
|---|---|
途中で切れたストリームでも、ターンが終わる前に必ず message_stop が届く（それより前は、message_stop なしでターンが終わることがあった） | Claude Code v2.1.290 以降 |
| 上の Claude Code を同梱する SDK | TypeScript Agent SDK v0.3.290 以降、Python Agent SDK v0.2.164 以降 |
原文は、ターンが終わった後も応答が「生成中」のまま表示され続けるなら、SDK を更新するよう勧めています。SDK を更新することが同梱の CLI を更新する方法だ、というのは第94回で見たとおりです。
3-1. ストリーミング出力を有効にする
オプションで include_partial_messages（Python）または includePartialMessages（TypeScript）を true にします。すると SDK は、通常の AssistantMessage と ResultMessage に加えて、到着した生の API イベントを含む StreamEvent を返すようになります。
コードがすべきことは3段階です。
- 各メッセージの種類を見て、
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);
}
}
}
}
言語ごとの違いを整理します。
| Python | TypeScript | |
|---|---|---|
| ストリームイベントの見分け方 | isinstance(message, StreamEvent)（claude_agent_sdk.types から import） | message.type === "stream_event" |
| イベントの中身の読み方 | 辞書なので event.get("type")、delta.get("text", "") | オブジェクトなので event.type、event.delta.text |
| 改行なしで出力 | print に end="" と flush=True を付ける | process.stdout.write |
表は筆者の整理です。Python の event は辞書で、.get() で安全に取り出しています。end="" と flush=True は、改行を入れずにすぐ画面に出すための指定で、これが無いと断片がまとまって遅れて表示されます（筆者補足）。
3-2. StreamEvent の中身
部分メッセージを有効にすると、Claude API の生のストリーミングイベントがオブジェクトに包まれて届きます。型の名前は SDK で違います。
| SDK | 型 |
|---|---|
| Python | StreamEvent（claude_agent_sdk.types から import） |
| TypeScript | type: 'stream_event' を持つ SDKPartialAssistantMessage |
どちらも生の Claude API のイベントで、たまったテキストではありません。 テキストの差分は自分で取り出してつなぎ合わせる必要があります。
いくつかのフィールドに注意があります。
parent_tool_use_idは常に空（Python はNone、TypeScript はnull）。ストリームイベントはメインのセッションについてしか出ず、サブエージェントの文字単位の差分は転送されません。出力をサブエージェントに結びつけたいなら、parent_tool_use_idを持つ完成したメッセージのほうを使います（第89回の「サブエージェントの呼び出しを見分ける」）。user_message_uuid：Claude Code は、ターンの最初の（ping でない）ストリームイベントでこれを設定し、ターンが応答している相手のメッセージが変わるときに設定し直します。Python のStreamEventはこのフィールドを持っていません。
event フィールドの中身は Claude API のストリーミングイベントそのものです。よく出る種類は次のとおりです。
| イベントの種類 | 意味 |
|---|---|
message_start | 新しいメッセージの開始 |
content_block_start | 新しいコンテンツブロック（テキストかツール使用）の開始 |
content_block_delta | コンテンツへの差分の更新 |
content_block_stop | コンテンツブロックの終了 |
message_delta | メッセージ単位の更新（停止の理由、使用量） |
message_stop | メッセージの終了 |
第92回で見た「生成中の出力トークン数は message_delta の usage から読む」は、この表の5行目のことです。
3-3. メッセージが届く順序
Claude Code は、空でないコンテンツブロックが1つ完成するたびに AssistantMessage を1つ出します。テキストとツール呼び出しを含む応答なら AssistantMessage は2つになり、それぞれ自分のブロックだけを持ち、同じメッセージ ID を共有します。ID は TypeScript では message.message.id、Python では message.message_id で読みます（第77回で見た「1つの応答＝1つのメッセージではない」の具体的な姿です）。
部分メッセージを有効にすると、各 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
この図から読み取れることを整理します。
- 完成したブロック（
AssistantMessage）は、そのブロックの終了イベントより先に来る。完成品を受け取った時点で、そのブロックの差分の表示を確定させてよい（筆者の解釈）。 - ツールが実際に動くのは
message_stopの後（図の... tool executes ...）。ツール入力の差分が流れ終わった時点では、まだツールは実行されていない。 - 次のターンのストリームイベントが続き、最後に
ResultMessageが来る。
部分メッセージを有効にしない場合は、StreamEvent 以外のメッセージがすべて届きます。よく出るのは、SystemMessage（セッションの初期化）、AssistantMessage（完成したコンテンツブロック）、ResultMessage（最終結果）、それに会話履歴が圧縮されたことを示す圧縮の境界のメッセージ（TypeScript は SDKCompactBoundaryMessage、Python は subtype が "compact_boundary" の SystemMessage）です。
途中で切れたストリーム
ターンを中断した場合や接続が切れた場合など、ストリームがメッセージの途中で切れても、ターンが終わる前にそのメッセージの message_stop は届きます（v2.1.290 以降）。ブロックごとの扱いは次のとおりです。
| 切れたブロック | content_block_stop |
|---|---|
| テキスト、思考（thinking） | 届く |
| ツール呼び出し | 届かない |
表は原文の記述を筆者がまとめたものです。原文の指示は、ツール呼び出しのブロックが開いたまま message_stop が来たら、その呼び出しの入力は不完全なものとして扱うことです。3-4 の例にこの扱いが入っていない点は、そこで触れます。
3-4. ツール呼び出しをストリーミングする
ツール呼び出しも差分で流れてきます。ツールが始まった時点、入力が作られていく様子、作り終えた時点を追えます。使うイベントは3つです。
| イベント | 意味 |
|---|---|
content_block_start | ツールの開始（content_block.type が tool_use） |
content_block_delta（input_json_delta） | 入力の断片が届いた（partial_json） |
content_block_stop | ツール呼び出しの完了 |
次の例は、いま呼ばれているツールを記録し、流れてくる JSON の入力をつなぎ合わせます。
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;
}
}
}
}
partial_json は文字通り「JSON の一部」なので、途中の tool_input は JSON として成り立っていないことがほとんどです。解釈（パース）するのは content_block_stop の後にする必要があります（筆者の指摘）。この例は表示しているだけなので問題になりませんが、途中で JSON.parse や json.loads をかけると失敗します。
筆者の指摘：3-3 の「切れたツール呼び出しには content_block_stop が届かない」をこの例に当てはめると、問題が1つ見えます。ツールの入力の途中で切れると、current_tool（TypeScript は currentTool）がリセットされないまま残ります。この例は content_block_stop を受けたときに current_tool が残っていれば「called with」を出す作りなので、その後に来る別のブロック（たとえば次のターンのテキスト）の content_block_stop で、切れた不完全な入力が完了したツール呼び出しとして表示されてしまう可能性があります。原文の指示どおり不完全として扱うには、message_stop を受けたときに current_tool が残っていれば「不完全」として捨てる処理を足すのが確実です。
3-5. ストリーミング UI を作る
テキストとツールのストリーミングを1つの画面にまとめる例です。エージェントがいまツールを実行しているかを in_tool フラグ（TypeScript は inTool）で記録し、ツールの間は [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 ---");
}
}
Python は end=""・flush=True 付きの print と、sys.stdout.write ＋ sys.stdout.flush() を使い分けていますが、どちらも改行なしで即時に出す目的です。最後に ResultMessage（TypeScript は message.type === "result"）で「— Complete —」を出します。
筆者の指摘：原文はこの例を「ツール完了は『完了』メッセージをトリガーする」と説明していますが、3-3 の順序図に照らすと、「done」が出るのはツールの入力を作り終えたとき（ツール呼び出しブロックの content_block_stop）で、ツールの実行が終わったときではありません。実際のツールの実行は、その後の message_stop の後です。つまり、時間のかかるコマンドを Bash で実行している最中も、画面には「[Using Bash...] done」と出ています。「実行中」を正しく見せたいなら、content_block_stop ではなく、次のターンのストリームイベントや、ツールの結果が届いたことを合図に「done」を出す必要があるでしょう（どのメッセージで結果を受け取るかは、このページには書かれていません）。
また、in_tool の間はテキストの差分を表示しない作りですが、1つのコンテンツブロックはテキストかツール使用のどちらかなので、ツールのブロックの間にテキストの差分が混ざることは通常ありません。この条件は「念のため」の保険と読めます（筆者の解釈）。
3-6. 既知の制限：構造化出力
原文が挙げる制限は1つです。部分メッセージを有効にすると、構造化出力の JSON は、ツール呼び出しの検証されていない input_json_delta の断片として流れてきます。検証済みの結果が届くのは、最後の ResultMessage.structured_output だけです。
第98回で見たとおり、構造化出力はスキーマに合わなければ SDK が作り直しを求めます。流れてきた JSON は、作り直しで捨てられる版かもしれません。画面に「生成中」として見せるのはよくても、アプリのデータとして使うのは structured_output に限る、ということです（筆者の指摘）。この流れ方から、構造化出力の仕組みが内部ではツール呼び出しの形を取っていることも読み取れます（筆者の解釈）。
4. まとめ
- 出力のストリーミングは
includePartialMessages（Python はinclude_partial_messages）をtrue。第91回の入力のストリーミングとは別の機能。 - 届くのは生の Claude API イベントを包んだ
StreamEvent（TS はSDKPartialAssistantMessage、type: 'stream_event'）。テキストはcontent_block_deltaのtext_deltaを自分でつなぐ。 - サブエージェントの差分は来ない（
parent_tool_use_idは常に空）。user_message_uuidは Python に無い。 AssistantMessageはブロックごとに1つ、同じメッセージ ID。各AssistantMessageはそのブロックのcontent_block_stopより前に来る。ツールが実行されるのはmessage_stopの後。- 切れたストリームでも
message_stopは来る（v2.1.290 以降、TS SDK v0.3.290／Py SDK v0.2.164 以降）。切れたツール呼び出しにはcontent_block_stopが来ないので、開いたままmessage_stopが来たら入力は不完全として扱う。 - ツールの入力は
input_json_deltaのpartial_jsonをつなぐ。パースは完了後。 - UI 例の「done」は入力の生成完了であって実行完了ではない。
- 構造化出力の途中の JSON は未検証。使うのは
structured_outputだけ。
次回予告
次回は 「TODO リストの追跡」（agent-sdk/todo-tracking） を取り上げる予定です。
今回、ツールの呼び出しを追って「いま何をしているか」を画面に出す方法を見ました。エージェントは、作業を進めるときに自分で TODO リストを作って項目を消していくことがあります。次回は、その TODO リストをアプリ側でどう受け取り、作業全体の進み具合として利用者に見せられるのかを見ていきます。
何か質問や相談があれば、コメントをお願いします。また、エンジニア案件の相談にも随時対応していますので、お気軽にお問い合わせください。
それでは、また明日お会いしましょう(^^)
コメント