An agent with no budget guardrails is a liability. One misrouted task, one unexpected recursion, one user who found the edge case your system prompt didn't cover — and you're looking at 500 LLM calls you didn't intend to pay for. This happens. The question is whether you built the circuit breaker before or after it happened.
Here are five patterns that actually work in production.
Pattern 1: Hard Token Budgets with Session Tracking
Track input and output tokens across every call in an agent session. Abort the session — gracefully, with a user-facing message — if the total exceeds a threshold you've defined per task type.
from anthropic import Anthropic
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class AgentBudget:
max_input_tokens: int
max_output_tokens: int
max_total_cost_usd: float
input_tokens_used: int = 0
output_tokens_used: int = 0
# Claude Sonnet 4.5 pricing
INPUT_COST_PER_TOKEN: float = 3.0 / 1_000_000
OUTPUT_COST_PER_TOKEN: float = 15.0 / 1_000_000
def record_usage(self, input_tokens: int, output_tokens: int):
self.input_tokens_used += input_tokens
self.output_tokens_used += output_tokens
@property
def cost_usd(self) -> float:
return (
self.input_tokens_used * self.INPUT_COST_PER_TOKEN +
self.output_tokens_used * self.OUTPUT_COST_PER_TOKEN
)
def check(self) -> Optional[str]:
if self.input_tokens_used > self.max_input_tokens:
return f"Input token budget exceeded ({self.input_tokens_used} > {self.max_input_tokens})"
if self.output_tokens_used > self.max_output_tokens:
return f"Output token budget exceeded ({self.output_tokens_used} > {self.max_output_tokens})"
if self.cost_usd > self.max_total_cost_usd:
return f"Cost budget exceeded (${self.cost_usd:.4f} > ${self.max_total_cost_usd:.4f})"
return None
client = Anthropic()
def run_agent_with_budget(
task: str,
tools: list,
budget: AgentBudget
) -> str:
messages = [{"role": "user", "content": task}]
while True:
# Check budget before each call
if violation := budget.check():
return f"Task aborted: {violation}. Partial work may have been completed."
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=messages,
tools=tools
)
budget.record_usage(
response.usage.input_tokens,
response.usage.output_tokens
)
print(f"[budget] cost=${budget.cost_usd:.4f} "
f"input={budget.input_tokens_used} output={budget.output_tokens_used}")
if response.stop_reason == "end_turn":
return response.content[0].text
if response.stop_reason == "tool_use":
# Process tool calls and continue loop
messages = handle_tool_calls(messages, response)
continue
break
return "Task completed"
# Usage: set per-task budgets based on task type
support_budget = AgentBudget(
max_input_tokens=50_000,
max_output_tokens=5_000,
max_total_cost_usd=0.20 # $0.20 max per support ticket
)
Set the budget based on what the task should cost at 2–3x the expected token usage. A support task that normally takes 20K tokens shouldn't be allowed to run to 500K. The budget is your safety net for unexpected loops, not your expected cost.
Pattern 2: Max Iterations Cap
Agents should not run indefinitely. Set a hard cap on the number of LLM calls regardless of token count.
MAX_ITERATIONS = {
"simple_task": 5,
"research_task": 15,
"complex_agent": 30
}
def run_agent_capped(task: str, task_type: str = "simple_task") -> str:
messages = [{"role": "user", "content": task}]
max_iter = MAX_ITERATIONS[task_type]
for iteration in range(max_iter):
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
messages=messages,
tools=tools
)
if response.stop_reason == "end_turn":
return response.content[0].text
if iteration == max_iter - 2:
# Warn the agent it's running out of iterations
messages.append({
"role": "user",
"content": "You have one more iteration. Provide your best answer now with what you have."
})
messages = handle_tool_calls(messages, response)
# Force a final response
final = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=512,
messages=messages + [{
"role": "user",
"content": "Maximum iterations reached. Summarize what you've found so far."
}]
)
return final.content[0].text
The "one more iteration" warning at iteration N-2 is important — it allows the agent to wrap up gracefully rather than being cut off mid-thought with no useful output.
Pattern 3: Model Tier Escalation
Start with a cheap model. Escalate to an expensive model only when needed. This is the opposite of how most teams build agents — they start with the best model and never reconsider.
TIER_MODELS = {
"economy": "claude-haiku-3-5", # $0.80/M input
"standard": "claude-sonnet-4-5", # $3.00/M input
"frontier": "claude-opus-4" # $15.00/M input
}
def tiered_agent(task: str, initial_tier: str = "economy") -> tuple[str, str]:
"""Returns (result, tier_used)."""
for tier in ["economy", "standard", "frontier"]:
if tier < initial_tier: # skip cheaper tiers if we're starting higher
continue
try:
result = attempt_task(task, TIER_MODELS[tier])
quality_ok = assess_quality(task, result) # lightweight quality check
if quality_ok:
return result, tier
print(f"[tier] {tier} insufficient quality, escalating...")
except Exception as e:
print(f"[tier] {tier} failed: {e}, escalating...")
# Final attempt with frontier model, accept whatever it returns
result = attempt_task(task, TIER_MODELS["frontier"])
return result, "frontier"
def assess_quality(task: str, response: str) -> bool:
"""Cheap quality gate using a small model."""
checker = client.messages.create(
model="claude-haiku-3-5", # use cheap model for quality check
max_tokens=5,
messages=[{
"role": "user",
"content": f"Does this response adequately answer the task? Task: {task[:200]}\nResponse: {response[:500]}\nReply: yes or no"
}]
)
return "yes" in checker.content[0].text.lower()
The quality checker itself uses the cheapest model — you don't need a frontier model to decide if a response is adequate. The escalation cost is only paid when economy tier genuinely can't handle the task.
Pattern 4: Structured Termination Conditions
Agents that decide in free-form text whether they're "done" loop longer than necessary. Force explicit termination by requiring a structured output with a finished flag.
import json
TERMINATION_TOOL = {
"name": "task_complete",
"description": "Call this tool ONLY when the task is fully complete. Do not call it speculatively.",
"input_schema": {
"type": "object",
"properties": {
"result": {
"type": "string",
"description": "The final answer or result for the user"
},
"confidence": {
"type": "string",
"enum": ["high", "medium", "low"],
"description": "Your confidence in the completeness of this result"
},
"tokens_could_have_saved": {
"type": "string",
"description": "Optional: note any unnecessary steps you took"
}
},
"required": ["result", "confidence"]
}
}
# Add this to your tools list alongside domain tools
# The agent must call task_complete to finish — no ambiguous "end_turn" endings
When task_complete is called, you know the agent considered the task done and committed to a result. This eliminates the pattern where an agent runs extra iterations "just to be sure" because it doesn't have a clear mechanism to declare completion.
Pattern 5: Tool Call Auditing
Log every tool call with its cost estimate. Alert on unexpected patterns — a tool being called far more than expected, or a specific tool dominating costs.
from collections import defaultdict
import time
class ToolCallAuditor:
def __init__(self):
self.call_counts = defaultdict(int)
self.call_sequence = []
self.start_time = time.time()
def record(self, tool_name: str, input_size: int):
self.call_counts[tool_name] += 1
self.call_sequence.append(tool_name)
# Detect repetition anomalies
if len(self.call_sequence) >= 6:
last_six = self.call_sequence[-6:]
if len(set(last_six)) == 1:
raise RuntimeError(
f"Tool loop detected: '{tool_name}' called 6 times in a row. Aborting."
)
def report(self):
elapsed = time.time() - self.start_time
print(f"\n--- Tool Call Audit ---")
print(f"Total elapsed: {elapsed:.1f}s")
for tool, count in sorted(self.call_counts.items(), key=lambda x: -x[1]):
print(f" {tool}: {count} calls")
print(f"Call sequence: {' -> '.join(self.call_sequence)}")
The repetition check is the most valuable — it catches the common failure mode where an agent gets stuck calling the same tool repeatedly because it's not processing the response correctly. Six identical consecutive calls is almost never intentional.
Putting It Together
These five patterns compound. An agent with: - A $0.20 token budget - A 25-iteration cap - Economy-first tiering - Explicit termination tooling - Tool call auditing
...will essentially never generate a surprise invoice. The worst case is $0.20 and a graceful error message. That's an acceptable operational failure mode.
The investment to implement all five is 2–3 hours of engineering. The cost of not implementing them is discovered when your first runaway task burns $50 on a single user request at 2am on a Saturday.