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

既定の動作はブロック単位、有効にするとトークン単位になる

Agent SDKは既定で、テキストブロックやツール呼び出しといった内容ブロックが1つ生成し終わるたびに、完成した AssistantMessage を返します。1回の応答全体ではなくブロック単位ですが、長い文章を書いている間は、ブロックが閉じるまで画面に何も出ません。

ストリーミング出力を有効にすると、SDKは AssistantMessage と ResultMessage に加えて、Claude APIの生のストリーミングイベントを包んだ StreamEvent を逐次返します。TypeScriptでの型名は SDKPartialAssistantMessage(type: "stream_event")です。

本記事は出力側(応答をどう受け取るか)の設定を扱います。プロンプトの送り方は別の設定なので、基本構成はAgent SDK入門で押さえておくと読み進めやすくなります。セッションの状態(作業中・入力待ち)だけを知りたいときは、session_state_changedの受け取り方が近道です。

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

テキストを取り出すまでに、型の判定が3段階あります。

どれかを省くと、テキスト以外のデータを文字列として出力してしまいます。

AssistantMessageはブロックごと、StreamEventの前に届く

ストリーミングを有効にしたときの順序は、完成メッセージが「応答の最後に1回」ではありません。AssistantMessage は、内容ブロックが1つ完成するたびに届きます。しかも、そのブロックの content_block_stop より前です。

StreamEvent (message_start)
StreamEvent (content_block_start) - テキストブロック
StreamEvent (content_block_delta) - テキストの断片…
AssistantMessage - テキストブロックの完成形
StreamEvent (content_block_stop)
StreamEvent (content_block_start) - tool_useブロック
StreamEvent (content_block_delta) - ツール入力の断片…
AssistantMessage - tool_useブロックの完成形
StreamEvent (content_block_stop)
StreamEvent (message_delta)
StreamEvent (message_stop)
... ツールが実行される ...
... 次のターンの StreamEvent が続く ...
ResultMessage - 最終結果

テキストとツール呼び出しを含む応答なら、AssistantMessage は2つです。それぞれが自分のブロックだけを持ち、同じメッセージIDを共有します。IDはTypeScriptで message.message.id、Pythonで message.message_id です。

ストリーミングを無効にしたときと比べて増えるのは StreamEvent だけで、完成メッセージの種類は変わりません。完成メッセージだけを見る既存コードは、有効にしても壊れません。

StreamEventの中身はAPIの生イベントそのもの

StreamEvent は、Pythonでは claude_agent_sdk.types から、TypeScriptでは SDKPartialAssistantMessage として扱います。どちらも中身はClaude APIの生イベントで、組み立て済みのテキストではなく差分が入っています。累積した文字列が欲しければ、自分で text_delta を連結します。

Managed AgentsのSSEイベントストリームも差分ベースですが、実装パターンはAgent SDKとは別物です。

event の種類は、よく使うもので次の6つです。

これらのイベントには、ブロックの位置を示す index が付きます。最終メッセージの content 配列の添字と対応するので、複数ブロックの差分を取り違えずに組み立てられます。差分の種類は、text_delta と input_json_delta のほかに、思考内容の thinking_delta などがあります。

ツール呼び出しの入力をJSON断片から組み立てる

テキストと同じく、ツールの入力も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ではなく、文字列の断片です。content_block_stop まで蓄積してからパースします。途中で JSON.parse すると失敗します。

「Bashを実行中…」を出すUIは1つのフラグで足りる

チャットUIで進捗を見せるには、テキストの差分とツールの開始・終了を組み合わせます。ツール実行中はテキストの表示を止めて [Using Read...] のような状態表示に切り替え、完了したら再開します。in_tool のようなフラグ1つで、2系統の情報を1画面に統合できます。Managed Agentsを人気のチャットフレームワークと組み合わせる場合も、この種のツール状態の表示ロジックはフロントエンド側が引き受けます。

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

サブエージェントの出力を追いたいとき

StreamEvent に流れるのはメインセッションのイベントだけです。サブエージェントのトークン単位の差分は転送されず、parent_tool_use_id も常に空です。サブエージェントを区別するには、parent_tool_use_id を持つ完成メッセージを使います。

TypeScriptには、サブエージェントのテキストと思考を完成メッセージとして受け取るオプション forwardSubagentText があります。既定は false で、無効のときはサブエージェントの tool_use と tool_result のブロックだけが出ます。すべての階層のサブエージェントが転送されるのはClaude Code v2.1.219以降です。

サブエージェントを呼ぶツールの名前は Agent です。Claude Code v2.1.63より前の tool_use ブロックでは Task でした。system:init のツール一覧には今も Task と出るため、block.name は両方を照合します。複数エージェントを編成する別の設計は、コーディネーターとスレッド分離で扱っています。

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 の中では、サブエージェントの区別は付きません。

ストリーミングと構造化出力を併用するとき

JSON Schemaで構造化出力を指定してストリーミングも有効にすると、JSONはツール呼び出しの input_json_delta の断片として流れます。この断片は検証前のものです。スキーマで検証された結果は、最後の ResultMessage.structured_output にだけ入ります。

画面に途中経過を出しつつ最後に構造化データを使う設計なら、テキストの表示と ResultMessage の受け取りを別の処理として書きます。途中で届く断片は、スキーマに合う保証がないので、画面表示のための素材にとどめます。

SDKのセットアップから最小構成まではClaude Agent SDK入門に、outputFormatの設定やZod・Pydanticでの型付けはAgent SDKの構造化出力入門にあります。

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

すべての処理にストリーミングが要るわけではありません。

画面を持つアプリなら価値が大きく、裏側で完結する処理では恩恵が少ない、という基準で選べます。

Claude CodeのCLIにも同名のフラグがある

Claude Code CLIの -p モードにも、--output-format stream-json というストリーミング出力があります。標準出力にJSONの行を流す仕組みで、プログラム内でオブジェクトとして受け取るAgent SDKの StreamEvent とは経路が違います。SDKの includePartialMessages に当たるのは、CLIの --include-partial-messages です。

claude --help | grep -A3 "include-partial-messages"

v2.1.287の出力は次のとおりです。

  --include-partial-messages            Include partial message chunks as they
                                        arrive (only works with --print and
                                        --output-format=stream-json)

「--print と --output-format=stream-json の両方が前提」という条件がヘルプに明記されています。--output-format の選択肢は text(既定)、json(単一の結果)、stream-json(リアルタイム)の3つです。CLIリファレンスにある --output-format stream-json の実行例には --verbose が付いています。--include-partial-messages の例も claude -p --output-format stream-json --verbose --include-partial-messages "query" の形です。

自分のアプリにライブラリとして組み込むなら本記事のAgent SDK、既存のCLIをパイプでつなぐなら、jqでの抽出方法も含めてClaude Codeの構造化出力とストリーミングをツールに組み込むが近道です。

よくあるつまずき

よくある質問

初回トークンまでの時間はどこで測れますか

TypeScriptでは、message_start の StreamEvent に付く ttft_ms が使えます。成功した ResultMessage にも ttft_ms と ttft_stream_ms があります。前者は最初の完成した AssistantMessage が届くまでの時間、後者は最初の message_start イベントが届くまでの時間で、後者のほうが短くなります。Pythonの StreamEvent には ttft_ms がありません。

無反応が続いても StreamEvent は届きますか

Claude APIは、応答の途中に ping イベントを挟むことがあります。ANTHROPIC_BASE_URL の先のゲートウェイが応答をkeep-aliveで保持している間も、includePartialMessages を有効にしたホストには ping が届きます。これは生存確認のフレームなので、沈黙とみなして切断する判定から外します。v2.1.257より前は、最後の実イベントから5分で ping が止まっていました。

複数のツールが並行して呼ばれるときの追跡はどうしますか

currentTool のような変数1つで追う設計は、1件ずつの前提です。各イベントの index をキーにして、ツールごとに入力文字列を分けて持つと、複数のブロックでも取り違えにくくなります。

まとめ

ストリーミング出力は includePartialMessages を true にするだけで有効になり、完成メッセージの処理はそのまま使えます。画面を持つアプリなら有効にする価値が大きく、結果だけを保存する処理や構造化出力が主目的の処理では、ResultMessage を受け取るだけで足ります。