こんにちは。よっしーです(^^)
背景
この連載では、Claude Codeの公式ドキュメントを1ページずつ読み解いていきます。公式ドキュメントは情報が網羅されている分、「結局どの機能を、どんな場面で使えばいいのか」は自分で考える必要があり、読むのに意外と時間がかかります。そこで、私が実務で使うために読み込んだ内容を「使う場面→実例コード」の順に整理して残していくことにしました。専門家の解説というより、一次情報を読んだ記録です。推測や動作を確認していない部分には、その都度そう書きます。
公式ドキュメント：https://code.claude.com/docs/ja/agent-sdk/streaming-vs-single-mode
1. 一言でいうと
Agent SDK でエージェントにメッセージを渡す方法は2つあります。1つの接続を開いたまま、メッセージを次々に送り込む「ストリーミング入力」と、1回の問い合わせで1つのメッセージを送って終わる「シングルメッセージ入力」です。公式が推奨しているのはストリーミング入力のほうです。
違いは、プロンプトに何を渡すかです。シングルメッセージ入力ではプロンプトに文字列を1つ渡します。ストリーミング入力では、メッセージを順に生み出すジェネレーターを渡し、SDK はそこから届くメッセージを、同じセッションの中で次々に処理します。
これまでの回で何度も「ストリーミング入力なら」という条件が出てきました。第81回では Claude の方向を丸ごと変える方法として、第90回では /compact と /clear が意味を持つ場面として登場しています。今回はその入力方式そのものの回です。
2. どういう場面で役立つか
シーン1：チャット画面のように対話を続けるアプリ
ユーザーが質問を送り、エージェントが答え、また質問を送る、という対話型のアプリです。ストリーミング入力なら、1つのセッションの中で会話の文脈を自然に保ったまま、次のメッセージを送り込めます。応答も最終結果だけでなく、生成されている途中から画面に出せます。
シーン2：作業の途中で画像を見せる
「このコードベースのセキュリティを調べて」と頼んだあとで、「ついでにこの設計図も見て」と画像を添えて送る、という使い方です。メッセージに画像を直接添付できるのは、ストリーミング入力だけです。
シーン3：サーバーレス環境での1回きりの処理
AWS Lambda のように、呼び出しごとに状態を持たない環境で、「この関数の仕組みを説明して」と1回問い合わせて結果を返すだけの処理です。この場合はシングルメッセージ入力が単純で向いています。
不要・向かないケース
- ストリーミング入力が不要な場面：1回の応答があれば足り、画像の添付も、セッション中の操作（割り込みや権限モードの切り替えなど）も要らないなら、シングルメッセージ入力で十分です。
- シングルメッセージ入力が不向きな場面：画像を添付したい、途中で割り込みたい、自然に何往復も会話したい、という場合です。原文はこれらをシングルメッセージ入力の制限として明記しています（3-3）。
3. コードと仕組みの解説
原文の fenced コードブロックは4本（Python と TypeScript の対訳2組）です。全数を引用します。コードは原文から機械的に抜き出したもので、入れ子に由来する先頭インデントを外した以外は変えていません。このページにはバージョン要件の記載がありません。
3-1. 2つのモードの比較
原文の記述を1つの表にまとめます（表の形は筆者の整理）。
| 項目 | ストリーミング入力 | シングルメッセージ入力 |
|---|---|---|
| 位置づけ | 推奨。エージェントの機能を全て使える | より単純だが、より制限が多い |
| セッションの形 | 長く動き続ける1つのセッション | 1回限りの問い合わせ。続きはセッションの状態を使って再開する |
| 画像の添付 | メッセージに直接添付できる | できない |
| メッセージのキュー | 複数のメッセージを順に処理し、割り込みもできる | できない |
| 割り込み | リアルタイムでできる | できない |
| 複数ターンの会話 | 文脈を自然に保ったまま続く | 自然にはできない（continue や resume で続ける） |
| 応答の見え方 | 生成中のものをリアルタイムで見られる | ― |
| 向いている場面 | 対話型のアプリ、長く動くエージェント | 1回きりの応答、ステートレスな環境（Lambda など） |
ストリーミング入力の利点として、原文はほかに「セッション中に全てのツールとカスタム MCP サーバーを使える」ことと、エージェントが「ユーザーの入力を受け取り、割り込みを処理し、権限の確認を表示し、セッションを管理する」長期プロセスとして動けることを挙げています。
連載のこれまでの回との関係。 次の機能は、どれもストリーミング入力（1つの接続を開いたままにする使い方）を前提にしていました。
| 回 | 機能 | ストリーミング入力との関係 |
|---|---|---|
| 第77回 | ターン数の上限と予算 | ストリーミング入力では、ターン数の上限はメッセージごとにリセットされ、予算は積み上がる |
| 第80回 | setPermissionMode() | セッションの途中で権限モードを切り替える |
| 第81回 | 完全なリダイレクト／Python での canUseTool | 方向転換はストリーミング入力で新しい指示を送る。Python の canUseTool には、プロンプトをジェネレーターで渡す回避策が必要だった |
| 第88回 | ストリーム中の巻き戻し | 接続が開いている間なら、そのまま rewindFiles() を呼べる |
| 第90回 | /compact・/clear | 1回きりの query() では意味が無く、ストリーミング入力で役に立つ |
第81回の Python の回避策（ダミーのフックとジェネレーター形式のプロンプト）が、まさに今回の「ストリーミング入力」の形だったことになります（この対応づけは筆者）。
3-2. ストリーミング入力の実装
次の例は、作業ディレクトリの diagram.png という画像を読み込みます。先にその名前で画像を1つ置くか、ファイル名を自分の画像に変えてから動かします。
import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";
async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
// First message
yield {
type: "user",
message: {
role: "user",
content: "Analyze this codebase for security issues"
},
parent_tool_use_id: null
};
// Wait for conditions or user input
await new Promise((resolve) => setTimeout(resolve, 2000));
// Follow-up with image
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
};
}
// Process streaming responses
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);
}
}
from claude_agent_sdk import (
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
)
import asyncio
import base64
async def streaming_analysis():
async def message_generator():
# First message
yield {
"type": "user",
"message": {
"role": "user",
"content": "Analyze this codebase for security issues",
},
}
# Wait for conditions
await asyncio.sleep(2)
# Follow-up with image
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,
},
},
],
},
}
# Use ClaudeSDKClient for streaming input
options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])
async with ClaudeSDKClient(options) as client:
# Send streaming input
await client.query(message_generator())
# Process responses
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())
流れは次のとおりです。
- ジェネレーターが最初のメッセージ「このコードベースのセキュリティ上の問題を分析して」を送る
- 2秒待つ（実際のアプリでは、ここでユーザーの入力や何かの条件を待つ）
- 「このアーキテクチャ図を見て」という文章と、base64 にした PNG 画像を1つのメッセージにまとめて送る
画像のブロックは、type: "image" の中の source に、type: "base64"、media_type: "image/png"、data（base64 の文字列）を入れる形です。第83回のカスタムツールが返す画像ブロック（data と mimeType を直接持つ形）とは書き方が違う点に注意してください。こちらは Claude にメッセージとして送る形、第83回はツールの結果として返す形です（形の違いの指摘は筆者）。
言語による違いがいくつかあります。
| 項目 | TypeScript | Python |
|---|---|---|
| 呼び出し方 | query() の prompt にジェネレーターを渡す | ClaudeSDKClient を開き、client.query() にジェネレーターを渡す |
| メッセージの形 | parent_tool_use_id: null を含める | 含めていない |
| 画像の読み込み | readFile("diagram.png", "base64") | ファイルをバイナリで読み、base64.b64encode でエンコードして文字列にする |
| 応答の受け取り | for await で結果メッセージを表示 | client.receive_response() でアシスタントの文章を表示 |
表示される内容が言語で違います。 TypeScript 版は、各応答が終わるたびに結果を表示します。Python 版の receive_response() のループは最初の結果メッセージで終わるので、セキュリティ分析の結果だけが表示されます。両方の応答を読みたいなら、Python リファレンスの「会話を続ける」例のように、メッセージ1つにつき query() と receive_response() を1組ずつ使います。同じ「ストリーミング入力の例」でも、Python 版は2つ目の応答を読まずに終わる、ということです。
画像の source が無いとき
画像ブロックの source が無い、またはオブジェクトでない場合、SDK はエラーを出しません。Claude Code は画像の代わりに [Image could not be processed: image block has no source object] のような文のメモを Claude に送り、セッションはそのまま続きます。
エラーにならないので、画像の組み立てを間違えても気づきにくいということです。Claude が「画像を処理できなかった」と答えてきたら、まず画像ブロックの形を疑うのがよいでしょう（筆者の補足）。
ジェネレーターが失敗したとき
メッセージを生み出すジェネレーターの中で例外が起きた場合の振る舞いが、言語で大きく違います。
| 言語 | 起きること | 調べ方 |
|---|---|---|
| TypeScript | ストリームは、元のエラーではなく 「Claude Code process aborted by user」 というエラーで終わる。その前に、同梱された SDK のソースの長い1行（縮小化されたもの）が表示されることもある | このメッセージが出たら、まずジェネレーターの中のコードを確認する。エラーの文は出力の最後まで読んで確かめる |
| Python | 例外はデバッグレベルのログに記録されるだけで、セッションは例外を出さずに止まったままになる | 出力が無いまま止まったら、デバッグログを有効にしてジェネレーターを確認する |
原文の例として挙がっているのは、読み込むファイルが見つからない場合です。上の例なら、diagram.png が無いと TypeScript では「ユーザーによって中断された」という、原因と関係の無さそうなエラーになり、Python では何も言わずに止まります。どちらも本当の原因が表に出にくいので、ジェネレーターの中では、ファイルの読み込みなど失敗しうる処理を自分で try で囲み、原因をログに出しておくのが安全です（筆者の提案）。
3-3. シングルメッセージ入力
シングルメッセージ入力は、次の場合に使います。
- 1回限りの応答が必要なとき
- 画像の添付や、セッション中の制御メソッドが要らないとき
- Lambda 関数のようなステートレスな環境で動かす必要があるとき
シングルメッセージ入力でできないことは、原文の警告にまとめられています。
- メッセージへの画像の直接添付
- 動的なメッセージのキュー
- リアルタイムの割り込み
- 自然な複数ターンの会話
エラーの扱い。 問い合わせが error_max_turns などのエラーの結果で終わった場合、シングルメッセージの query() 呼び出しは、最後の結果メッセージを出したあとで、失敗の文を含むエラーを投げます。後続の処理を続ける必要があるなら、ループを try で囲みます。これは第83回以降の例でずっと出てきた「1回きりの query() はエラーの結果の後で例外を投げる」の出どころです。結果の種類（サブタイプ）は第77回の範囲です。
import { query } from "@anthropic-ai/claude-agent-sdk";
// Simple one-shot query
// query() throws after an error result, such as error_max_turns
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}`);
}
// Continue conversation with session management
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():
# Simple one-shot query using query() function
# query() raises ResultError after an error result, such as error_max_turns
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}")
# Continue conversation with session management
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())
実行すると、それぞれの問い合わせが最後の結果の文章を表示します。最初に認証の流れの説明、次に認可の仕組みの説明が出ます。
2回目の問い合わせは continue: true（Python は continue_conversation=True）で、直前のセッションの続きとして送っています（第87回の continue）。シングルメッセージ入力でも、こうしてセッションを続ければ複数の問い合わせで文脈を共有できます。ただし、問い合わせのたびに新しく接続を開くことになり、途中の割り込みや画像の添付はできません。
Python 版のコメントには、エラーの結果のあとに投げられるのが ResultError だと書かれています。TypeScript 版のコメントは「エラーの結果のあとで投げる」とだけ書いています。
読むときの注意を1つ挙げておきます。2回目の問い合わせには allowedTools がありません。第87回で見たように、許可リストは呼び出しのたびに渡すもので、セッションに保存されて引き継がれるわけではないと読めます。2回目に Claude がファイルを読もうとすると、権限の評価に回ることになります（第80回）。続きの問い合わせでもツールを使わせたいなら、allowedTools をもう一度渡しておくのが確実です（筆者の指摘）。
3-4. どちらを選ぶか
原文の内容を、判断の順に並べ直すと次のようになります（整理は筆者）。
- 画像を添付したい、途中で割り込みたい、セッション中に権限モードやモデルを切り替えたい → ストリーミング入力
- ユーザーと何往復も対話するアプリ → ストリーミング入力（文脈が自然に続き、応答を途中から表示できる）
- Lambda などのステートレスな環境で、1回の応答を返すだけ → シングルメッセージ入力
- 迷ったら → 公式の推奨はストリーミング入力
シングルメッセージ入力でも continue や resume でセッションを続けられるので、「1回ずつ問い合わせるが、文脈は前の続き」という作り方もできます。ステートレスな環境で会話を続けたい場合は、第87回で見たセッションストアや、必要な結果だけを次の問い合わせに渡す方法と組み合わせることになります。
4. まとめ
- 入力の方法は2つ。ストリーミング入力はジェネレーターでメッセージを次々に送る長く続くセッション、シングルメッセージ入力は文字列1つの1回きりの問い合わせ。推奨はストリーミング入力。
- ストリーミング入力でしかできないこと：画像の直接添付、メッセージのキュー、リアルタイムの割り込み、自然な複数ターンの会話。権限モードの切り替えや
/compact・/clearなど、これまでの回で出てきた「セッション中の操作」もこちらが前提。 - 画像はメッセージの
contentにtype: "image"とsource（base64・media_type・data）で入れる。sourceが無いとエラーにならず、「処理できなかった」というメモが Claude に届く。 - Python の
ClaudeSDKClientのreceive_response()は最初の結果で終わる。複数の応答を読むには、メッセージごとにquery()とreceive_response()を組にする。 - ジェネレーターの例外は、TypeScript では「Claude Code process aborted by user」に化け、Python では何も言わずに止まる。ジェネレーターの中で自分で例外を捕まえておく。
- シングルメッセージ入力は、エラーの結果のあとで例外を投げる（Python は
ResultError）。続けるならtryで囲む。続きの問い合わせは continue で文脈を共有できるが、allowedToolsは毎回渡す。 - 1回の応答で足りる、ステートレスな環境で動かす、ならシングルメッセージ入力。それ以外はストリーミング入力。
次回予告
次回は 「コストの追跡」（agent-sdk/cost-tracking） を取り上げる予定です。
第89回では、サブエージェントの費用が問い合わせの total_cost_usd に加算され、maxBudgetUsd で支出に上限を設けられると見ました。第86回では、プロンプトキャッシュが効いているかを結果メッセージのトークン数で確かめられると触れています。次回は、エージェントがどれだけのトークンを使い、いくらかかったのかを、SDK のコードからどう読み取り、どう集計するのかを見ていきます。
何か質問や相談があれば、コメントをお願いします。また、エンジニア案件の相談にも随時対応していますので、お気軽にお問い合わせください。
それでは、また明日お会いしましょう(^^)
コメント