Chapter 28

Building an agent from scratch

Agents from scratch, the engineering chapter of Part IX (Agents). From the 80/20 framing of agent engineering (LLM is 20%; everything else is 80%) through tool design principles, tool implementation patterns, schemas and structured calls (function calling, MCP), error handling and recovery (retries, circuit breakers, fallback tools), observability and trace debugging, and agent scaffolding (system prompts, tool descriptions, few-shot examples). The chapter that turns conceptual agent knowledge into working production code.

Chapter 27 covered what an agent is: an LLM acting as a controller in a loop with an environment. ReAct as the foundational pattern, AutoGPT as the cautionary tale, memory and patterns and the $2025$ stack. All conceptual. This chapter is engineering, how to take those concepts and assemble a working agent out of real code that survives real traffic. Production agents are $80\%$ engineering, $20\%$ LLM. The $20\%$ is the model and the basic loop. The $80\%$ is everything in this chapter: tools, schemas, error handling, observability, scaffolding.

The work breaks into six concerns. Tool design, atomic, single-responsibility, clear-contract functions the LLM can call effectively. Tool implementation: sync vs async, timeouts, output truncation, sandboxing for anything dangerous. Schemas, JSON Schema, OpenAI function calling, Anthropic MCP; how the LLM knows what tools exist and how to call them. Error handling, retries, circuit breakers, fallback tools, surfacing errors to the LLM. Observability, structured logging, distributed tracing, debugging from traces. Scaffolding, system prompts, tool descriptions, few-shot examples, output validation, cost budgets.

Tools should be designed so the LLM can use them effectively, not so you feel clever. A minimal ReAct agent is fifty lines; a production agent is $500$ – $5{,}000$ . The difference is everything in this chapter. By the end, you’ll have the operational toolkit to build agents that hold up in production.

From concept to code

The 80/20

Production agents are $80\%$ engineering on top of $20\%$ LLM. The $20\%$ is the model itself, the prompt, and the basic loop, all covered in Ch 27. The $80\%$ is everything else: tools, schemas, error handling, observability, scaffolding. Most engineers leaving a conceptual chapter on agents intuit that the LLM is the hard part. Ch 28 wants to correct that intuition. The LLM is the easy part, pick a frontier model, write a competent prompt, run a ReAct loop. The engineering around it is where the months of work live.

What goes into a real agent

The composition is short to state and long to build.

$\text{production agent} \;=\; \text{ReAct loop} \;+\; \text{tools} \;+\; \text{schemas} \;+\; \text{error handling} \;+\; \text{observability} \;+\; \text{scaffolding}$

(28.real-agent)

Each addend requires deliberate design. Skip one and the agent is fragile. Skip two and it never gets to production. The empirical anchors as of early $2025$ : a minimal ReAct agent runs in ~ $50$ lines of Python; a production agent runs $500$ – $5{,}000$ . Production agents typically expose $5$ – $50$ well-designed tools, not hundreds. Per-turn latency sits in the $1$ – $10$ second range; total task latency for non-trivial tasks is $10$ – $300$ seconds. Per-task cost runs $\$ 0.01 $–$ $10 $depending on model and complexity. The reliability bar for production deployment is ~$ 95%$+ task completion, anything lower and operators stop trusting the agent.

Engineering agents well means engineering with empathy for the LLM. Tools should be designed so the LLM can use them effectively. Schemas should be obvious to anyone reading them, including the model. Errors should be informative enough that the agent can plan its next move. Traces should be readable by humans debugging at $3$ AM. The agent is a system; this chapter is its system design.

Tool design principles

Six principles

A tool is a function the agent can call. Designing good tools is the most important non-LLM skill in agent engineering, and the place engineers most consistently underinvest.

Single responsibility. Each tool does one thing. Avoid grab-bag utilities like a file_operations tool that reads, writes, lists, copies, and chmods. Split it: read_file, write_file, list_directory, copy_file. The LLM picks tools by name, clear names beat clever abstractions.

Atomic operations. Tools should complete fully or fail cleanly. Avoid partial-completion semantics that the LLM has to reason about. A tool that writes half a file before erroring forces the agent to model the world in finer detail than it can reliably handle.

Clear contracts. Every tool has five parts: a name in verb-noun form (read_file, search_web, send_email); a one-to-three-sentence description; parameters with types and per-parameter descriptions; a return value with predictable structure; and documented error modes. Skip any of the five and the LLM has to guess.

Idempotency where possible. Calling twice has the same effect as calling once. This is critical for retry logic. Reads are naturally idempotent; writes often aren’t. When you can’t make a tool idempotent, make the non-idempotency visible, an explicit force=True parameter or a confirmation step.

Bounded outputs. Tools should return bounded amounts of data. A search_web that returns $100$ results blows up the context window in a single call. Limit, paginate, summarize. The right output size is “enough for the LLM to make the next decision, not more.”

Informative errors. Errors should tell the agent what went wrong AND what to try next. “Connection failed” is bad; “Connection failed: timeout after 30s; retry or try a different tool” is good. The LLM uses the error text to plan recovery; vague errors lead to vague recoveries.

Design anti-patterns

Four patterns that recur in production audits and need to be designed out. Kitchen-sink tools, one tool with twenty parameters and a dozen modes; impossible to describe; impossible for the LLM to use correctly. Magic parameters: parameter names that don’t tell the LLM what they do (flag, opts, mode); the description is the only thing the model sees. Silent failures: tools that return empty strings or null on error instead of an explicit failure signal; the agent thinks the call succeeded and acts on garbage. Unbounded outputs: tools that dump entire databases into context; one call exhausts the window.

Implementing tools

Synchronous tools

The simplest implementation: a synchronous Python function that takes parameters, does its work, and returns a result. Pros: easy to write, test, and debug. Cons: blocks the agent loop; tool calls can’t run in parallel. Synchronous tools are the right default for development and for tools whose latency is small relative to LLM latency.

Async, timeouts, and truncation

For I/O-bound tools, anything that waits on a network or disk, async unlocks parallel execution. The cost is operational complexity: the agent loop has to be async-aware, debugging is harder, and error semantics get subtler. Use async when parallelism actually pays for itself; stay sync otherwise.

Timeouts on every tool, always. Infinite hangs are an anti-pattern, they kill the agent loop without producing a useful error. Every network call gets a hard timeout, usually $5$ – $30$ seconds depending on the operation. Output truncation, always. Cap the size of every tool’s return value; if the underlying call produces more, truncate and tell the LLM how much was hidden. Side-effect tools with confirmation, anything that mutates state (send email, delete file, run code, charge a card) should require explicit confirmation or run sandboxed. The LLM might call them by mistake, and the cost of a mistaken side effect can be much higher than a missing answer.

Tool registries

A registry maps tool names to functions, schemas, and descriptions. Single source of truth for what tools exist. A registry pattern means you define a tool once and the OpenAI schema, the Anthropic schema, the execution wrapper, and the error handling all read from the same place.

# Production-style tool registry: function + schema + description.
# This is the pattern at the heart of every well-engineered agent.

from typing import Callable
import math

class Tool:
    def __init__(self, name: str, description: str, parameters_schema: dict, function: Callable):
        self.name = name
        self.description = description
        self.parameters_schema = parameters_schema
        self.function = function

    def execute(self, arguments: dict) -> dict:
        """Execute with structured error handling. Returns {'status': 'ok'|'error', ...}."""
        try:
            result = self.function(**arguments)
            return {'status': 'ok', 'result': result}
        except Exception as e:
            return {
                'status': 'error',
                'error_type': type(e).__name__,
                'message': str(e),
            }


# Tool registry: maps names to Tool objects
TOOLS: dict[str, Tool] = {}


def register(name: str, description: str, parameters_schema: dict):
    """Decorator to register a tool."""
    def wrapper(fn):
        TOOLS[name] = Tool(name, description, parameters_schema, fn)
        return fn
    return wrapper


@register(
    name='get_weather',
    description='Get current weather for a city. Returns temperature and conditions.',
    parameters_schema={
        'type': 'object',
        'properties': {
            'city': {'type': 'string', 'description': 'City name (e.g. "Tokyo")'},
        },
        'required': ['city'],
    },
)
def get_weather(city: str) -> dict:
    return {'city': city, 'temp_c': 18, 'conditions': 'partly cloudy'}


@register(
    name='calculate',
    description='Evaluate a mathematical expression. Returns the result.',
    parameters_schema={
        'type': 'object',
        'properties': {
            'expression': {'type': 'string', 'description': 'Math expression (e.g. "2 + 2")'},
        },
        'required': ['expression'],
    },
)
def calculate(expression: str) -> float:
    return eval(expression, {'__builtins__': {}}, {'sqrt': math.sqrt, 'pi': math.pi})


# Execute via the registry
print(TOOLS['get_weather'].execute({'city': 'Tokyo'}))
print(TOOLS['calculate'].execute({'expression': '5 * 7'}))
print(TOOLS['calculate'].execute({'expression': '1/0'}))   # error case

# Observations:
# - Registry pattern: single source of truth for tool name + schema + function
# - execute() wraps in structured error handling (never raises to the agent)
# - Errors are structured data, returned as observations to the LLM
# - Provider-specific schema conversion (OpenAI / Anthropic) can be added later

Tool schemas and structured calls

The schema problem

The LLM does not see Python types. It sees text. Schemas tell the LLM how to call tools in a structured, parseable way. Two dominant approaches exist in $2025$ : function calling (OpenAI’s API, now widely adopted across providers) and MCP (Anthropic’s open Model Context Protocol). Both produce structured JSON tool calls the developer parses and executes.

Function calling vs MCP

OpenAI function calling works as follows. The developer declares each tool with a JSON Schema giving its name, description, and parameters. The LLM’s response contains a tool_calls array with the tool name and argument JSON. The developer parses the arguments, executes the function, and feeds the result back as a tool message in the next turn. A minimal schema looks like:

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "Return current weather for a city.",
    "parameters": {
      "type": "object",
      "properties": {
        "city": { "type": "string", "description": "City name (e.g. 'Tokyo')" }
      },
      "required": ["city"]
    }
  }
}

Anthropic tool use is conceptually identical (schemas plus structured tool-use blocks), with explicit parallel tool calls, reasoning tokens, and slightly different message formatting. Model Context Protocol (MCP) sits a layer above function calling. MCP servers expose tools; MCP clients (LLM applications) discover and invoke them. The protocol standardizes tool discovery, authentication, schemas, and error semantics across providers. Tools become portable between LLM applications without per-application integration work.

OpenAPI integration is the third leg. Many existing REST APIs auto-convert to function-calling schemas via mature tooling. An agent can use any documented REST API with minimal glue code, point a converter at the OpenAPI spec, get tool schemas out, drop them in the registry.

Tool schema builder

6 tools · Python → OpenAI / Anthropic schemas · sample LLM call

Pick a tool:

get_weather

simple

Python source

def get_weather(city: str) -> dict:
    """Get current weather for a city.
 
    Returns temperature in Celsius and conditions.
 
    Args:
        city: City name (e.g. "Tokyo", "Paris", "San Francisco")
    """
    response = requests.get(
        f"https://api.weather.com/{city}",
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

Schema

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "Get current weather for a city. Returns temperature in Celsius and conditions.",
    "parameters": {
      "type": "object",
      "properties": {
        "city": {
          "type": "string",
          "description": "City name (e.g. \"Tokyo\", \"Paris\", \"San Francisco\")"
        }
      },
      "required": [
        "city"
      ]
    }
  }
}

Schema design note

The simplest pattern: one required string parameter. Notice that the LLM emits `arguments` as a JSON-encoded string; your code must parse it before calling the function. The description carries the examples: that's where the LLM picks up the right input format.

Cycle through the tools. Each Python function compiles to a schema the LLM reads to decide whether and how to call it. OpenAI and Anthropic schemas are nearly identical: minor key differences (function vs name), same underlying mechanism. The sample LLM call shows what the LLM actually emits: a structured tool_calls array with the function name and JSON-encoded arguments. Schema design IS contract design: descriptions, type constraints, enums, and required fields are what stops the LLM from passing malformed inputs. The most under-appreciated piece of agent engineering, now visible.

Six curated Python tool functions, each a different schema pattern: simple, optional params, side-effect, bounded output, restricted eval, complex shape. For each, see the Python source side by side with the OpenAI function-calling schema, the Anthropic tool-use schema, and a sample LLM tool-call response as JSON. Demonstrates the function → schema → invocation pipeline that every agent framework runs on. The schema is the contract between engineer and LLM, descriptions, type constraints, and required fields are what stops the LLM from passing malformed inputs.

Best practices

Five practices separate schemas the LLM uses correctly from schemas it doesn’t. Descriptions matter: the LLM picks tools by reading them; vague descriptions guarantee wrong choices. Parameter examples in descriptions: include sample values inline; the LLM mirrors them. Required vs optional: be explicit; ambiguity here triggers junk arguments. Enums for constrained values: if a parameter is one of five strings, declare it as an enum, not a free string. Object types for complex inputs: when an argument is structured, declare it as an object with its own schema, not a stringified JSON blob.

Error handling and recovery

The reality of tool errors

Production tool calls fail constantly. Network failures. Services unavailable. Rate limits. Invalid arguments because the LLM got it slightly wrong. Timeouts. Authentication failures. Permission denied. Every one of these happens regularly in any agent that handles real traffic, and a robust agent handles every one of them.

Six patterns

Retry with exponential backoff for transient failures. Wait progressively longer between attempts, $1$ s, $2$ s, $4$ s, capped at a maximum delay. Categorize errors first: retry transient failures (network, rate-limit, timeout); don’t retry permanent ones (invalid arguments, $4xx$ auth errors).

Circuit breakers for cascading failures. If a tool fails repeatedly, stop calling it for a cooldown period. Prevents one broken service from taking the whole agent down, an agent that keeps trying a dead endpoint just burns iterations and cost.

Fallback tools for redundancy. When one tool fails, try a backup, two web search providers, a cached lookup before a live API, a cheaper model when the expensive one rate-limits. Fallbacks need their own design (when does the fallback fire? what does the LLM see?), but for high-availability agents they’re standard.

Surface errors to the LLM, never swallow them. The agent uses error messages to plan recovery. Return errors as structured observations the model can read.

Bounded iteration. Even with perfect error handling, cap iterations. No agent should loop indefinitely searching for a non-existent solution.

Cost budgets. Cap total cost per task. If an agent burns $\$ 10$ in inference without progress, something is wrong and the system should stop it.

Anti-patterns that recur. Swallowing errors silently, the agent never sees them and never recovers. Retrying non-retryable errors, invalid arguments fail forever no matter how patiently you wait. Catching too broadly, except Exception hides bugs as readily as it hides transients. Infinite retry without backoff or cap, a denial-of-service against your own infrastructure.

# Combining circuit breakers and retry with backoff.
# Production agents use this layered defense.

import time
import random

class CircuitBreaker:
    """Stop calling a service after N failures; reset after cooldown."""
    def __init__(self, threshold=3, cooldown=5.0):
        self.failures = 0
        self.open_until = 0.0
        self.threshold = threshold
        self.cooldown = cooldown

    def call(self, fn, *args, **kwargs):
        if time.time() < self.open_until:
            raise Exception("CircuitOpen: cooldown active")
        try:
            result = fn(*args, **kwargs)
            self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.threshold:
                self.open_until = time.time() + self.cooldown
                print(f"  [CIRCUIT OPEN] {self.failures} failures; cooldown {self.cooldown}s")
            raise


def with_retry(fn, *args, max_retries=3, base_wait=0.001, **kwargs):
    """Retry with exponential backoff. Returns (result, error)."""
    for attempt in range(max_retries):
        try:
            return fn(*args, **kwargs), None
        except Exception as e:
            wait = base_wait * (2 ** attempt)
            print(f"  Attempt {attempt + 1}: {e}. Waiting {wait*1000:.0f}ms...")
            if attempt < max_retries - 1:
                time.sleep(wait)
    return None, f"Failed after {max_retries} attempts"


# Mock flaky tool: fails 60% of the time
def flaky_search(query):
    if random.random() < 0.6:
        raise ConnectionError("Service temporarily unavailable")
    return f"Search results for: {query}"


breaker = CircuitBreaker(threshold=2, cooldown=0.5)


def safe_search(query):
    """Wrap the tool in circuit breaker + retry layers."""
    try:
        result, error = with_retry(
            lambda q: breaker.call(flaky_search, q),
            query,
            max_retries=2,
        )
        return result if not error else f"Error: {error}"
    except Exception as e:
        return f"Error: {e}"


# Run multiple calls
random.seed(42)
for i in range(6):
    print(f"\nCall {i+1}: searching for 'topic_{i}'")
    print(f"Result: {safe_search(f'topic_{i}')}")
    time.sleep(0.1)

# Observations:
# - Layered defense: retry handles transient flakiness; circuit breaker prevents cascade
# - Real production: also add fallback tools, error categorization, structured logging
# - The LLM should see structured errors (not exceptions) so it can recover

Observability: tracing and debugging

The trace problem

When an agent fails in production, the trace is the only diagnostic. Without good traces, debugging is impossible. A failed agent run can involve dozens of LLM calls and tool invocations spread across minutes; reading raw logs to reconstruct what happened is impractical, and “rerun and see” is unreliable because most failures are non-deterministic.

A good trace shows every LLM call (model name, prompt tokens, completion tokens, latency, cost), every tool call (name, arguments, result, latency, success/failure), the thought-action-observation sequence, errors and recoveries, and cost-and-time accumulating per task. All of that, structured, queryable, and visualizable.

Spans and structured logs

The standard pattern, borrowed from distributed systems: trace IDs unique per task; spans as nested timed operations (LLM call → tool call → result processing); parent-child relationships so each span knows where it sits in the tree; attributes as structured metadata per span. Tools like OpenTelemetry standardize the data model; LangSmith, Helicone, and Braintrust render it.

Structured logging as a baseline. Every event is JSON, not free-text, {event: tool_call, task_id, iteration, tool, arguments, latency_ms, success}. Structured logs are queryable; free-text isn’t.

Agent trace inspector

4 traces · flame-graph-style nested spans · click for detail

Pick a trace:

CLEAN SUCCESS

clean

"What's the weather in Tokyo, and what date is it today?"

⏱️2.29stotal

💰$0.019cost

📊1,515tokens

✓completed

Span flame graph · click for detail

0ms1145ms2290ms

✓agent_turn 1

950ms

✓llm_call

720ms

✓parse

2ms

✓get_weather

195ms

✓agent_turn 2

840ms

✓llm_call

680ms

✓parse

1ms

✓get_date

140ms

✓agent_turn 3

480ms

✓llm_call

460ms

✓final_answer

5ms

Selected span detail

Click a span above to see its full attributes.

Insights

Total task time: ~2.3 seconds, dominated by LLM latency (1.9s of 2.3s)
Tool calls are fast (140-200ms each); the bottleneck is the model, not the tools
Cost: ~$0.019 total for 3 LLM calls, manageable for most production use cases
Three iterations match the 3 thought-action-observation triples in the chapter's opener

Scenario note

A clean 3-turn ReAct trace: two tool calls (weather, date) followed by a final answer. No retries, no errors, modest cost. This is what production agents look like on the happy path.

This is what production observability looks like. Each agent task is a tree of spans (LLM calls, tool calls, parses, retries, final answers), each with timing, attributes (model, tokens, cost), and status. Clean traces tell you nothing; they're the baseline. The interesting traces are the failure modes: transient retries (engineering working as designed), hallucinated tool calls (the LLM recovers via structured errors), cost-blown runaways (context bloat caught by hard caps). Without traces, every agent failure is mysterious; with them, the cause is visible in seconds. Production tools like LangSmith, Helicone, and Braintrust render this view at scale.

# A simple structured-logging trace layer.
# Production: use OpenTelemetry; this shows the conceptual pattern.

import time
import uuid
from contextlib import contextmanager

class Tracer:
    def __init__(self):
        self.spans = []
        self.task_id = None
        self.span_stack = []

    def start_task(self, name: str) -> str:
        self.task_id = str(uuid.uuid4())
        self.spans = []
        self.span_stack = []
        return self.task_id

    @contextmanager
    def span(self, name: str, **attrs):
        span_id = str(uuid.uuid4())[:8]
        parent_id = self.span_stack[-1] if self.span_stack else None
        start = time.time()
        self.span_stack.append(span_id)
        record = {
            'span_id': span_id,
            'parent_id': parent_id,
            'task_id': self.task_id,
            'name': name,
            'start': start,
            'attributes': attrs,
        }
        self.spans.append(record)
        try:
            yield record
            record['status'] = 'ok'
        except Exception as e:
            record['status'] = 'error'
            record['error'] = str(e)
            raise
        finally:
            record['end'] = time.time()
            record['duration_ms'] = (record['end'] - start) * 1000
            self.span_stack.pop()

    def report(self):
        print(f"=== Trace for task {self.task_id} ===")
        for span in self.spans:
            depth = 0
            pid = span['parent_id']
            while pid:
                depth += 1
                pid = next((s['parent_id'] for s in self.spans if s['span_id'] == pid), None)
            indent = '  ' * depth
            icon = 'OK' if span.get('status') == 'ok' else 'ER'
            print(f"{indent}[{icon}] {span['name']:>15} ({span['duration_ms']:.1f}ms)  {span['attributes']}")


# Use the tracer
tracer = Tracer()
tracer.start_task("answer_user_query")

with tracer.span("agent_turn", iteration=1):
    with tracer.span("llm_call", model="claude-sonnet-4", tokens_in=450, tokens_out=120):
        time.sleep(0.05)
    with tracer.span("tool_call", tool="get_weather", city="Tokyo"):
        time.sleep(0.02)

with tracer.span("agent_turn", iteration=2):
    with tracer.span("llm_call", model="claude-sonnet-4", tokens_in=620, tokens_out=85):
        time.sleep(0.04)
    with tracer.span("tool_call", tool="get_date"):
        time.sleep(0.01)

tracer.report()

# Observations:
# - Spans nest via context managers (with statement)
# - Each span has start, end, duration, attributes
# - Real production: OpenTelemetry standardizes this; LangSmith renders it
# - Visualization turns hours of debugging into seconds of inspection

Debugging from traces

The mapping from failure mode to trace signature is short and worth memorizing. Hung agent: look for spans without end times. Wrong answer: look at the thought-action sequence for missteps. Cost spike: look for excessive iterations or oversized contexts. Slow response: look for slow tools or model-side latency. Most production debugging is pattern-matching from this list.

Scaffolding the agent

What goes into scaffolding

Scaffolding is everything that’s not the loop or the tools, the supporting structure that the user never sees but that makes the agent work. Seven components earn their keep in production. System prompt: sets the agent’s role, behavior, and constraints. The most important single piece of agent code. Tool descriptions: read by the LLM at every call; write them as if for a colleague who has to do the work. Few-shot examples, examples of correct agent behavior embedded in the system prompt. The most-impactful single intervention for many tasks: two or three good examples often outperform paragraphs of instructions. Output validation: after the LLM produces output, validate it; reject and retry on invalid output. Memory layer: what state does the agent preserve across turns and sessions? (Ch 27 §5.) Conversation management: for chat-style agents, the layer that maintains user identity, session state, and conversation history. Cost and rate limits: hard caps that prevent runaway spending.

A sample production system prompt, drawn down to the irreducible:

You are a customer-support agent for [Company]. Your goals:
1. Answer user questions accurately using the tools provided.
2. If you can't help, escalate to a human via the `escalate` tool.
3. Be concise and friendly.

You have access to these tools:
- search_kb(query): search the knowledge base
- check_order(order_id): look up order status
- escalate(reason): hand off to a human agent

Guidelines:
- Never invent information; if unsure, use search_kb or escalate.
- Confirm sensitive actions before taking them.
- Keep responses under 200 words.

Output format: ReAct (Thought/Action/Observation).

Production patterns

Five patterns recur across production agents. System-prompt versioning, prompts evolve; version them like code and record which version produced each task. Dynamic few-shot retrieval, for complex tasks, retrieve relevant examples per task instead of hard-coding the same ones (Ch 22’s RAG patterns applied to the prompt itself). Output validation with retry: reject malformed JSON; ask the LLM to try again with the error as context. Sanitized logging: redact PII before traces leave the application. Sandboxing dangerous tools, Docker, VMs, or restricted environments for anything that runs code or modifies external state.

The anti-patterns are mirror images. Massive system prompts (> $2{,}000$ tokens) the LLM can’t reliably follow. Vague tool descriptions that force the model to guess. No few-shot examples when the task is unusual enough to need them. Skipping validation because the LLM “usually gets it right”, usually isn’t always, and the failures are silent.

Exercises

Four exercises that lock in the engineering toolkit. Each is a self-contained problem with a starting template; hints are collapsed by default; try the problem first.

The exercises trace the chapter’s arc: build a tool registry (Ex 1) → implement a retry pattern (Ex 2) → add structured tracing (Ex 3) → tie everything together in a complete agent harness (Ex 4). After these, the reader has all the components of a production agent.

Exercise 1 (easy): Tool registry with execute()

Build a tool registry where tools are registered by name, schemas are generated from metadata, and execution is wrapped in structured error handling.

Hint

The pattern:

A Tool class holds: name, description, parameters schema, function
A register decorator adds tools to a global registry
Each tool’s execute(arguments) method calls the function and wraps any exception in a structured {'status': 'error', 'error_type': ..., 'message': ...} response
On success: {'status': 'ok', 'result': ...}

The point is that the LLM never sees a Python exception, it sees a structured observation it can reason about.

from typing import Callable
import math

class Tool:
    def __init__(
        self,
        name: str,
        description: str,
        parameters_schema: dict,
        function: Callable,
    ):
        self.name = name
        self.description = description
        self.parameters_schema = parameters_schema
        self.function = function

    def execute(self, arguments: dict) -> dict:
        """
        Execute the tool with given arguments.
        Returns a structured response:
          {'status': 'ok', 'result': ...}      on success
          {'status': 'error', 'error_type': str, 'message': str}  on exception
        """
        # TODO:
        # try:
        #     result = self.function(**arguments)
        #     return {'status': 'ok', 'result': result}
        # except Exception as e:
        #     return {
        #         'status': 'error',
        #         'error_type': type(e).__name__,
        #         'message': str(e),
        #     }
        pass

    def to_openai_schema(self) -> dict:
        """Convert to OpenAI function-calling schema format."""
        return {
            'type': 'function',
            'function': {
                'name': self.name,
                'description': self.description,
                'parameters': self.parameters_schema,
            },
        }


TOOLS: dict[str, Tool] = {}


def register(name: str, description: str, parameters_schema: dict):
    """Decorator: register a function as a Tool in the global registry."""
    def wrapper(fn):
        TOOLS[name] = Tool(name, description, parameters_schema, fn)
        return fn
    return wrapper


@register(
    name='get_weather',
    description='Get current weather for a city.',
    parameters_schema={
        'type': 'object',
        'properties': {
            'city': {'type': 'string', 'description': 'City name'},
        },
        'required': ['city'],
    },
)
def get_weather(city: str) -> dict:
    return {'city': city, 'temp_c': 18, 'conditions': 'partly cloudy'}


@register(
    name='calculate',
    description='Evaluate a mathematical expression.',
    parameters_schema={
        'type': 'object',
        'properties': {
            'expression': {'type': 'string', 'description': 'Math expression'},
        },
        'required': ['expression'],
    },
)
def calculate(expression: str) -> float:
    safe = {'__builtins__': {}}
    locals_ = {'sqrt': math.sqrt, 'pi': math.pi}
    return eval(expression, safe, locals_)


# Test
# print("Registered tools:", list(TOOLS.keys()))
# print()
#
# # Successful call
# result = TOOLS['get_weather'].execute({'city': 'Tokyo'})
# print(f"get_weather(Tokyo): {result}")
#
# # Error: division by zero
# result = TOOLS['calculate'].execute({'expression': '1/0'})
# print(f"calculate(1/0):     {result}")
#
# # Error: wrong argument name
# result = TOOLS['get_weather'].execute({'town': 'Tokyo'})
# print(f"get_weather(town):  {result}")
#
# # Observations:
# # - Successful calls return {status:'ok', result:...}
# # - All exceptions become {status:'error', ...}: never raised to the agent
# # - Schema generation uses metadata only (no inspection); explicit is better
# # - Real production: also generate Anthropic schema; add OpenAPI conversion

Solution

The exception boundary lives entirely in execute(): call the function, wrap success as {'status': 'ok', ...}, catch any exception and wrap it as {'status': 'error', ...} instead of letting it propagate.

def execute(self, arguments: dict) -> dict:
    try:
        result = self.function(**arguments)
        return {'status': 'ok', 'result': result}
    except Exception as e:
        return {
            'status': 'error',
            'error_type': type(e).__name__,
            'message': str(e),
        }

Running the three test calls gives:

get_weather(Tokyo): {'status': 'ok', 'result': {'city': 'Tokyo', 'temp_c': 18, 'conditions': 'partly cloudy'}}
calculate(1/0):     {'status': 'error', 'error_type': 'ZeroDivisionError', 'message': 'division by zero'}
get_weather(town):  {'status': 'error', 'error_type': 'TypeError', 'message': "get_weather() got an unexpected keyword argument 'town'"}

The third call is the important one: calling with the wrong argument name raises a TypeError inside self.function(**arguments), and execute() catches it like any other exception, so a malformed tool call never crashes the agent loop.

Exercise 2 (medium): Retry with exponential backoff

Implement a retry helper that re-executes a flaky function with exponential backoff, returning a structured (result, error) tuple after success or exhaustion.

Hint

The pattern:

Attempt up to max_retries times
On exception, sleep for base_wait * 2^attempt (exponential backoff)
Return (result, None) on success, (None, "error message") on exhaustion
The structured tuple lets the caller decide what to do, no exceptions raised

Real production also adds: error categorization (retryable vs not), circuit breakers, jitter (small random offset to avoid thundering-herd retries).

import time
import random


def with_retry(fn, *args, max_retries=3, base_wait=0.001, **kwargs):
    """
    Retry fn with exponential backoff.
    Returns (result, error_message). Exactly one of these is None.
    """
    # TODO:
    # For attempt in range(max_retries):
    #   try:
    #     result = fn(*args, **kwargs)
    #     return (result, None)
    #   except Exception as e:
    #     wait = base_wait * (2 ** attempt)
    #     print(f"  Attempt {attempt + 1} failed: {e}. Waiting {wait*1000:.0f}ms...")
    #     if attempt < max_retries - 1: time.sleep(wait)
    # Return (None, f"Failed after {max_retries} attempts")
    pass


# Mock flaky tool
def flaky_search(query):
    """Fails 50% of the time."""
    if random.random() < 0.5:
        raise ConnectionError("Service temporarily unavailable")
    return f"Search results for: {query}"


# Test
# random.seed(42)
#
# for i in range(4):
#     print(f"\nCall {i + 1}: query='topic_{i}'")
#     result, error = with_retry(flaky_search, f'topic_{i}', max_retries=3)
#     if error:
#         print(f"  Final: {error}")
#     else:
#         print(f"  Final: {result}")
#
# # Observations:
# # - Exponential backoff: waits 1ms, 2ms, 4ms between retries (scaled down for demo)
# # - Some calls succeed on retry; some exhaust their budget
# # - The (result, error) tuple lets the caller decide what to do next
# # - Real production: add jitter, circuit breakers, error categorization

Solution

Loop up to max_retries, return immediately on success, sleep with exponential backoff between attempts (but not after the last one), and return a structured error once the budget is exhausted.

def with_retry(fn, *args, max_retries=3, base_wait=0.001, **kwargs):
    for attempt in range(max_retries):
        try:
            result = fn(*args, **kwargs)
            return (result, None)
        except Exception as e:
            wait = base_wait * (2 ** attempt)
            print(f"  Attempt {attempt + 1} failed: {e}. Waiting {wait*1000:.0f}ms...")
            if attempt < max_retries - 1:
                time.sleep(wait)
    return (None, f"Failed after {max_retries} attempts")

With random.seed(42) and the given flaky_search, the four calls resolve as: call 1 succeeds immediately, call 2 fails all 3 attempts (waiting 1ms, 2ms, 4ms) and returns "Failed after 3 attempts", calls 3 and 4 each succeed on the first attempt. The if attempt < max_retries - 1 guard matters: without it, the helper sleeps one extra time after the final failed attempt, delaying an error the caller already has enough information to act on.

Exercise 3 (medium): Structured tracing layer

Add a Tracer class that records spans (timed operations) with attributes. Spans nest via Python’s context-manager protocol. Use it to trace a mock agent loop.

Hint

The pattern:

Tracer keeps a list of span records and a stack of currently-open spans
tracer.span(name, **attrs) returns a context manager that:
- on enter: pushes a new span onto the stack with start time + attrs
- on exit: pops, records end time and status
Each span records: id, parent_id, name, start, duration, attributes, status
report() prints the tree with indentation

Real production uses OpenTelemetry, but the conceptual model is identical.

import time
import uuid
from contextlib import contextmanager


class Tracer:
    def __init__(self):
        self.spans = []
        self.span_stack = []
        self.task_id = None

    def start_task(self, name: str) -> str:
        self.task_id = str(uuid.uuid4())
        self.spans = []
        self.span_stack = []
        return self.task_id

    @contextmanager
    def span(self, name: str, **attrs):
        """
        Context manager. On enter: open a span. On exit: close it.
        Spans nest naturally based on the call stack.
        """
        # TODO:
        # 1. Generate span_id (uuid4 first 8 chars)
        # 2. parent_id = span_stack[-1] if span_stack else None
        # 3. record = {span_id, parent_id, task_id, name, start: time.time(), attributes: attrs}
        # 4. Append record to self.spans
        # 5. Push span_id onto self.span_stack
        # 6. try: yield record; record['status'] = 'ok'
        #    except: record['status'] = 'error'; record['error'] = str(e); raise
        #    finally: record['end'] = time.time(); record['duration_ms'] = (end-start)*1000
        #             self.span_stack.pop()
        pass

    def report(self):
        print(f"=== Trace for task {self.task_id} ===")
        for span in self.spans:
            # Compute depth by walking parent chain
            depth = 0
            pid = span['parent_id']
            while pid:
                depth += 1
                pid = next((s['parent_id'] for s in self.spans if s['span_id'] == pid), None)
            indent = '  ' * depth
            icon = 'OK' if span.get('status') == 'ok' else 'ER'
            attrs_str = ' '.join(f'{k}={v}' for k, v in span['attributes'].items())
            print(f"{indent}[{icon}] {span['name']:<20} ({span['duration_ms']:.1f}ms)  {attrs_str}")


# Test
# tracer = Tracer()
# tracer.start_task("answer_user_query")
#
# with tracer.span("agent_turn", iteration=1):
#     with tracer.span("llm_call", model="claude-sonnet-4", tokens_in=320, tokens_out=95):
#         time.sleep(0.04)
#     with tracer.span("tool_call", tool="get_weather", city="Tokyo"):
#         time.sleep(0.02)
#
# with tracer.span("agent_turn", iteration=2):
#     with tracer.span("llm_call", model="claude-sonnet-4", tokens_in=455, tokens_out=60):
#         time.sleep(0.03)
#     with tracer.span("tool_call", tool="get_date"):
#         time.sleep(0.01)
#
# tracer.report()
#
# # Observations:
# # - Spans nest via context managers (with statement)
# # - Each span has start, end, duration, attributes
# # - Trace structure mirrors the call structure of the agent
# # - Real production: OpenTelemetry standardizes this; LangSmith renders it
# # - Visualization turns hours of debugging into seconds of inspection

Solution

span() is a context manager: build the record on enter (with the current stack top as parent_id), push it, then on exit stamp status/duration and pop. The try/except/finally makes error spans and normal spans share the same bookkeeping.

@contextmanager
def span(self, name: str, **attrs):
    span_id = str(uuid.uuid4())[:8]
    parent_id = self.span_stack[-1] if self.span_stack else None
    record = {
        'span_id': span_id,
        'parent_id': parent_id,
        'task_id': self.task_id,
        'name': name,
        'start': time.time(),
        'attributes': attrs,
    }
    self.spans.append(record)
    self.span_stack.append(span_id)
    try:
        yield record
        record['status'] = 'ok'
    except Exception as e:
        record['status'] = 'error'
        record['error'] = str(e)
        raise
    finally:
        record['end'] = time.time()
        record['duration_ms'] = (record['end'] - record['start']) * 1000
        self.span_stack.pop()

Running the two-turn mock trace produces nested output like:

=== Trace for task <uuid> ===
[OK] agent_turn           (78.4ms)  iteration=1
  [OK] llm_call             (48.2ms)  model=claude-sonnet-4 tokens_in=320 tokens_out=95
  [OK] tool_call            (30.0ms)  tool=get_weather city=Tokyo
[OK] agent_turn           (53.4ms)  iteration=2
  [OK] llm_call             (38.3ms)  model=claude-sonnet-4 tokens_in=455 tokens_out=60
  [OK] tool_call            (15.0ms)  tool=get_date

(exact millisecond values vary run to run). llm_call and tool_call indent one level under agent_turn because parent_id points at whatever span is on top of the stack when they open, which report()’s parent-chain walk turns into indentation.

Exercise 4 (hard): Complete agent harness

Tie everything together: build a complete agent harness that uses a tool registry (Ex 1), retry helper (Ex 2), and tracer (Ex 3) to execute a multi-turn ReAct task.

Hint

The harness:

Agent(tools, tracer, max_iterations) holds the components
run(task) executes the ReAct loop with tracing
Each iteration: traced agent_turn span; traced llm_call; parse; traced tool_call (with retry)
Returns the final answer (or error) when the loop ends

The point is that all the pieces compose: the tool registry executes safely; retry handles flakiness; the tracer records everything; the agent loop drives the iteration.

For the LLM, you’ll mock responses, focus on the orchestration, not the model.

import time
import re
from contextlib import contextmanager


# Reuse: Tool, Tracer from previous exercises
# (For brevity in this exercise, we'll define minimal versions inline.)

class Tool:
    def __init__(self, name, description, function):
        self.name = name; self.description = description; self.function = function

    def execute(self, arguments):
        try:
            return {'status': 'ok', 'result': self.function(**arguments)}
        except Exception as e:
            return {'status': 'error', 'error_type': type(e).__name__, 'message': str(e)}


class Tracer:
    def __init__(self):
        self.spans = []; self.stack = []

    @contextmanager
    def span(self, name, **attrs):
        span = {'name': name, 'start': time.time(), 'attributes': attrs, 'depth': len(self.stack)}
        self.spans.append(span); self.stack.append(span)
        try:
            yield span
            span['status'] = 'ok'
        except Exception as e:
            span['status'] = 'error'; span['error'] = str(e); raise
        finally:
            span['duration_ms'] = (time.time() - span['start']) * 1000
            self.stack.pop()

    def report(self):
        for s in self.spans:
            icon = 'OK' if s.get('status') == 'ok' else 'ER'
            attrs = ' '.join(f'{k}={v}' for k, v in s['attributes'].items())
            print(f"{'  ' * s['depth']}[{icon}] {s['name']:<20} ({s['duration_ms']:.1f}ms)  {attrs}")


# Mock tools
def tool_weather(city):
    return f"18C, partly cloudy in {city}"

def tool_date():
    return "2025-05-22"

TOOLS = {
    'get_weather': Tool('get_weather', 'Weather', tool_weather),
    'get_date':    Tool('get_date',    'Date',    tool_date),
}


def mock_llm(prompt):
    """Mock LLM that drives a 3-turn task."""
    if 'weather' in prompt and 'Tokyo' not in prompt:
        return 'Thought: weather first\nAction: get_weather(city="Tokyo")'
    if 'Tokyo' in prompt and '2025-05-22' not in prompt:
        return 'Thought: date next\nAction: get_date()'
    if '2025-05-22' in prompt:
        return 'Thought: ready to answer\nAction: final_answer("Today in Tokyo: 18C, partly cloudy.")'
    return 'Action: final_answer("Unable to answer.")'


def parse_react(text):
    t = re.search(r'Thought:\s*(.+?)(?=\n|$)', text)
    a = re.search(r'Action:\s*(.+?)(?=\n|$)', text)
    return (t.group(1) if t else '', a.group(1) if a else '')


def parse_action(action_str):
    m = re.match(r'(\w+)\((.*)\)', action_str)
    if not m: return None, {}
    name = m.group(1)
    args_str = m.group(2)
    # Naive: parse "key=value" pairs
    args = {}
    for pair in re.findall(r'(\w+)\s*=\s*"([^"]*)"', args_str):
        args[pair[0]] = pair[1]
    return name, args


class Agent:
    def __init__(self, tools, tracer, max_iterations=8):
        self.tools = tools
        self.tracer = tracer
        self.max_iterations = max_iterations

    def run(self, task):
        """
        Run the ReAct loop with tracing.
        Returns the final answer string.
        """
        # TODO:
        # 1. history = [f"Task: {task}"]
        # 2. For iteration in range(self.max_iterations):
        #      with self.tracer.span("agent_turn", iteration=iteration+1):
        #        with self.tracer.span("llm_call", tokens=420):  # mock attributes
        #          response = mock_llm("\n".join(history))
        #        thought, action = parse_react(response)
        #        history.extend([f"Thought: {thought}", f"Action: {action}"])
        #        if action.startswith('final_answer'):
        #          answer = re.search(r'final_answer\(["\'](.*)["\']\)', action)
        #          return answer.group(1) if answer else "Done"
        #        tool_name, tool_args = parse_action(action)
        #        if tool_name not in self.tools:
        #          history.append(f"Observation: Error - unknown tool {tool_name}")
        #          continue
        #        with self.tracer.span("tool_call", tool=tool_name):
        #          result = self.tools[tool_name].execute(tool_args)
        #        if result['status'] == 'error':
        #          history.append(f"Observation: Error - {result['message']}")
        #        else:
        #          history.append(f"Observation: {result['result']}")
        # 3. Return "Max iterations reached."
        pass


# Test
# tracer = Tracer()
# agent = Agent(TOOLS, tracer)
#
# answer = agent.run("What's the weather and the date today?")
# print(f"\n=== Final answer: {answer}")
# print(f"\n=== Trace:")
# tracer.report()
#
# # Observations:
# # - The agent orchestrates: tools execute safely; tracer records everything;
# #   loop terminates on final_answer or max_iterations
# # - Each component is independent; the agent composes them
# # - Real production agents add: retry on tool errors, cost tracking,
# #   structured logging, observability platform integration
# # - This is the engineering 80% the chapter covers, in compact form

Solution

The ReAct loop: mock the LLM call, parse thought/action, terminate on final_answer, otherwise dispatch to the matching tool and append the observation to history for the next turn. Everything is wrapped in traced spans.

def run(self, task):
    history = [f"Task: {task}"]
    for iteration in range(self.max_iterations):
        with self.tracer.span("agent_turn", iteration=iteration + 1):
            with self.tracer.span("llm_call", tokens=420):
                response = mock_llm("\n".join(history))
            thought, action = parse_react(response)
            history.extend([f"Thought: {thought}", f"Action: {action}"])
            if action.startswith('final_answer'):
                answer = re.search(r'final_answer\(["\'](.*)["\']\)', action)
                return answer.group(1) if answer else "Done"
            tool_name, tool_args = parse_action(action)
            if tool_name not in self.tools:
                history.append(f"Observation: Error - unknown tool {tool_name}")
                continue
            with self.tracer.span("tool_call", tool=tool_name):
                result = self.tools[tool_name].execute(tool_args)
            if result['status'] == 'error':
                history.append(f"Observation: Error - {result['message']}")
            else:
                history.append(f"Observation: {result['result']}")
    return "Max iterations reached."

With agent.run("What's the weather and the date today?"), mock_llm gates its branches on what’s already in the transcript: turn 1’s action text (get_weather(city="Tokyo")) is the first thing to mention Tokyo, so get_weather fires first; turn 2 sees 'Tokyo' in prompt and '2025-05-22' not in prompt and calls get_date; turn 3 sees the date and returns final_answer. The final answer is "Today in Tokyo: 18C, partly cloudy.", and the trace shows all three agent turns with both tools actually called (get_weather then get_date), not a hardcoded answer that skips retrieval. (Note the task string itself must not contain “Tokyo”: if it did, the first mock_llm check would see “Tokyo” already in the prompt on turn 1 and skip straight past get_weather.)

Part IX: chapter map

Chapter status

Chapter	Topic	Status
Ch 27	Agent foundations	done
Ch 28 (this)	Agents from scratch	closing here
Ch 29	Multi-agent systems	next
Ch 30	Agent eval and frameworks	closes the curriculum

What’s left

Ch 29 (Multi-agent systems) covers the cases where a single-agent setup isn’t enough: orchestration patterns, agent-to-agent communication, frameworks (CrewAI, AutoGen, OpenAI Swarm). The composition arc, in code. Ch 30 (Agent eval and frameworks) brings Part VIII’s evaluation discipline back to bear on agent systems, how to quantitatively evaluate agents at scale (SWE-bench, GAIA, OSWorld), the production frameworks that wrap everything in this chapter, and the deployment patterns. Ch 30 closes the curriculum.

The trajectory is short to state. Ch 27 conceptual → Ch 28 engineering (you are here) → Ch 29 composition → Ch 30 evaluation. Each chapter builds on the prior; each delivers a layer of the stack that the next one composes on top of. After Ch 30 the reader has the full picture, from numpy primitives in Ch 1 through transformer internals (Ch 4-6), pretraining (Ch 7-10), alternative architectures (Ch 11-12), post-training (Ch 13-16), inference (Ch 17-19), modern capabilities (Ch 20-23), safety and interpretability and evaluation (Ch 24-26), and finally production agent systems (Ch 27-30). Every layer covered, every layer with its own working code.

Production agents are $80\%$ engineering on top of $20\%$ LLM. This chapter covered the $80\%$ . Tool design, atomic, single-responsibility, clear-contract functions. Tool implementation, sync, async, timeouts, output bounds, sandboxing. Schemas, function calling, MCP, the structured-call pipeline. Error handling, retry, circuit breakers, fallback, surfacing errors to the LLM. Observability, structured logs, distributed traces, debugging from spans. Scaffolding, system prompts, tool descriptions, few-shot examples, output validation, cost budgets. All of it is engineering you’d recognize from any production system, applied with the LLM in mind.

Ch 29 composes multiple agents when a single agent doesn’t fit and orchestration starts to matter. Ch 30 brings Part VIII’s evaluation discipline back to bear on agent systems, and closes the curriculum. You’re now equipped to build a real agent. The next chapters compose them, evaluate them, and bring the whole arc home.