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

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())

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())

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())

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

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);

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

# 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("エージェントがツールエラーから回復できませんでした")

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

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()))

#\!/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()

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

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())

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")

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