{project-root}/
├── .vscode/
│   └── launch.json                 # Debug config
├── .gitignore
├── requirements.txt
├── {agent-location}/               # Adaptive (see above)
│   ├── .claude/
│   │   └── settings.local.json
│   ├── src/
│   │   ├── __init__.py
│   │   ├── agent.py
│   │   ├── schemas.py
│   │   └── exceptions.py
│   ├── logs/                       # Execution logs
│   ├── output/                     # Message history (JSONL)
│   ├── task.md
│   └── claude_agent.py

claude-agent-sdk
pydantic
python-dotenv
colorlog

# Required: Your Anthropic API key
ANTHROPIC_API_KEY=your-api-key-here

# Optional: Agent configuration
# AGENT_MODEL=claude-sonnet-4-5
# AGENT_MAX_TURNS=50
# AGENT_TIMEOUT_SECONDS=1800
# AGENT_MAX_BUDGET_USD=10.0
# AGENT_PERMISSION_MODE=bypassPermissions  # default: bypassPermissions (allows all tools)

# Agent logs and output
logs/
output/
*.jsonl

# Environment and secrets
.env
.env.local

# Python
__pycache__/
*.pyc
*.pyo
.venv/
venv/
*.egg-info/

# IDE
.idea/
*.swp
*.swo

{
  "permissions": {
    "allow": [
      "Bash(curl:*)",
      "Bash(wget:*)",
      "Bash(git:*)",
      "Bash(find:*)",
      "Bash(grep:*)",
      "Bash(ls:*)",
      "Bash(cat:*)",
      "Bash(python:*)",
      "Bash(jq:*)",
      "Bash(date:*)"
    ],
    "deny": [
      "Bash(rm -rf:*)",
      "Bash(sudo:*)",
      "Bash(chmod 777:*)",
      "Bash(> /dev:*)"
    ]
  }
}

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "Run {agent-name}",
      "type": "debugpy",
      "request": "launch",
      "python": "${command:python.interpreterPath}",
      "program": "${workspaceFolder}/{agent-path}/claude_agent.py",
      "args": [],
      "console": "integratedTerminal",
      "justMyCode": true,
      "env": {
        "PYTHONPATH": "${workspaceFolder}/{agent-path}",
        "PYTHONUNBUFFERED": "1",
      },
      "cwd": "${workspaceFolder}/{agent-path}",
      "envFile": "${workspaceFolder}/.env"
    }
  ]
}

# Agent Task

## Objective
Describe what the agent should accomplish.

## Instructions
1. Step one
2. Step two
3. Step three

## Expected Output
Describe what output.json should contain.

"""Agent source package."""

"""Agent exceptions."""


class AgentError(Exception):
    """Base exception."""
    pass


class ConfigurationError(AgentError):
    """Configuration error."""
    pass


class SessionIncompleteError(AgentError):
    """Session did not complete."""
    pass


class MaxTurnsExceededError(AgentError):
    """Exceeded turn limit."""
    pass


class ServerError(AgentError):
    """API error."""
    pass

"""Data models."""
from enum import StrEnum
from typing import Any, Optional

from pydantic import BaseModel, Field


class TaskStatus(StrEnum):
    SUCCESS = "SUCCESS"
    PARTIAL = "PARTIAL"
    FAILED = "FAILED"
    ERROR = "ERROR"


class LLMMetrics(BaseModel):
    """Session metrics."""
    model: str
    duration_ms: int
    num_turns: int
    session_id: str
    total_cost_usd: Optional[float] = None


class AgentResult(BaseModel):
    """Agent execution result."""
    status: TaskStatus
    summary: str
    details: Optional[str] = None
    data: dict[str, Any] = Field(default_factory=dict)
    metrics: Optional[LLMMetrics] = None
    error: Optional[str] = None
    message_history: list[dict[str, Any]] = Field(default_factory=list)

"""Agent execution logic."""
import asyncio
import logging
import os
import time
from typing import Any, Optional

from claude_agent_sdk import query
from claude_agent_sdk.types import (
    AssistantMessage,
    ClaudeAgentOptions,
    ResultMessage,
    SystemMessage,
    UserMessage,
)

from .exceptions import (
    AgentError,
    MaxTurnsExceededError,
    ServerError,
    SessionIncompleteError,
)
from .schemas import AgentResult, LLMMetrics, TaskStatus

logger = logging.getLogger(__name__)

# ANSI color codes for verbose output
BLUE = "\033[94m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
MAGENTA = "\033[95m"
RESET = "\033[0m"
BOLD = "\033[1m"


def serialize_block(block: Any) -> dict[str, Any]:
    """Convert SDK message block to serializable dict."""
    block_type = getattr(block, 'type', None)
    if block_type == 'tool_use':
        return {
            "type": "tool_use",
            "name": getattr(block, 'name', ''),
            "input": getattr(block, 'input', {}),
        }
    elif block_type == 'tool_result':
        content = getattr(block, 'content', '')
        return {
            "type": "tool_result",
            "is_error": getattr(block, 'is_error', False),
            "content": str(content)[:500] if content else "",
        }
    elif block_type == 'text':
        return {"type": "text", "text": getattr(block, 'text', '')}
    return {"type": str(block_type), "raw": str(block)[:200]}


def print_verbose(prefix: str, content: str, color: str = RESET) -> None:
    """Print verbose output with colored prefix."""
    print(f"{color}{BOLD}[{prefix}]{RESET} {content}", flush=True)


async def run_agent_with_timeout(
    task: str,
    working_dir: str = ".",
    additional_dirs: Optional[list[str]] = None,
    verbose: bool = True,
) -> AgentResult:
    """Run the agent with configuration from environment."""
    max_turns = int(os.getenv("AGENT_MAX_TURNS", "50"))
    timeout_seconds = int(os.getenv("AGENT_TIMEOUT_SECONDS", "1800"))
    max_budget = float(os.getenv("AGENT_MAX_BUDGET_USD", "10.0"))
    model = os.getenv("AGENT_MODEL", "claude-sonnet-4-5")
    permission_mode = os.getenv("AGENT_PERMISSION_MODE", "bypassPermissions")

    logger.info(f"Starting agent with model={model}, max_turns={max_turns}")

    # add_dirs grants the agent read/write access to directories outside cwd.
    # Use this when the agent needs to access external resources, configs, or data.
    options = ClaudeAgentOptions(
        model=model,
        cwd=working_dir,
        permission_mode=permission_mode,
        max_turns=max_turns,
        max_budget_usd=max_budget,
        add_dirs=additional_dirs or [],  # Additional allowed directories
    )

    try:
        result_text = ""
        session_id = ""
        num_turns = 0
        total_cost = None
        message_history: list[dict[str, Any]] = []

        async def run_query():
            nonlocal result_text, session_id, num_turns, total_cost
            turn_count = 0

            async for message in query(prompt=task, options=options):
                if isinstance(message, AssistantMessage):
                    turn_count += 1
                    content = getattr(message, 'message', None)
                    serialized_blocks: list[dict[str, Any]] = []
                    if content:
                        msg_content = getattr(content, 'content', None)
                        if msg_content:
                            for block in msg_content:
                                serialized_blocks.append(serialize_block(block))
                                if verbose:
                                    block_type = getattr(block, 'type', None)
                                    if block_type == 'text':
                                        text = getattr(block, 'text', '')
                                        if text:
                                            preview = text[:500] + ("..." if len(text) > 500 else "")
                                            print_verbose("ASSISTANT", preview, GREEN)
                                    elif block_type == 'tool_use':
                                        tool_name = getattr(block, 'name', 'unknown')
                                        tool_input = getattr(block, 'input', {})
                                        print_verbose("TOOL", f"{tool_name}", YELLOW)
                                        if tool_name == 'Bash' and 'command' in tool_input:
                                            print_verbose("CMD", tool_input['command'][:200], MAGENTA)
                    message_history.append({
                        "turn": turn_count,
                        "role": "assistant",
                        "timestamp": time.time(),
                        "content": serialized_blocks,
                    })
                    if verbose:
                        print_verbose(f"TURN {turn_count}", "", CYAN)

                elif isinstance(message, UserMessage):
                    content = getattr(message, 'message', None)
                    if content:
                        msg_content = getattr(content, 'content', None)
                        if msg_content and verbose:
                            for block in msg_content:
                                block_type = getattr(block, 'type', None)
                                if block_type == 'tool_result':
                                    result_content = getattr(block, 'content', '')
                                    if isinstance(result_content, str) and result_content:
                                        preview = result_content[:300] + ("..." if len(result_content) > 300 else "")
                                        print_verbose("RESULT", preview, BLUE)

                elif isinstance(message, SystemMessage):
                    if verbose:
                        print_verbose("SYSTEM", str(message)[:200], CYAN)

                if isinstance(message, ResultMessage):
                    result_text = getattr(message, 'result', str(message))
                    session_id = getattr(message, 'session_id', '')
                    num_turns = getattr(message, 'num_turns', turn_count)
                    total_cost = getattr(message, 'total_cost_usd', None)
                    if verbose:
                        cost_str = f"${total_cost:.4f}" if total_cost else "N/A"
                        print_verbose("COMPLETE", f"Turns: {num_turns}, Cost: {cost_str}", GREEN)

        await asyncio.wait_for(run_query(), timeout=timeout_seconds)

        metrics = LLMMetrics(
            model=model,
            duration_ms=0,
            num_turns=num_turns,
            session_id=session_id,
            total_cost_usd=total_cost,
        )

        return AgentResult(
            status=TaskStatus.SUCCESS,
            summary="Task completed successfully",
            details=result_text,
            metrics=metrics,
            message_history=message_history,
        )

    except asyncio.TimeoutError:
        raise AgentError(f"Agent timed out after {timeout_seconds} seconds")
    except Exception as e:
        error_msg = str(e)
        if "max_turns" in error_msg.lower():
            raise MaxTurnsExceededError(f"Exceeded {max_turns} turns")
        if "server" in error_msg.lower() or "api" in error_msg.lower():
            raise ServerError(f"API error: {error_msg}")
        if "incomplete" in error_msg.lower():
            raise SessionIncompleteError(f"Session incomplete: {error_msg}")
        raise AgentError(f"Agent execution failed: {error_msg}")

#!/usr/bin/env python3
"""
Claude Agent entry point.

Usage:
    python claude_agent.py                    # Run with default task.md
    python claude_agent.py --task-file x.md   # Custom task file
    python claude_agent.py --output out.json  # Save output to file
    python claude_agent.py -v                 # Verbose output

Prerequisites:
    pip install -r requirements.txt
    Set ANTHROPIC_API_KEY in .env file
"""
import argparse
import asyncio
import json
import logging
import sys
from datetime import datetime
from pathlib import Path

import colorlog
from dotenv import load_dotenv

from src.agent import run_agent_with_timeout
from src.schemas import AgentResult, TaskStatus

# Load environment variables from .env file
load_dotenv()


def setup_logging(logs_dir: Path) -> logging.Logger:
    """Configure logging with console and file handlers."""
    logs_dir.mkdir(parents=True, exist_ok=True)

    # Console handler with colors
    console_handler = colorlog.StreamHandler()
    console_handler.setFormatter(colorlog.ColoredFormatter(
        "%(log_color)s%(asctime)s [%(levelname)s]%(reset)s %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        log_colors={
            "DEBUG": "cyan",
            "INFO": "green",
            "WARNING": "yellow",
            "ERROR": "red",
            "CRITICAL": "red,bg_white",
        },
    ))

    # File handler for persistent logs
    log_file = logs_dir / f"agent_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
    file_handler = logging.FileHandler(log_file, encoding="utf-8")
    file_handler.setFormatter(logging.Formatter(
        "%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    ))

    logger = colorlog.getLogger(__name__)
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    logger.setLevel(logging.INFO)
    return logger


def load_task_file(task_file: Path) -> str:
    if not task_file.exists():
        raise FileNotFoundError(f"Task file not found: {task_file}")
    return task_file.read_text(encoding="utf-8")


def save_message_history(
    result: AgentResult,
    output_dir: Path,
    session_id: str,
    logger: logging.Logger,
) -> None:
    """Save message history to JSONL file."""
    if not result.message_history:
        return
    output_dir.mkdir(parents=True, exist_ok=True)
    output_file = output_dir / f"session_{session_id}.jsonl"
    with output_file.open("w", encoding="utf-8") as f:
        for msg in result.message_history:
            f.write(json.dumps(msg) + "\n")
    logger.info(f"Message history saved to: {output_file}")


def main() -> int:
    base_dir = Path(__file__).parent
    logs_dir = base_dir / "logs"
    output_dir = base_dir / "output"
    logger = setup_logging(logs_dir)

    parser = argparse.ArgumentParser(description="Execute Claude agent task")
    parser.add_argument(
        "--dir",
        default=".",
        help="Working directory (default: current)",
    )
    parser.add_argument(
        "--task-file",
        type=Path,
        default=Path("task.md"),
        help="Task file (default: task.md)",
    )
    parser.add_argument(
        "--output",
        help="Output file path (default: stdout)",
    )
    parser.add_argument(
        "--add-dir",
        action="append",
        default=[],
        help="Additional directories",
    )
    parser.add_argument(
        "-v", "--verbose",
        action="store_true",
        help="Enable verbose output",
    )
    args = parser.parse_args()

    if args.verbose:
        logger.setLevel(logging.DEBUG)

    try:
        task: str = load_task_file(args.task_file)
    except FileNotFoundError as e:
        logger.error(str(e))
        return 1

    logger.info(f"Task file: {args.task_file}")
    logger.info(f"Working directory: {args.dir}")

    try:
        result = asyncio.run(
            run_agent_with_timeout(
                task=task,
                working_dir=args.dir,
                additional_dirs=args.add_dir,
                verbose=args.verbose,
            )
        )

        # Save message history to JSONL
        session_id = (
            result.metrics.session_id
            if result.metrics
            else datetime.now().strftime("%Y%m%d_%H%M%S")
        )
        save_message_history(result, output_dir, session_id, logger)

        # Output result JSON (excluding message_history to avoid duplication)
        output_json: str = result.model_dump_json(
            indent=2, exclude={"message_history"}
        )

        if args.output:
            Path(args.output).write_text(output_json, encoding="utf-8")
            logger.info(f"Result saved to: {args.output}")
        else:
            print(output_json)

        if result.metrics:
            logger.info(
                f"Turns: {result.metrics.num_turns}, "
                f"Cost: ${result.metrics.total_cost_usd:.4f}"
                if result.metrics.total_cost_usd
                else f"Turns: {result.metrics.num_turns}"
            )

        return 0 if result.status == TaskStatus.SUCCESS else 1

    except Exception as e:
        logger.error(f"Agent failed: {e}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

✓ Agent "{agent-name}" created at {agent-path}/
✓ Virtual environment created and dependencies installed

To run:
  # Copy .env.example to .env and set your API key
  cp .env.example .env
  # Edit .env and add your ANTHROPIC_API_KEY
  
  # Activate virtual environment
  source .venv/bin/activate      # Unix/macOS
  # .\.venv\Scripts\Activate.ps1  # Windows PowerShell
  
  # Run the agent
  python {agent-path}/claude_agent.py

Or press F5 in VS Code/Cursor (select .venv interpreter first).

# Copy example and edit
cp .env.example .env

# Add your key to .env
ANTHROPIC_API_KEY=sk-ant-api03-xxxxx...

curl -s https://api.anthropic.com/v1/messages \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "Content-Type: application/json" \
  -d '{"model":"claude-sonnet-4-5-20250929","max_tokens":10,"messages":[{"role":"user","content":"Hi"}]}'