Build a Claude Code Custom Agent with Tool Use 2026

pip install anthropic --upgrade
export ANTHROPIC_API_KEY="sk-ant-..."

python -c "import anthropic; print(anthropic.__version__)"
# 0.25.0 or higher

# tools.py
TOOLS = [
    {
        "name": "bash",
        "description": (
            "Run a shell command and return stdout + stderr. "
            "Use for file listing, git operations, running tests, or any CLI task. "
            "Do NOT use for file reads — use read_file instead."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The shell command to execute.",
                }
            },
            "required": ["command"],
        },
    },
    {
        "name": "read_file",
        "description": "Read the full contents of a file by path. Returns the text as a string.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Absolute or relative path to the file.",
                }
            },
            "required": ["path"],
        },
    },
    {
        "name": "write_file",
        "description": "Write or overwrite a file with the given content.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "Path where the file will be written.",
                },
                "content": {
                    "type": "string",
                    "description": "Full file content to write.",
                },
            },
            "required": ["path", "content"],
        },
    },
]

# executor.py
import subprocess
from pathlib import Path


def execute_tool(name: str, tool_input: dict) -> str:
    """Run the tool Claude requested and return a string result."""

    if name == "bash":
        return _run_bash(tool_input["command"])

    if name == "read_file":
        return _read_file(tool_input["path"])

    if name == "write_file":
        return _write_file(tool_input["path"], tool_input["content"])

    return f"ERROR: Unknown tool '{name}'"


def _run_bash(command: str) -> str:
    # timeout=30 prevents infinite loops from locking up your agent
    result = subprocess.run(
        command,
        shell=True,
        capture_output=True,
        text=True,
        timeout=30,
    )
    output = result.stdout
    if result.stderr:
        output += f"\n[stderr]\n{result.stderr}"
    if result.returncode != 0:
        output += f"\n[exit code: {result.returncode}]"
    return output or "(no output)"


def _read_file(path: str) -> str:
    try:
        return Path(path).read_text(encoding="utf-8")
    except FileNotFoundError:
        return f"ERROR: File not found: {path}"
    except Exception as e:
        return f"ERROR: {e}"


def _write_file(path: str, content: str) -> str:
    try:
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        return f"OK: wrote {len(content)} chars to {path}"
    except Exception as e:
        return f"ERROR: {e}"

# agent.py
import anthropic
from tools import TOOLS
from executor import execute_tool

client = anthropic.Anthropic()

MAX_TURNS = 20  # Hard stop — prevents runaway billing on stuck loops


def run_agent(user_prompt: str, system: str = "") -> str:
    """
    Run the full agentic loop for a single task.
    Returns Claude's final text response.
    """
    messages = [{"role": "user", "content": user_prompt}]

    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=system,
            tools=TOOLS,
            messages=messages,
        )

        print(f"[turn {turn + 1}] stop_reason={response.stop_reason}")

        # Append Claude's full response to history
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            # No tool calls — Claude is done. Extract and return the text.
            return _extract_text(response.content)

        if response.stop_reason == "tool_use":
            # Process every tool call in this response
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  → calling tool: {block.name}({block.input})")
                    result = execute_tool(block.name, block.input)
                    print(f"  ← result: {result[:120]}...")

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

            # Feed all results back in a single user turn
            messages.append({"role": "user", "content": tool_results})

        else:
            # max_tokens or other stop — return whatever text we have
            return _extract_text(response.content)

    return "ERROR: MAX_TURNS reached without a final response."


def _extract_text(content: list) -> str:
    return "\n".join(
        block.text for block in content if hasattr(block, "text")
    )

# main.py
from agent import run_agent

SYSTEM = """
You are a precise coding agent. When asked to make changes:
1. Read the relevant files first.
2. Write the updated files.
3. Run tests to verify.
4. Report what you changed and the test result.
Never skip reading before writing.
"""

task = """
Read src/utils.py. Add a function `clamp(value, min_val, max_val)`
that clamps a number between min and max. Write the updated file.
Then run: python -m pytest tests/test_utils.py -v
"""

result = run_agent(task, system=SYSTEM)
print("\n=== Agent Result ===")
print(result)

import re

BLOCKED_PATTERNS = [
    r"\brm\s+-rf\b",       # Recursive delete
    r"\bsudo\b",           # Privilege escalation
    r"\bcurl\b.*\|\s*sh",  # Pipe-to-shell downloads
    r"\bdd\b.*of=/dev/",   # Disk overwrite
]


def _run_bash(command: str) -> str:
    for pattern in BLOCKED_PATTERNS:
        if re.search(pattern, command):
            return f"BLOCKED: command matches unsafe pattern '{pattern}'"

    result = subprocess.run(
        command, shell=True, capture_output=True, text=True, timeout=30
    )
    # ... rest of implementation

# smoke_test.py
from agent import run_agent

result = run_agent("Create a file /tmp/hello.txt with content 'agent works'. Then read it back and confirm.")
assert "agent works" in result, f"Unexpected result: {result}"
print("✅ Smoke test passed:", result)