ガイド

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

テキストやツール呼び出しがストリーミングされる際に、Agent SDKからリアルタイムでレスポンスを受け取る

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

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

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

ストリーミングを有効にするには、オプションでinclude_partial_messages（Python）またはincludePartialMessages（TypeScript）をtrueに設定します。これにより、SDKは通常のAssistantMessageやResultMessageに加えて、生のAPIイベントを含むStreamEventメッセージを到着時に返すようになります。

コードでは以下を行う必要があります：

各メッセージのタイプをチェックして、StreamEventを他のメッセージタイプと区別する
StreamEventの場合、eventフィールドを抽出してそのtypeをチェックする
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())

StreamEventリファレンス

部分メッセージが有効な場合、生のClaude APIストリーミングイベントがオブジェクトにラップされて受信されます。型名は各SDKで異なります：

Python: StreamEvent（claude_agent_sdk.typesからインポート）
TypeScript: SDKPartialAssistantMessage（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 # サブエージェントからの場合の親ツール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（最終結果）、CompactBoundaryMessage（会話履歴が圧縮されたことを示す）があります。

テキストレスポンスのストリーミング

テキストが生成されるたびに表示するには、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())

ツール呼び出しのストリーミング

ツール呼び出しもインクリメンタルにストリーミングされます。ツールの開始を追跡し、入力が生成されるたびに受信し、完了を確認できます。以下の例では、現在呼び出されているツールを追跡し、ストリーミングされるJSON入力を蓄積します。3つのイベントタイプを使用します：

content_block_start: ツールの開始
content_block_delta（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())

ストリーミングUIの構築

この例では、テキストとツールのストリーミングを統合的なUIに組み合わせています。エージェントが現在ツールを実行中かどうかを追跡し（in_toolフラグを使用）、ツール実行中は[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())

既知の制限事項

一部のSDK機能はストリーミングと互換性がありません：

拡張思考: max_thinking_tokens（Python）またはmaxThinkingTokens（TypeScript）を明示的に設定した場合、StreamEventメッセージは発行されません。各ターン後に完全なメッセージのみを受信します。なお、SDKではデフォルトで思考は無効になっているため、有効にしない限りストリーミングは動作します。
構造化出力: JSON結果は最終的なResultMessage.structured_outputにのみ表示され、ストリーミングデルタとしては表示されません。詳細は構造化出力を参照してください。

次のステップ

テキストとツール呼び出しをリアルタイムでストリーミングできるようになったので、以下の関連トピックを探索してください：

インタラクティブ vs ワンショットクエリ: ユースケースに合った入力モードを選択する
構造化出力: エージェントから型付きJSONレスポンスを取得する
権限: エージェントが使用できるツールを制御する

Was this page helpful?

ガイド

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

テキストやツール呼び出しがストリーミングされる際に、Agent SDKからリアルタイムでレスポンスを受け取る

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

コードでは以下を行う必要があります：

各メッセージのタイプをチェックして、StreamEventを他のメッセージタイプと区別する
StreamEventの場合、eventフィールドを抽出してそのtypeをチェックする
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_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リファレンス

部分メッセージが有効な場合、生のClaude APIストリーミングイベントがオブジェクトにラップされて受信されます。型名は各SDKで異なります：

Python: StreamEvent（claude_agent_sdk.typesからインポート）
TypeScript: SDKPartialAssistantMessage（type: 'stream_event'）

@dataclass
class StreamEvent:
    uuid: str                      # このイベントの一意識別子
    session_id: str                # セッション識別子
    event: dict[str, Any]          # 生のClaude APIストリームイベント
    parent_tool_use_id: str | None # サブエージェントからの場合の親ツール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 - 最終結果

テキストレスポンスのストリーミング

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

ツール呼び出しのストリーミング

content_block_start: ツールの開始
content_block_delta（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())

ストリーミングUIの構築

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

既知の制限事項

一部のSDK機能はストリーミングと互換性がありません：

拡張思考: max_thinking_tokens（Python）またはmaxThinkingTokens（TypeScript）を明示的に設定した場合、StreamEventメッセージは発行されません。各ターン後に完全なメッセージのみを受信します。なお、SDKではデフォルトで思考は無効になっているため、有効にしない限りストリーミングは動作します。
構造化出力: JSON結果は最終的なResultMessage.structured_outputにのみ表示され、ストリーミングデルタとしては表示されません。詳細は構造化出力を参照してください。

次のステップ

テキストとツール呼び出しをリアルタイムでストリーミングできるようになったので、以下の関連トピックを探索してください：

インタラクティブ vs ワンショットクエリ: ユースケースに合った入力モードを選択する
構造化出力: エージェントから型付きJSONレスポンスを取得する
権限: エージェントが使用できるツールを制御する

Was this page helpful?