Master the Claude Agent SDK

pip install claude-agent-sdk httpx

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    async for message in query(
        prompt="What files are in this directory?",
        options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"]),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    SystemMessage, AssistantMessage, ResultMessage
)

async def main():
    async for message in query(
        prompt="List all Python files in this directory",
        options=ClaudeAgentOptions(allowed_tools=["Bash", "Glob"]),
    ):
        if isinstance(message, SystemMessage):
            print(f"[SYSTEM] subtype={message.subtype}")
            if message.subtype == "init":
                print(f"  Session ID: {message.data['session_id']}")

        elif isinstance(message, AssistantMessage):
            print(f"[ASSISTANT] {message.content[:100]}...")

        elif isinstance(message, ResultMessage):
            print(f"[RESULT] status={message.subtype}")
            print(message.result)

asyncio.run(main())

from claude_agent_sdk import tool, create_sdk_mcp_server

@tool(
    "get_temperature",
    "Get the current temperature at a location",
    {"latitude": float, "longitude": float},
)
async def get_temperature(args):
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": args["latitude"],
                "longitude": args["longitude"],
                "current": "temperature_2m",
                "temperature_unit": "fahrenheit",
            },
        )
        data = response.json()
    return {
        "content": [
            {
                "type": "text",
                "text": f"Temperature: {data['current']['temperature_2m']}°F",
            }
        ]
    }

# Create the MCP server
weather_server = create_sdk_mcp_server(
    name="weather",
    version="1.0.0",
    tools=[get_temperature],
)

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_temperature"],
    )
    async for message in query(
        prompt="What's the temperature in San Francisco?",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

@tool(
    "fetch_stock_price",
    "Look up a stock price by ticker symbol",
    {"ticker": str},
)
async def fetch_stock_price(args):
    try:
        # ... fetch stock price logic ...
        price = 182.63
        return {
            "content": [{"type": "text", "text": f"{args['ticker']}: ${price}"}]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True,
        }

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition

async def main():
    async for message in query(
        prompt="Review this codebase for security issues and code quality",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep", "Agent"],
            agents={
                "security-reviewer": AgentDefinition(
                    description="Security expert that finds vulnerabilities.",
                    prompt=(
                        "Analyze code for security issues: "
                        "injection, auth flaws, data exposure. "
                        "Report each finding with severity and recommendation."
                    ),
                    tools=["Read", "Glob", "Grep"],
                ),
                "quality-reviewer": AgentDefinition(
                    description="Code quality expert for maintainability reviews.",
                    prompt=(
                        "Review code for readability, DRY, SOLID principles. "
                        "Identify code smells, complex functions, and "
                        "suggest improvements."
                    ),
                    tools=["Read", "Glob", "Grep"],
                ),
            },
        ),
    ):
        if hasattr(message, "result"):
            print(message.result)

asyncio.run(main())

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    SystemMessage, ResultMessage
)

async def main():
    session_id = None

    # First query: read and analyze
    print("--- Query 1: Analyzing auth module ---")
    async for message in query(
        prompt="Read and analyze the auth module in this project",
        options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"]),
    ):
        if isinstance(message, SystemMessage) and message.subtype == "init":
            session_id = message.data["session_id"]
            print(f"Session ID: {session_id}")
        if isinstance(message, ResultMessage):
            print(message.result)

    # Second query: follow up with full context
    print("\n--- Query 2: Follow-up with context ---")
    async for message in query(
        prompt="Now find all security issues in what you just read",
        options=ClaudeAgentOptions(resume=session_id),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

# CLAUDE.md - Project context for agents

## Project: MyApp API Server
- Framework: FastAPI
- Database: PostgreSQL with SQLAlchemy ORM
- Auth: JWT tokens with bcrypt password hashing
- Python version: 3.12

## Coding Conventions
- Use type hints on all function signatures
- Prefer async/await for I/O operations
- Tests go in tests/ directory, mirroring src/ structure
- Use pydantic models for request/response validation

## Important Notes
- Never modify migration files directly
- The auth module in src/auth/ is security-critical
- Environment variables are in .env (not committed)

# Conceptual example - Managed Agents API
# Memory stores persist data across sessions

# Agent saves a learned preference
await memory.set("user_preference_language", "python")
await memory.set("last_reviewed_file", "src/auth/middleware.py")

# In a later session, the agent retrieves it
lang = await memory.get("user_preference_language")
# Returns: "python"

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"],
            }
        },
    )

    async for message in query(
        prompt="Go to https://example.com and tell me the page title",
        options=options,
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

options = ClaudeAgentOptions(
    mcp_servers={
        "filesystem": {
            "command": "npx",
            "args": [
                "-y",
                "@modelcontextprotocol/server-filesystem",
                "/path/to/allowed/directory",
            ],
        }
    },
)

import asyncio
from claude_agent_sdk import (
    query, ClaudeAgentOptions, ResultMessage,
    tool, create_sdk_mcp_server,
)

# Custom in-process tool
@tool("analyze_sentiment", "Analyze text sentiment", {"text": str})
async def analyze_sentiment(args):
    # Simplified example
    text = args["text"].lower()
    if any(w in text for w in ["great", "love", "excellent"]):
        sentiment = "positive"
    elif any(w in text for w in ["bad", "hate", "terrible"]):
        sentiment = "negative"
    else:
        sentiment = "neutral"
    return {"content": [{"type": "text", "text": f"Sentiment: {sentiment}"}]}

custom_server = create_sdk_mcp_server(
    name="analysis", version="1.0.0", tools=[analyze_sentiment]
)

async def main():
    options = ClaudeAgentOptions(
        # Built-in tools
        allowed_tools=[
            "Read", "Glob", "Grep",
            "mcp__analysis__*",      # Custom tools
            "mcp__playwright__*",     # External MCP tools
        ],
        mcp_servers={
            # In-process custom server
            "analysis": custom_server,
            # External MCP server (Playwright)
            "playwright": {
                "command": "npx",
                "args": ["@playwright/mcp@latest"],
            },
        },
    )

    async for message in query(
        prompt="Visit example.com and analyze the sentiment of the page text",
        options=options,
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

import asyncio
from datetime import datetime
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AgentDefinition,
    tool,
    create_sdk_mcp_server,
    HookMatcher,
    SystemMessage,
    ResultMessage,
)


# ── Custom Metrics Tool ──────────────────────────────────────
@tool(
    "calculate_complexity",
    "Calculate cyclomatic complexity of a Python file",
    {"file_path": str},
)
async def calculate_complexity(args):
    """Analyze a Python file and return a complexity score."""
    try:
        with open(args["file_path"], "r") as f:
            code = f.read()

        # Count complexity indicators
        branches = sum(
            code.count(kw)
            for kw in ["if ", "elif ", "for ", "while ", "except ", "with "]
        )
        functions = code.count("def ")
        classes = code.count("class ")

        score = branches + functions
        level = (
            "low" if score < 10
            else "moderate" if score < 25
            else "high"
        )

        return {
            "content": [{
                "type": "text",
                "text": (
                    f"File: {args['file_path']}\n"
                    f"Complexity score: {score} ({level})\n"
                    f"Branches: {branches}, Functions: {functions}, "
                    f"Classes: {classes}"
                ),
            }]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True,
        }


@tool(
    "count_lines",
    "Count lines of code, comments, and blanks in a file",
    {"file_path": str},
)
async def count_lines(args):
    """Count different types of lines in a source file."""
    try:
        with open(args["file_path"], "r") as f:
            lines = f.readlines()

        total = len(lines)
        blank = sum(1 for ln in lines if ln.strip() == "")
        comments = sum(1 for ln in lines if ln.strip().startswith("#"))
        code = total - blank - comments

        return {
            "content": [{
                "type": "text",
                "text": (
                    f"File: {args['file_path']}\n"
                    f"Total: {total}, Code: {code}, "
                    f"Comments: {comments}, Blank: {blank}"
                ),
            }]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True,
        }


# Create the metrics MCP server
metrics_server = create_sdk_mcp_server(
    name="metrics",
    version="1.0.0",
    tools=[calculate_complexity, count_lines],
)


# ── Audit Hook ────────────────────────────────────────────────
async def audit_hook(input_data, tool_use_id, context):
    """Log every tool call for auditing purposes."""
    timestamp = datetime.now().isoformat()
    tool_name = input_data.get("tool_name", "unknown")
    with open("./agent_audit.log", "a") as f:
        f.write(
            f"{timestamp} | tool={tool_name} | "
            f"id={tool_use_id}\n"
        )
    return {}


# ── Main Application ─────────────────────────────────────────
async def main():
    session_id = None

    print("=" * 60)
    print("  Codebase Health Checker")
    print("=" * 60)

    # Phase 1: Initial comprehensive analysis
    print("\n[Phase 1] Running comprehensive health check...\n")

    async for message in query(
        prompt=(
            "Perform a comprehensive health check on this codebase. "
            "Use the security auditor to find vulnerabilities, "
            "the test coverage analyzer to review tests, and "
            "the metrics tools to measure complexity. "
            "Provide a detailed health report."
        ),
        options=ClaudeAgentOptions(
            allowed_tools=[
                "Read", "Glob", "Grep", "Bash", "Agent",
                "mcp__metrics__*",
            ],
            mcp_servers={"metrics": metrics_server},
            agents={
                "security-auditor": AgentDefinition(
                    description=(
                        "Finds security vulnerabilities in code."
                    ),
                    prompt=(
                        "Scan for common security issues: "
                        "SQL injection, XSS, hardcoded secrets, "
                        "insecure dependencies, auth flaws. "
                        "Rate each finding as critical/high/medium/low."
                    ),
                    tools=["Read", "Glob", "Grep"],
                ),
                "test-coverage-analyzer": AgentDefinition(
                    description=(
                        "Analyzes test coverage and suggests "
                        "improvements."
                    ),
                    prompt=(
                        "Review test files and identify gaps. "
                        "Check for missing edge cases, untested "
                        "functions, and suggest new test cases."
                    ),
                    tools=["Read", "Glob", "Grep", "Bash"],
                ),
            },
            hooks={
                "PostToolUse": [
                    HookMatcher(
                        matcher=".*",
                        hooks=[audit_hook],
                    )
                ]
            },
            permission_mode="acceptEdits",
        ),
    ):
        if isinstance(message, SystemMessage):
            if message.subtype == "init":
                session_id = message.data["session_id"]
        if isinstance(message, ResultMessage):
            print(message.result)

    # Phase 2: Follow-up with session context
    if session_id:
        print("\n" + "=" * 60)
        print("[Phase 2] Generating improvement plan...\n")

        async for message in query(
            prompt=(
                "Based on the health check you just performed, "
                "create a prioritized improvement plan. "
                "List the top 5 actions ranked by impact."
            ),
            options=ClaudeAgentOptions(resume=session_id),
        ):
            if isinstance(message, ResultMessage):
                print(message.result)

    print("\n" + "=" * 60)
    print("  Health check complete. See agent_audit.log for details.")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(main())