Build Your Own Agent Loop

#!/usr/bin/env python3
"""A minimal agentic coder. 150 lines. Under 200 if you're strict."""
import os, sys, subprocess, json
from anthropic import Anthropic

client = Anthropic()
MODEL  = "claude-sonnet-4-6"
ROOT   = os.path.realpath(".")

# ---- Tool definitions ----
TOOLS = [
    {
        "name": "bash",
        "description": "Run a bash command in the project root. Use for tests, git, installs.",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
    {
        "name": "read_file",
        "description": "Read a file. Returns content.",
        "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]},
    },
    {
        "name": "edit_file",
        "description": "Replace old_str with new_str in path. Fails if old_str not unique.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "old_str": {"type": "string"},
                "new_str": {"type": "string"},
            },
            "required": ["path", "old_str", "new_str"],
        },
    },
    {
        "name": "search",
        "description": "Search file contents with ripgrep. Returns matches.",
        "input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]},
    },
    {
        "name": "done",
        "description": "Signal that the task is complete. Include a summary of what was accomplished.",
        "input_schema": {"type": "object", "properties": {"summary": {"type": "string"}}, "required": ["summary"]},
    },
]

# ---- Tool executors ----
DENY = ["rm -rf /", "sudo", "curl | sh"]

def exec_bash(command: str) -> str:
    if any(d in command for d in DENY):
        return f"DENIED: {command}"
    print(f"$ {command}")
    try:
        p = subprocess.run(command, shell=True, cwd=ROOT, capture_output=True, text=True, timeout=120)
        out = (p.stdout + p.stderr)[-20000:]
        return f"exit={p.returncode}\n{out}"
    except subprocess.TimeoutExpired:
        return "TIMEOUT"

def exec_read(path: str) -> str:
    full = os.path.realpath(os.path.join(ROOT, path))
    if not full.startswith(ROOT): return "ERROR: path escape"
    try:
        with open(full) as f: return f.read()[:30000]
    except Exception as e: return f"ERROR: {e}"

def exec_edit(path: str, old_str: str, new_str: str) -> str:
    full = os.path.realpath(os.path.join(ROOT, path))
    if not full.startswith(ROOT): return "ERROR: path escape"
    with open(full) as f: content = f.read()
    if content.count(old_str) != 1:
        return f"ERROR: old_str appears {content.count(old_str)} times, must be exactly 1"
    with open(full, "w") as f: f.write(content.replace(old_str, new_str))
    return f"Edited {path}"

def exec_search(pattern: str) -> str:
    p = subprocess.run(["rg", "-n", pattern, ROOT], capture_output=True, text=True)
    return "\n".join(p.stdout.split("\n")[:50])

EXECUTORS = {"bash": exec_bash, "read_file": exec_read, "edit_file": exec_edit, "search": exec_search}

# ---- Agent loop ----
def run_agent(task: str, max_iters: int = 20, max_tokens: int = 100_000) -> dict:
    messages = [{"role": "user", "content": task}]
    tokens = 0

    for i in range(max_iters):
        resp = client.messages.create(
            model=MODEL,
            max_tokens=4096,
            tools=TOOLS,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": resp.content})
        tokens += resp.usage.input_tokens + resp.usage.output_tokens

        # Stop 1: explicit done
        done_blocks = [b for b in resp.content if getattr(b, "type", None) == "tool_use" and b.name == "done"]
        if done_blocks:
            return {"status": "complete", "iters": i + 1, "tokens": tokens, "summary": done_blocks[0].input.get("summary", "")}

        # Stop 3: budget (stop 2 success check omitted — pass test cmd via task instead)
        if tokens >= max_tokens:
            return {"status": "budget", "iters": i + 1, "tokens": tokens}

        # Execute tool calls
        tool_results = []
        for b in resp.content:
            if getattr(b, "type", None) == "tool_use":
                fn = EXECUTORS.get(b.name)
                result = fn(**b.input) if fn else f"unknown tool: {b.name}"
                tool_results.append({"type": "tool_result", "tool_use_id": b.id, "content": result})
        if tool_results:
            messages.append({"role": "user", "content": tool_results})

    # Stop 4: max iters
    return {"status": "max_iters", "iters": max_iters, "tokens": tokens}

# ---- Main ----
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: agent.py '<task>'")
        sys.exit(1)
    task = " ".join(sys.argv[1:])
    result = run_agent(task)
    print(json.dumps(result, indent=2))

export ANTHROPIC_API_KEY="sk-ant-..."

# Create a worktree to work in, then cd there
cd ~/code/myproject
git worktree add ../myproject-agent-test exp/agent-test
cd ../myproject-agent-test

# Run the agent
python agent.py "Add a docstring to every function in src/utils.py"

# Inspect what it did
git diff