リアルタイムでレスポンスをストリーミングする

デフォルトでは、Agent SDK は Claude がレスポンスの生成を完了した後に、完全な AssistantMessage オブジェクトを返します。テキストとツール呼び出しが生成されるときにインクリメンタルな更新を受け取るには、オプションで include_partial_messages（Python）または includePartialMessages（TypeScript）を true に設定して、部分的なメッセージストリーミングを有効にします。

このページは出力ストリーミング（リアルタイムでトークンを受け取ること）について説明しています。入力モード（メッセージの送信方法）については、エージェントにメッセージを送信するを参照してください。また、CLI 経由で Agent SDK を使用してレスポンスをストリーミングすることもできます。

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

ストリーミングを有効にするには、オプションで 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: SDKPartialAssistantMessage with type: 'stream_event'

どちらも生の Claude API イベントを含み、蓄積されたテキストではありません。テキストデルタを自分で抽出して蓄積する必要があります。各タイプの構造は以下の通りです：

@dataclass
class StreamEvent:
    uuid: str  # このイベントの一意の識別子
    session_id: str  # セッション識別子
    event: dict[str, Any]  # 生の Claude API ストリームイベント
    parent_tool_use_id: str | None  # 常に None

type SDKPartialAssistantMessage = {
  type: "stream_event";
  event: BetaRawMessageStreamEvent; // Anthropic SDK から
  parent_tool_use_id: string | null;
  uuid: UUID;
  session_id: string;
  ttft_ms?: number; // メッセージ開始イベントにのみ存在する、最初のトークンまでの時間（ミリ秒）
};

parent_tool_use_id フィールドは Python では常に None、TypeScript では null です。ストリームイベントはメインセッションのみに対して発行されます。サブエージェントからのトークンレベルのデルタは転送されません。出力をサブエージェントに属性付けするには、parent_tool_use_id を含む完全なメッセージを使用してください。サブエージェント呼び出しの検出を参照してください。 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_delta with input_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 レスポンスを取得する
権限: エージェントが使用できるツールを制御する

Agent SDK

コアコンセプト

入力と出力

ツールで拡張

動作をカスタマイズ

制御と可観測性

デプロイメント

SDK リファレンス

リアルタイムでレスポンスをストリーミングする

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

StreamEvent リファレンス

メッセージフロー

テキストレスポンスをストリーミングする

ツール呼び出しをストリーミングする

ストリーミング UI を構築する

既知の制限事項

次のステップ

​ストリーミング出力を有効にする

​StreamEvent リファレンス

​メッセージフロー

​テキストレスポンスをストリーミングする

​ツール呼び出しをストリーミングする

​ストリーミング UI を構築する

​既知の制限事項

​次のステップ

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

StreamEvent リファレンス

メッセージフロー

テキストレスポンスをストリーミングする

ツール呼び出しをストリーミングする

ストリーミング UI を構築する

既知の制限事項

次のステップ