単一エージェントに巨大なプロンプトを詰め込む構成は、プロトタイプでは動いても本番では崩れがちです。役割ごとにエージェントを分けたくなりますが、そこに認証・権限制御・そして「モデルの出力が毎回違う中でどう動作確認するのか」という本番特有の壁が立ちはだかります。本稿は Claude Agent SDK（Python）で、サブエージェントを束ねるオーケストレーターを空ディレクトリから完成まで通しで作り、権限フックで危険なコマンドを止め、決定的な信号だけを投影して動作確認する手順までを示します。対象は claude-agent-sdk（Python 3.10 以上）で、記述は 2026-09-01 時点です。API はバージョンで変わり得ます。

SDK が提供するもの（citation-first）

Agent SDK は、Claude Code を動かすのと同じツール・エージェントループ・コンテキスト管理を、プログラムから扱える形で提供します。提供形態はライブラリで、対応言語は限定されています。

The SDK is available as a library for Python and TypeScript only.

Python と TypeScript の 2 言語のみがライブラリとして提供されている、という記述です。それ以外の言語からは、CLI をサブプロセスとして駆動する経路が用意されています。

本番のマルチエージェントで使う中心機能は次の 4 つです。サブエージェントは「特化したサブタスクのために専門化されたエージェントをスポーンする」機能、Hooks は「エージェントライフサイクルの主要ポイントでカスタムコードを実行する」機能、Permissions は「どのツールを自動実行し、どれを承認必須にするか」を制御する機能、Sessions は「会話をまたいでコンテキストを維持し、後から再開・フォークする」機能です。外部接続は MCP（Model Context Protocol）で行います。組み込みツールはファイルの読み書き・編集、コマンド実行、Web 検索を実行できます。

実装：空ディレクトリから通しで作る

前提は Python 3.10 以上（2026-09-01 時点）と、claude CLI（Python SDK が内部で起動します）、ANTHROPIC_API_KEY です。

mkdir agent-fleet && cd agent-fleet
python -m venv .venv && source .venv/bin/activate
pip install "claude-agent-sdk" anyio
npm install -g @anthropic-ai/claude-code
export ANTHROPIC_API_KEY=sk-ant-...
python -c "import claude_agent_sdk as s; print('sdk import ok')"

最後の行の期待出力は次のとおりです。これは自作の print 文の出力なので、環境が整っていれば必ず一致します。

sdk import ok

guard.py：危険な Bash を止める権限フック

# guard.py
DANGEROUS = ("rm -rf", "sudo", "curl ", "wget ", "> /dev/")

async def deny_dangerous_bash(input_data, tool_use_id, context):
    command = input_data.get("tool_input", {}).get("command", "")
    if any(token in command for token in DANGEROUS):
        print(f"[guard] denied Bash: {command}", flush=True)
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": "blocked by fleet policy",
            }
        }
    return {}

フックの判定ロジックは純粋な Python なので、単体で決定的にテストできます。

# guard_test.py
import anyio
from guard import deny_dangerous_bash

async def main():
    result = await deny_dangerous_bash(
        {"tool_input": {"command": "rm -rf /tmp/demo"}}, "tool-1", None
    )
    print(f"decision={result['hookSpecificOutput']['permissionDecision']}")

anyio.run(main)

python guard_test.py の期待出力です。モデルを一切介さない自作コードの出力なので、逐語で再現します。

[guard] denied Bash: rm -rf /tmp/demo
decision=deny

orchestrator.py：サブエージェントを束ねる

# orchestrator.py
import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AgentDefinition,
    AssistantMessage,
    ResultMessage,
    TextBlock,
    HookMatcher,
)
from guard import deny_dangerous_bash

AGENTS = {
    "researcher": AgentDefinition(
        description="Gathers facts from the local repo and the web.",
        prompt="You collect concise, sourced facts. Do not write prose.",
        tools=["Read", "Grep", "WebSearch"],
        model="sonnet",
    ),
    "writer": AgentDefinition(
        description="Turns researched facts into a short draft.",
        prompt="You write a tight draft from the facts you are given.",
        tools=["Read", "Write"],
        model="sonnet",
    ),
}

def build_options():
    return ClaudeAgentOptions(
        agents=AGENTS,
        allowed_tools=["Read", "Grep", "Write", "Bash", "WebSearch", "Task"],
        permission_mode="acceptEdits",
        hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[deny_dangerous_bash])]},
        max_turns=12,
    )

async def run_fleet(task: str) -> bool:
    is_error = True
    async for message in query(prompt=task, options=build_options()):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"[log] {block.text.strip().replace(chr(10), ' ')[:80]}")
        elif isinstance(message, ResultMessage):
            is_error = message.is_error
    print(f"[fleet] is_error={is_error}")
    return not is_error

if __name__ == "__main__":
    ok = anyio.run(
        run_fleet,
        "Use the researcher subagent to list the files in this repo, "
        "then the writer subagent to summarise them in three bullet points.",
    )
    raise SystemExit(0 if ok else 1)

モデルが生成する自然文は [log] 行に流し、動作確認では触りません。確認対象は決定的な末尾マーカーだけです。

python orchestrator.py | grep '^\[fleet\]'

成功時の期待出力（grep で決定的な信号だけを投影しています）。

[fleet] is_error=False

別言語・障害分離のための CLI ドライバ

Python/TypeScript 以外から、あるいはエージェントごとにプロセスを分離したいとき（1 体のクラッシュを全体から隔離する）は、CLI をサブプロセスとして駆動します。JSON 出力の可変部分（result 本文やコスト）ではなく、決定的なエンベロープ項目だけを jq で投影して検証します。

claude -p "reply with exactly: pong" --output-format json > out.json
jq -r '.type, .subtype, .is_error' out.json

成功時の期待出力です。モデルの答えが何であれ、成功したリクエストのエンベロープ項目は固定されます。

result
success
false

本番でつまずく3点（docs に無い設計判断）

以下は私見を含む運用上の指針です。事実主張と設計判断を分けて書きます。

第一に、非決定性のテスト設計です。エージェントの自然文出力を期待値と突き合わせる方式はテストを壊れやすくします。私の設計判断は、動作確認を「決定的な信号の投影」に一本化することです。本稿の 3 つの確認はすべてこの形になっています。フックは純ロジックを直接呼ぶ、CLI は jq で type/subtype/is_error を抜く、オーケストレーターは grep で [fleet] マーカーを抜く。モデルの散文はログに流し、CI の assert 対象から外します。これで再現可能な合否判定になります。

第二に、認証・課金のコンプライアンスです。事前承認がない限り、Anthropic はサードパーティ開発者が自社製品で claude.ai ログインやレートリミットを提供することを許可していません。第三者に配布するエージェントは API 課金（ANTHROPIC_API_KEY など）で動かす前提にし、claude.ai の認証に依存させない設計にします。

第三に、障害分離とコスト上限です。すべてのエージェントを 1 プロセスの query ループに載せると、1 体の暴走が全体を巻き込みます。役割の粒度が粗い、または信頼境界が異なるエージェントは CLI サブプロセスに切り出し、max_turns や allowed_tools を各プロセスで絞るのが安全です。フック（deny）は許可リスト（allowed_tools）の後段に置く最後の防波堤として二重化します。

既知の限界

--output-format json の result 本文・total_cost_usd・session_id・duration_ms は毎回変わるため、逐語比較には使えません（検証は決定的項目に限定します）。サブエージェントが実際に起動するか、危険コマンドを実行しようとするかはタスク次第で、フックの発火はモデル挙動に依存します。だからこそ本稿はフック判定を単体テストで固定しています。API の関数・引数はバージョンで変化し得ます（本稿は 2026-09-01 時点）。

References

[1] https://code.claude.com/docs/en/agent-sdk/overview