Real-World Scenarios

import anyio
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def automated_code_review(project_path: Path):
    """Perform automated code review and generate report."""
    
    # Define structured output schema
    review_schema = {
        "type": "json_schema",
        "json_schema": {
            "name": "code_review",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "files_reviewed": {"type": "integer"},
                    "issues": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "severity": {"type": "string", "enum": ["critical", "major", "minor"]},
                                "file": {"type": "string"},
                                "line": {"type": "integer"},
                                "description": {"type": "string"}
                            },
                            "required": ["severity", "file", "description"],
                            "additionalProperties": False
                        }
                    },
                    "score": {"type": "integer", "minimum": 0, "maximum": 100}
                },
                "required": ["files_reviewed", "issues", "score"],
                "additionalProperties": False
            }
        }
    }
    
    options = ClaudeAgentOptions(
        cwd=str(project_path),
        allowed_tools=["Read", "Grep", "Glob"],
        output_format=review_schema,
        max_budget_usd=2.0
    )
    
    prompt = """
    Review all Python files in the src/ directory.
    Focus on:
    - Security vulnerabilities
    - Code quality issues
    - Performance problems
    - Best practice violations
    """
    
    async for message in query(prompt=prompt, options=options):
        if isinstance(message, ResultMessage) and message.structured_output:
            review = message.structured_output
            
            # Generate report
            print(f"\n{'='*60}")
            print(f"CODE REVIEW REPORT")
            print(f"{'='*60}")
            print(f"Files Reviewed: {review['files_reviewed']}")
            print(f"Overall Score: {review['score']}/100")
            print(f"Issues Found: {len(review['issues'])}\n")
            
            for issue in review["issues"]:
                print(f"[{issue['severity'].upper()}] {issue['file']}")
                if 'line' in issue:
                    print(f"  Line {issue['line']}")
                print(f"  {issue['description']}\n")
            
            return review

anyio.run(lambda: automated_code_review(Path("/path/to/project")))

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def generate_documentation():
    """Generate comprehensive project documentation."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Glob", "Grep"],
        permission_mode="acceptEdits",
        system_prompt="You are a technical documentation expert"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Generate API documentation
        print("Generating API documentation...")
        await client.query("""
        Analyze all Python files in the src/ directory and create
        comprehensive API documentation in docs/api.md.
        Include:
        - Module descriptions
        - Class documentation
        - Function signatures
        - Parameter descriptions
        - Return values
        - Usage examples
        """)
        async for msg in client.receive_response():
            pass
        
        # Generate README
        print("\nGenerating README...")
        await client.query("""
        Create a comprehensive README.md with:
        - Project overview
        - Installation instructions
        - Quick start guide
        - Usage examples
        - Contributing guidelines
        """)
        async for msg in client.receive_response():
            pass
        
        # Generate architecture docs
        print("\nGenerating architecture documentation...")
        await client.query("""
        Create docs/architecture.md documenting:
        - System architecture
        - Component interactions
        - Data flow
        - Design decisions
        """)
        async for msg in client.receive_response():
            pass
        
        print("\nDocumentation generation complete!")

anyio.run(generate_documentation)

import anyio
import sys
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def ci_cd_check():
    """Run CI/CD checks with Claude."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Bash", "Grep"],
        permission_mode="bypassPermissions",  # Non-interactive for CI
        max_budget_usd=1.0,
        model="haiku",  # Fast and cheap for CI
        setting_sources=["project"]  # Reproducible
    )
    
    checks_passed = True
    
    # Test execution
    print("Running tests...")
    async for msg in query("Run all tests with pytest and report results", options=options):
        if isinstance(msg, ResultMessage):
            if msg.is_error:
                print("❌ Tests failed")
                checks_passed = False
            else:
                print("✅ Tests passed")
    
    # Linting
    print("\nRunning linters...")
    async for msg in query("Run pylint and flake8 on all Python files", options=options):
        if isinstance(msg, ResultMessage):
            if msg.is_error:
                print("❌ Linting failed")
                checks_passed = False
            else:
                print("✅ Linting passed")
    
    # Security scan
    print("\nRunning security scan...")
    async for msg in query("Scan for security vulnerabilities using bandit", options=options):
        if isinstance(msg, ResultMessage):
            if msg.is_error:
                print("❌ Security issues found")
                checks_passed = False
            else:
                print("✅ Security scan passed")
    
    # Exit with appropriate code
    sys.exit(0 if checks_passed else 1)

anyio.run(ci_cd_check)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def database_migration_assistant():
    """Interactive assistant for database migrations."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="default",  # Require confirmation for changes
        enable_file_checkpointing=True  # Allow rollback
    )
    
    checkpoints = []
    
    async with ClaudeSDKClient(options=options) as client:
        # Analyze current schema
        print("Step 1: Analyzing current database schema...")
        await client.query("""
        Analyze the current database schema from models.py.
        Identify the tables, columns, and relationships.
        """)
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("analyze", msg.uuid))
        
        # Generate migration
        print("\nStep 2: Generating migration...")
        await client.query("""
        Generate a migration script to:
        1. Add new 'status' column to users table
        2. Add index on email column
        3. Create new audit_log table
        
        Save the migration to migrations/001_add_status_and_audit.py
        """)
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints.append(("generate", msg.uuid))
        
        # Review migration
        print("\nStep 3: Review the migration...")
        await client.query("""
        Review the migration script and verify:
        - No data loss
        - Proper indexes
        - Backward compatibility
        """)
        async for msg in client.receive_response():
            pass
        
        # Run migration
        print("\nStep 4: Ready to run migration...")
        user_approval = input("Apply migration? (yes/no): ")
        
        if user_approval.lower() == "yes":
            await client.query("Run the migration script")
            async for msg in client.receive_response():
                if isinstance(msg, ResultMessage) and msg.is_error:
                    print("\n❌ Migration failed! Rolling back...")
                    if checkpoints:
                        await client.rewind_files(checkpoints[-1][1])
                    print("Files restored to pre-migration state")
                else:
                    print("\n✅ Migration completed successfully")
        else:
            print("Migration cancelled")

anyio.run(database_migration_assistant)

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, UserMessage

async def refactoring_assistant(target_module: str):
    """Assist with large-scale refactoring."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "MultiEdit", "Bash", "Grep"],
        permission_mode="acceptEdits",
        enable_file_checkpointing=True,
        sandbox={"enabled": True}  # Sandbox bash commands
    )
    
    checkpoints = {}
    
    async with ClaudeSDKClient(options=options) as client:
        # Phase 1: Analysis
        print(f"Analyzing {target_module}...")
        await client.query(f"""
        Analyze the {target_module} module for refactoring opportunities:
        - Code duplication
        - Long functions/methods
        - Complex conditionals
        - Naming inconsistencies
        """)
        async for msg in client.receive_response():
            if isinstance(msg, UserMessage) and msg.uuid:
                checkpoints["analysis"] = msg.uuid
        
        # Phase 2: Plan
        print("\nCreating refactoring plan...")
        await client.set_permission_mode("plan")
        await client.query(f"""
        Create a detailed refactoring plan for {target_module} with:
        1. List of changes
        2. Order of operations
        3. Risk assessment
        4. Testing strategy
        """)
        async for msg in client.receive_response():
            pass
        
        # Phase 3: Execute incrementally
        print("\nExecuting refactoring...")
        await client.set_permission_mode("acceptEdits")
        
        refactoring_steps = [
            "Extract duplicate code into helper functions",
            "Break down long functions",
            "Simplify complex conditionals",
            "Improve variable naming",
            "Add type hints"
        ]
        
        for i, step in enumerate(refactoring_steps, 1):
            print(f"\nStep {i}/{len(refactoring_steps)}: {step}")
            await client.query(f"Apply refactoring: {step}")
            
            async for msg in client.receive_response():
                if isinstance(msg, UserMessage) and msg.uuid:
                    checkpoints[f"step_{i}"] = msg.uuid
            
            # Run tests after each step
            print(f"Running tests after step {i}...")
            await client.query("Run tests for this module")
            
            test_passed = True
            async for msg in client.receive_response():
                if isinstance(msg, ResultMessage) and msg.is_error:
                    test_passed = False
            
            if not test_passed:
                print(f"❌ Tests failed after step {i}! Rolling back...")
                prev_checkpoint = checkpoints.get(f"step_{i-1}" if i > 1 else "analysis")
                if prev_checkpoint:
                    await client.rewind_files(prev_checkpoint)
                print("Rolled back to previous checkpoint")
                break
            else:
                print(f"✅ Step {i} completed successfully")
        
        print("\nRefactoring complete!")

anyio.run(lambda: refactoring_assistant("authentication"))

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def interactive_debugger():
    """Interactive debugging assistant."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Bash", "Grep", "Write"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        print("Interactive Debugging Assistant")
        print("Describe your issue or type 'exit' to quit\n")
        
        while True:
            issue = input("Issue: ").strip()
            if issue.lower() == 'exit':
                break
            
            # Investigate issue
            await client.query(f"""
            Help me debug this issue: {issue}
            
            Steps:
            1. Examine relevant code
            2. Check logs if available
            3. Identify potential causes
            4. Suggest fixes
            """)
            
            print("\nClaude's analysis:")
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(block.text)
            
            # Apply fix if user wants
            apply_fix = input("\nApply suggested fix? (y/n): ")
            if apply_fix.lower() == 'y':
                await client.query("Apply the fix we discussed")
                async for msg in client.receive_response():
                    pass
                print("Fix applied. Test the changes.")
            
            print("\n" + "="*60 + "\n")

anyio.run(interactive_debugger)

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def generate_tests(source_file: str, test_file: str):
    """Generate comprehensive tests for a module."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write"],
        permission_mode="acceptEdits",
        system_prompt="""You are a testing expert. Generate comprehensive tests including:
        - Unit tests for all functions
        - Edge cases and error conditions
        - Integration tests where applicable
        - Fixtures and mocks
        - Clear test documentation
        """
    )
    
    prompt = f"""
    Generate comprehensive tests for {source_file} and save them to {test_file}.
    
    Include:
    - Test for all public functions and methods
    - Edge cases (empty inputs, None, boundaries)
    - Error conditions (exceptions, invalid inputs)
    - Integration tests for complex workflows
    - Appropriate fixtures and mocks
    - Docstrings explaining each test
    
    Use pytest framework.
    """
    
    async for msg in query(prompt=prompt, options=options):
        print(msg)

anyio.run(lambda: generate_tests("src/auth.py", "tests/test_auth.py"))

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

async def performance_optimizer():
    """Analyze and optimize code performance."""
    
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash", "Grep"],
        permission_mode="acceptEdits"
    )
    
    async with ClaudeSDKClient(options=options) as client:
        # Profile current performance
        print("Step 1: Profiling current performance...")
        await client.query("""
        Profile the application to identify performance bottlenecks:
        1. Run with cProfile
        2. Analyze hotspots
        3. Identify slow functions
        4. Check for N+1 queries
        """)
        async for msg in client.receive_response():
            pass
        
        # Optimize bottlenecks
        print("\nStep 2: Optimizing bottlenecks...")
        await client.query("""
        Optimize the identified performance issues:
        - Add caching where appropriate
        - Optimize database queries
        - Use more efficient algorithms
        - Add indexes
        - Implement lazy loading
        """)
        async for msg in client.receive_response():
            pass
        
        # Verify improvements
        print("\nStep 3: Verifying improvements...")
        await client.query("""
        Run performance tests again and compare:
        - Before/after execution times
        - Memory usage
        - Database query counts
        Generate a performance improvement report
        """)
        async for msg in client.receive_response():
            pass

anyio.run(performance_optimizer)