Claude Code Agent SDK — カスタムAIエージェント構築ガイド｜Python実装とマルチエージェント設計

pip install claude-agent-sdk

import os
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."

project/
├── agents/
│   ├── reviewer.py    # コードレビューエージェント
│   ├── doc_gen.py     # ドキュメント生成エージェント
│   └── utils.py       # 共通ユーティリティ
├── .env               # APIキー（Gitignore推奨）
└── requirements.txt

from claude_agent_sdk import query

result = query(
    "Analyze the codebase and identify potential security vulnerabilities",
    cwd="/path/to/project"
)
print(result["output"])

from claude_agent_sdk import ClaudeSDKClient, Tool

client = ClaudeSDKClient()
result = client.query(
    prompt="Review this PR",
    tools=[custom_tool],
    hooks=[pre_commit_hook],
    cwd="/path/to/project"
)

from claude_agent_sdk import Tool

def check_test_coverage(path: str) -> dict:
    """Calculate test coverage for a given file."""
    # pytest-cov等を使った実装
    coverage = calculate_coverage(path)
    return {"coverage_percent": coverage, "status": "pass" if coverage > 80 else "fail"}

coverage_tool = Tool(
    name="check_test_coverage",
    description="Calculate test coverage percentage for a source file",
    input_schema={
        "type": "object",
        "properties": {"path": {"type": "string"}},
        "required": ["path"]
    },
    function=check_test_coverage
)

client = ClaudeSDKClient()
result = client.query(
    "Check test coverage for all files in src/",
    tools=[coverage_tool]
)

from claude_agent_sdk import Hook

def validate_before_commit(context):
    """Run validation checks before git commit."""
    if "git commit" in context.tool_input:
        # lint check
        result = subprocess.run(["npm", "run", "lint"], capture_output=True)
        if result.returncode != 0:
            return {"allow": False, "message": "Lint check failed. Fix errors before committing."}
    return {"allow": True}

commit_hook = Hook(
    event="PreToolUse",
    tool="Bash",
    function=validate_before_commit
)

client = ClaudeSDKClient()
result = client.query(
    "Implement feature X and commit",
    hooks=[commit_hook]
)

from claude_agent_sdk import ClaudeSDKClient

client = ClaudeSDKClient()

# 親エージェント
result = client.query("""
Review this pull request:
1. Create a sub-agent to analyze code quality
2. Create another sub-agent to check security vulnerabilities
3. Aggregate results and generate a summary report
""")

from claude_agent_sdk import ClaudeSDKClient
import asyncio

async def orchestrate_migration():
    lead = ClaudeSDKClient()
    
    # リードエージェントがタスク分割
    plan = await lead.query("""
    Plan migration from MongoDB to PostgreSQL:
    1. Analyze schema
    2. Generate migration scripts
    3. Create test fixtures
    4. Write documentation
    Output: JSON task list
    """)
    
    tasks = parse_tasks(plan["output"])
    
    # ワーカーエージェントを並列起動
    workers = [ClaudeSDKClient() for _ in tasks]
    results = await asyncio.gather(*[
        worker.query(task["prompt"]) for worker, task in zip(workers, tasks)
    ])
    
    # 結果をマージ
    final = await lead.query(f"Aggregate results: {results}")
    return final

result = asyncio.run(orchestrate_migration())

client = ClaudeSDKClient(permission_mode="restricted")
client.add_allowed_commands(["npm test", "git status"])

client = ClaudeSDKClient(
    sandbox=True,
    allowed_paths=["/project/src", "/project/tests"],
    blocked_network=True
)

name: AI Code Review
on:
  pull_request:
    types: [opened, synchronize]

jobs:
  ai-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install claude-agent-sdk
      - name: Run AI review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python scripts/ai_reviewer.py --pr ${{ github.event.pull_request.number }}

from claude_agent_sdk import ClaudeSDKClient
import os

pr_number = os.environ["PR_NUMBER"]
client = ClaudeSDKClient(permission_mode="restricted")

result = client.query(f"""
Review PR #{pr_number}:
1. Check code quality
2. Identify security issues
3. Suggest improvements
Post review comments via GitHub API
""")

from claude_agent_sdk import ClaudeSDKClient, Tool
import subprocess
import json

def get_pr_diff(pr_number: int) -> str:
    """Fetch PR diff from GitHub."""
    result = subprocess.run(
        ["gh", "pr", "diff", str(pr_number)],
        capture_output=True, text=True
    )
    return result.stdout

def post_review_comment(pr_number: int, body: str, path: str, line: int):
    """Post review comment to GitHub PR."""
    subprocess.run([
        "gh", "pr", "review", str(pr_number),
        "--comment", "--body", body,
        "--path", path, "--line", str(line)
    ])

gh_tools = [
    Tool(name="get_pr_diff", function=get_pr_diff, ...),
    Tool(name="post_review_comment", function=post_review_comment, ...)
]

client = ClaudeSDKClient()
result = client.query(
    f"Review PR #{pr_number}. For each issue found, post a review comment with specific line references and improvement suggestions.",
    tools=gh_tools
)

from claude_agent_sdk import ClaudeSDKClient

client = ClaudeSDKClient()

result = client.query("""
Analyze all API endpoints in src/api/ and generate OpenAPI 3.0 specification:
1. Extract route definitions, parameters, response types
2. Read JSDoc/docstrings for descriptions
3. Generate openapi.yaml with complete schemas
4. Validate the generated spec
""")

if result["success"]:
    print("API documentation generated at docs/openapi.yaml")