Building Your First AI Agent From Scratch: A Complete Engineering Guide

   ┌─────────────────────────────────────────────┐
   │                                               │
   │   ┌──────────┐      ┌──────────┐             │
   └──▶│  REASON  │─────▶│   ACT    │             │
       │ (think)  │      │(use tool)│             │
       └──────────┘      └──────────┘             │
            ▲                  │                   │
            │                  ▼                   │
            │            ┌──────────┐               │
            └────────────│ OBSERVE  │───────────────┘
                          │ (result) │
                          └──────────┘

   Loop continues until:
   → Model reasons "I have enough to answer" → EXIT with final answer
   → Max iterations reached → EXIT with failure/escalation
   → Unrecoverable tool error → EXIT with failure/escalation

tools = [
    {
        "name": "search_web",
        "description": (
            "Search the web for current information. Use this when the "
            "question requires up-to-date facts, current events, or "
            "information not likely to be in your training data."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "The search query"}
            },
            "required": ["query"],
        },
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression and return the numeric result.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "A math expression, e.g. '47 * 12 + 8'"}
            },
            "required": ["expression"],
        },
    },
]

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=AGENT_SYSTEM_PROMPT,
    tools=tools,
    messages=conversation_history,
)

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """
    Dispatches a model-requested tool call to its real implementation.
    Returns a string result that will be shown back to the model.
    """
    if tool_name == "search_web":
        try:
            results = web_search_api(tool_input["query"])
            return format_search_results(results)
        except TimeoutError:
            return "ERROR: search request timed out. Try a more specific query."
        except Exception as e:
            return f"ERROR: search failed unexpectedly: {e}"

    elif tool_name == "calculate":
        try:
            # Never use raw eval() on model-supplied input in production —
            # use a restricted math expression parser instead.
            result = safe_eval_math(tool_input["expression"])
            return str(result)
        except (ValueError, SyntaxError):
            return f"ERROR: '{tool_input['expression']}' is not a valid expression."

    return f"ERROR: unknown tool '{tool_name}'"

conversation_history.append({"role": "assistant", "content": response.content})
conversation_history.append({
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": tool_use_block.id,
        "content": tool_result_string,
    }],
})

MAX_ITERATIONS = 8

def run_agent(user_task: str) -> str:
    conversation_history = [{"role": "user", "content": user_task}]

    for iteration in range(MAX_ITERATIONS):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=AGENT_SYSTEM_PROMPT,
            tools=tools,
            messages=conversation_history,
        )

        tool_use_blocks = [b for b in response.content if b.type == "tool_use"]

        if not tool_use_blocks:
            # Model produced a final answer with no further tool calls.
            return extract_text(response.content)

        conversation_history.append({"role": "assistant", "content": response.content})

        # Execute every requested tool call and append results.
        tool_results = []
        for block in tool_use_blocks:
            result_str = execute_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result_str,
            })
        conversation_history.append({"role": "user", "content": tool_results})

    # Hit the iteration cap without a final answer — fail explicitly rather
    # than silently returning nothing or looping forever.
    return "AGENT_INCOMPLETE: reached maximum reasoning steps without resolving the task."

import json
import sqlite3

class AgentMemory:
    """
    Durable, thread-scoped conversation history. A crash mid-task loses
    nothing — the next run reconstructs state from the last write.
    """
    def __init__(self, db_path: str = "agent_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS threads (
                thread_id TEXT PRIMARY KEY,
                history TEXT NOT NULL,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

    def load(self, thread_id: str) -> list:
        row = self.conn.execute(
            "SELECT history FROM threads WHERE thread_id = ?", (thread_id,)
        ).fetchone()
        return json.loads(row[0]) if row else []

    def save(self, thread_id: str, history: list) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO threads (thread_id, history) VALUES (?, ?)",
            (thread_id, json.dumps(history)),
        )
        self.conn.commit()

def query_database(filters: dict) -> str:
    results = db.query(filters)
    if not results:
        # Distinguish "confirmed empty" from "possibly malformed query" —
        # don't let the model conclude "there is no data" when the real
        # problem might be a bad filter.
        applied_filter_summary = ", ".join(f"{k}={v}" for k, v in filters.items())
        return (
            f"No rows matched filters: {applied_filter_summary}. "
            f"If this is unexpected, verify the filter values are correct "
            f"(e.g., check date formats, exact spelling of category names)."
        )
    return format_results(results)

def create_support_ticket(issue_summary: str, idempotency_key: str) -> str:
    """
    idempotency_key should be deterministic given the task context
    (e.g., a hash of thread_id + issue_summary), not random per call,
    so a retried call with the same key is a safe no-op.
    """
    existing = tickets_table.find_one({"idempotency_key": idempotency_key})
    if existing:
        return f"Ticket already exists: #{existing['ticket_id']} (no duplicate created)"

    ticket_id = tickets_table.insert({
        "summary": issue_summary,
        "idempotency_key": idempotency_key,
    })
    return f"Created ticket #{ticket_id}"

SENSITIVE_TOOLS = {"issue_refund", "send_email", "delete_record"}

def execute_tool(tool_name: str, tool_input: dict, approval_callback) -> str:
    if tool_name in SENSITIVE_TOOLS:
        approved = approval_callback(tool_name, tool_input)  # blocks for human input
        if not approved:
            return f"DENIED: human reviewer did not approve {tool_name} with {tool_input}"
    return _execute_tool_impl(tool_name, tool_input)