Agent SDKのストリーミング出力を有効にする

Agent SDKの既定動作とストリーミング出力の違い

Agent SDKは既定では、Claudeが1回の応答を生成し終えてから AssistantMessage をまとめて返します。応答が長い場合やツール呼び出しを含む処理では、結果が届くまで画面に何も表示されない待ち時間が生まれます。トークンが1文字ずつ流れてくるチャットUIのような体験を作りたいときは、この既定動作をストリーミング出力に切り替える必要があります。

ストリーミング出力を有効にすると、SDKは完成した AssistantMessage や ResultMessage に加えて、Claude APIの生のストリーミングイベントをそのまま含む StreamEvent(TypeScriptでは SDKPartialAssistantMessage、type: "stream_event")を逐次yieldするようになります。テキストもツール呼び出しの入力も、トークン単位で組み立てられていく過程がそのまま見えるのが最大の違いです。

本記事が扱うのは出力側(応答をどう受け取るか)のストリーミングです。プロンプトをどう送るかという入力側の設定は別物で、まずはAgent SDK入門で基本構成を押さえておくと理解しやすくなります。

includePartialMessagesでストリーミングを有効にする

ストリーミングを有効にする設定は1つだけです。TypeScriptでは includePartialMessages、Pythonでは include_partial_messages を options に true で渡します。

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版も構造は同じです。

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())

受信したメッセージがまず StreamEvent かどうかを判定し、次に event.type が content_block_delta か、さらに delta.type が text_delta かという3段階のネストしたチェックを通すのがこのAPIの基本形です。テキストチャンク以外にも複数種類のイベントが同じ event フィールドに流れてくるため、この判定を省略すると意図しないデータを文字列として扱ってしまいます。

StreamEventの構造とイベント種別

StreamEvent はPython・TypeScriptで型名が異なります。Pythonは claude_agent_sdk.types から StreamEvent としてインポートし、TypeScriptは SDKPartialAssistantMessage(判定は type === "stream_event")という名前になります。どちらも中身はClaude APIの生イベントで、すでに組み立てられたテキストではなく差分(delta)そのものが入っている点は共通です。累積した文字列が欲しい場合は、自分で text_delta を連結する必要があります。

event フィールドの中身でよく使う種類は次の6つです。

サブエージェントのトークン単位の差分は転送されません。parent_tool_use_id は StreamEvent では常に空で、サブエージェントが呼ばれたかどうかを見分けるには完成した AssistantMessage 側の parent_tool_use_id を使います。

サブエージェントの呼び出しを見分ける

サブエージェントのトークン単位の差分がストリーミングされない以上、「今どのサブエージェントが動いているか」を知るには別の手がかりが要ります。サブエージェントは Agent という名前のツール(v2.1.63 より前のSDKでは Task という名前だった)として呼び出されるため、tool_use ブロックの name が Agent(または互換性のため Task も)かどうかを確認すれば呼び出しを検出できます。そして、そのサブエージェントの実行中に届くメッセージには parent_tool_use_id が付きます。

for await (const message of query({
  prompt: "Use the code-reviewer agent to review this codebase",
  options: {
    allowedTools: ["Read", "Glob", "Grep", "Agent"],
    agents: {
      "code-reviewer": {
        description: "Expert code reviewer.",
        prompt: "Analyze code quality and suggest improvements.",
        tools: ["Read", "Glob", "Grep"]
      }
    }
  }
})) {
  const msg = message as any;
  for (const block of msg.message?.content ?? []) {
    if (block.type === "tool_use" && (block.name === "Task" || block.name === "Agent")) {
      console.log(`Subagent invoked: ${block.input.subagent_type}`);
    }
  }
  if (msg.parent_tool_use_id) {
    console.log("  (running inside subagent)");
  }
}

この判定は完成したメッセージ側で行うのであって、StreamEvent の中では行いません。StreamEvent の parent_tool_use_id は仕様上常に空なので、サブエージェントの区別はここでは付けられない、という制約を思い出しておくと実装で迷いません。

メッセージが届く順序

ストリーミングを有効にすると、1回のやり取りは次の順序でメッセージが届きます。

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 - すべての内容を含む完成メッセージ
... ツールが実行される ...
... 次のターンの StreamEvent が続く ...
ResultMessage - 最終結果

ストリーミングを無効にしたときと比べ、StreamEvent が追加されるだけで AssistantMessage や ResultMessage は変わらず届きます。既存のコードがこれらの完成メッセージだけを見ているなら、ストリーミングを有効にしても壊れません。

ストリーミングを無効にした場合に届くメッセージ

比較のため、includePartialMessages を指定しない既定の状態で届くメッセージも押さえておきます。届くのは StreamEvent 以外のすべてで、具体的にはセッション初期化を示す SystemMessage、完成した応答である AssistantMessage、最終結果の ResultMessage、そして会話履歴が圧縮されたタイミングを示す境界メッセージ(TypeScriptでは SDKCompactBoundaryMessage、Pythonでは subtype が "compact_boundary" の SystemMessage)です。ストリーミングを有効にするかどうかで変わるのは StreamEvent の有無だけで、それ以外のメッセージの種類は共通しています。

ツール呼び出しをリアルタイムに追跡する

テキストと同じように、ツール呼び出しの入力もJSON断片として順に届きます。使うイベントは3種類です。

• content_block_start: ツール呼び出しが始まる(content_block.type が tool_use)
• content_block_delta の input_json_delta: 入力JSONの断片が届く
• content_block_stop: ツール呼び出しの完了

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" && event.content_block.type === "tool_use") {
      currentTool = event.content_block.name;
      toolInput = "";
    } else if (event.type === "content_block_delta" && event.delta.type === "input_json_delta") {
      toolInput += event.delta.partial_json;
    } else if (event.type === "content_block_stop" && currentTool) {
      console.log(`Tool ${currentTool} called with: ${toolInput}`);
      currentTool = null;
    }
  }
}

partial_json は完成したJSONではなく文字列の断片です。ツールの呼び出しが完了するまで蓄積してからパースする必要があり、途中経過をそのまま JSON.parse しようとすると失敗します。

テキストとツール状態を組み合わせたUIを作る

チャットUIで「Bashを実行中…」のような進捗表示を出したい場合、テキストのストリーミングとツール呼び出しの開始・終了を組み合わせます。ツール実行中はテキスト表示を止めて [Using Read...] のようなステータスに切り替え、完了したら再開するのが典型的なパターンです。in_tool のようなフラグ1つで、テキストとツール状態という2系統の情報を1つの画面に統合できます。

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":
                print(f"\n[Using {content_block.get('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", ""))
        elif event_type == "content_block_stop" and in_tool:
            print(" done", flush=True)
            in_tool = False
    elif isinstance(message, ResultMessage):
        print("\n\n--- Complete ---")

多段階のタスクを進めるエージェントの進捗を画面に見せたいときに、このパターンがそのまま使えます。

ストリーミングでは構造化出力が使えない制限

ストリーミングには既知の制限が1つあります。JSON Schemaで構造化出力を指定していても、その結果は最終の ResultMessage.structured_output にしかまとまりません。構造化出力そのものをストリーミングデルタとして受け取ることはできず、確定した値は最後のメッセージを待つ必要があります。画面に途中経過を出しながら最後に構造化データを使う設計なら、テキストデルタの表示と ResultMessage の受け取りを別の処理として書くことになります。SDKのセットアップから最小構成の組み立てまではClaude Agent SDK入門で扱っています。

ストリーミングを有効にする場面としない場面

すべてのユースケースでストリーミングが必要なわけではありません。判断の目安を早見表にまとめます。

対話的な画面を持つアプリでは有効にする価値が大きく、裏側で完結するバッチ処理では判定コードが増えるだけで恩恵が少ない、という基準で選ぶとシンプルです。

Claude CodeのCLIにもストリーミングはあるが仕組みが違う

Claude Code CLIの -p モードにも --output-format stream-json というストリーミング出力があります。ただしこちらはサブプロセスの標準出力にJSON行を吐き出す仕組みで、Agent SDKの StreamEvent のようにプログラム内でオブジェクトとして受け取る経路とは異なります。CLIの出力フォーマットとjqでの抽出方法はClaude Codeの構造化出力とストリーミングをツールに組み込むにまとめています。ライブラリとして自分のアプリに組み込むなら本記事のAgent SDK、既存のCLIをパイプでつなぐならそちらの記事が近道です。

よくあるつまずき

• StreamEvent の判定を忘れてテキストを二重に処理する: ストリーミングを有効にすると AssistantMessage も従来どおり届くため、StreamEvent のテキスト断片と完成メッセージのテキストを両方画面に出すと同じ内容が二重に表示されます
• partial_json を都度パースしようとする: 断片は完成したJSONではないため、content_block_stop が届くまで文字列として蓄積してからパースします
• サブエージェントのトークンが流れてこないと勘違いする: 仕様上サブエージェントの差分は転送されません。サブエージェントの出力を追いたい場合は完成メッセージ側の parent_tool_use_id を見ます

よくある質問

ストリーミングを有効にすると料金は変わりますか

includePartialMessages は届き方を変える設定で、AssistantMessage / ResultMessage の内容は変わりません。

StreamEventだけを見ていればAssistantMessageは無視してよいですか

用途によります。テキストをその場で表示するだけならStreamEventの断片だけで十分ですが、会話履歴として保存する、あるいは後続の処理に渡す場合は、確定済みの内容がまとまっている AssistantMessage を使うほうが安全です。

TypeScriptとPythonでStreamEventの中身に違いはありますか

型名(SDKPartialAssistantMessage と StreamEvent)が違うだけで、event フィールドが持つイベント種別や content_block_delta の構造は共通です。ttft_ms(初回トークン時間)はTypeScript側にのみ存在します。

複数のツールが並行して呼ばれるときイベントはどう届きますか

ツールごとに content_block_start から content_block_stop までの一連のイベントが独立して届きます。currentTool のような変数1つで1件ずつ追跡する設計だと、並行呼び出しがあった場合に取りこぼす可能性があるため、ツールごとに状態を分けて管理するのが安全です。

会話履歴の圧縮境界メッセージはストリーミングと関係がありますか

関係ありません。長い会話で自動的に履歴が圧縮されたタイミングを示すメッセージで、includePartialMessages の有効・無効にかかわらず届きます。ストリーミングを有効にしている場合は、このメッセージの前後にも通常どおり StreamEvent が流れます。

ツール名がTaskとAgentの両方をチェックする必要があるのはなぜですか

サブエージェントを呼ぶツールの名前は v2.1.63 で Task から Agent に変わりましたが、system:init のツール一覧や result.permission_denials[].tool_name には今も Task という古い名前が残る箇所があります。両方の値を確認しておけば、手元のSDKのバージョンが新旧どちらでも検出漏れが起きるのを防げます。

まとめ

ストリーミング出力は includePartialMessages / include_partial_messages を true にするだけで有効になり、既存の AssistantMessage / ResultMessage の処理はそのまま残ります。追加で届く StreamEvent を「種別を3段階でチェックして断片を蓄積する」という型で扱えば、テキストとツール呼び出し両方をリアルタイムに追跡できます。サブエージェントの検出だけは完成メッセージ側の役割だと切り分けておけば、実装で迷う場面は大きく減ります。チャットUIや進捗表示を作るなら、早い段階で組み込む価値のある機能です。