Mini-Project: Wrap a Run and See Inside

Description

Seven lessons built, separately, every piece of vocabulary that was needed: the problem (lesson 03), the four signals (lesson 04), cost and latency calculated for the first time (lesson 05), the boundary of what this guide does and doesn't do (lesson 06), and the complete brief that justifies all of it (lesson 07). This mini-project brings them together in one place: a function, run_and_observe, that wraps run_reservo_agent — without touching a single line of its code — and returns, alongside the usual response, a complete report with the module's four signals.

It's not Module 2's structured logger — there's no trace_id, no JSON format persisted to a file, no per-step events. It's, deliberately, the simplest possible version of "wrapping a run with instrumentation": a function that calls the usual function, measures what it can measure, and hands it to you along with the response. By the end of this lesson, you'll have the first real answer — still minimal — to lesson 07's brief.

Connection to the module

This is the synthesis of the eight lessons. run_and_observe doesn't invent any new technique: it reuses estimate_cost_cents and estimate_run_tokens from lesson 05, the counting logic from lesson 04, and lesson 05's latency model — all together, around run_reservo_agent, run over a batch of real runs.

Worked example: run_and_observe, over a batch of four runs

The function that wraps, without touching the agent

import itertools
import json
import logging
import statistics
from dataclasses import dataclass, asdict

import reservo_agent as ra

logging.basicConfig(level=logging.INFO, format="%(message)s")

INPUT_PRICE_CENTS_PER_MILLION_TOKENS = 300
OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS = 1500

TOOL_LATENCY_MS = {
    "list_rooms": 40,
    "get_quote": 25,
    "book_room": 120,
    "cancel_booking": 90,
}

_run_ids = itertools.count(1)


def estimate_cost_cents(input_tokens, output_tokens):
    return (
        input_tokens * INPUT_PRICE_CENTS_PER_MILLION_TOKENS
        + output_tokens * OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS
    ) // 1_000_000


@dataclass
class RunReport:
    """Todo lo que este módulo puede decir de UN run, en un solo lugar."""
    run_id: int
    question: str
    steps: int
    tool_calls: int
    tool_errors: int
    input_tokens: int
    output_tokens: int
    cost_cents: int
    latency_ms: int
    completed: bool

    @property
    def tool_fail_rate(self):
        return self.tool_errors / self.tool_calls if self.tool_calls else 0.0


def run_and_observe(question, model_script, max_iterations=10):
    """Envuelve run_reservo_agent (agent-fundamentals M8, SIN tocar su
    lógica) y mide lo que la lección 03 mostró que no se podía ver: pasos,
    tool calls, errores, tokens estimados, costo, latencia modelada."""
    run_id = next(_run_ids)
    completed = True
    try:
        final, history = ra.run_reservo_agent(question, model_script, max_iterations=max_iterations)
    except RuntimeError:
        completed = False
        raise

    input_chars = output_chars = 0
    tool_calls = tool_errors = 0
    latency_ms = 0
    tool_use_name = {}
    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            input_chars += len(content)
            continue
        for block in content:
            if block["type"] == "tool_use":
                tool_calls += 1
                output_chars += len(json.dumps(block["input"]))
                tool_use_name[block["id"]] = block["name"]
            elif block["type"] == "tool_result":
                input_chars += len(block["content"])
                if block.get("is_error"):
                    tool_errors += 1
                else:
                    latency_ms += TOOL_LATENCY_MS.get(tool_use_name.get(block["tool_use_id"]), 0)
            elif block["type"] == "text":
                output_chars += len(block["text"])

    input_tokens, output_tokens = input_chars // 4, output_chars // 4
    report = RunReport(
        run_id=run_id, question=question, steps=len(history),
        tool_calls=tool_calls, tool_errors=tool_errors,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
        latency_ms=latency_ms, completed=completed,
    )
    logging.info("run %s completado: %s pasos, %s tool calls (%s errores), %sms, %s centavos",
                 report.run_id, report.steps, report.tool_calls, report.tool_errors,
                 report.latency_ms, report.cost_cents)
    return final, report

Read the signature carefully: run_and_observe(question, model_script, max_iterations=10) — exactly the same parameters as run_reservo_agent. Inside, the first real line is final, history = ra.run_reservo_agent(...) — the only call to the agent's logic in the entire function. Everything that follows — counting tool calls, summing errors, estimating tokens, calculating cost, summing modeled latency — is the wrapper: it reads what run_reservo_agent already produced, it never changes how it produces it. logging.info(...) is the only use of the logging module in this module — a single line per run, no JSON format or trace_id yet: that's, deliberately, Module 2's job.

Running the batch: four tasks, none repeated on purpose

script_a = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Focus", "tier": "premium", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_04", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #1."}]},
]

script_b = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 1}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "book_room",
         "input": {"room": "Boardroom", "tier": "pro", "hours": 1, "member": "Sofía"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Boardroom pro por 1 hora para Sofía. Total $64.00. Confirmación #2."}]},
]

script_c = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "list_rooms", "input": {}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "get_quote",
         "input": {"room": "Focus", "tier": "premium", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 0}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_04", "name": "get_quote",
         "input": {"room": "Focus", "tier": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_05", "name": "book_room",
         "input": {"room": "Focus", "tier": "pro", "hours": 3, "member": "Ana"}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé Focus pro por 3 horas para Ana. Total $60.00. Confirmación #3."}]},
]

script_d = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "book_room",
         "input": {"room": "Studio", "tier": "basic", "hours": 1, "member": "Diego"}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_02", "name": "cancel_booking", "input": {"id": 4}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Reservé y luego cancelé Studio basic 1h para Diego."}]},
]

reports = []
for question, script in [
    ("Reserva Focus pro 3h para Ana", script_a),
    ("Reserva Boardroom pro 1h para Sofía", script_b),
    ("Reserva Focus pro 3h para Ana", script_c),
    ("Reserva y cancela Studio basic 1h para Diego", script_d),
]:
    final, report = run_and_observe(question, script)
    reports.append(report)

print()
print("=== RunReport de cada run ===")
for r in reports:
    print(asdict(r))

print()
print("=== resumen del lote ===")
print("runs                    :", len(reports))
print("tool calls totales      :", sum(r.tool_calls for r in reports))
print("tool errors totales     :", sum(r.tool_errors for r in reports))
print("costo total (centavos)  :", sum(r.cost_cents for r in reports))
print("latencia promedio (ms)  :", round(statistics.mean(r.latency_ms for r in reports), 1))

What to expect:

run 1 completado: 10 pasos, 4 tool calls (1 errores), 185ms, 0 centavos
run 2 completado: 8 pasos, 3 tool calls (0 errores), 185ms, 0 centavos
run 3 completado: 12 pasos, 5 tool calls (2 errores), 185ms, 0 centavos
run 4 completado: 6 pasos, 2 tool calls (0 errores), 210ms, 0 centavos

=== RunReport de cada run ===
{'run_id': 1, 'question': 'Reserva Focus pro 3h para Ana', 'steps': 10, 'tool_calls': 4, 'tool_errors': 1, 'input_tokens': 64, 'output_tokens': 56, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 2, 'question': 'Reserva Boardroom pro 1h para Sofía', 'steps': 8, 'tool_calls': 3, 'tool_errors': 0, 'input_tokens': 53, 'output_tokens': 49, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 3, 'question': 'Reserva Focus pro 3h para Ana', 'steps': 12, 'tool_calls': 5, 'tool_errors': 2, 'input_tokens': 70, 'output_tokens': 67, 'cost_cents': 0, 'latency_ms': 185, 'completed': True}
{'run_id': 4, 'question': 'Reserva y cancela Studio basic 1h para Diego', 'steps': 6, 'tool_calls': 2, 'tool_errors': 0, 'input_tokens': 24, 'output_tokens': 31, 'cost_cents': 0, 'latency_ms': 210, 'completed': True}

=== resumen del lote ===
runs                    : 4
tool calls totales      : 14
tool errors totales     : 3
costo total (centavos)  : 0
latencia promedio (ms)  : 191.2

Read the batch summary with lesson 04's four signals in mind: the batch's error rate is 0% (completed=True on all four RunReports); per-tool failure rate is 3/14 = 21.4%; total cost is 0 cents — again, the honest answer for runs this size, not a bug; average latency, calculated with statistics.mean over the four latency_ms values, is 191.2 ms. Notice the average is not equal to any of the four individual values (185, 185, 185, 210) — it's sensitive to run 4, which ran book_room and cancel_booking instead of the usual list_rooms → get_quote → book_room sequence, and so has a different mix of real tools behind it.

The limit that persists: not even this wrapper saves a run that fails

run_and_observe is a big improvement over having nothing — but it inherits, on purpose, the exact limit lesson 03 named: if run_reservo_agent exhausts the iteration cap, the exception propagates before run_and_observe can build any RunReport, because all the counting logic lives after the call that can fail.

stuck_script = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": f"toolu_0{n}", "name": "list_rooms", "input": {}}]}
    for n in range(1, 4)
]
try:
    final, report = run_and_observe("Reserva algo", stuck_script, max_iterations=2)
except RuntimeError as exc:
    print("RuntimeError capturado en run_and_observe:", exc)

What to expect:

RuntimeError capturado en run_and_observe: max_iterations alcanzado (2)

Not a single RunReport, not a single logging.info, no trace of the attempt at all — exactly what you already saw in lesson 03, now confirmed on this mini-project's complete wrapper. This isn't a flaw in run_and_observe: it's proof that capturing information at the end of a run — no matter how complete the report it builds at the end is — is never enough for the runs you most need to diagnose, the ones that don't reach a clean end. Truly solving this — recording every step as it happens, not at the end — is, precisely, the problem Module 2 (structured logging and tracing a run) builds from its first lesson.

Common mistakes

1. Thinking run_and_observe "is already" Module 2's logger. It isn't — it doesn't persist anything to a file, has no trace_id, and, as you just saw, loses everything if the run fails. It is, deliberately, the simplest possible wrapper: enough to minimally answer lesson 07's brief, insufficient for real production.

2. Modifying run_reservo_agent to make it easier to wrap. No need — and it isn't done in this guide. run_and_observe demonstrates, precisely, that a complete measurement layer can be built without touching a single line of the agent agent-fundamentals already delivered.

3. Calculating tool_fail_rate before summing tool_calls across the whole batch. The individual RunReport exposes tool_fail_rate as a per-run @property — for the whole batch, you need to sum tool_errors and tool_calls across all reports first, and divide after (sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports)), not average each run's individual rate, which gives a different, less accurate number when runs have different numbers of tool calls.

4. Using statistics.mean without being clear on what it hides. This batch's 191.2 ms average is correct, but an average over only four runs can be misleading — three of the four came out to exactly 185 ms, and the fourth (210) shifted the average. With a batch of thousands of real, far more varied runs, a single average hides even more: that's why Module 4 develops percentiles (p50, p95), not just the average.

5. Forgetting that completed=False inside the except RuntimeError never gets to build a RunReport. run_and_observe's code sets completed = False and re-raises the exception — it doesn't swallow it. That's intentional: the function never lies by saying a run completed when it didn't, even though the cost of that honesty is not producing any report for that case.

Exercises

Exercise 1: Calculate the batch's correct per-tool failure rate (Easy)

With the worked example's four RunReports, calculate the batch's per-tool failure rate the correct way (summing before dividing) and compare it with the — incorrect — result of averaging the four individual rates.

correct_rate = sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports)
wrong_rate = statistics.mean(r.tool_fail_rate for r in reports)
print(f"tasa correcta (suma/suma)      : {correct_rate:.1%}")
print(f"tasa incorrecta (promedio de %) : {wrong_rate:.1%}")

tasa correcta (suma/suma)      : 21.4%
tasa incorrecta (promedio de %) : 16.2%

Exercise 2: Add a fifth clean run and observe how the summary changes (Medium)

Run a fifth run — any task with no is_error at all, for example quoting Studio pro 2h without booking anything — with run_and_observe, add it to reports, and recalculate the whole batch's summary (tool calls, tool errors, total cost, average latency).

script_e = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_01", "name": "get_quote",
         "input": {"room": "Studio", "tier": "pro", "hours": 2}}]},
    {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": "Studio pro 2h cuesta $64.00."}]},
]
final_e, report_e = run_and_observe("¿Cuánto cuesta Studio pro 2h?", script_e)
reports.append(report_e)

print("tool calls totales      :", sum(r.tool_calls for r in reports))
print("tool errors totales     :", sum(r.tool_errors for r in reports))
print("tool-fail rate del lote :", f"{sum(r.tool_errors for r in reports) / sum(r.tool_calls for r in reports):.1%}")
print("costo total (centavos)  :", sum(r.cost_cents for r in reports))
print("latencia promedio (ms)  :", round(statistics.mean(r.latency_ms for r in reports), 1))

tool calls totales      : 15
tool errors totales     : 3
tool-fail rate del lote : 20.0%
costo total (centavos)  : 0
latencia promedio (ms)  : 158

Exercise 3: Fix run_and_observe so it survives a RuntimeError with a partial report (Hard)

The limit in "The limit that persists" section is real: today, a RuntimeError produces no RunReport at all. Without building Module 2's complete logger, propose — in code — a minimal change to run_and_observe that catches the RuntimeError, and returns a RunReport marked with completed=False and the rest of the fields at 0, instead of letting the exception propagate with no report at all. Run your version over the worked example's stuck_script and confirm you now do get a report.

def run_and_observe_v2(question, model_script, max_iterations=10):
    run_id = next(_run_ids)
    try:
        final, history = ra.run_reservo_agent(question, model_script, max_iterations=max_iterations)
    except RuntimeError as exc:
        report = RunReport(
            run_id=run_id, question=question, steps=0, tool_calls=0, tool_errors=0,
            input_tokens=0, output_tokens=0, cost_cents=0, latency_ms=0, completed=False,
        )
        logging.info("run %s NO completado: %s", report.run_id, exc)
        return None, report

    # ... el resto es idéntico a run_and_observe.
    input_chars = output_chars = 0
    tool_calls = tool_errors = 0
    latency_ms = 0
    tool_use_name = {}
    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            input_chars += len(content)
            continue
        for block in content:
            if block["type"] == "tool_use":
                tool_calls += 1
                output_chars += len(json.dumps(block["input"]))
                tool_use_name[block["id"]] = block["name"]
            elif block["type"] == "tool_result":
                input_chars += len(block["content"])
                if block.get("is_error"):
                    tool_errors += 1
                else:
                    latency_ms += TOOL_LATENCY_MS.get(tool_use_name.get(block["tool_use_id"]), 0)
            elif block["type"] == "text":
                output_chars += len(block["text"])

    input_tokens, output_tokens = input_chars // 4, output_chars // 4
    report = RunReport(
        run_id=run_id, question=question, steps=len(history),
        tool_calls=tool_calls, tool_errors=tool_errors,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
        latency_ms=latency_ms, completed=True,
    )
    logging.info("run %s completado: %s pasos, %s tool calls (%s errores), %sms, %s centavos",
                 report.run_id, report.steps, report.tool_calls, report.tool_errors,
                 report.latency_ms, report.cost_cents)
    return final, report


stuck_script = [
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": f"toolu_0{n}", "name": "list_rooms", "input": {}}]}
    for n in range(1, 4)
]
final_stuck, report_stuck = run_and_observe_v2("Reserva algo", stuck_script, max_iterations=2)
print(report_stuck)

run N NO completado: max_iterations alcanzado (2)
RunReport(run_id=N, question='Reserva algo', steps=0, tool_calls=0, tool_errors=0, input_tokens=0, output_tokens=0, cost_cents=0, latency_ms=0, completed=False)

Summary and next step

• We built run_and_observe: this guide's first instrumentation wrapper, which calls run_reservo_agent without touching it and returns, alongside the usual response, a RunReport with the module's four signals.
• We ran it over a real batch of four tasks, none repeated from an earlier lesson: 14 total tool calls, 3 errors (21.4% per-tool failure rate), 0 cents total cost, 191.2 ms average latency.
• We confirmed, with real execution, that lesson 03's limit persists even with this wrapper: a RuntimeError still leaves no RunReport at all, because all the measurement happens after a call that can fail before finishing. That is, precisely, the problem Module 2 solves.

With this, Module 1 closes. You have the complete vocabulary — the four signals —, a clear boundary — what's already built, what this guide operates, what's infrastructure —, the brief that justifies the seven remaining modules, and your first instrumentation wrapper, run over real data.

Next module: Module 2 — Structured Logging and Tracing a Run. There, this lesson's deliberately open limit gets fully resolved: a logger that records every step as it happens, with a deterministic trace_id that correlates a run end to end, able to leave a trail even when the entire run ends in RuntimeError.

Additional resources

1. Python — logging — The module this lesson uses minimally, and that Module 2 develops in depth, with its own structured formatter.
2. Python — statistics — statistics.mean, used here over a batch of four; statistics.median and statistics.quantiles are Module 4's central content.
3. Python — dataclasses — RunReport, the structure that brings the four signals together into a single per-run object.
4. Anthropic — Building effective agents — On why measuring an agentic system is a discipline of its own, separate from building it — this module's complete argument, closed.
5. Python 3.14 — What's New — The version every line of code in this module ran on, including this last lesson's real RuntimeError.