Cost per Run, in Cents

Description

The previous two lessons built the pieces separately: estimate_tokens (lesson 03) converts text into a token count; claude-sonnet-5's fixed pricing (lesson 04) converts tokens into cents. This lesson combines them into estimate_cost_cents — the whole module's central formula — and builds cost_for_run, the function that walks a real Reservo run's history and calculates, with a breakdown per tool call, how much that entire run cost.

This is the lesson where Module 2's trace_id and this module's cost meet for the first time: every CostReport this lesson produces is identified by the same deterministic trace_id you already saw in RUN_LOG.jsonl — the same correlation as always, now applied to money.

Connection to the module

This lesson delivers observability/cost_calculator.py's central piece: estimate_cost_cents, StepCost, CostReport, and cost_for_run. It's, precisely, the point where this module stops being preparation and becomes the real answer to the question that opened lesson 01: "how much did this run cost?"

estimate_cost_cents: the complete formula, run for real

With lesson 04's two constants already fixed, the formula combines input and output tokens, each with its own price, and rounds down at the end — the same integer-arithmetic discipline as always:

INPUT_PRICE_CENTS_PER_MILLION_TOKENS = 300    # $3.00 / 1M tokens -- claude-sonnet-5, precio de lista
OUTPUT_PRICE_CENTS_PER_MILLION_TOKENS = 1500  # $15.00 / 1M tokens -- claude-sonnet-5, precio de lista


def estimate_tokens(text):
    return len(text) // 4


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

Notice the order of operations: first, each token amount gets multiplied by its price (two potentially large numbers), then both products get summed, and at the end it's divided by 1_000_000. Reversing that order — dividing before summing, for example — would produce different, incorrect results because of each partial division's rounding down. This is the same formula this guide's DISEÑO fixed from the start; this lesson is simply the first time it actually runs over a complete run.

StepCost and CostReport: the breakdown, structured

A single number of cents per run is useful, but it doesn't say where that cost was spent. StepCost stores an individual tool call's cost; CostReport brings together every StepCost in a run, plus its totals:

from dataclasses import dataclass, field


@dataclass
class StepCost:
    """El costo estimado de UN tool call dentro de un run."""
    step: int
    tool: str
    input_tokens: int
    output_tokens: int
    cost_cents: int


@dataclass
class CostReport:
    """El costo estimado de un run completo, con su desglose por paso."""
    trace_id: str
    question: str
    steps: list = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    cost_cents: int = 0

trace_id is, deliberately, CostReport's first field — the same deterministic identifier from run_logger.py (Module 2), never a new id invented for this module. Any CostReport can be cross-referenced, by that field, against that same run's RUN_LOG.jsonl lines.

cost_for_run: walking history, without touching run_reservo_agent

cost_for_run receives the trace_id (from traced_run, Module 2), the question, and the history run_reservo_agent returns (agent-fundamentals M8, without touching its logic) — the same "wrap, don't rebuild" technique you already used in Module 1 with run_and_observe.

import itertools
import json


def cost_for_run(trace_id, question, history):
    """Recorre history (de run_reservo_agent, SIN tocarlo) y calcula el
    costo estimado del run completo, con desglose por tool call. Input =
    texto que el agente LEE (la pregunta + cada tool_result); output =
    texto que el agente PRODUCE (el input de cada tool_use + el texto
    final)."""
    steps = []
    total_input_text = question
    total_output_text = ""
    pending = {}
    step_counter = itertools.count(1)

    for turn in history:
        content = turn["content"]
        if isinstance(content, str):
            continue  # ya se contó como `question`, arriba
        for block in content:
            if block["type"] == "tool_use":
                step = next(step_counter)
                args_text = json.dumps(block["input"])
                total_output_text += args_text
                pending[block["id"]] = {"step": step, "tool": block["name"], "args_text": args_text}
            elif block["type"] == "tool_result":
                entry = pending[block["tool_use_id"]]
                result_text = block["content"]
                total_input_text += result_text
                step_in = estimate_tokens(result_text)
                step_out = estimate_tokens(entry["args_text"])
                steps.append(StepCost(
                    step=entry["step"], tool=entry["tool"],
                    input_tokens=step_in, output_tokens=step_out,
                    cost_cents=estimate_cost_cents(step_in, step_out),
                ))
            elif block["type"] == "text":
                total_output_text += block["text"]

    input_tokens = estimate_tokens(total_input_text)
    output_tokens = estimate_tokens(total_output_text)
    return CostReport(
        trace_id=trace_id, question=question, steps=steps,
        input_tokens=input_tokens, output_tokens=output_tokens,
        cost_cents=estimate_cost_cents(input_tokens, output_tokens),
    )

Read the classification carefully, because it's the same one you already saw in Module 1, lesson 08 (run_and_observe), applied here in more detail: the text the agent reads — the original question and every tool_result — counts as input; the text the agent produces — every tool_use's arguments (serialized as the model "wrote" them) and the final response — counts as output. pending, a dictionary indexed by the tool_use's id, is what lets each tool_use get paired with its matching tool_result to build a complete StepCost for every step, without assuming they arrive in any particular order.

Notice a real design decision: the run's totals (CostReport's input_tokens, output_tokens) are calculated over the complete concatenated text (total_input_text, total_output_text), applying estimate_tokens once at the end — not by summing each individual StepCost's input_tokens/output_tokens. Lesson 03 already previewed why: summing already-rounded estimates loses precision compared to concatenating first and rounding once. The per-step breakdown (steps) is still useful for seeing where the cost concentrated, but the run's official total is never calculated by summing those already-rounded figures.

Worked example: the cost of Ana's canonical run

Run the usual script — list_rooms → get_quote (rejected) → get_quote (corrected) → book_room — this time wrapped in Module 2's traced_run, and calculate its cost right after:

import logging
import reservo_agent as ra
import run_logger as rl

rl.logger.setLevel(logging.INFO)

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."}]},
]
with rl.traced_run("Reserva Focus pro 3h para Ana", 1) as trace_id:
    final, history = ra.run_reservo_agent("Reserva Focus pro 3h para Ana", script_a)

print()
print("=== costo del run", trace_id, "===")
report = cost_for_run(trace_id, "Reserva Focus pro 3h para Ana", history)
print("input_tokens :", report.input_tokens)
print("output_tokens:", report.output_tokens)
print("cost_cents   :", report.cost_cents)
print()
print("--- desglose por tool call ---")
for s in report.steps:
    print(f"  paso {s.step}: {s.tool:<12} in={s.input_tokens:>3} out={s.output_tokens:>3} cost_cents={s.cost_cents}")

What to expect:

{"seq": 1, "trace_id": "run-8487582448eb", "event": "run_started", "question": "Reserva Focus pro 3h para Ana", "tool_errors": 0, "error": ""}
{"seq": 2, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 1, "tool": "list_rooms", "is_error": false, "content": ""}
{"seq": 4, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 1, "tool": "list_rooms", "is_error": false, "content": "[{\"room\": \"Focus\", \"rate_cents\": 2500}, {\"room\": \"Studio\", \"rate_cents\": 4000}, {\"room\": \"Boardroom\", \"rate_cents\": 8000}]"}
{"seq": 6, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 2, "tool": "get_quote", "is_error": false, "content": ""}
{"seq": 8, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 2, "tool": "get_quote", "is_error": true, "content": "'tier'='premium' no está en enum ['basic', 'pro']"}
{"seq": 10, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 3, "tool": "get_quote", "is_error": false, "content": ""}
{"seq": 12, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 3, "tool": "get_quote", "is_error": false, "content": "{\"price_cents\": 6000}"}
{"seq": 14, "trace_id": "run-8487582448eb", "event": "tool_use", "step": 4, "tool": "book_room", "is_error": false, "content": ""}
{"seq": 16, "trace_id": "run-8487582448eb", "event": "tool_result", "step": 4, "tool": "book_room", "is_error": false, "content": "{\"booking_id\": 1, \"confirmed\": true}"}
{"seq": 18, "trace_id": "run-8487582448eb", "event": "run_finished", "question": "Reserva Focus pro 3h para Ana", "tool_errors": 1, "error": ""}

=== costo del run run-8487582448eb ===
input_tokens : 64
output_tokens: 56
cost_cents   : 0

--- desglose por tool call ---
  paso 1: list_rooms   in= 30 out=  0 cost_cents=0
  paso 2: get_quote    in= 12 out= 12 cost_cents=0
  paso 3: get_quote    in=  5 out= 11 cost_cents=0
  paso 4: book_room    in=  9 out= 15 cost_cents=0

Read the breakdown closely, because it reveals something that isn't obvious before calculating it: step 2 — the get_quote with the invalid tier, rejected by check_input_v2 — cost tokens just like any other step (12 input, 12 output). The error wasn't free. The model (concept) spent output tokens proposing tier="premium", and the system spent input tokens reading the error message that came back. The whole run's cost_cents is still 0 — the honest answer for a run this size, the same one you already saw in Module 1 — but the per-step breakdown already tells you, precisely, that a quarter of this run's total cost was spent on an attempt that ended up rejected.

Common mistakes

1. Summing each StepCost's cost_cents to get the run's total cost. As lesson 03 already warned, summing already-rounded values separately loses precision compared to concatenating the text first and applying estimate_tokens/estimate_cost_cents once. CostReport.cost_cents is calculated over the run's totals, never by summing steps.

2. Thinking a tool_result with is_error: True costs nothing. The worked example disproves it with numbers: step 2, the rejected attempt, cost exactly the same in tokens as a successful step of similar size. An agent that makes a mistake and self-corrects (agent-fundamentals M7) pays for both attempts, not just the one that worked.

3. Forgetting to initialize total_input_text with question. history's first turn is {"role": "user", "content": question} — a string, not a list of blocks — so cost_for_run's for turn in history explicitly skips it (if isinstance(content, str): continue) because it was already counted when initializing total_input_text = question. Forgetting that initialization would leave the original question out of the run's cost.

4. Confusing entry["args_text"] (the tool_use's arguments, already serialized) with result_text (the tool_result's content). The former counts as output (what the model generated when requesting the tool); the latter counts as input (what the system feeds back for the model to read). Reversing this classification would also reverse which part of the cost gets attributed to each direction.

5. Running cost_for_run over the history of a run that raised RuntimeError. As you already saw in Module 1 with run_and_observe, if run_reservo_agent raises an exception, it never reaches a return, and there's no history to capture from outside the call. cost_for_run assumes, as a precondition, that history is the result of a run that did finish — well or badly, but finished; it isn't designed for a run that didn't even get to produce a complete history.

Exercises

Exercise 1: Calculate Sofía's run's cost (Easy)

Run cost_for_run over Sofía's script — list_rooms → get_quote (Boardroom, pro, 1h) → book_room, with no error at all — wrapped in traced_run with sequence_number=2. Confirm input_tokens, output_tokens, and cost_cents.

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."}]},
]
with rl.traced_run("Reserva Boardroom pro 1h para Sofía", 2) as trace_id_b:
    final_b, history_b = ra.run_reservo_agent("Reserva Boardroom pro 1h para Sofía", script_b)

report_b = cost_for_run(trace_id_b, "Reserva Boardroom pro 1h para Sofía", history_b)
print("input_tokens :", report_b.input_tokens)
print("output_tokens:", report_b.output_tokens)
print("cost_cents   :", report_b.cost_cents)

input_tokens : 53
output_tokens: 49
cost_cents   : 0

Exercise 2: Find Ana's run's most expensive step (Medium)

Using the worked example's report (Ana's run), find the StepCost with the highest total tokens (input_tokens + output_tokens), without assuming in advance which one it is.

busiest = max(report.steps, key=lambda s: s.input_tokens + s.output_tokens)
print(f"paso más costoso: paso {busiest.step} ({busiest.tool}), "
      f"{busiest.input_tokens + busiest.output_tokens} tokens totales")

paso más costoso: paso 1 (list_rooms), 30 tokens totales

Exercise 3: Quantify the cost of the error and its correction (Hard)

Run cost_for_run over an error-free version of Ana's task — list_rooms → get_quote (tier="pro" directly, no rejected attempt) → book_room. Compare its input_tokens + output_tokens against the worked example's run (with the error), and confirm the difference matches, exactly, the tokens from the erroring run's step 2.

script_clean = [
    {"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": "pro", "hours": 3}}]},
    {"stop_reason": "tool_use", "content": [
        {"type": "tool_use", "id": "toolu_03", "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."}]},
]
final_clean, history_clean = ra.run_reservo_agent("Reserva Focus pro 3h para Ana", script_clean)
report_clean = cost_for_run("run-clean", "Reserva Focus pro 3h para Ana", history_clean)

total_con_error = report.input_tokens + report.output_tokens          # del ejemplo trabajado
total_sin_error = report_clean.input_tokens + report_clean.output_tokens
diferencia = total_con_error - total_sin_error

paso_2 = next(s for s in report.steps if s.step == 2)
tokens_paso_2 = paso_2.input_tokens + paso_2.output_tokens

print("tokens CON error (run original)   :", total_con_error)
print("tokens SIN error (run limpio)     :", total_sin_error)
print("diferencia                        :", diferencia)
print("tokens del paso 2 (el rechazado)  :", tokens_paso_2)
print("coinciden exactamente             :", diferencia == tokens_paso_2)

tokens CON error (run original)   : 120
tokens SIN error (run limpio)     : 96
diferencia                        : 24
tokens del paso 2 (el rechazado)  : 24
coinciden exactamente             : True

Summary and next step

• We built estimate_cost_cents, StepCost, CostReport, and cost_for_run — the heart of observability/cost_calculator.py, combining estimate_tokens (L03) with the fixed pricing (L04).
• We ran cost_for_run over Ana's canonical run, wrapped in Module 2's traced_run: 64 input tokens, 56 output, 0 cents — with a four-step breakdown that precisely shows where every token concentrated.
• We confirmed, with real execution, that a step rejected on validation (is_error: True) costs tokens just like a successful one — the error isn't free, and Exercise 3 quantified it exactly: 24 tokens, the full cost of a failed attempt and its correction.
• Every CostReport is identified by run_logger.py's same deterministic trace_id — Module 2's correlation, now applied to cost.

Next lesson: 06 — Scaling Cost to Thousands of Runs. With an individual run's cost already calculated — and often 0 cents, the honest answer for runs this size — we scale that figure to batches of 1,000, 10,000, and 100,000 runs, and confirm why summing tokens before rounding is the only correct way to do it.

Additional resources

1. Anthropic — Token counting — The real count of input and output tokens, the reference this lesson measures its own estimate against.
2. Python — dataclasses — StepCost and CostReport, and field(default_factory=list) to avoid the classic shared-mutable-default-value bug.
3. Python — itertools.count — The counter used to number every step inside cost_for_run, the same technique you already saw in run_logger.py.
4. Anthropic — Tool use (function calling) overview — The tool_use/tool_result protocol cost_for_run walks without altering it, exactly as run_and_observe did in Module 1.
5. Python 3.14 — What's New — The version every line of code in this lesson ran on.