#!/usr/bin/env python3 """ s04_subagent.py - Subagents Spawn a child agent with fresh messages=[]. The child works in its own context, sharing the filesystem, then returns only a summary to the parent. Parent agent Subagent +------------------+ +------------------+ | messages=[...] | | messages=[] | <-- fresh | | dispatch | | | tool: task | ---------->| while tool_use: | | prompt="..." | | call tools | | description="" | | append results | | | summary | | | result = "..." | <--------- | return last text | +------------------+ +------------------+ | Parent context stays clean. Subagent context is discarded. Key insight: "Process isolation gives context isolation for free." """ import os import subprocess from pathlib import Path from anthropic import Anthropic from dotenv import load_dotenv load_dotenv(override=True) if os.getenv("ANTHROPIC_BASE_URL"): os.environ.pop("ANTHROPIC_AUTH_TOKEN", None) WORKDIR = Path.cwd() client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL")) MODEL = os.environ["MODEL_ID"] SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks." SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings." # -- Tool implementations shared by parent and child -- def safe_path(p: str) -> Path: path = (WORKDIR / p).resolve() if not path.is_relative_to(WORKDIR): raise ValueError(f"Path escapes workspace: {p}") return path def run_bash(command: str) -> str: dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"] if any(d in command for d in dangerous): return "Error: Dangerous command blocked" try: r = subprocess.run(command, shell=True, cwd=WORKDIR, capture_output=True, text=True, timeout=120) out = (r.stdout + r.stderr).strip() return out[:50000] if out else "(no output)" except subprocess.TimeoutExpired: return "Error: Timeout (120s)" def run_read(path: str, limit: int = None) -> str: try: lines = safe_path(path).read_text().splitlines() if limit and limit < len(lines): lines = lines[:limit] + [f"... ({len(lines) - limit} more)"] return "\n".join(lines)[:50000] except Exception as e: return f"Error: {e}" def run_write(path: str, content: str) -> str: try: fp = safe_path(path) fp.parent.mkdir(parents=True, exist_ok=True) fp.write_text(content) return f"Wrote {len(content)} bytes" except Exception as e: return f"Error: {e}" def run_edit(path: str, old_text: str, new_text: str) -> str: try: fp = safe_path(path) content = fp.read_text() if old_text not in content: return f"Error: Text not found in {path}" fp.write_text(content.replace(old_text, new_text, 1)) return f"Edited {path}" except Exception as e: return f"Error: {e}" TOOL_HANDLERS = { "bash": lambda **kw: run_bash(kw["command"]), "read_file": lambda **kw: run_read(kw["path"], kw.get("limit")), "write_file": lambda **kw: run_write(kw["path"], kw["content"]), "edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]), } # Child gets all base tools except task (no recursive spawning) CHILD_TOOLS = [ {"name": "bash", "description": "Run a shell command.", "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}}, {"name": "read_file", "description": "Read file contents.", "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}}, {"name": "write_file", "description": "Write content to file.", "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}, {"name": "edit_file", "description": "Replace exact text in file.", "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}}, ] # -- Subagent: fresh context, filtered tools, summary-only return -- def run_subagent(prompt: str) -> str: sub_messages = [{"role": "user", "content": prompt}] # fresh context for _ in range(30): # safety limit response = client.messages.create( model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages, tools=CHILD_TOOLS, max_tokens=8000, ) sub_messages.append({"role": "assistant", "content": response.content}) if response.stop_reason != "tool_use": break results = [] for block in response.content: if block.type == "tool_use": handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown tool: {block.name}" results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]}) sub_messages.append({"role": "user", "content": results}) # Only the final text returns to the parent -- child context is discarded return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)" # -- Parent tools: base tools + task dispatcher -- PARENT_TOOLS = CHILD_TOOLS + [ {"name": "task", "description": "Spawn a subagent with fresh context. It shares the filesystem but not conversation history.", "input_schema": {"type": "object", "properties": {"prompt": {"type": "string"}, "description": {"type": "string", "description": "Short description of the task"}}, "required": ["prompt"]}}, ] def agent_loop(messages: list): while True: response = client.messages.create( model=MODEL, system=SYSTEM, messages=messages, tools=PARENT_TOOLS, max_tokens=8000, ) messages.append({"role": "assistant", "content": response.content}) if response.stop_reason != "tool_use": return results = [] for block in response.content: if block.type == "tool_use": if block.name == "task": desc = block.input.get("description", "subtask") print(f"> task ({desc}): {block.input['prompt'][:80]}") output = run_subagent(block.input["prompt"]) else: handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown tool: {block.name}" print(f" {str(output)[:200]}") results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)}) messages.append({"role": "user", "content": results}) if __name__ == "__main__": history = [] while True: try: query = input("\033[36ms04 >> \033[0m") except (EOFError, KeyboardInterrupt): break if query.strip().lower() in ("q", "exit", ""): break history.append({"role": "user", "content": query}) agent_loop(history) response_content = history[-1]["content"] if isinstance(response_content, list): for block in response_content: if hasattr(block, "text"): print(block.text) print()