ホーム / チュートリアル / Agent SDK のストリーミング入力と単発入力の違い・使い分け

Agent SDK のストリーミング入力と単発入力の違い・使い分け

Claude Agent SDK エンジニアリング 6分で読めます

Claude Agent SDK でエージェントに指示を渡す方法は2つあります。長く動き続けるセッションにメッセージを順に流し込むストリーミング入力と、1回の問い合わせで完結する単発メッセージ入力です。この記事では公式ドキュメント「Streaming Input」に沿って、2つのモードの違い、ストリーミング入力でできること、非同期ジェネレータを使った実装例、単発入力の使いどころと制限、エラー処理と会話の継続を整理します。

2つの入力モードの違い

公式ドキュメントは、Agent SDK の入力モードを次の2つに分けています。

どちらを使うかは query() の prompt に何を渡すかで決まります。TypeScript のリファレンスでは prompt の型が string | AsyncIterable<SDKUserMessage> と定義されており、文字列を渡せば単発入力、ユーザーメッセージを順に返す非同期イテラブル（非同期ジェネレータ）を渡せばストリーミング入力になります。Python では、ストリーミング入力の例に ClaudeSDKClient が使われています。

Python リファレンスにある query() と ClaudeSDKClient の比較表の要点は次のとおりです。

項目query()ClaudeSDKClient
セッション既定で毎回新しいセッションを作る同じセッションを使い回す
会話1回のやり取り同じ文脈で複数回のやり取り
ストリーミング入力対応対応
割り込み（interrupt）非対応対応
会話の継続continue_conversation または resume で手動自動
向いている用途単発のタスク続けて行う会話

ストリーミング入力モードでできること

ストリーミング入力モードでは、エージェントが長く動き続けるプロセスとして、ユーザーの入力を受け取り、割り込みを処理し、権限の確認を表に出し、セッションを管理します。公式が挙げる利点は次の5つです。

TypeScript の Query オブジェクトには、ストリーミング入力モードでだけ使えるメソッドがあります。リファレンスでは interrupt()（実行中の問い合わせを止める）、setPermissionMode()（権限モードの変更）、setModel()（モデルの変更）などが「ストリーミング入力モードでのみ利用可能」と明記されています。セッションの途中で止めたり設定を切り替えたりしたいなら、ストリーミング入力を選ぶ必要があります。権限確認の扱いは「Agent SDK の承認とユーザー入力」で扱っています。

非同期ジェネレータで送る実装例

公式の例は、最初にコードベースのセキュリティ確認を依頼し、2秒待ってから画像付きの2通目を送るものです。作業ディレクトリにある diagram.png を読み込むので、試すときは画像を用意するかファイル名を変えてください。TypeScript ではユーザーメッセージを yield する非同期ジェネレータを作り、そのまま prompt に渡します。

import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";

async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
  // 1通目
  yield {
    type: "user",
    message: {
      role: "user",
      content: "Analyze this codebase for security issues"
    },
    parent_tool_use_id: null
  };

  // 条件やユーザー入力を待つ
  await new Promise((resolve) => setTimeout(resolve, 2000));

  // 画像付きの2通目
  yield {
    type: "user",
    message: {
      role: "user",
      content: [
        { type: "text", text: "Review this architecture diagram" },
        {
          type: "image",
          source: {
            type: "base64",
            media_type: "image/png",
            data: await readFile("diagram.png", "base64")
          }
        }
      ]
    },
    parent_tool_use_id: null
  };
}

for await (const message of query({
  prompt: generateMessages(),
  options: {
    maxTurns: 10,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

Python では ClaudeSDKClient を async with で開き、client.query() にジェネレータを渡してから receive_response() で応答を読みます。

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio
import base64


async def streaming_analysis():
    async def message_generator():
        # 1通目
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Analyze this codebase for security issues",
            },
        }

        # 条件を待つ
        await asyncio.sleep(2)

        # 画像付きの2通目
        with open("diagram.png", "rb") as f:
            image_data = base64.b64encode(f.read()).decode()

        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Review this architecture diagram"},
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                ],
            },
        }

    options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])

    async with ClaudeSDKClient(options) as client:
        await client.query(message_generator())

        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)


asyncio.run(streaming_analysis())

実行結果には言語で差があります。TypeScript 版は応答が完了するたびに結果を表示します。Python 版の receive_response() は最初の結果メッセージ（ResultMessage）までで終わるため、表示されるのはセキュリティ分析だけです。両方の応答を読みたい場合は、メッセージごとに query() と receive_response() を1組ずつ使います。

画像ブロックの source が無い、またはオブジェクトでない場合、SDK はエラーを返しません。Claude Code は画像の代わりに [Image could not be processed: image block has no source object] のようなテキストを Claude に送り、セッションはそのまま続きます。画像が無視されているように見えるときは、まずブロックの形を確認してください。

ジェネレータ内で例外が起きたときの挙動にも注意が必要です。

単発メッセージ入力の使いどころと制限

単発メッセージ入力は、より簡単ですが機能は限られます。公式が挙げる使いどころは次のとおりです。

一方、単発メッセージ入力では次の機能が使えません。

途中で追加の指示を送る、画像を見せる、止める、といった対話的な操作が必要なら、ストリーミング入力を選びます。逆に、バッチ処理やサーバーレス関数から決まった指示を1回投げるだけなら、単発入力のほうが構成は単純です。

エラー処理と会話の継続

単発メッセージの query() は、error_max_turns のようなエラーの結果で終わると、最後の結果メッセージを返したあとに失敗内容を含むエラーを送出します。処理を続けたいコードでは、ループを try で囲みます。続けて同じ会話を進めたいときは、TypeScript では continue: true、Python では continue_conversation=True を指定すると直前の会話の続きになります（どちらもリファレンスで「直近の会話を続ける」オプションと説明されています）。

import { query } from "@anthropic-ai/claude-agent-sdk";

// 1回きりの問い合わせ（error_max_turns などで終わると throw する）
try {
  for await (const message of query({
    prompt: "Explain the authentication flow",
    options: {
      maxTurns: 5,
      allowedTools: ["Read", "Grep"]
    }
  })) {
    if (message.type === "result" && message.subtype === "success") {
      console.log(message.result);
    }
  }
} catch (error) {
  console.error(`Query failed: ${error}`);
}

// 直前の会話を続ける
try {
  for await (const message of query({
    prompt: "Now explain the authorization process",
    options: {
      continue: true,
      maxTurns: 5
    }
  })) {
    if (message.type === "result" && message.subtype === "success") {
      console.log(message.result);
    }
  }
} catch (error) {
  console.error(`Query failed: ${error}`);
}
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio


async def single_message_example():
    try:
        async for message in query(
            prompt="Explain the authentication flow",
            options=ClaudeAgentOptions(max_turns=5, allowed_tools=["Read", "Grep"]),
        ):
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(message.result)
    except Exception as e:
        print(f"Query failed: {e}")

    # 直前の会話を続ける
    try:
        async for message in query(
            prompt="Now explain the authorization process",
            options=ClaudeAgentOptions(continue_conversation=True, max_turns=5),
        ):
            if isinstance(message, ResultMessage) and message.subtype == "success":
                print(message.result)
    except Exception as e:
        print(f"Query failed: {e}")


asyncio.run(single_message_example())

実行すると、認証フローの説明、続いて認可処理の説明が順に表示されます。結果メッセージの subtype は success のほか error_during_execution・error_max_turns・error_max_budget_usd などがあり、詳しくは「Agent SDK のエージェントループ」で扱っています。特定のセッション ID を指定して再開する方法は「Agent SDK のセッション」を参照してください。

本ページは Streaming Input を中心に、Agent SDK reference - TypeScript（prompt の型、interrupt() などのストリーミング専用メソッド、continue）と Agent SDK reference - Python（query() と ClaudeSDKClient の比較、receive_response()、continue_conversation）を照合した非公式の日本語解説です（確認日 2026-11-01）。最新の仕様は公式ドキュメントを確認してください。

関連チュートリアル / 次に読む

Agent SDK のストリーミング出力 Agent SDK のセッション Agent SDK のエージェントループ Agent SDK の承認とユーザー入力 Claude Agent SDK でエージェントを作る
← チュートリアル一覧へ戻る カテゴリで探す