How Not to Overspend with AI Agents: Budget Guardrails

An agent with no budget guardrails can make hundreds of LLM calls on one task. A misrouted task can start it, and so can an unexpected recursion or a user who hits a case your system prompt did not cover. A guardrail sets how much a task may spend before it starts. It stops the task at that amount. Below are five patterns with Python code for the Anthropic SDK, followed by the provider-side caps that sit above them.

Prices and model names in this article were updated on 31 October 2026.

The code uses Claude Sonnet 5.5 (claude-sonnet-5-5, $2 per million input tokens, $10 per million output) as the default model. The patterns work the same with any vendor's SDK; only the usage field names differ. For a worked example of where agent tokens go, see the hidden token tax.

1. Hard token budgets per session

Track input and output tokens across every call in an agent session. Stop the session when the total passes a threshold you set per task type, and show the user a message they can read.

Count cache reads and cache writes too. Anthropic's usage object reports them in separate fields, and input_tokens covers only the tokens after the last cache breakpoint.

from dataclasses import dataclass
from typing import Optional
from anthropic import Anthropic

client = Anthropic()
MODEL = "claude-sonnet-5-5"

# Claude Sonnet 5.5 list prices, USD per 1M tokens, October 2026
PRICE_PER_M = {"input": 2.00, "cache_write": 2.50, "cache_read": 0.10, "output": 10.00}


@dataclass
class AgentBudget:
    max_input_tokens: int
    max_output_tokens: int
    max_total_cost_usd: float

    input_tokens_used: int = 0   # uncached + cache writes + cache reads
    output_tokens_used: int = 0
    cost_usd: float = 0.0

    def record_usage(self, usage) -> None:
        cache_write = usage.cache_creation_input_tokens or 0
        cache_read = usage.cache_read_input_tokens or 0
        self.input_tokens_used += usage.input_tokens + cache_write + cache_read
        self.output_tokens_used += usage.output_tokens
        self.cost_usd += (
            usage.input_tokens * PRICE_PER_M["input"]
            + cache_write * PRICE_PER_M["cache_write"]
            + cache_read * PRICE_PER_M["cache_read"]
            + usage.output_tokens * PRICE_PER_M["output"]
        ) / 1_000_000

    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


def text_of(response) -> str:
    """Join the text blocks. Responses can also hold thinking and tool_use blocks."""
    return "".join(block.text for block in response.content if block.type == "text")


def handle_tool_calls(messages: list, response) -> list:
    """Append Claude's turn, then one user turn with a tool_result per tool_use block.
    run_tool is your own dispatcher."""
    results = [
        {"type": "tool_result", "tool_use_id": block.id, "content": run_tool(block.name, block.input)}
        for block in response.content
        if block.type == "tool_use"
    ]
    return messages + [
        {"role": "assistant", "content": response.content},
        {"role": "user", "content": results},
    ]


def run_agent_with_budget(task: str, tools: list, budget: AgentBudget) -> str:
    messages = [{"role": "user", "content": task}]

    while True:
        # Check the budget before each call
        if violation := budget.check():
            return f"Task stopped: {violation}. Partial work may have been completed."

        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            messages=messages,
            tools=tools,
        )
        budget.record_usage(response.usage)

        print(f"[budget] cost=${budget.cost_usd:.4f} "
              f"input={budget.input_tokens_used} output={budget.output_tokens_used}")

        if response.stop_reason == "tool_use":
            messages = handle_tool_calls(messages, response)
            continue

        return text_of(response)


# 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 at two to three times the tokens a task should use. A support task that normally takes 20,000 tokens should not be allowed to run to 500,000. The budget is a safety net for loops; the expected cost is a separate, lower number.

The check runs before each call, so the last call can take the total past the cap by its own cost. Take a call with 20,000 input tokens and the full 1,024 output tokens. At Sonnet 5.5 prices it costs 20,000 × $2 / 1M + 1,024 × $10 / 1M = $0.040 + $0.010 = $0.050. Size the cap with that overshoot in mind.

2. A cap on iterations

Agents should not run without limit. Set a hard cap on the number of LLM calls, whatever the token count.

MAX_ITERATIONS = {
    "simple_task": 5,
    "research_task": 15,
    "complex_agent": 30,
}

def run_agent_capped(task: str, tools: list, 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=MODEL,
            max_tokens=1024,
            messages=messages,
            tools=tools,
        )

        if response.stop_reason != "tool_use":
            return text_of(response)

        messages = handle_tool_calls(messages, response)

        if iteration == max_iter - 2:
            # Warn the agent it has one call left. Text goes after the tool_result blocks.
            messages[-1]["content"].append({
                "type": "text",
                "text": "You have one more step. Give your best answer now with what you have.",
            })

    # Cap reached: ask for a summary and allow no more tool calls
    messages[-1]["content"].append({
        "type": "text",
        "text": "Maximum steps reached. Summarise what you have found so far.",
    })
    final = client.messages.create(
        model=MODEL,
        max_tokens=512,
        messages=messages,
        tools=tools,
        tool_choice={"type": "none"},
    )
    return text_of(final)

The warning one step before the cap lets the agent wrap up instead of being cut off with nothing useful to show. Two API rules shape the code. Anthropic's tool-use docs say tool_result blocks "must come FIRST in the content array" of the user turn, with any text after them. So the code appends the warning to that turn instead of sending a new message. The final call keeps the tools list and sets tool_choice to none, which "prevents Claude from using any tools".

3. Model tier escalation

Start with a cheap model. Move to a more expensive one only when a check says the answer is not good enough. The alternative, running every task on the most capable model, pays the top price for tasks a small model can do.

The first step up in the Claude 5.5 family is the big one. Haiku 5.5 is priced at $0.10 / $0.50 per million tokens (prompts up to 100,000 tokens), Sonnet 5.5 at $2 / $10 and Opus 5.5 at $4 / $20. Sonnet is 20 times the price of Haiku, and Opus twice the price of Sonnet.

# Claude 5.5 family, USD per 1M input / output tokens, October 2026
TIERS = [
    ("economy", "claude-haiku-5-5"),    # $0.10 / $0.50 (prompts up to 100K tokens)
    ("standard", "claude-sonnet-5-5"),  # $2 / $10
    ("frontier", "claude-opus-5-5"),    # $4 / $20
]

def tiered_agent(task: str, start_tier: str = "economy") -> tuple[str, str]:
    """Returns (result, tier_used)."""
    names = [name for name, _ in TIERS]
    start = names.index(start_tier)

    for name, model in TIERS[start:-1]:
        try:
            result = attempt_task(task, model)  # your agent loop, run on this model
            if assess_quality(task, result):
                return result, name
            print(f"[tier] {name}: quality check failed, escalating")
        except Exception as e:
            print(f"[tier] {name} failed: {e}, escalating")

    # Top tier: accept whatever it returns
    name, model = TIERS[-1]
    return attempt_task(task, model), name


def assess_quality(task: str, response: str) -> bool:
    """Cheap quality gate on the smallest model."""
    checker = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=1024,  # leave room: thinking tokens count toward max_tokens
        messages=[{
            "role": "user",
            "content": (
                f"Does this response adequately answer the task?\n"
                f"Task: {task[:200]}\nResponse: {response[:500]}\n"
                f"Reply with one word: yes or no"
            ),
        }],
    )
    return text_of(checker).strip().lower().startswith("yes")

The quality check runs on the cheapest model too. Deciding whether an answer is adequate does not need the top tier. You pay for escalation only when the economy tier cannot handle the task. max_tokens sits well above the one-word answer for a reason. Thinking tokens count toward max_tokens (and are billed as output), and thinking is on by default on models such as Haiku 5.5. A limit of 5 tokens can end the response before any text appears. For routing beyond a single vendor, see model routing and the October 2026 pricing snapshot.

4. Structured termination conditions

Agents that decide in free text whether they are done tend to run extra calls. Give them an explicit way to finish, such as a tool they call once with the result.

TERMINATION_TOOL = {
    "name": "task_complete",
    "description": "Call this tool ONLY when the task is fully complete. Do not call it speculatively.",
    "strict": True,
    "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"],
        "additionalProperties": False,
    },
}

# Add this to your tools list alongside the domain tools.
# When the agent calls task_complete, stop the loop and return its "result".

A call to task_complete commits the agent to a result. An agent with no clear way to declare completion runs extra calls "just to be sure", and this tool removes that habit.

You cannot force this call with tool_choice on the newest models. Anthropic's docs say that any and tool "return a 400 error" on Claude Opus 5.5, Sonnet 5.5, Fable 5.1 and Mythos 5.1. The docs recommend auto with strict tool use instead. That is why the definition sets "strict": True and the loop keeps the default auto.

5. Tool call auditing

Log every tool call. Raise an alert when one tool is called far more often than planned, or when it takes most of the spend.

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
        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("\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 catches a common failure, where an agent keeps calling the same tool because it misreads the response. Six identical calls in a row are rarely intended. Call record from handle_tool_calls, once per tool_use block.

Putting it together

The five patterns stack. An agent with:

• a $0.20 cost budget per task
• a 25-call cap
• economy-first tiering
• a task_complete tool
• tool call auditing

has a known worst case per task. It spends $0.20 plus the cost of the one call that crossed the line, then stops with a message the user can read.

Per-task budgets do not protect you from a bug that starts a million tasks. Set a cap at the provider too:

Coding agents ship the same idea. Claude Code has a --max-budget-usd flag for print mode (claude -p). It stops the run once Claude Code's own cost estimate reaches the amount, and that estimate can differ from the bill. The Claude Code cost guide covers it with the other spend controls.

What to do next

1. Log tokens per task, including cache reads and writes, for a week.
2. Set the per-task budget at two to three times the median cost you measured.
3. Add the iteration cap and the task_complete tool in the same change.
4. Set a workspace or project spend limit at the provider.
5. Run the LLM cost audit checklist once a quarter.

Sources

• Claude pricing (platform.claude.com)
• Prompt caching (Claude Platform Docs)
• Handle tool calls (Claude Platform Docs)
• Define tools (Claude Platform Docs)
• Strict tool use (Claude Platform Docs)
• Anthropic workspaces and spend limits
• Anthropic rate limits
• OpenAI usage tiers
• Gemini API rate limits
• Claude Code: manage costs effectively
• Claude Code CLI reference