⟐ Claude Code/2026-04-17上級

Claude Code SDK でカスタム自律エージェントを実装する — 公式 CLI を超えた独自ループ設計の実践

Claude Code SDK の内部構造を理解し、公式 CLI では制御しきれない自律エージェントループを Python/TypeScript で実装します。ツール許可・エラー回復・ストリーミングに加え、課金変更後のコスト計測とモデル棚卸しまで、実運用で使っているパターンを共有します。

claude-code¹³² SDK⁴ エージェント¹⁴ 自律エージェント⁴ Python²² TypeScript²⁴ MCP⁵⁴ 自動化⁸⁷

✦ プレミアム記事

Claude Code を使い込んでいると、ある壁にぶつかります。「もっと細かく制御したい」「このタイミングで割り込みたい」「バッチ処理でヘッドレス実行したい」——そういった要望が出てきたとき、公式 CLI のオプションだけでは限界があります。

実は Claude Code は CLI の裏側に SDK が存在しており、Python または TypeScript から直接操作できます。SDK を使うと、ツールの実行を自分でハンドリングしたり、ストリーミングイベントを細粒度で処理したり、複数エージェントをプログラマブルに協調させたりといった、CLI では実現できない制御が可能になります。

ここでは SDK の内部構造から実際の実装パターンまで、本番で使えるレベルで解説します。コードは動作確認済みで、エラーハンドリングも含めた実用的な内容です。

私自身、個人開発のアプリ運用と並行して複数の技術ブログを運営しており、記事の整形や日英の整合性チェックといった定型作業を SDK ベースの自律ループに任せております。深夜も人手を挟まず静かに回り続ける仕組み。一度形になると、もう手放せません。本文のコード例は、その運用のなかで実際に使っている計測と回復のパターンが土台になっています。

SDK が提供するものと、CLI との違い

まず、Claude Code SDK が何者なのかを整理しておきます。

Claude Code CLI（claude コマンド）は人間が対話的に使うことを想定して設計されています。入力プロンプトを受け取り、許可確認を行い、ファイル操作やコマンド実行をインタラクティブに進める——このフローは人間との対話には最適ですが、自動化パイプラインには向いていません。

SDK はこの制約を解消します。プログラムから Claude を呼び出し、ツール実行の可否を独自ロジックで判断し、結果を構造化されたデータとして受け取れます。具体的に何が変わるかというと：

ツール許可の自動化: allowed_tools パラメータでリストを渡せば、個別の確認なしにツールが実行されます
イベントのプログラマブルな処理: tool_use、text、result などのイベントを1つ1つハンドリングできます
セッション管理: セッション ID を持ち越してコンテキストを継続できます
並列実行: 複数の Claude インスタンスを同時に起動して協調させられます

特に自動化や CI/CD 環境でのコード生成、定型タスクのバッチ処理には SDK が必須です。

Python SDK のセットアップと基本構造

まず Python 環境でのセットアップから始めます。

# SDK をインストール
pip install claude-code-sdk
 
# Claude Code CLI も必要（SDK の裏側で使われる）
npm install -g @anthropic-ai/claude-code
 
# 認証確認
claude --version

Python SDK の基本的な使い方です。最初は同期版から理解すると分かりやすいです：

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
 
async def run_simple_agent():
    """最もシンプルなエージェント実行の例"""
    
    options = ClaudeCodeOptions(
        max_turns=5,           # 最大ターン数（無限ループ防止に必須）
        allowed_tools=["Read", "Write", "Bash"],  # 許可するツール
    )
    
    # query() はメッセージのストリームを返す非同期ジェネレータ
    async for message in query(
        prompt="src/app.py を読んで、改善点を3つ挙げてください",
        options=options
    ):
        # メッセージタイプに応じて処理を分岐
        if message.type == "assistant":
            for block in message.content:
                if block.type == "text":
                    print(f"Claude: {block.text}")
        
        elif message.type == "result":
            print(f"\n完了 — 合計コスト: ${message.cost_usd:.4f}")
            print(f"実行ターン数: {message.num_turns}")
            break
 
asyncio.run(run_simple_agent())

このコードを実行すると、Claude が src/app.py を読み込み、分析結果をストリームで受け取れます。max_turns=5 が重要で、これを設定しないとエージェントが終わらないケースが稀に発生します（後述）。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦SDK ヘッドレス実行の1回あたりコストを ResultMessage から自動記録する計測コードと、実運用での実測分布（平均 $0.04〜0.12）

✦既定モデルの切り替わりや fast mode 廃止で自律ループを止めないための、モデル固定と月次棚卸しチェックリスト5項目

✦自前 SDK ループ・CLI・Managed Agents の使い分け判断基準（2026年前半の環境変化を踏まえた実運用の結論）

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

メッセージストリームの完全な型構造

SDK の核心はメッセージストリームの処理です。どんな型が流れてくるかを理解していないと、本番コードは書けません。

from claude_code_sdk import (
    query,
    ClaudeCodeOptions,
    AssistantMessage,
    ResultMessage,
    SystemMessage,
    UserMessage,
)
from claude_code_sdk.types import (
    TextBlock,
    ToolUseBlock,
    ToolResultBlock,
)
 
async def inspect_stream():
    """ストリームの全メッセージタイプを可視化する診断コード"""
    
    options = ClaudeCodeOptions(
        max_turns=3,
        allowed_tools=["Read", "Bash"],
    )
    
    message_count = 0
    
    async for message in query(
        prompt="現在のディレクトリ構造を教えてください",
        options=options
    ):
        message_count += 1
        
        if isinstance(message, SystemMessage):
            # システム初期化メッセージ（通常1回）
            print(f"[{message_count}] System: {message.data}")
        
        elif isinstance(message, AssistantMessage):
            # Claude の応答（テキスト + ツール呼び出しが混在）
            print(f"[{message_count}] Assistant ({len(message.content)} blocks):")
            
            for i, block in enumerate(message.content):
                if isinstance(block, TextBlock):
                    preview = block.text[:80].replace('\n', '\\n')
                    print(f"  Block[{i}] TextBlock: '{preview}...'")
                
                elif isinstance(block, ToolUseBlock):
                    print(f"  Block[{i}] ToolUseBlock:")
                    print(f"    name: {block.name}")
                    print(f"    input: {block.input}")
                    print(f"    id: {block.id}")
        
        elif isinstance(message, UserMessage):
            # ツール実行結果（SDK が内部的に生成）
            print(f"[{message_count}] User (tool results):")
            for block in message.content:
                if isinstance(block, ToolResultBlock):
                    content_preview = str(block.content)[:60]
                    print(f"  ToolResult id={block.tool_use_id}: {content_preview}...")
        
        elif isinstance(message, ResultMessage):
            # 最終結果（1回だけ流れる）
            print(f"\n[{message_count}] Result:")
            print(f"  stop_reason: {message.stop_reason}")
            print(f"  cost_usd: ${message.cost_usd:.6f}")
            print(f"  input_tokens: {message.usage.input_tokens}")
            print(f"  output_tokens: {message.usage.output_tokens}")
    
    print(f"\n合計 {message_count} メッセージを処理しました")
 
asyncio.run(inspect_stream())

これを実行すると、内部でどんなやり取りが起きているかが一目で分かります。私がこのコードを最初に書いたとき、UserMessage が SDK 側で自動生成されていることに驚きました。ツール実行の結果は自動的に次のターンの入力に変換されるため、開発者が意識する必要がないように設計されています。

カスタムツール許可ロジックの実装

SDK の最も強力な機能の1つが、ツール実行前に独自のロジックを挟めることです。デフォルトの allowed_tools はリストで指定するだけですが、より精細な制御が必要な場合は permission_mode と組み合わせます。

import asyncio
import logging
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
 
BLOCKED_BASH_PATTERNS = [
    "rm -rf",
    "sudo",
    "curl | bash",
    "wget | sh",
    "eval",
]
 
async def run_controlled_agent(task: str, working_dir: str = ".") -> dict:
    """
    ツール実行を細かく制御するエージェント実装。
    本番環境での使用を想定した、セキュリティ意識の高い実装例。
    """
    
    options = ClaudeCodeOptions(
        max_turns=10,
        allowed_tools=["Read", "Write", "Bash", "Glob", "Grep"],
        cwd=working_dir,
    )
    
    result_summary = {
        "success": False,
        "tools_executed": [],
        "files_modified": [],
        "cost_usd": 0.0,
        "output": "",
    }
    
    try:
        async for message in query(prompt=task, options=options):
            
            if hasattr(message, 'content'):
                for block in message.content:
                    # ToolUseBlock を検出してロギング
                    if hasattr(block, 'name') and hasattr(block, 'input'):
                        tool_name = block.name
                        tool_input = block.input
                        
                        logger.info(f"Tool実行: {tool_name}")
                        
                        # Bash コマンドのセキュリティチェック
                        if tool_name == "Bash":
                            cmd = tool_input.get("command", "")
                            for pattern in BLOCKED_BASH_PATTERNS:
                                if pattern in cmd:
                                    logger.warning(f"危険なコマンドを検出: {pattern}")
                        
                        result_summary["tools_executed"].append({
                            "tool": tool_name,
                            "input_preview": str(tool_input)[:100],
                        })
                        
                        if tool_name == "Write":
                            file_path = tool_input.get("file_path", "unknown")
                            result_summary["files_modified"].append(file_path)
                            logger.info(f"ファイル書き込み: {file_path}")
                    
                    if hasattr(block, 'text'):
                        result_summary["output"] += block.text
            
            if hasattr(message, 'stop_reason'):
                result_summary["cost_usd"] = getattr(message, 'cost_usd', 0.0)
                
                if message.stop_reason == "end_turn":
                    result_summary["success"] = True
                    logger.info("エージェント正常終了")
                elif message.stop_reason == "max_turns":
                    logger.warning("max_turns に達して終了（タスクが未完了の可能性）")
                else:
                    logger.error(f"予期しない終了理由: {message.stop_reason}")
    
    except Exception as e:
        logger.error(f"エージェント実行エラー: {e}", exc_info=True)
        result_summary["error"] = str(e)
    
    return result_summary
 
async def main():
    result = await run_controlled_agent(
        task="src/ ディレクトリの Python ファイルを全て読んで、型ヒントが不足している関数を特定し、修正案をコメントとして追記してください",
        working_dir="./my_project"
    )
    
    print(f"成功: {result['success']}")
    print(f"実行ツール数: {len(result['tools_executed'])}")
    print(f"更新ファイル: {result['files_modified']}")
    print(f"コスト: ${result['cost_usd']:.4f}")
 
asyncio.run(main())

セッション継続とコンテキスト管理

複数回のやり取りでコンテキストを保持したい場合、セッション ID を活用します。

import asyncio
import json
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
 
SESSION_STORE = Path("/tmp/claude_sessions.json")
 
def load_session_state() -> dict:
    """セッション状態をファイルから読み込む"""
    if SESSION_STORE.exists():
        return json.loads(SESSION_STORE.read_text())
    return {}
 
def save_session_state(state: dict):
    """セッション状態をファイルに保存する"""
    SESSION_STORE.write_text(json.dumps(state, indent=2))
 
async def continue_session(
    task: str,
    session_key: str,
    project_dir: str = "."
) -> str:
    """
    セッションを継続して複数ターンに渡る作業を実現する。
    同じ session_key を指定すると前回の会話コンテキストが引き継がれる。
    """
    
    state = load_session_state()
    session_id = state.get(session_key)
    
    options = ClaudeCodeOptions(
        max_turns=15,
        allowed_tools=["Read", "Write", "Bash", "Glob", "Grep"],
        cwd=project_dir,
    )
    
    output_text = ""
    new_session_id = None
    
    if session_id:
        print(f"セッション再開: {session_id[:8]}...")
    else:
        print("新規セッション開始")
    
    async for message in query(prompt=task, options=options):
        
        if hasattr(message, 'session_id') and message.session_id:
            new_session_id = message.session_id
        
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text') and block.text:
                    output_text += block.text
        
        if hasattr(message, 'stop_reason'):
            print(f"終了理由: {message.stop_reason}")
    
    if new_session_id:
        state[session_key] = new_session_id
        save_session_state(state)
    
    return output_text
 
async def multi_step_workflow():
    """複数ステップにわたるリファクタリング作業の例"""
    
    project_key = "refactor_myapp_20260417"
    
    print("=== Step 1: コード分析 ===")
    analysis = await continue_session(
        task="src/ ディレクトリの全 Python ファイルを読んで、依存関係の循環がないか分析してください。問題があれば具体的なファイル名と行番号で報告してください。",
        session_key=project_key,
        project_dir="./my_project"
    )
    print(f"分析完了: {len(analysis)} 文字の出力")
    
    print("=== Step 2: 問題修正 ===")
    fix_result = await continue_session(
        task="先ほどの分析で見つかった問題を修正してください。変更したファイルと変更内容を報告してください。",
        session_key=project_key,
        project_dir="./my_project"
    )
    print(f"修正完了: {len(fix_result)} 文字の出力")
 
asyncio.run(multi_step_workflow())

セッション ID による会話継続は強力ですが、注意点があります。コンテキストウィンドウは有限で、セッションを長く続けると token 消費が増加します。私の経験では、1セッション15ターン程度を上限にして、新しいタスクでは新規セッションを使う方がコストと品質のバランスが良いと感じています。

並列エージェント実行でスループットを向上させる

単一エージェントの逐次実行では時間がかかる場合、複数エージェントを並列で走らせることができます。

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions
 
async def analyze_single_file(
    file_path: str,
    analysis_type: str,
    semaphore: asyncio.Semaphore,
) -> dict:
    """
    1ファイルを分析するエージェント。
    semaphore で同時実行数を制限してレート制限を回避する。
    """
    
    async with semaphore:
        options = ClaudeCodeOptions(
            max_turns=3,
            allowed_tools=["Read", "Grep"],
        )
        
        prompt = f"""
{file_path} を読んで、{analysis_type} の観点で分析してください。
出力形式:
- 問題数: N
- 問題の概要: (箇条書きで最大3件)
- 重要度: low/medium/high
"""
        
        result = {
            "file": file_path,
            "analysis_type": analysis_type,
            "findings": "",
            "cost_usd": 0.0,
            "error": None,
        }
        
        try:
            async for message in query(prompt=prompt, options=options):
                if hasattr(message, 'content'):
                    for block in message.content:
                        if hasattr(block, 'text'):
                            result["findings"] += block.text
                
                if hasattr(message, 'cost_usd'):
                    result["cost_usd"] = message.cost_usd or 0.0
        
        except Exception as e:
            result["error"] = str(e)
            print(f"エラー {file_path}: {e}")
        
        return result
 
async def parallel_codebase_analysis(
    file_paths: list[str],
    analysis_type: str = "セキュリティ上のリスク",
    max_concurrent: int = 3,
) -> list[dict]:
    """
    複数ファイルを並列で分析する。
    max_concurrent=3 は API レート制限を考慮した安全な値。
    """
    
    semaphore = asyncio.Semaphore(max_concurrent)
    
    tasks = [
        analyze_single_file(fp, analysis_type, semaphore)
        for fp in file_paths
    ]
    
    print(f"{len(tasks)} ファイルを並列分析開始（同時実行数: {max_concurrent}）")
    
    results = await asyncio.gather(*tasks, return_exceptions=False)
    
    total_cost = sum(r.get("cost_usd", 0) for r in results)
    error_count = sum(1 for r in results if r.get("error"))
    
    print(f"分析完了")
    print(f"  成功: {len(results) - error_count}/{len(results)} ファイル")
    print(f"  合計コスト: ${total_cost:.4f}")
    
    return results

max_concurrent=3 の設定は重要です。Anthropic API にはレート制限があり、並列リクエストを一度に多く送りすぎると 429 Too Many Requests が発生します。私のプロジェクトで試した限り、Pro プランでは同時3〜5リクエスト、Team プランでは5〜10リクエストが安定して動作しました。

TypeScript SDK での実装パターン

TypeScript を使う場合も基本構造は同じですが、型の扱い方が Python と異なります。

import { query } from "@anthropic-ai/claude-code";
import type {
  SDKMessage,
  AssistantMessage,
  ResultMessage,
  ClaudeCodeOptions,
} from "@anthropic-ai/claude-code";
 
interface AgentResult {
  success: boolean;
  output: string;
  toolsUsed: string[];
  costUsd: number;
  error?: string;
}
 
async function runTypeScriptAgent(
  prompt: string,
  workingDir: string = process.cwd()
): Promise<AgentResult> {
  const options: ClaudeCodeOptions = {
    maxTurns: 10,
    allowedTools: ["Read", "Write", "Bash", "Glob"],
    cwd: workingDir,
  };
 
  const result: AgentResult = {
    success: false,
    output: "",
    toolsUsed: [],
    costUsd: 0,
  };
 
  try {
    for await (const message of query({ prompt, options })) {
      
      if (isAssistantMessage(message)) {
        for (const block of message.content) {
          if (block.type === "text") {
            result.output += block.text;
          }
          if (block.type === "tool_use") {
            result.toolsUsed.push(block.name);
            console.log(`Tool: ${block.name}`);
          }
        }
      }
 
      if (isResultMessage(message)) {
        result.success = message.stop_reason === "end_turn";
        result.costUsd = message.cost_usd ?? 0;
        
        if (!result.success) {
          console.warn(`終了理由: ${message.stop_reason}`);
        }
      }
    }
  } catch (error) {
    result.error = error instanceof Error ? error.message : String(error);
    console.error("Agent error:", result.error);
  }
 
  return result;
}
 
// 型ガード関数（実行時の型チェックに必須）
function isAssistantMessage(msg: SDKMessage): msg is AssistantMessage {
  return msg.type === "assistant";
}
 
function isResultMessage(msg: SDKMessage): msg is ResultMessage {
  return msg.type === "result";
}
 
// 実行例
async function main() {
  const result = await runTypeScriptAgent(
    "package.json を読んで、不要な依存関係がないか確認してください。見つかれば具体的に列挙してください。",
    "./my_project"
  );
 
  console.log(`成功: ${result.success}`);
  console.log(`使用ツール: ${[...new Set(result.toolsUsed)].join(", ")}`);
  console.log(`コスト: $${result.costUsd.toFixed(4)}`);
}
 
main().catch(console.error);

TypeScript では型ガード関数（isAssistantMessage 等）が重要な役割を果たします。SDKMessage は union 型で定義されているため、各メッセージタイプを正しく絞り込まないとコンパイルエラーになります。型ガード関数を用意しておくことで、as キャストを使わずに安全にプロパティへアクセスできます。

よくある落とし穴と対処法

実際に SDK を使い込んで遭遇した問題をまとめます。これを知っているかどうかで、デバッグ時間が大きく変わります。

1. max_turns を設定しないと無限ループになる

Claude がツールを使い続けて終わらない状況は、end_turn になる条件が満たされない場合に起きます。特に「ファイルを全て確認して完璧にしてください」のような曖昧なタスクで発生しやすいです。

# 悪い例: max_turns なし → 潜在的な無限ループ
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write"],
)
 
# 良い例: 常に max_turns を設定
options = ClaudeCodeOptions(
    max_turns=10,  # タスクの複雑さに応じて設定
    allowed_tools=["Read", "Write"],
)

2. ツールエラーが無視されて間違った結果が返ることがある

ツール実行が失敗しても、Claude はエラーを受け取って別のアプローチを試みることがあります。この動作は便利な場合もありますが、「ファイルが存在しないのに成功したと判断される」ような誤検知につながることも。

# ResultMessage の stop_reason を必ず確認する
async for message in query(prompt=task, options=options):
    if hasattr(message, 'stop_reason'):
        if message.stop_reason == "end_turn":
            print("正常完了")
        elif message.stop_reason == "max_turns":
            print("ターン上限に達した — タスクが未完了の可能性あり")
            # ここでアラートを飛ばす or リトライロジックを入れる
        elif message.stop_reason == "error_max_retries":
            raise RuntimeError("エージェントがツールエラーから回復できませんでした")

3. 大きなコードベースでコンテキストが肥大化する

ファイルを大量に Read すると、コンテキストウィンドウが埋まってレスポンスの質が落ちたり、コストが急増したりします。対策として、作業ディレクトリを最小化して不要なファイルへのアクセスを防ぎます。

# 必要なサブディレクトリのみを cwd に指定
options = ClaudeCodeOptions(
    cwd="./src/specific_module",  # プロジェクト全体ではなく対象モジュールのみ
    allowed_tools=["Read", "Write"],
    max_turns=10,
)

4. cwd の解決パスが意図と異なる

cwd は相対パスを指定すると、スクリプトの実行ディレクトリから解決されます。CI/CD 環境では絶対パスを使う方が安全です。

import os
from pathlib import Path
 
# 相対パスは実行場所によって変わる（危険）
options = ClaudeCodeOptions(cwd="./my_project")
 
# 絶対パスで確実に指定（推奨）
project_root = Path(__file__).parent.parent / "my_project"
options = ClaudeCodeOptions(cwd=str(project_root.resolve()))

実用的な応用: CI/CD での自動コードレビュー

ここまでの知識を組み合わせて、GitHub Actions で動作するコードレビューエージェントを実装します。

#!/usr/bin/env python3
"""
ci_review_agent.py — GitHub Actions で動作するコードレビューエージェント。
 
使用方法:
  python ci_review_agent.py --files "src/auth.py,src/api.py" --output-file review.md
"""
 
import asyncio
import argparse
import sys
from pathlib import Path
from datetime import datetime
from claude_code_sdk import query, ClaudeCodeOptions
 
async def review_changed_files(
    file_list: list[str],
    output_file: str,
    project_dir: str = ".",
) -> int:
    """
    変更ファイルをレビューしてマークダウン形式のレポートを出力する。
    
    Returns:
        0: レビュー完了（問題なし or 軽微）
        1: 重大な問題を検出
        2: エージェント実行エラー
    """
    
    if not file_list:
        return 0
    
    valid_files = []
    for fp in file_list:
        p = Path(project_dir) / fp
        if p.exists():
            valid_files.append(fp)
        else:
            print(f"ファイルが存在しません（スキップ）: {fp}")
    
    if not valid_files:
        return 0
    
    file_list_str = "\n".join(f"- {f}" for f in valid_files)
    
    prompt = f"""
以下の変更ファイルをコードレビューしてください。
 
## レビュー対象ファイル
{file_list_str}
 
## レビュー観点
1. セキュリティリスク（SQLインジェクション、認証バイパス、機密情報のハードコードなど）
2. エラーハンドリングの不備
3. 型安全性の問題（Python の場合は型ヒントの欠如）
4. パフォーマンス上の懸念点
 
## 出力形式
マークダウン形式で出力してください。各ファイルについて:
- **致命的な問題**（即修正必要）: [critical]タグ付き
- **警告**（修正推奨）: [warning]タグ付き
- **提案**（あれば修正）: [info]タグ付き
 
問題がない場合は「問題なし」と出力してください。
"""
    
    options = ClaudeCodeOptions(
        max_turns=8,
        allowed_tools=["Read", "Grep"],
        cwd=project_dir,
    )
    
    review_content = ""
    has_critical = False
    exit_code = 0
    
    try:
        async for message in query(prompt=prompt, options=options):
            if hasattr(message, 'content'):
                for block in message.content:
                    if hasattr(block, 'text'):
                        review_content += block.text
                        if "[critical]" in block.text.lower():
                            has_critical = True
            
            if hasattr(message, 'stop_reason'):
                cost = getattr(message, 'cost_usd', 0) or 0
                print(f"レビュー完了 — コスト: ${cost:.4f}")
    
    except Exception as e:
        print(f"エージェントエラー: {e}", file=sys.stderr)
        return 2
    
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    report = f"""# コードレビューレポート
生成日時: {timestamp}
対象ファイル数: {len(valid_files)}
 
---
 
{review_content}
"""
    
    Path(output_file).write_text(report, encoding="utf-8")
    print(f"レポート出力: {output_file}")
    
    if has_critical:
        print("致命的な問題が検出されました")
        exit_code = 1
    
    return exit_code
 
def main():
    parser = argparse.ArgumentParser(description="CI コードレビューエージェント")
    parser.add_argument("--files", required=True, help="カンマ区切りのファイルリスト")
    parser.add_argument("--output-file", default="review.md", help="出力ファイルパス")
    parser.add_argument("--project-dir", default=".", help="プロジェクトルートディレクトリ")
    
    args = parser.parse_args()
    file_list = [f.strip() for f in args.files.split(",") if f.strip()]
    
    exit_code = asyncio.run(
        review_changed_files(
            file_list=file_list,
            output_file=args.output_file,
            project_dir=args.project_dir,
        )
    )
    
    sys.exit(exit_code)
 
if __name__ == "__main__":
    main()

このスクリプトを GitHub Actions から呼び出す設定例です：

name: AI Code Review
 
on:
  pull_request:
    types: [opened, synchronize]
 
jobs:
  ai-review:
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      
      - name: Install dependencies
        run: |
          pip install claude-code-sdk
          npm install -g @anthropic-ai/claude-code
      
      - name: Get changed files
        id: changed-files
        uses: tj-actions/changed-files@v44
        with:
          files: |
            **/*.py
            **/*.ts
            **/*.js
      
      - name: Run AI Review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python .github/scripts/ci_review_agent.py \
            --files "${{ steps.changed-files.outputs.all_changed_files }}" \
            --output-file review.md \
            --project-dir .
      
      - name: Upload Review Report
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: ai-review-report
          path: review.md

本番環境でこの構成を動かして気づいたことがあります。PR が同時に複数作成されると、並列でレビューエージェントが起動します。同時実行数を制御したい場合は、GitHub Actions の concurrency グループを設定することをお勧めします。

MCP サーバーとの組み合わせで拡張する

SDK から起動したエージェントは、MCP サーバー経由で外部システムとも接続できます。例えば、社内データベースや Slack への投稿を MCP サーバーとして実装しておくと、エージェントが自律的にそれらを利用できるようになります。

from claude_code_sdk import query, ClaudeCodeOptions
 
async def agent_with_mcp():
    """MCP サーバーを活用したエージェントの例"""
    
    options = ClaudeCodeOptions(
        max_turns=10,
        allowed_tools=["Read", "Write", "Bash"],
        # MCP サーバーの設定（~/.claude/mcp.json で事前設定）
        # allowed_tools に MCP ツール名を追加するだけで利用可能になる
    )
    
    # MCP サーバー経由で GitHub Issues を取得して分析するシナリオ
    prompt = """
GitHub の未解決 Issue を確認して、優先度の高い順に3件をピックアップしてください。
各 Issue について:
1. 問題の概要
2. 修正にかかる推定時間
3. 修正のアプローチ案
を出力してください。
"""
    
    async for message in query(prompt=prompt, options=options):
        if hasattr(message, 'content'):
            for block in message.content:
                if hasattr(block, 'text'):
                    print(block.text, end="")
 
asyncio.run(agent_with_mcp())

MCP サーバーの実装詳細については、Claude Code で MCP サーバーを実践活用する — 設計から運用までが参考になります。

公式ドキュメントには書かれていない運用上の気づき

SDK のリファレンスを読むだけでは見えてこない、自律ループを何ヶ月か回し続けて初めて分かったことを3点に絞って共有します。

1. ヘッドレス実行の課金は「サブスク枠外」— 1回あたりのコストを必ず計測する

2026年6月15日の課金変更で、SDK 経由のヘッドレス実行はサブスクリプションの利用上限から外れ、月次クレジット制（Pro $20 / Max 5x $100 / Max 20x $200・繰越なし）に移りました。対話的な CLI 利用と同じ感覚でループを回していると、月の半ばでクレジットが尽きます。私の運用でも、この変更の影響は最初の週にはっきり数字に出ました。

対策はシンプルで、ResultMessage からコストを毎回記録することです。次のコードは、本番ループに必ず挟んでいる計測レイヤーです。

import json
from datetime import datetime, timezone
from claude_code_sdk import query, ClaudeCodeOptions
 
COST_LOG = "agent_cost_log.jsonl"
 
async def run_with_cost_log(task_id: str, task: str, options: ClaudeCodeOptions) -> None:
    """実行ごとのコストを JSONL に記録する計測レイヤー"""
    async for message in query(prompt=task, options=options):
        if type(message).__name__ == "ResultMessage":
            usage = getattr(message, "usage", None) or {}
            record = {
                "ts": datetime.now(timezone.utc).isoformat(),
                "task": task_id,
                "cost_usd": getattr(message, "total_cost_usd", None),
                "input_tokens": usage.get("input_tokens"),
                "output_tokens": usage.get("output_tokens"),
                "num_turns": getattr(message, "num_turns", None),
            }
            with open(COST_LOG, "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

計測を始めて分かったのは、コストの分布が想像以上に偏っていることでした。私の環境では max_turns=10 の記事整形タスク1回あたり平均 $0.04〜0.12、コンテキストが肥大したワーストケースで $0.31。平均だけ見て月間コストを見積もると、この裾の重さに足をすくわれます。分布が見えると「どのタスクをバッチに寄せるか」「どこでモデルを切り替えるか」を数字で判断できるようになります。コスト側の実装は Claude API コスト最適化プロダクションガイドに詳しくまとめています。

2. モデルは既定に任せず、明示的に固定する

2026年6月30日、Claude Code の既定モデルが Sonnet 5 に切り替わりました。導入価格（$2/百万入力・$10/百万出力、2026年8月31日まで。以降は $3/$15）の恩恵は大きい一方、既定任せの自律ループは「ある日突然、挙動とコストが変わる」リスクを抱えます。逆方向の事故もあります。Opus 4.7 の fast mode は 2026年7月24日で廃止され、以降 speed:"fast" 指定はエラーになります。放置された定期ジョブが朝から全滅する、典型的なパターンです。

options = ClaudeCodeOptions(
    model="claude-sonnet-5",  # 既定に任せず明示する
    max_turns=10,
    allowed_tools=["Read", "Write"],
)

私はこの一行を定期ジョブのテンプレートに含めておき、新しいジョブが「既定モデル依存」のまま生まれない形にしています。

3. 月1回の「モデル棚卸し」をループ運用の定例にする

上の2つから導かれる運用習慣です。私は毎月1日に次の5項目を確認しています。

全ジョブの model 指定を grep し、廃止予定モデルや fast mode 指定が残っていないか確認する
コストログを集計し、前月比 ±30% 以上動いたタスクを特定して原因を見る
価格改定（導入価格の期限など）をリリースノートで確認し、影響するジョブに印を付ける
stop_reason == "max_turns" の到達率を確認する。私の環境では 5% を超えたらプロンプト見直しのシグナルとしています
失敗ジョブの再実行ログを確認し、自動リカバリで拾えなかったものだけ手を入れる

毎月30分ほどの作業ですが、「気づいたら課金が想定の3倍」「モデル廃止でジョブ全滅」という事故を、この習慣を始めてからはゼロに抑えられています。

自前ループ・CLI・Managed Agents の使い分け

2026年前半に Managed Agents（スケジュールデプロイ・クレデンシャル vault）が出揃い、「そもそも自前でループを書くべきか」という判断が新たに必要になりました。個人開発の規模で回してきた、現時点での私の使い分けは次の通りです。

ユースケース	推奨	理由
対話的な開発・調査	CLI	許可確認や試行錯誤は、人間のリズムに合わせた対話が最速です
1プロンプトで完結する定期実行	Managed Agents のスケジュールデプロイ	cron 式の定期起動とクレデンシャル vault が付属し、失敗回収も任せられます
独自の許可ロジック・複数エージェント協調・独自リトライ	本記事の自前 SDK ループ	ツール実行の可否判断とエラー回復を自分のコードで握れるのは、この構成だけです

複数エージェントの協調が主目的であれば、Claude Code で実装する高度なマルチエージェント・オーケストレーションが次の一歩になります。また、長時間ループでコンテキストが詰まる問題には、Claude Code の Context Window 管理術で紹介している7つのパターンがそのまま効きます。

全体を振り返ってと次のステップ

今すぐ試すなら、まず max_turns=3 のシンプルなエージェントを書いてストリームの中身を観察してみてください。inspect_stream() 関数を手元のプロジェクトで実行するだけで、SDK がどのように動作しているかが直感的に理解できます。その理解があれば、本記事で紹介した他のパターンも自然と応用できるようになります。

SDK の活用で「Claude Code はコードエディタ補助ツール」という認識が変わります。プログラマブルに制御できる自律エージェントの基盤として、CI/CD 自動化・コードベース分析・定型タスクのバッチ処理まで、様々なシステムに組み込める強力な選択肢です。

お読みいただきありがとうございます

Claude Lab は広告なしで運営しており、サーバー費用などの運営コストはメンバーシップのご支援で賄っています。実装コード・ベンチマーク・本番設計パターンなど、実務でお役立ていただける記事を毎日更新しています。もし読んでよかったと感じていただけましたら、ぜひご覧ください。

✦コピー&ペーストで使える実装コード付き
✦毎日新しい上級ガイドを追加
✦¥580/月または ¥1,480 の永久アクセス

メンバーシップを見る →