Cost Optimization on the Claude API

AI adoption is starting to take a familiar shape: you build on AI to stay at the intelligence frontier, but scaling your product makes token costs unsustainable. The zeitgeist refers to this phenomenon as a shift from "tokenmaxxing" to "budgetmaxxing." Yet, the defensible solution isn't to downgrade your models or shrink the scope of your AI use cases. Instead, we see successful teams focus on optimizing the architecture surrounding their models to reach the Pareto frontier of intelligence and cost.

This cookbook lays out the checklist Anthropic's Applied AI team runs builders through to bring their API spend into line without sacrificing product quality. Here's a bird's eye view:

Before you start optimizing – get it working on a capable model, build an eval, and measure a baseline
Prompt caching – reuse your tokens on subsequent turns instead of reprocessing them
Input token management – let the model discover context via tools
Agent-loop efficiency – keep multi-turn context from compounding
Output token management – hand the model tighter generation constraints
Batch API – defer non-interactive workloads to an async queue
Model selection and effort – find the cheapest tier that still clears your quality bar

Notice how model selection sits near the bottom. It's the easiest lever to pull but directly constrains the intelligence of your product. The goal of this cookbook is to maintain your intelligence ceiling while recouping cost where the dollars are not driving any material gains in product performance.

Prerequisites

Note: Running this cookbook top to bottom costs ~$40 in API credits. We recommend only running cells live when you want to experiment with the inputs.

To run this notebook, you will need:

Anthropic API key to call the Claude API (get your API key(opens in new tab))
Python 3.10+ with the anthropic, matplotlib, pillow, and python-dotenv packages
The assets/ directory in the same location as this notebook, which can be pulled from the cookbooks repo(opens in new tab)

Setup

To get started, create a .env file in this directory and drop in your Anthropic API key with ANTHROPIC_API_KEY=your-anthropic-api-key. Then install and import the handful of dependencies, including Anthropic's Python SDK(opens in new tab).

%%capture

%pip install --upgrade anthropic matplotlib pillow python-dotenv

import base64

import copy

import csv

import io

import json

import textwrap

import time

from datetime import datetime

from functools import partial

from pathlib import Path

import anthropic

import matplotlib.pyplot as plt

from dotenv import load_dotenv

from IPython.display import HTML, display

from PIL import Image

load_dotenv()

client = anthropic.Anthropic()

With that in place, define the base models and pricing schemas. We'll use Opus as the default workhorse for this notebook, but your use case might warrant a different intelligence ceiling – the baseline eval in the next section will help us identify the right starting point. Here we just load in some helper functions to make our eval results easily legible.

To optimize cost, we need to ensure we're measuring cost at every turn. Anthropic's API carries a usage block with input, output, and cache token counts. Below we define usage_cost() to turn those raw counts into dollar amounts using the current per-million-token prices(opens in new tab). Note that the PRICING table below is a hardcoded snapshot of list prices at the time of writing, and per-token rates are subject to change as new models ship or introductory pricing expires.

FABLE = "claude-fable-5"

OPUS = "claude-opus-5"

SONNET = "claude-sonnet-5"

HAIKU = "claude-haiku-4-5"

# Our default model for this notebook

MODEL = OPUS

# $/M tokens

PRICING = {

FABLE: {"in": 10.00, "out": 50.00},

OPUS: {"in": 5.00, "out": 25.00},

SONNET: {"in": 2.00, "out": 10.00},

HAIKU: {"in": 1.00, "out": 5.00},

}

def usage_cost(usage, model=MODEL, batch=False):

p = PRICING[model]

m = 1_000_000

# Cache-write tokens may be split by TTL; fall back to the aggregate

cc = getattr(usage, "cache_creation", None)

w5m = getattr(cc, "ephemeral_5m_input_tokens", 0) or 0

w1h = getattr(cc, "ephemeral_1h_input_tokens", 0) or 0

if not (w5m or w1h):

w5m = usage.cache_creation_input_tokens or 0

cost = (

(usage.input_tokens or 0) * p["in"] / m

+ w5m * p["in"] * 1.25 / m

+ w1h * p["in"] * 2.00 / m

+ (usage.cache_read_input_tokens or 0) * p["in"] * 0.10 / m

+ (usage.output_tokens or 0) * p["out"] / m

)

# Advisor iterations bill at the advisor model's rate, separately

for it in getattr(usage, "iterations", None) or []:

if getattr(it, "type", None) == "advisor_message":

cost += usage_cost(it, model=it.model)

return cost * (0.5 if batch else 1.0)

_hdr_printed = False

def print_usage(label, usage, model=MODEL, batch=False, cache=True):

global _hdr_printed

cw = f" {'cache_w':>8s} {'cache_r':>8s}" if cache else ""

if not _hdr_printed:

print(f"{'':32s} {'in':>7s}{cw} {'out':>7s} {'cost':>10s}")

_hdr_printed = True

c = usage_cost(usage, model, batch)

cv = (

(f" {usage.cache_creation_input_tokens or 0:>8d} {usage.cache_read_input_tokens or 0:>8d}")

if cache

else ""

)

print(

f"{label:32s} {usage.input_tokens or 0:>7d}{cv} "

f"{usage.output_tokens or 0:>7d} {f'${c:.4f}':>10s}"

)

return c

def reset_usage_header():

global _hdr_printed

_hdr_printed = False

print(f"{'model':<18s} {'$/M in':>8s} {'$/M out':>9s}")

for m, p in PRICING.items():

tag = " <- our default" if m == MODEL else ""

print(f"{m:<18s} {p['in']:>8.2f} {p['out']:>9.2f}{tag}")

model                $/M in   $/M out
claude-fable-5        10.00     50.00
claude-opus-5          5.00     25.00  <- our default
claude-sonnet-5        2.00     10.00
claude-haiku-4-5       1.00      5.00

# Shared eval output format

def results_table(rows, *, numbered=False, ref="baseline"):

"""rows: [(label, trials)], each trial {"correct", "n", "per_task", "turns"}.

The first row is the cost baseline"""

base = sum(t["per_task"] for t in rows[0][1]) / len(rows[0][1])

head = [

"config",

"trials (pass/n)",

"mean pass",

"requests",

"$ / task",

"$ / 10k tasks",

f"vs {ref}",

]

if numbered:

head.insert(0, "#")

body = []

for i, (label, trials) in enumerate(rows, 1):

n = trials[0]["n"]

passes = [t["correct"] for t in trials]

mean = sum(passes) / (len(passes) * n)

cost = sum(t["per_task"] for t in trials) / len(trials)

turns = sum(t["turns"] for t in trials) / len(trials)

delta = "—" if i == 1 else f"{cost / base - 1:+.0%}"

vals = ([f"[{i:2d}]"] if numbered else []) + [

label,

" · ".join(f"{p}/{n}" for p in passes),

f"{mean:.0%}",

f"{turns:.0f}",

f"${cost:.4f}",

f"${cost * 10_000:,.0f}",

delta,

]

tds = "".join(

f'<td style="white-space:pre;text-align:{"left" if j <= 1 + numbered else "right"}">{v}</td>'

for j, v in enumerate(vals)

)

body.append(f"<tr>{tds}</tr>")

ths = "".join(f"<th>{h}</th>" for h in head)

return HTML(f"<table><thead><tr>{ths}</tr></thead><tbody>{''.join(body)}</tbody></table>")

_TRIAL_COLS = "{cfg:<34}{trial:>6}{passed:>9}{turns:>10}{ptask:>11}{p10k:>15} {misses}"

def trial_header():

print(

_TRIAL_COLS.format(

cfg="config",

trial="trial",

passed="pass/n",

turns="requests",

ptask="$ / task",

p10k="$ / 10k tasks",

misses="misses",

flush=True,

)

def trial_row(label, trial, res, note=""):

"""res: one trial dict — {"correct", "n", "per_task", "turns", "misses"}"""

misses = ", ".join(res.get("misses") or []) or "—"

print(

_TRIAL_COLS.format(

cfg=label,

trial=trial,

passed=f"{res['correct']}/{res['n']}",

turns=res["turns"],

ptask=f"${res['per_task']:.4f}",

p10k=f"${res['per_task'] * 10_000:,.0f}",

misses=misses,

)

+ (f" ({note})" if note else ""),

flush=True,

)

def context_header(width=44):

print(

f"{'':<32s} {'context window':<{width}s} {'cache_w':>10s} {'cache_r':>10s} tokens in context",

flush=True,

)

def context_bar(label, seen, prev=None, cache_w=None, cache_r=None, *, full=180_000, width=44):

"""Per-turn context size: solid bar for tokens the model saw. `prev` shows

the turn-over-turn delta; cache_w/cache_r log how much of the turn was a

cache write vs a cache read."""

solid = min(width, round(seen / full * width))

bar = "█" * solid

delta = f" ({seen - prev:+,})" if prev is not None else ""

cache = f"{cache_w:>10,} {cache_r:>10,} " if cache_w is not None else ""

print(f"{label:<32s} {bar:<{width}s} {cache}{seen:>8,}{delta}", flush=True)

To anchor this cost optimization exercise in reality, we'll run through a use case that embodies what our Applied AI team might face in the field: a claims adjuster agent for Acme Insurance, a fictional auto-insurance company. The cell below loads synthetic data from assets/, which serves as environment context for our agent to work with. It mocks a real production backend without all of the underlying infra.

The agent's system prompt and tool schema (known as its "prefix") are defined in the following cell. The system prompt assigns the agent's persona, specifies the desired response shape, and embeds a full ~12K-token underwriting manual from assets/policy_manual.md. The agent has eight investigation actions it can call on via tools (pull the claim, policy, fraud score, repair estimate, photos) plus four terminal verdicts (approve_claim, deny_claim, route_to_supervisor, refer_to_siu) that the loop watches for. count_tokens at the end shows what this fixed prefix weighs before a single user turn is added.

# Load in mock data from CSVs in the Cookbook's assets/ directory

with open("assets/claims.csv") as f:

CLAIMS = {r["id"]: r for r in csv.DictReader(f)}

with open("assets/repair_parts.csv") as f:

PARTS = list(csv.DictReader(f))

PHOTOS = {}

with open("assets/damage_photos.csv") as f:

for r in csv.DictReader(f):

PHOTOS.setdefault(r["claim_id"], []).append(r)

PRIORS = {}

with open("assets/prior_claims.csv") as f:

for r in csv.DictReader(f):

PRIORS.setdefault(r["policyholder_id"], []).append(

{k: r[k] for k in ("prior_ref", "loss_date", "type", "amount_usd", "shop", "flagged")}

)

# Each tool projects a slice of the wide claims row so payloads stay scoped

_CLAIM_COLS = (

"id",

"policyholder_id",

"loss_date",

"reported_date",

"loss_state",

"description",

"other_parties",

"estimate_usd",

)

_POLICY_COLS = (

"policyholder_id",

"collision",

"comprehensive",

"deductible_usd",

"status",

"inception_date",

"exclusions",

"excluded_drivers",

)

_HIST_COLS = (

"policyholder_id",

"prior_claims_12mo",

"prior_claims_5yr",

"tenure_years",

"on_watch_list",

)

_FRAUD_COLS = ("siu_score", "indicators", "fraud_note")

def _slice(cid, cols):

return json.dumps({k: CLAIMS.get(cid, {}).get(k, "") for k in cols})

def _estimate_text(cid):

"""Fallback itemized estimate: catalog parts filtered to the claim's damage

area (and tier-1 only for minor damage), apportioned to sum to estimate_usd"""

c = CLAIMS.get(cid, {})

total = float(c.get("estimate_usd", 0))

areas = set((c.get("damage_area") or "front").split(";"))

parts = [

for p in PARTS

if p["area"] in areas and (c.get("severity") == "major" or p["tier"] == "1")

] or PARTS

hrs = [float(p["labor_hrs"]) for p in parts]

rows, remaining = [], total

for i, (p, h) in enumerate(zip(parts, hrs, strict=True), 1):

amt = round(total * h / sum(hrs), 2) if i < len(parts) else round(remaining, 2)

remaining -= amt

rows.append(

f"{i:02d} {p['part']:22s} R&R labor {p['labor_hrs']}h "

f"@ ${p['rate_per_hr']}/hr part OEM #{p['oem_no']} ${amt:>9,.2f}"

)

return (

f"CERTIFIED COLLISION CENTER — ESTIMATE OF RECORD\n"

f"Claim {cid} · prepared {CLAIMS.get(cid, {}).get('reported_date', '')}\n\n"

+ "\n".join(rows)

+ f"\n\nSUBTOTAL PARTS+LABOR ${total:,.2f}\n"

"Notes: OEM where available; supplements pending teardown; "

"photos on file; labor rates per §3.4 regional caps."

)

def _photo_metadata(cid):

return "\n".join(

f"IMG_{cid[-3:]}_{r['seq']:>02s}.jpg 3024x4032 {r['timestamp']} appraiser: {r['note']}"

for r in PHOTOS.get(cid, [])

)

def _manual_section(query):

"""Slice the manual by section number ('6', '5.2') or title keyword; unmatched queries get the contents"""

q = str(query).lower().replace("section", "").strip(" .:#")

lines = POLICY_MANUAL.split("\n")

heads = [

(i, len(ln) - len(ln.lstrip("#")), ln.lstrip("# ").removeprefix("Section "))

for i, ln in enumerate(lines)

if ln.startswith(("## ", "### "))

]

for k, (i, level, head) in enumerate(heads):

num, title = head.replace(":", "").split(" ", 1)

if q and (q == num or q in title.lower()):

end = next((j for j, lv, _ in heads[k + 1 :] if lv <= level), len(lines))

return "\n".join(lines[i:end])

return "No matching section. Contents:\n" + "\n".join(h for _, lv, h in heads if lv == 2)

def execute_tool(name, args):

cid = args.get("claim_id", "")

ph = args.get("policyholder_id", "")

if ph and not cid:

cid = next((k for k, v in CLAIMS.items() if v["policyholder_id"] == ph), "")

if name in TERMINAL:

return json.dumps({"recorded": name, **args})

if name == "get_claim":

return _slice(cid, _CLAIM_COLS)

if name == "get_policy":

return _slice(cid, _POLICY_COLS)

if name == "get_customer_history":

hist = {k: CLAIMS.get(cid, {}).get(k, "") for k in _HIST_COLS}

hist["prior_claims"] = PRIORS.get(hist.get("policyholder_id", ""), [])

return json.dumps(hist)

if name == "check_fraud_signals":

return _slice(cid, _FRAUD_COLS)

if name == "lookup_repair_estimate":

doc = Path(f"assets/estimates/{cid}.txt")

return doc.read_text() if doc.exists() else _estimate_text(cid)

if name == "get_damage_photos":

return _photo_metadata(cid) or "No photos on file."

if name == "read_manual":

return _manual_section(args.get("section", ""))

if name == "calculate_payout":

c = CLAIMS.get(cid, {})

est, ded = float(c.get("estimate_usd", 0)), float(c.get("deductible_usd", 0))

return json.dumps(

{"claim_id": cid, "estimate": est, "deductible": ded, "payout": max(est - ded, 0)}

)

return json.dumps({"tool": name, "args": args, "result": "ok"})

POLICY_MANUAL = Path("assets/policy_manual.md").read_text()

SYSTEM_PROMPT = (

"You are a Senior Adjuster in Acme Insurance's auto claims unit; your payment "

"and denial authority is set out in Section 6 of the underwriting manual below. "

"Standard investigation on every claim: retrieve the claim record, the policy, "

"the policyholder's claim history, and the SIU fraud-indicator score before "

"deciding; pull the estimate, photos, or additional documents only when the "

"decision turns on them. Adjudicate strictly per the manual: decide claims within "

"your authority yourself, and escalate only where the manual requires it: "

"`route_to_supervisor` for the supervisor review or higher-authority approval the "

"manual specifies, `refer_to_siu` for the manual's mandatory SIU referrals. Conclude "

"every adjudication by calling exactly one of `approve_claim`, `deny_claim`, "

"`route_to_supervisor`, or `refer_to_siu`.\n\n"

"=== UNDERWRITING MANUAL ===\n"

f"{POLICY_MANUAL}\n"

"=== END MANUAL ==="

)

def _tool(name, description, params):

return {

"name": name,

"description": description,

"input_schema": {

"type": "object",

"properties": {k: {"type": "string", "description": v} for k, v in params.items()},

"required": list(params),

}

TOOLS = [

_tool("get_claim", "Retrieve the full claim record.", {"claim_id": "Claim ID, e.g. CLM-001"}),

_tool("get_policy", "Retrieve the policy bound to a claim.", {"claim_id": "Claim ID"}),

_tool(

"get_customer_history",

"List the policyholder's prior claims.",

{"policyholder_id": "Policyholder ID"},

_tool(

"lookup_repair_estimate",

"Get the itemized repair estimate for a claim.",

{"claim_id": "Claim ID"},

_tool(

"check_fraud_signals",

"Run SIU fraud-indicator scoring on a claim.",

{"claim_id": "Claim ID"},

_tool(

"get_damage_photos", "Retrieve damage-photo metadata for a claim.", {"claim_id": "Claim ID"}

_tool(

"calculate_payout",

"Compute payout after deductible and depreciation.",

{"claim_id": "Claim ID"},

_tool(

"request_docs",

"Request additional documentation from the insured.",

{"claim_id": "Claim ID", "docs": "Comma-separated doc types"},

_tool(

"approve_claim",

"Record an APPROVE decision.",

{"claim_id": "Claim ID", "amount": "Payout in USD"},

_tool(

"deny_claim",

"Record a DENY decision with an exclusion code.",

{"claim_id": "Claim ID", "code": "Exclusion code"},

_tool(

"route_to_supervisor",

"Send the claim up for the supervisor review or higher-authority approval the manual "

"requires (fraud-indicator score in the 3-5 band; payment above your authority).",

{"claim_id": "Claim ID", "reason": "Why"},

_tool(

"refer_to_siu",

"File the mandatory referral to the fraud unit (SIU, form SIU-REF-1) when the manual requires it: "

"fraud-indicator score of 6 or higher, or any Section 5.2 mandatory trigger.",

{"claim_id": "Claim ID", "reason": "Why"},

]

TERMINAL = {

"approve_claim": "APPROVE",

"deny_claim": "DENY",

"route_to_supervisor": "SUPERVISOR",

"refer_to_siu": "FRAUD",

}

n_tokens = client.messages.count_tokens(

model=MODEL,

system=SYSTEM_PROMPT,

tools=TOOLS,

messages=[{"role": "user", "content": "hi"}],

).input_tokens

print(f"--- system prompt ({len(SYSTEM_PROMPT):,} chars) ---\n")

print(SYSTEM_PROMPT[:280] + "...\n")

print(f"--- tools ({len(json.dumps(TOOLS)):,} chars) ---\n")

print(textwrap.fill(", ".join(t["name"] for t in TOOLS), 84))

print(f"\nStatic prefix (system + tools): {n_tokens:,} tokens")

--- system prompt (30,395 chars) ---

You are a Senior Adjuster in Acme Insurance's auto claims unit; your payment and denial authority is set out in Section 6 of the underwriting manual below. Standard investigation on every claim: retrieve the claim record, the policy, the policyholder's claim history, and the SIU ...

--- tools (3,342 chars) ---

get_claim, get_policy, get_customer_history, lookup_repair_estimate,
check_fraud_signals, get_damage_photos, calculate_payout, request_docs,
approve_claim, deny_claim, route_to_supervisor, refer_to_siu

Static prefix (system + tools): 13,027 tokens

Before you start optimizing

First ensure that the agent actually works, regardless of whether or not the unit economics are scalable. If you can't make an agent that solves your problem, you definitely won't be able to make a cost-optimal one. Run the agent on Opus or Fable to start; you should maintain a high intelligence ceiling and pull on other cost levers before sacrificing the agent's overall reasoning capabilities.

Determining to what extent the agent works requires an eval that measures both pass rate and cost per task. We measure cost per task instead of cost per token because a model with a higher sticker price can be the cheaper option if it finishes the job in fewer turns. Pass rate allows us to establish a quality bar such that we don't degrade the quality of the agent itself as we reduce cost per task.

The eval doesn't need to be elaborate; a few representative tasks with known-good answers is enough to catch regressions. Check out this blog post(opens in new tab) for additional direction on building your eval suite. For the purposes of this cookbook, we'll simply use ten claims from claims.csv with human-labeled correct verdicts.

with open("assets/eval_set.csv") as f:

EVAL_CLAIMS = list(csv.DictReader(f))

print(f"Eval set: {len(EVAL_CLAIMS)} claims")

for c in EVAL_CLAIMS:

print(f" {c['id']} → expected {c['expected']}")

Eval set: 10 claims
  CLM-001 → expected APPROVE
  CLM-002 → expected DENY
  CLM-003 → expected APPROVE
  CLM-004 → expected FRAUD
  CLM-005 → expected DENY
  CLM-006 → expected APPROVE
  CLM-007 → expected SUPERVISOR
  CLM-008 → expected SUPERVISOR
  CLM-009 → expected FRAUD
  CLM-010 → expected FRAUD

The adjudicate() function below defines our agent config. We'll run the baseline eval in an unoptimized state – assume this is how we've inherited the agent. It uses Opus at effort=high, full tool loop, no caching, and no context management. Opus High should be able to correctly classify all eval claims with enough intelligence headroom to flex to harder problems. run_eval() walks it across all ten claims and prints the scoreboard.

def adjudicate(

claim,

model=MODEL,

effort="high",

tools=None,

betas=None,

cache=False,

cache_system=False,

system_prompt=None,

system_suffix=None,

max_turns=6,

"""Agentic loop: investigate with tools, conclude via a terminal tool call.

Returns (verdict, total_cost, n_turns)"""

create = client.beta.messages.create if betas else client.messages.create

system = (

f"[trial {getattr(run_eval, 'nonce', 0)}]\n\n"

+ (system_prompt or SYSTEM_PROMPT)

+ (f"\n\n{system_suffix}" if system_suffix else "")

)

extra = {}

# Haiku takes neither thinking nor effort

if model != HAIKU:

extra |= {"thinking": {"type": "adaptive"}, "output_config": {"effort": effort}}

if cache:

extra["cache_control"] = {"type": "ephemeral"}

# Explicit breakpoint so later claims read the static prefix instead of rewriting it

if cache_system:

system = [{"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}]

if betas:

extra["betas"] = betas

messages = [{"role": "user", "content": f"Adjudicate claim {claim['id']}: {claim['summary']}"}]

cost, verdict, turns = 0.0, None, 0

for turns in range(1, max_turns + 1): # noqa: B007

r = create(

model=model,

max_tokens=4096,

system=system,

tools=(tools if tools is not None else TOOLS),

messages=messages,

**extra,

)

cost += usage_cost(r.usage, model)

messages.append({"role": "assistant", "content": r.content})

calls = [b for b in r.content if b.type == "tool_use"]

for tu in calls:

if tu.name in TERMINAL:

verdict = TERMINAL[tu.name]

if verdict or not calls:

break

messages.append(

{

"role": "user",

"content": [

{

"type": "tool_result",

"tool_use_id": tu.id,

"content": execute_tool(tu.name, tu.input),

}

for tu in calls

}

)

return verdict or "NONE", cost, turns

def run_eval(adjudicate_fn, label, *, quiet=False, trial=1):

"""One pass over the eval set. Verbose mode traces every claim (used once,

for the baseline); quiet mode prints a single trial_row(). Either way the

trial's numbers land in run_eval.last for the summary tables"""

# Reset the cache on each trial for a fair comparison

run_eval.nonce = time.time_ns()

if not quiet:

print(label, flush=True)

total, correct, turns, misses = 0.0, 0, 0, []

for claim in EVAL_CLAIMS:

decision, cost, n = adjudicate_fn(claim)

ok = decision == claim["expected"]

total += cost

turns += n

correct += ok

if not ok:

misses.append(claim["id"])

if not quiet:

exp = f"(exp {claim['expected']})"

print(

f" {'✓' if ok else '✗'} {claim['id']} {decision:<10} {exp:<16} "

f"{n} turns ${cost:.4f}",

flush=True,

)

per_task = total / len(EVAL_CLAIMS)

run_eval.last = {

"correct": correct,

"n": len(EVAL_CLAIMS),

"per_task": per_task,

"turns": turns,

"misses": misses,

}

if quiet:

trial_row(label, trial, run_eval.last)

else:

print(

f" → {correct}/{len(EVAL_CLAIMS)} correct · {turns} turns · "

f"${per_task:.4f}/task · ${total:.4f} total\n",

flush=True,

)

return per_task

# Baseline run on Opus High

run_eval(

lambda c: adjudicate(c),

"baseline (opus · effort=high)",

)

BASELINE_TRIAL = dict(run_eval.last)

baseline (opus · effort=high)
  ✓ CLM-001  APPROVE    (exp APPROVE)    4 turns  $0.2844
  ✓ CLM-002  DENY       (exp DENY)       3 turns  $0.2355
  ✓ CLM-003  APPROVE    (exp APPROVE)    4 turns  $0.3117
  ✓ CLM-004  FRAUD      (exp FRAUD)      3 turns  $0.2444
  ✓ CLM-005  DENY       (exp DENY)       3 turns  $0.2451
  ✓ CLM-006  APPROVE    (exp APPROVE)    4 turns  $0.3001
  ✓ CLM-007  SUPERVISOR (exp SUPERVISOR) 4 turns  $0.3305
  ✓ CLM-008  SUPERVISOR (exp SUPERVISOR) 5 turns  $0.4194
  ✓ CLM-009  FRAUD      (exp FRAUD)      3 turns  $0.2643
  ✓ CLM-010  FRAUD      (exp FRAUD)      3 turns  $0.2709
  → 10/10 correct · 36 turns · $0.2906/task · $2.9063 total

Our baseline run lands at 10/10 correct for **~ $0.29/task**. At a million claims a year, every cent off cost per task will save us ~$ 10K in annual spend. But our quality bar is high – when it comes to insurance claims, mis-labels are more costly than inference savings, so we'll hold our quality bar at near-100% accuracy. That means that as we optimize, we'll look to achieve less than ~$0.29/task while getting all 10 eval cases correct.

Note: Model outputs are nondeterministic, so the same configuration can land at a different pass rate and cost per task from one run to the next. Your live eval runs in this notebook may return different results than what we've hardcoded in text. When conducting a cost optimization audit for your own agent, be sure to run many parallel trials so you're making decisions based on a representative distribution rather than a single sample.

Prompt caching

The first cost optimization strategy we'll employ is prompt caching. Right now our system prompt and tool schemas weigh ~13K tokens, and we're paying full input price to reprocess all of it on every turn of every claim. But that content doesn't change, so why should we repeatedly pay full price for sampling it?

That's where prompt caching(opens in new tab) comes in – it lets Claude reuse an already-processed prefix across calls by temporarily storing it server-side. You can cache for five minutes at 1.25× the normal input rate or for one hour at a 2× multiple, and then reprocessing those tokens only costs 0.1× (so long as the prefix block doesn't change). You declare the static prefix by placing a "breakpoint" after it. Prompt caching is a great cost saver in multi-turn conversations and doesn't come at the expense of output quality, which is why we usually pull this lever first.

The simplest way to get started is top-level auto caching(opens in new tab). Pass cache_control={"type": "ephemeral"} on the request and the API places a single breakpoint after the last cacheable block, sliding it forward as the conversation grows. Below, we ask two follow-ups on the same claim with and without auto-caching. Watch the cache_r column – without caching, it stays at zero and every call re-pays for the same system prompt and tool definitions; with caching, call 1 pays a small write premium and call 2 reads back the whole prefix at a tenth of the price.

claim = EVAL_CLAIMS[0]

followups = [

"Is this loss covered under the policy?",

"What payout amount would you recommend?",

]

def converse(label, *, cache=False):

"""Ask each follow-up in a growing conversation. The nonce keeps the prefix

cold: fresh per run when caching, fresh per call when not (so the uncached

variant's call 2 can't read call 1's implicit entry either)"""

run_nonce = f"(cache demo session {time.time_ns()})\n\n"

messages = [{"role": "user", "content": f"Claim {claim['id']}: {claim['summary']}"}]

total = 0.0

for i, q in enumerate(followups, 1):

prefix = run_nonce if cache else f"(cache demo session {time.time_ns()})\n\n"

messages.append({"role": "user", "content": q})

r = client.messages.create(

model=MODEL,

max_tokens=512,

system=prefix + SYSTEM_PROMPT,

tools=TOOLS,

tool_choice={"type": "none"}, # keep tools in the (cached) prefix; answer in prose

messages=messages,

**({"cache_control": {"type": "ephemeral"}} if cache else {}),

)

total += print_usage(f"{label} · claim A · call {i}", r.usage)

messages.append({"role": "assistant", "content": r.content})

return total

# BEFORE — no cache_control, every turn re-processes the full manual + tools

reset_usage_header()

t1 = converse("no caching")

print(f"→ total ${t1:.4f}\n")

# AFTER — top-level auto caching

reset_usage_header()

t2 = converse("auto cache", cache=True)

print(f"→ total ${t2:.4f}")

print(f"→ {(1 - t2 / t1):.0%} cheaper vs no caching")

in  cache_w  cache_r     out       cost
no caching · claim A · call 1      13099        0        0      17    $0.0659
no caching · claim A · call 2      13129        0        0     126    $0.0688
→ total $0.1347

                                      in  cache_w  cache_r     out       cost
auto cache · claim A · call 1          2    13097        0      13    $0.0822
auto cache · claim A · call 2          2       26    13097      82    $0.0088
→ total $0.0910
→ 32% cheaper vs no caching

Auto caching only pays off if the prefix is actually stable. The most common way to "break the cache" is having something dynamic (a timestamp, a request ID, a user name) in the system prompt or a tool description. The prefix might look the same to you, but the cache key is different every call and you'll see cache_read stay at zero.

The fix is to keep the static prefix byte-stable and push anything volatile below the breakpoint. In the cell below, the "before" run stamps datetime.now() into the system prompt and never gets a hit; the "after" run freezes the system prompt by placing an explicit cache_control marker on that block and then moving the timestamp into the user turn instead. The same information reaches the model but now the static content is only paid for once.

# Both variants build off the same base, the only delta is where the timestamp goes

STABLE_PREFIX = f"(byte-stable demo {time.time_ns()})\n\n" + SYSTEM_PROMPT

reset_usage_header()

# Anti-pattern: a dynamic timestamp at the top of the system prompt

def ask_unstable(question):

stamped = f"Current adjuster session started: {datetime.now().isoformat()}\n\n" + STABLE_PREFIX

return client.messages.create(

model=MODEL,

max_tokens=512,

tools=TOOLS,

system=[{"type": "text", "text": stamped, "cache_control": {"type": "ephemeral"}}],

messages=[

{"role": "user", "content": f"Claim {claim['id']}: {claim['summary']}\n\n{question}"}

)

t1 = 0.0

for i, q in enumerate(followups, 1):

t1 += print_usage(f"unstable · claim A · call {i}", ask_unstable(q).usage)

print(f"→ total ${t1:.4f}\n")

reset_usage_header()

# Fix: freeze the system prompt; move volatile context into the user turn

def ask_stable(question):

return client.messages.create(

model=MODEL,

max_tokens=512,

tools=TOOLS,

system=[{"type": "text", "text": STABLE_PREFIX, "cache_control": {"type": "ephemeral"}}],

messages=[

{

"role": "user",

"content": (

f"[session_started={datetime.now().isoformat()}]\n"

f"Claim {claim['id']}: {claim['summary']}\n\n{question}"

}

)

t2 = 0.0

for i, q in enumerate(followups, 1):

t2 += print_usage(f"byte-stable · claim A · call {i}", ask_stable(q).usage)

print(f"→ total ${t2:.4f}")

print(f"→ {(1 - t2 / t1):.0%} cheaper vs unstable prefix")

in  cache_w  cache_r     out       cost
unstable · claim A · call 1          132    12995        0     179    $0.0864
unstable · claim A · call 2          131    12995        0     120    $0.0849
→ total $0.1712

                                      in  cache_w  cache_r     out       cost
byte-stable · claim A · call 1       154    12969        0     177    $0.0863
byte-stable · claim A · call 2       153        0    12969     124    $0.0103
→ total $0.0966
→ 44% cheaper vs unstable prefix

Auto caching gives you one breakpoint that slides. Sometimes you want more control than that, typically when different layers of your prefix change at different rates. Explicit breakpoints(opens in new tab) let you place up to four cache_control markers yourself, so an upstream tier (the policy manual in the system prompt) can keep hitting even when a downstream one (per-claim context) changes.

The cell below asks the same question about three different claims. In the first pass a dynamic per-claim intake packet leads the system block, so the prefix diverges from the very first token. In the second pass the manual comes first with the breakpoint after it and the per-claim packet is an uncached suffix, so the manual is written once on claim 1 and read on claims 2 and 3.

# One question about three claims

MANUAL = f"(intake session {time.time_ns()})\n\n" + SYSTEM_PROMPT

claim_ids = ["CLM-001", "CLM-002", "CLM-003"]

def intake_packet(cid):

"""Full per-claim intake context (~1.1K tokens)"""

return "\n\n".join(

[

f"=== CLAIM {cid} INTAKE ===",

execute_tool("get_claim", {"claim_id": cid}),

execute_tool("get_policy", {"claim_id": cid}),

execute_tool("get_customer_history", {"claim_id": cid}),

execute_tool("check_fraud_signals", {"claim_id": cid}),

execute_tool("get_damage_photos", {"claim_id": cid}),

execute_tool("lookup_repair_estimate", {"claim_id": cid}),

]

)

def run(label, system_for):

reset_usage_header()

total = 0.0

for i, cid in enumerate(claim_ids, 1):

r = client.messages.create(

model=MODEL,

max_tokens=512,

tools=TOOLS,

tool_choice={"type": "none"},

system=system_for(cid),

messages=[{"role": "user", "content": followups[0]}],

)

total += print_usage(f"{label} · claim {i}", r.usage)

print(f"→ total ${total:.4f}")

return total

# BEFORE — variable content first, bytes diverge from the start of the system block

t1 = run(

"variable-first",

lambda cid: [

{

"type": "text",

"text": intake_packet(cid) + "\n\n" + MANUAL,

"cache_control": {"type": "ephemeral"},

}

)

print()

# AFTER — stable manual first with the breakpoint after it

t2 = run(

"stable-first",

lambda cid: [

{"type": "text", "text": MANUAL, "cache_control": {"type": "ephemeral"}},

{"type": "text", "text": intake_packet(cid)},

)

print(f"→ {(1 - t2 / t1):.0%} cheaper vs variable-first")

in  cache_w  cache_r     out       cost
variable-first · claim 1              86    14062        0     226    $0.0940
variable-first · claim 2              86    14036        0     191    $0.0929
variable-first · claim 3              86    13925        0     163    $0.0915
→ total $0.2784

                                      in  cache_w  cache_r     out       cost
stable-first · claim 1              1181    12965        0     101    $0.0895
stable-first · claim 2              1155        0    12965     247    $0.0184
stable-first · claim 3              1044        0    12965     310    $0.0195
→ total $0.1273
→ 54% cheaper vs variable-first

Explicit breakpoints are also useful when you want mixed TTLs(opens in new tab) (a 1-hour cache on persistent content like the policy manual, 5-minute on something ephemeral like conversation history) or when you have a busy agent loop that appends more than ~20 content blocks between calls, since cache lookup only scans back that far from a breakpoint.

If you have a multi-turn agent or reuse the same prompts and tools across conversations, prompt caching can offer significant cost savings. Keep these common cache-breakers in mind to maximize your hit rate:

Placing dynamic content above a breakpoint – Any byte change like timestamps or a new set of tools incurs a new cache write, so you should push dynamic content into the user turn.
Switching models mid-conversation – Caches are per-model, so it resets when you route a turn to a different model string.
Compacting on every turn – Rewriting history invalidates the cache from that point forward, so previous messages should be pruned infrequently instead (more on that in §Agent-loop efficiency).

We recommend tracking your cache hit rate during and after cost optimization to ensure you don't introduce any breaking changes. The Claude Console(opens in new tab) shows your cache read/write token split, or you can log usage.cache_read_input_tokens and usage.cache_creation_input_tokens off each response and watch the ratio yourself.

Input token management

Caching makes repeat tokens cheap, but do we really have to send those tokens to the model in the first place? Ideally the model gets only essential context upfront, and everything else stays within reach if it turns out to matter. This is progressive disclosure, the same way you'd brief a colleague by handing over just enough information for the decision at hand without burying them in context.

Our baseline agent started with all context passed upfront: a full underwriting manual embedded in the system prompt, restated tool definitions, and twelve tool schemas loaded. Let's streamline this so the model only uses the tokens it absolutely needs. We'll call count_tokens(opens in new tab) to measure how prefix bloat changes over time.

tool_recap = "\n\nAVAILABLE TOOLS:\n" + "\n".join(

f"- {t['name']}: {t['description']}" for t in TOOLS

)

bloated = client.messages.count_tokens(

model=MODEL,

system=SYSTEM_PROMPT + tool_recap,

tools=TOOLS,

messages=[{"role": "user", "content": f"Claim {claim['id']}: {claim['summary']}"}],

)

print(f"{'bloated prefix':30s} {bloated.input_tokens:>6,} tokens")

bloated prefix                 13,462 tokens

First we'll move rarely-used reference material behind a tool. The underwriting manual is ~11K tokens sitting in every prefix. Instead, we should expose it via a read_manual tool so Claude pulls the section it needs only when a claim actually turns on a specific rule. The trade-off is that the manual now costs a tool turn when it is needed. If most calls consult it anyway, keeping it in the (cached) system prompt is cheaper; retrieval wins when reference context is large and rarely hit.

SYSTEM_LEAN = (

"You are a Senior Adjuster in Acme Insurance's auto claims unit; your payment "

"and denial authority is set out in Section 6 of the underwriting manual, which "

"you can consult via `read_manual` when you need a specific rule. "