#!/usr/bin/env python3 """A minimal agent that can search the web, call tools, and keep state on disk. Requirements: Python 3.10+, ``pip install "anthropic>=1.6"``, ``ANTHROPIC_API_KEY``. Run a task: python minimal_search_agent.py --task "..." Resume after a crash (the loop checkpoints ``.agent/session.json`` after every tool round): python minimal_search_agent.py --resume Simulate a crash after N tool rounds, then resume: python minimal_search_agent.py --task "..." --crash-after 1 python minimal_search_agent.py --resume """ from __future__ import annotations import argparse import json import os import sys import time from pathlib import Path import anthropic MODEL = "claude-opus-5" WORKSPACE = Path("agent_workspace") STATE_DIR = WORKSPACE / ".agent" # reserved: tools may not read or write here SESSION_FILE = STATE_DIR / "session.json" MEMORY_FILE = STATE_DIR / "memory.json" SYSTEM_PROMPT = """You are a small research agent. Use web_search when you need facts you do not already know. Use write_file to save results the user asked for. Paths are relative to the workspace. Use remember to store short facts that should survive this session. When the task is done, reply with a short summary of what you did and where you saved things.""" # --- tools ------------------------------------------------------------------ # A server-side tool: Anthropic runs the search, results come back in the same response. WEB_SEARCH = {"type": "web_search_20260209", "name": "web_search", "max_uses": 5} # Client-side tools: the model asks, this process runs them, the loop returns results. CLIENT_TOOLS = [ { "name": "read_file", "description": "Read a UTF-8 text file inside the workspace.", "input_schema": { "type": "object", "properties": {"path": {"type": "string", "description": "Relative path"}}, "required": ["path"], "additionalProperties": False, }, "strict": True, }, { "name": "write_file", "description": "Create or overwrite a UTF-8 text file inside the workspace.", "input_schema": { "type": "object", "properties": { "path": {"type": "string", "description": "Relative path"}, "content": {"type": "string"}, }, "required": ["path", "content"], "additionalProperties": False, }, "strict": True, }, { "name": "remember", "description": "Store a short fact under a key so future runs can use it without searching again.", "input_schema": { "type": "object", "properties": {"key": {"type": "string"}, "value": {"type": "string"}}, "required": ["key", "value"], "additionalProperties": False, }, "strict": True, }, ] def safe_path(relative: str) -> Path: """Refuse paths that escape the workspace or touch the agent's own state files. Errors go back to the model as text. """ target = (WORKSPACE / relative).resolve() if WORKSPACE.resolve() not in target.parents and target != WORKSPACE.resolve(): raise ValueError(f"path {relative!r} is outside the workspace") if target == STATE_DIR.resolve() or STATE_DIR.resolve() in target.parents: raise ValueError(f"path {relative!r} is reserved for agent state") return target def load_memory() -> dict[str, str]: if MEMORY_FILE.exists(): return json.loads(MEMORY_FILE.read_text(encoding="utf-8")) return {} def run_tool(name: str, args: dict) -> str: if name == "read_file": return safe_path(args["path"]).read_text(encoding="utf-8") if name == "write_file": target = safe_path(args["path"]) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(args["content"], encoding="utf-8") return f"wrote {len(args['content'])} chars to {args['path']}" if name == "remember": memory = load_memory() memory[args["key"]] = args["value"] MEMORY_FILE.write_text(json.dumps(memory, ensure_ascii=False, indent=2), encoding="utf-8") return f"remembered {args['key']}" raise ValueError(f"unknown tool {name}") # --- state ------------------------------------------------------------------ def save_session(messages: list, status: str, rounds: int) -> None: """Atomic checkpoint: write to a temp file, then rename over the old one.""" tmp = SESSION_FILE.with_suffix(".tmp") tmp.write_text( json.dumps({"status": status, "rounds": rounds, "messages": messages}, ensure_ascii=False), encoding="utf-8", ) os.replace(tmp, SESSION_FILE) def load_session() -> dict: return json.loads(SESSION_FILE.read_text(encoding="utf-8")) def to_plain(blocks) -> list: """Response content blocks are SDK objects; store them as plain JSON for the checkpoint.""" return [b.model_dump(exclude_none=True) for b in blocks] # --- the loop --------------------------------------------------------------- def run(messages: list, rounds: int, crash_after: int | None) -> None: client = anthropic.Anthropic() memory = load_memory() system = SYSTEM_PROMPT if memory: system += "\n\nFacts remembered from earlier runs:\n" + json.dumps(memory, ensure_ascii=False) usage = {"input": 0, "output": 0, "cache_read": 0, "searches": 0, "calls": 0} started = time.time() while True: usage["calls"] += 1 response = client.messages.create( model=MODEL, max_tokens=16000, system=system, tools=[WEB_SEARCH, *CLIENT_TOOLS], messages=messages, ) u = response.usage usage["input"] += u.input_tokens usage["output"] += u.output_tokens usage["cache_read"] += u.cache_read_input_tokens or 0 if u.server_tool_use: usage["searches"] += u.server_tool_use.web_search_requests or 0 print(f"[call {usage['calls']}] stop_reason={response.stop_reason} " f"in={u.input_tokens} out={u.output_tokens}") messages.append({"role": "assistant", "content": to_plain(response.content)}) if response.stop_reason == "pause_turn": # Server-side search hit its iteration cap; resend as-is and it continues. save_session(messages, "running", rounds) continue if response.stop_reason == "refusal": save_session(messages, "refused", rounds) print("model declined the request:", response.stop_details) return if response.stop_reason == "max_tokens": save_session(messages, "truncated", rounds) print("hit max_tokens; raise the limit or split the task") return if response.stop_reason != "tool_use": final = "".join(b.text for b in response.content if b.type == "text") save_session(messages, "done", rounds) print("\n=== result ===\n" + final) break results = [] for block in response.content: if block.type != "tool_use": continue # server_tool_use blocks were already executed by the API try: output = run_tool(block.name, block.input) results.append({"type": "tool_result", "tool_use_id": block.id, "content": output}) print(f" tool {block.name}({json.dumps(block.input, ensure_ascii=False)[:80]}) ok") except Exception as exc: # the model sees the error and can recover results.append({"type": "tool_result", "tool_use_id": block.id, "content": f"Error: {exc}", "is_error": True}) print(f" tool {block.name} failed: {exc}") messages.append({"role": "user", "content": results}) rounds += 1 save_session(messages, "running", rounds) if crash_after is not None and rounds >= crash_after: print(f"--- simulated crash after {rounds} tool round(s); run with --resume ---") os._exit(1) elapsed = time.time() - started print(f"\ncalls={usage['calls']} tool_rounds={rounds} searches={usage['searches']} " f"input_tokens={usage['input']} output_tokens={usage['output']} " f"cache_read={usage['cache_read']} elapsed={elapsed:.1f}s") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--task", help="what the agent should do") parser.add_argument("--resume", action="store_true", help="continue from session.json") parser.add_argument("--crash-after", type=int, help="exit after N tool rounds (demo)") args = parser.parse_args() sys.stdout.reconfigure(line_buffering=True) # keep logs intact even if we os._exit STATE_DIR.mkdir(parents=True, exist_ok=True) if args.resume: state = load_session() if state["status"] != "running": sys.exit(f"nothing to resume: last session ended with status {state['status']!r}") messages, rounds = state["messages"], state["rounds"] print(f"resuming after {rounds} tool round(s), {len(messages)} messages on disk") elif args.task: messages, rounds = [{"role": "user", "content": args.task}], 0 else: sys.exit("give --task or --resume") run(messages, rounds, args.crash_after) if __name__ == "__main__": main()