Agent SDKストリーミング入力とシングルメッセージ入力の使い分け

Agent SDKには2つの入力モードがある

Agent SDKでエージェントに指示を渡す方法は、ストリーミング入力とシングルメッセージ入力の2つです。選ぶ基準は、実行中のエージェントに後から手を出す必要があるかどうかに尽きます。

実行中に手を出したいときはストリーミング入力

ストリーミング入力に切り替える理由の大半は、実行中の制御です。ストリーミング入力は、ユーザー入力を受け取りながら割り込みを処理し、権限の確認を利用者に示し、セッションを管理する長寿命プロセスとしてエージェントを動かします。

ストリーミング入力では、複数のメッセージを送ると順番に処理され、処理中に割り込むこともできます。応答が生成される様子もリアルタイムで受け取れます。

次の例は、1通目に文章、2通目に画像をyieldするジェネレータをquery()のpromptへ渡します。

import { readFile } from "fs/promises";
import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
 
async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
  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));
 
  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);
  }
}

作業ディレクトリにdiagram.pngが無いとreadFileが失敗します。手元で試すときは、同名の画像を置くか読み込み先を書き換えてください。

画像ブロックのsourceが欠けていたり、オブジェクトでなかったりしても、SDKはエラーを出しません。画像の代わりに[Image could not be processed: image block has no source object]のような文面がClaudeに渡り、セッションは続きます。画像が読まれた手応えが無いときは、ブロックの形を先に疑います。

割り込みや設定変更はストリーミング入力の専用機能

TypeScriptのクエリオブジェクトには、実行中のセッションに働きかけるメソッドがあります。interrupt()、setPermissionMode()、setModel()に加えapplyFlagSettings()も、ストリーミング入力モードでだけ使えるとTypeScriptのリファレンスに明記されています。

Pythonでは、同じ役割をClaudeSDKClientが担います。query()関数もストリーミング入力は受け付けますが、割り込みに対応するのはClaudeSDKClientだけです。Pythonのquery()関数は「新しいセッションで1回のやり取り」、ClaudeSDKClientは「同じセッションで複数のやり取り」という分担になっています。

Pythonで割り込むときは、後始末に注意が要ります。

読み切りを省いてinterrupt()の直後に新しいquery()を投げ、receive_response()を1回だけ回すと、返ってくるのは新しい指示ではなく中断済みタスクのメッセージです。

途中で送ったメッセージがいつ届くかはpriorityで決まる

TypeScriptのSDKUserMessageにはpriorityフィールドがあり、実行中のターンに送ったメッセージがいつClaudeに届くかを決められます。

shouldQuery: falseを付けたメッセージは、トランスクリプトに追記されるだけで、応答のターンを起こしません。次にターンを起こすメッセージへ合流するので、手元で実行したコマンドの出力のような文脈を、モデル呼び出しを使わずに足せます。

呼び出しごとに完結させたいときはシングルメッセージ入力

シングルメッセージ入力が向くのは、1回きりの応答が欲しいとき、画像添付やセッション途中の制御が要らないとき、Lambda関数のようなステートレスな環境で動かすときです。直接の画像添付、動的なメッセージキュー、リアルタイムの割り込み、自然な複数ターンの会話には対応しません。

// 1回きりのクエリ
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}`);
}
 
// 前回のセッションの続きとして、新しいクエリを1件投げる
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);
  }
}

continue: true(Pythonではcontinue_conversation=True)は、現在のディレクトリで最後に使われたセッションを探して、その履歴の上に新しいクエリを足します。セッションIDを管理しなくて済む反面、複数のセッションを並行して持つ場面には向きません。ユーザーごとにセッションを分けるマルチユーザーのアプリでは、resumeにセッションIDを渡す形を使います。

セッションの持ち方は言語でも違います。TypeScriptにはセッションを保持するクライアントオブジェクトが無く、2回目以降のquery()ごとにcontinue: trueを渡します。PythonのClaudeSDKClientは、client.query()を呼ぶたびに自動で同じセッションを続けるので、IDもcontinue_conversationも要りません。複数ターンの会話を1つのプロセスで続けるなら、PythonはClaudeSDKClient、TypeScriptはcontinue: trueが公式の整理です。

つまり、Slack botのように1つのプロセスが複数ユーザーの会話を持つ構成は、モードの選択とは別に、セッションIDの管理設計が必要になります。具体的な実例はAgent SDKでSlack常駐botを作るにあります。

上限に達したときの出方

クエリがerror_max_turnsのようなエラー結果で終わると、最終結果メッセージをyieldした後で、失敗内容を含むエラーが投げられます。PythonではResultErrorです。処理を続けたいコードはtryブロックで囲みます。

maxTurnsが数えるのはツール使用のターンだけです。最大ターン数に達した時点でキューに残っているメッセージは、キューに残ったまま新しいターンとして処理され、そのターンのmaxTurnsの数え直しが始まります。maxBudgetUsdの累計は数え直されず、積み上がり続けます。費用で止めたいときは、maxBudgetUsd(Pythonではmax_budget_usd)を併用でき、超えるとerror_max_budget_usdの結果になります。

ジェネレータが失敗したときの見え方

エラーの出方は、モードより言語で違いが大きく出ます。

TypeScriptのストリーミング入力では、メッセージジェネレータの中で例外が起きると(存在しないファイルを読むなど)、ストリームは元のエラーではなくClaude Code process aborted by userで終わります。この文面を見たら、まずジェネレータの中身を疑います。バンドルされたSDKソースの長いミニファイ済みの行が前に出ることもあるので、出力は末尾まで読みます。

Pythonのストリーミング入力では、ジェネレータの例外はデバッグレベルのログに残るだけで、raiseされません。セッションは応答なしで止まります。出力がないままハングしたら、デバッグログを有効にしてジェネレータを調べます。

Pythonのcan_use_toolの例には回避策のフックが付く

Pythonでcan_use_toolコールバックを使う公式の例は、何もしないPreToolUseフックを1つ添えています。コード中のコメントは「Required workaround: dummy hook keeps the stream open for can_use_tool」で、フックはストリームを開いたままにするためのものです。承認コールバックを組み込むときは、この例のとおりにフックを足すところから始めます。

承認や確認質問を処理する手順は、Agent SDK canUseToolで承認と確認質問を処理するにまとめています。

CLIにも同じ区別がある

SDKの2モードは、claudeコマンドの非対話モードにも対応する入口があります。v2.1.289のclaude --helpには、--input-formatが「--printのときだけ有効」で、値はtext(既定)かstream-json(リアルタイムのストリーミング入力)と表示されます。

claude --help | grep -A3 -e "--input-format"

  --input-format <format>               Input format (only works with --print):
                                        "text" (default), or "stream-json"
                                        (realtime streaming input) (choices:
                                        "text", "stream-json")

--replay-user-messagesはstream-jsonの入力と出力を両方使うときだけ有効で、標準入力から受けたユーザーメッセージを標準出力へ返し、受信の確認に使えます。SDKを使わずにシェルスクリプトからストリーミング入力を試したい場合の入口です。

どちらを選ぶかの早見表

セッションの保存先を自前のストレージに移す話は、Agent SDK SessionStoreでセッションをS3やRedisに永続化するで扱っています。最小構成のクエリから試すなら、Claude Agent SDK入門が入口になります。

まとめ

分かれ目は、実行中に外から手を入れる必要があるかどうかです。割り込み、画像、承認のどれか1つでも要るなら、最初からストリーミング入力で組むほうが書き直しを避けられます。呼び出しごとに完結する処理なら、シングルメッセージ入力の単純さが活きます。