1# Check Python version (3.9+ required)
2python --version
3 
4# Create project
5mkdir my-agent && cd my-agent
6python -m venv venv
7source venv/bin/activate  # Windows: venv\Scripts\activate
8 
9# Install dependencies
10pip install anthropic python-dotenv

ANTHROPIC_API_KEY=your-api-key-here

1import os
2from dotenv import load_dotenv
3 
4load_dotenv()
5 
6ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
7MODEL = "claude-sonnet-4-20250514"
8MAX_TOKENS = 4096

1import os
2 
3FILE_TOOLS = [
4    {
5        "name": "read_file",
6        "description": "Read contents of a file",
7        "input_schema": {
8            "type": "object",
9            "properties": {
10                "path": {"type": "string", "description": "Path to file"}
11            },
12            "required": ["path"]
13        }
14    },
15    {
16        "name": "write_file",
17        "description": "Write content to a file",
18        "input_schema": {
19            "type": "object",
20            "properties": {
21                "path": {"type": "string", "description": "Path to file"},
22                "content": {"type": "string", "description": "Content to write"}
23            },
24            "required": ["path", "content"]
25        }
26    },
27    {
28        "name": "list_directory",
29        "description": "List files in a directory",
30        "input_schema": {
31            "type": "object",
32            "properties": {
33                "path": {"type": "string", "description": "Directory path"}
34            },
35            "required": ["path"]
36        }
37    }
38]
39 
40def read_file(path: str) -> str:
41    try:
42        with open(path, 'r') as f:
43            return f.read()
44    except Exception as e:
45        return f"Error: {str(e)}"
46 
47def write_file(path: str, content: str) -> str:
48    try:
49        with open(path, 'w') as f:
50            f.write(content)
51        return f"Wrote {len(content)} characters to {path}"
52    except Exception as e:
53        return f"Error: {str(e)}"
54 
55def list_directory(path: str) -> str:
56    try:
57        return "\n".join(os.listdir(path))
58    except Exception as e:
59        return f"Error: {str(e)}"

1import requests
2 
3WEB_TOOLS = [{
4    "name": "fetch_url",
5    "description": "Fetch content from a URL",
6    "input_schema": {
7        "type": "object",
8        "properties": {
9            "url": {"type": "string", "description": "URL to fetch"}
10        },
11        "required": ["url"]
12    }
13}]
14 
15def fetch_url(url: str) -> str:
16    try:
17        response = requests.get(url, timeout=10)
18        response.raise_for_status()
19        return response.text[:5000]  # Limit size
20    except Exception as e:
21        return f"Error: {str(e)}"

1import subprocess
2import tempfile
3 
4CODE_TOOLS = [{
5    "name": "run_python",
6    "description": "Execute Python code and return output",
7    "input_schema": {
8        "type": "object",
9        "properties": {
10            "code": {"type": "string", "description": "Python code"}
11        },
12        "required": ["code"]
13    }
14}]
15 
16def run_python(code: str) -> str:
17    try:
18        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
19            f.write(code)
20            temp_path = f.name
21 
22        result = subprocess.run(
23            ['python', temp_path],
24            capture_output=True,
25            text=True,
26            timeout=30
27        )
28 
29        output = result.stdout
30        if result.stderr:
31            output += f"\nErrors:\n{result.stderr}"
32        return output or "Code executed (no output)"
33 
34    except subprocess.TimeoutExpired:
35        return "Error: Execution timed out (30s limit)"
36    except Exception as e:
37        return f"Error: {str(e)}"

1SYSTEM_PROMPT = """You solve problems step by step.
2 
3For each step:
41. THOUGHT: What do I need to do?
52. ACTION: Use a tool
63. OBSERVATION: What did I learn?
74. REPEAT until done
8 
9Always think before you act."""
10 
11def react_agent(task: str) -> str:
12    messages = [{"role": "user", "content": task}]
13 
14    response = client.messages.create(
15        model=MODEL,
16        system=SYSTEM_PROMPT,
17        tools=TOOLS,
18        messages=messages
19    )
20    # ... continue agent loop

1def plan_and_execute(task: str) -> str:
2    # Step 1: Create plan
3    plan_response = client.messages.create(
4        model=MODEL,
5        system="Create a step-by-step plan. Return numbered steps.",
6        messages=[{"role": "user", "content": task}]
7    )
8    plan = extract_text(plan_response)
9 
10    # Step 2: Execute each step
11    results = []
12    for step in parse_plan(plan):
13        result = execute_step(step)
14        results.append(result)
15 
16    # Step 3: Synthesize
17    return synthesize_results(results)

1def tree_of_thoughts(problem: str, branches: int = 3) -> str:
2    # Generate approaches
3    approaches = [generate_approach(problem) for _ in range(branches)]
4 
5    # Evaluate each
6    scored = [(a, evaluate(a)) for a in approaches]
7 
8    # Execute best
9    best = max(scored, key=lambda x: x[1])
10    return execute_approach(best[0])

1class ShortTermMemory:
2    def __init__(self, max_messages: int = 20):
3        self.messages = []
4        self.max_messages = max_messages
5 
6    def add(self, role: str, content: str):
7        self.messages.append({"role": role, "content": content})
8        if len(self.messages) > self.max_messages:
9            self.messages = self.messages[-self.max_messages:]
10 
11    def get_messages(self):
12        return self.messages.copy()

1# pip install chromadb sentence-transformers
2import chromadb
3from sentence_transformers import SentenceTransformer
4 
5class LongTermMemory:
6    def __init__(self):
7        self.client = chromadb.Client()
8        self.collection = self.client.create_collection("agent_memory")
9        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
10 
11    def store(self, text: str, metadata: dict = None):
12        embedding = self.encoder.encode(text).tolist()
13        self.collection.add(
14            embeddings=[embedding],
15            documents=[text],
16            metadatas=[metadata or {}],
17            ids=[str(hash(text))]
18        )
19 
20    def search(self, query: str, n: int = 5) -> list:
21        embedding = self.encoder.encode(query).tolist()
22        results = self.collection.query(
23            query_embeddings=[embedding],
24            n_results=n
25        )
26        return results["documents"][0]

1class MemoryAgent:
2    def __init__(self):
3        self.short_term = ShortTermMemory()
4        self.long_term = LongTermMemory()
5 
6    def run(self, message: str) -> str:
7        # Retrieve relevant memories
8        memories = self.long_term.search(message)
9        context = "\n".join(memories) if memories else ""
10 
11        # Add to short-term
12        self.short_term.add("user", message)
13 
14        # Run with context
15        system = f"Relevant context:\n{context}" if context else ""
16        response = run_agent_with_system(system, self.short_term.get_messages())
17 
18        # Store response
19        self.long_term.store(response)
20        self.short_term.add("assistant", response)
21 
22        return response

1import time
2from anthropic import APIError, RateLimitError
3 
4def run_with_retry(message: str, max_retries: int = 3) -> str:
5    for attempt in range(max_retries):
6        try:
7            return run_agent(message)
8 
9        except RateLimitError:
10            wait = 2 ** attempt
11            print(f"Rate limited. Waiting {wait}s...")
12            time.sleep(wait)
13 
14        except APIError as e:
15            print(f"API error: {e}")
16            if attempt == max_retries - 1:
17                raise
18 
19    raise Exception("Max retries exceeded")

1def run_with_loop_detection(message: str, max_iterations: int = 20) -> str:
2    messages = [{"role": "user", "content": message}]
3    seen_calls = []
4    iteration = 0
5 
6    while iteration < max_iterations:
7        response = client.messages.create(
8            model=MODEL,
9            tools=TOOLS,
10            messages=messages
11        )
12 
13        if response.stop_reason == "tool_use":
14            current = [(b.name, str(b.input)) for b in response.content if b.type == "tool_use"]
15 
16            # Check for loop
17            if current in seen_calls[-3:]:
18                messages.append({
19                    "role": "user",
20                    "content": "You're repeating yourself. Try a different approach or give your final answer."
21                })
22            else:
23                seen_calls.append(current)
24                # Process tools normally...
25 
26        else:
27            return extract_text(response)
28 
29        iteration += 1
30 
31    return "Agent reached max iterations"

1def execute_tool_safely(name: str, inputs: dict) -> str:
2    try:
3        result = execute_tool(name, inputs)
4        return result
5    except Exception as e:
6        return f"Tool '{name}' failed: {str(e)}. Try a different approach."

1from dataclasses import dataclass
2from typing import List
3import logging
4 
5logging.basicConfig(level=logging.INFO)
6logger = logging.getLogger(__name__)
7 
8@dataclass
9class Note:
10    source: str
11    content: str
12 
13class ResearchAgent:
14    def __init__(self):
15        self.client = Anthropic()
16        self.notes: List[Note] = []
17        self.visited: set = set()
18 
19    def research(self, question: str) -> str:
20        logger.info(f"Researching: {question}")
21 
22        # Phase 1: Generate search queries
23        queries = self._generate_queries(question)
24 
25        # Phase 2: Search and gather
26        for query in queries:
27            results = self._search(query)
28            for result in results:
29                if result.url not in self.visited:
30                    content = self._fetch_and_extract(result.url)
31                    self._take_notes(result.url, content)
32                    self.visited.add(result.url)
33 
34        # Phase 3: Synthesize
35        report = self._synthesize(question)
36 
37        logger.info(f"Done. {len(self.notes)} notes from {len(self.visited)} sources")
38        return report
39 
40    def _generate_queries(self, question: str) -> List[str]:
41        response = self.client.messages.create(
42            model=MODEL,
43            system="Generate 3-5 search queries to research this question.",
44            messages=[{"role": "user", "content": question}]
45        )
46        return parse_queries(extract_text(response))
47 
48    def _search(self, query: str) -> List[dict]:
49        # Implement with your search API
50        pass
51 
52    def _fetch_and_extract(self, url: str) -> str:
53        # Fetch URL and extract key information
54        pass
55 
56    def _take_notes(self, source: str, content: str):
57        # Use Claude to extract key points
58        response = self.client.messages.create(
59            model=MODEL,
60            system="Extract 3-5 key facts from this content.",
61            messages=[{"role": "user", "content": content}]
62        )
63        self.notes.append(Note(source=source, content=extract_text(response)))
64 
65    def _synthesize(self, question: str) -> str:
66        notes_text = "\n\n".join(f"[{n.source}]\n{n.content}" for n in self.notes)
67 
68        response = self.client.messages.create(
69            model=MODEL,
70            system="Synthesize these notes into a comprehensive answer. Cite sources.",
71            messages=[{"role": "user", "content": f"Question: {question}\n\nNotes:\n{notes_text}"}]
72        )
73        return extract_text(response)

1import pytest
2from tools import calculate, execute_tool
3 
4def test_calculator():
5    assert calculate("2 + 2") == "4"
6    assert calculate("sqrt(16)") == "4.0"
7    assert "Error" in calculate("invalid")
8 
9def test_execute_tool():
10    result = execute_tool("calculator", {"expression": "10 * 5"})
11    assert result == "50"
12 
13def test_unknown_tool():
14    result = execute_tool("unknown", {})
15    assert "Unknown" in result

1def test_agent_math():
2    result = run_agent("What is 15 + 27?")
3    assert "42" in result
4 
5def test_agent_multi_step():
6    result = run_agent("Calculate 10^2, then take the square root")
7    assert "10" in result