Claude Agent SDK: Build a Production AI Agent

pip install claude-agent-sdk
export ANTHROPIC_API_KEY=your-api-key

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions


async def review_pull_request(pr_url: str):
    """Review a GitHub pull request for code quality and security issues."""

    options = ClaudeAgentOptions(
        system_prompt="""You are an expert code reviewer. Analyze the provided
        code changes for: code quality issues, security vulnerabilities,
        performance problems, and style inconsistencies. Be specific about
        line numbers and provide actionable suggestions.""",
        allowed_tools=["Read", "Glob", "Grep", "Bash"],
        max_turns=20,
    )

    async for message in query(
        prompt=f"Review the code changes in this PR: {pr_url}",
        options=options,
    ):
        if hasattr(message, "result"):
            return message.result


result = asyncio.run(review_pull_request("https://github.com/org/repo/pull/42"))
print(result)

from claude_agent_sdk import tool, create_sdk_mcp_server
from typing import Any
import json


@tool(
    "analyze_complexity",
    "Calculate cyclomatic complexity for a Python function",
    {"code": str, "function_name": str},
)
async def analyze_complexity(args: dict[str, Any]) -> dict[str, Any]:
    code = args["code"]
    func_name = args["function_name"]

    # Count decision points: if, elif, for, while, except, and, or
    decision_keywords = ["if ", "elif ", "for ", "while ", "except ", " and ", " or "]
    complexity = 1  # Base complexity
    for keyword in decision_keywords:
        complexity += code.count(keyword)

    risk = "low" if complexity <= 5 else "medium" if complexity <= 10 else "high"

    return {
        "content": [{
            "type": "text",
            "text": json.dumps({
                "function": func_name,
                "cyclomatic_complexity": complexity,
                "risk_level": risk,
                "recommendation": f"Consider refactoring"
                    if complexity > 10 else "Complexity acceptable",
            }),
        }]
    }


@tool(
    "check_security_patterns",
    "Scan code for common security anti-patterns",
    {"code": str, "language": str},
)
async def check_security_patterns(args: dict[str, Any]) -> dict[str, Any]:
    code = args["code"]
    findings = []

    patterns = {
        "SQL injection": ["execute(f\"", "execute(f'", ".format(", "% ("],
        "Hardcoded secrets": ["password =", "api_key =", "secret =", "token ="],
        "Unsafe deserialization": ["pickle.loads", "yaml.load(", "eval("],
        "Command injection": ["os.system(", "subprocess.call(shell=True"],
    }

    for vuln_type, indicators in patterns.items():
        for indicator in indicators:
            if indicator in code:
                findings.append({
                    "type": vuln_type,
                    "indicator": indicator,
                    "severity": "high" if vuln_type in
                        ["SQL injection", "Command injection"] else "medium",
                })

    return {
        "content": [{
            "type": "text",
            "text": json.dumps({
                "total_findings": len(findings),
                "findings": findings,
            }),
        }]
    }


@tool(
    "generate_review_comment",
    "Format a structured review comment for posting to GitHub",
    {"file_path": str, "line_number": int, "severity": str,
     "category": str, "message": str, "suggestion": str},
)
async def generate_review_comment(args: dict[str, Any]) -> dict[str, Any]:
    severity_labels = {"high": "CRITICAL", "medium": "WARNING", "low": "SUGGESTION"}
    label = severity_labels.get(args["severity"], "NOTE")

    formatted = (
        f"**[{label}]** {args['category']}\n\n"
        f"{args['message']}\n\n"
        f"**Suggested fix:**\n{args['suggestion']}"
    )

    return {
        "content": [{
            "type": "text",
            "text": json.dumps({
                "file": args["file_path"],
                "line": args["line_number"],
                "body": formatted,
            }),
        }]
    }


# Bundle all tools into an MCP server
review_tools = create_sdk_mcp_server(
    name="code-review",
    version="1.0.0",
    tools=[analyze_complexity, check_security_patterns, generate_review_comment],
)

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher
import time

# Track token usage across the session
usage_tracker = {"input_tokens": 0, "output_tokens": 0, "tool_calls": 0}


async def enforce_read_only(input_data, tool_use_id, context):
    """Block any tool call that could modify files or run destructive commands."""
    tool_name = input_data.get("tool_name", "")

    # Block write operations entirely
    if tool_name in ["Write", "Edit"]:
        return {
            "systemMessage": "This is a read-only review. Do not modify files.",
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Code review agent is read-only",
            },
        }

    # For Bash, block destructive commands
    if tool_name == "Bash":
        command = input_data.get("tool_input", {}).get("command", "")
        destructive = ["rm ", "mv ", "git push", "git commit", "chmod", "chown"]
        if any(cmd in command for cmd in destructive):
            return {
                "hookSpecificOutput": {
                    "hookEventName": input_data["hook_event_name"],
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Destructive command blocked: {command}",
                },
            }

    return {}


async def track_usage(input_data, tool_use_id, context):
    """Log every tool call for observability."""
    usage_tracker["tool_calls"] += 1
    tool_name = input_data.get("tool_name", "unknown")
    print(f"[AUDIT] Tool call #{usage_tracker['tool_calls']}: {tool_name}")
    return {}


async def cost_guard(input_data, tool_use_id, context):
    """Stop the agent if it exceeds a tool call budget."""
    if usage_tracker["tool_calls"] > 50:
        return {
            "continue_": False,
            "systemMessage": "Tool call budget exceeded. Wrap up your review.",
        }
    return {}

options = ClaudeAgentOptions(
    system_prompt="You are an expert code reviewer...",
    allowed_tools=[
        "Read", "Glob", "Grep", "Bash",
        "mcp__code-review__analyze_complexity",
        "mcp__code-review__check_security_patterns",
        "mcp__code-review__generate_review_comment",
    ],
    mcp_servers={"code-review": review_tools},
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Write|Edit|Bash", hooks=[enforce_read_only]),
            HookMatcher(hooks=[track_usage]),
            HookMatcher(hooks=[cost_guard]),
        ],
        "PostToolUse": [
            HookMatcher(hooks=[track_usage]),
        ],
    },
    max_turns=30,
)

import os

options = ClaudeAgentOptions(
    system_prompt="""You are an expert code reviewer. For each PR:
    1. Fetch the PR details and changed files using GitHub MCP tools
    2. Read each changed file and analyze the diff
    3. Run security and complexity analysis on modified functions
    4. Generate structured review comments
    5. Post a summary review""",
    mcp_servers={
        "code-review": review_tools,
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]},
        },
    },
    allowed_tools=[
        "Read", "Glob", "Grep", "Bash",
        "mcp__code-review__*",
        "mcp__github__*",
    ],
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Write|Edit|Bash", hooks=[enforce_read_only]),
            HookMatcher(hooks=[cost_guard]),
        ],
    },
    max_turns=30,
)

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_TOKEN": "${GITHUB_TOKEN}"
      }
    }
  }
}

from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

options = ClaudeAgentOptions(
    system_prompt="""You are a lead code reviewer orchestrating a team of
    specialized reviewers. For each PR:
    1. Delegate style review to the style-reviewer agent
    2. Delegate security review to the security-reviewer agent
    3. Delegate performance review to the performance-reviewer agent
    4. Aggregate their findings into a final review""",
    allowed_tools=["Read", "Glob", "Grep", "Task", "mcp__code-review__*"],
    mcp_servers={"code-review": review_tools},
    agents={
        "style-reviewer": AgentDefinition(
            description="Reviews code for style, naming conventions, and patterns.",
            prompt="""Analyze code style: variable naming, function length,
            import organization, docstrings, and adherence to PEP 8.
            Report findings with severity levels.""",
            tools=["Read", "Glob", "Grep"],
            model="sonnet",
        ),
        "security-reviewer": AgentDefinition(
            description="Scans code for security vulnerabilities and anti-patterns.",
            prompt="""Scan for: SQL injection, XSS, hardcoded secrets, unsafe
            deserialization, command injection, path traversal, and insecure
            cryptography. Rate each finding as critical, high, or medium.""",
            tools=["Read", "Glob", "Grep", "mcp__code-review__check_security_patterns"],
        ),
        "performance-reviewer": AgentDefinition(
            description="Analyzes code for performance issues and complexity.",
            prompt="""Identify: O(n^2) or worse algorithms, unnecessary database
            queries, missing caching opportunities, memory leaks, and excessive
            complexity. Include Big O analysis where relevant.""",
            tools=["Read", "Glob", "Grep", "mcp__code-review__analyze_complexity"],
            model="sonnet",
        ),
    },
    max_turns=40,
)

async def track_subagent(input_data, tool_use_id, context):
    agent_id = input_data.get("agent_id", "unknown")
    transcript = input_data.get("agent_transcript_path", "")
    print(f"[SUBAGENT] Completed: {agent_id} | Transcript: {transcript}")
    return {}

options.hooks["SubagentStop"] = [HookMatcher(hooks=[track_subagent])]

async def ci_review(pr_number: int, repo: str) -> dict:
    """Run a code review as part of CI/CD pipeline."""

    result = {"status": "unknown", "findings": [], "approved": False}

    async for message in query(
        prompt=f"Review PR #{pr_number} in {repo}. "
               f"Output JSON with fields: findings (array), "
               f"approved (bool), summary (string).",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Bash"],
            max_turns=20,
            permission_mode="acceptEdits",
        ),
    ):
        if hasattr(message, "result"):
            result["review"] = message.result
            result["status"] = "completed"

    return result

async def interactive_review():
    session_id = None
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
    )

    # First pass: initial review
    async for message in query(
        prompt="Review the changes in src/auth.py",
        options=options,
    ):
        if hasattr(message, "subtype") and message.subtype == "init":
            session_id = message.session_id
        if hasattr(message, "result"):
            print(message.result)

    # Follow-up: agent remembers context from the first query
    async for message in query(
        prompt="What about the error handling in that module?",
        options=ClaudeAgentOptions(resume=session_id),
    ):
        if hasattr(message, "result"):
            print(message.result)

# Container-level isolation: restrict network to specific domains
options = ClaudeAgentOptions(
    allowed_tools=["Read", "Glob", "Grep", "Bash"],
    # The SDK respects network restrictions from the container
    # Set --network none on the Docker container, then use a
    # proxy to allowlist specific domains (GitHub API, etc.)
)

async for message in query(prompt="Review the PR", options=options):
    # Check init message for MCP connection failures
    if hasattr(message, "subtype") and message.subtype == "init":
        mcp_servers = getattr(message, "data", {}).get("mcp_servers", [])
        failed = [s for s in mcp_servers if s.get("status") != "connected"]
        if failed:
            print(f"MCP servers failed to connect: {failed}")

    # Check for execution errors
    if hasattr(message, "subtype") and message.subtype == "error_during_execution":
        print("Agent encountered an error during execution")
        # Implement retry logic or fallback