Why Every Agent Loop Needs a Step Limit and Spend Cap

Your agent calls a tool, the tool returns a 500, and the model decides to try again. Then again. Nobody is watching, so it keeps going until the context window fills or someone opens the invoice. An agent step limit, paired with a spend cap, is the cheapest insurance you can add, and it only works if it lives in your code, not in the prompt.

By the end you will have a small loop wrapper that enforces three hard limits and returns a clean "stopped" result the caller can act on.

How a runaway AI agent happens

An agent loop is simple: call the model, run the tools it asks for, feed the results back, repeat until the model answers without asking for a tool. Nothing in that loop counts laps. The model alone decides when to stop.

That is a weak design for two reasons. Tool errors come back as text, and a model often reads "request failed" as an invitation to retry. And every step re-sends the conversation so far, so step 200 costs far more than step 2.

Telling the model "give up after five attempts" in the prompt does not fix it. You are asking the stuck component to count its own laps.

Put an agent step limit in code, outside the model

Enforce three counters per task:

• Steps: model calls. Catches slow loops.
• Tool calls: every execution, parallel ones included. Catches a model that fans out ten calls per step, which a step limit alone misses.
• Spend: dollars, computed from the token usage the API reports. Catches the case where a few steps are very expensive.

from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass(frozen=True)
class Limits:
    max_steps: int = 20          # model calls
    max_tool_calls: int = 40     # tool executions, parallel calls included
    max_cost_usd: float = 1.50   # model spend; add paid tool costs to the same tally


@dataclass
class AgentResult:
    status: str                  # "done" or "stopped"
    stop_reason: str | None      # "max_steps", "max_tool_calls", "max_cost_usd"
    output: str | None
    steps: int
    tool_calls: int
    cost_usd: float
    messages: list[dict[str, Any]] = field(default_factory=list)


def run_agent(task: str, call_model: Callable, tools: dict[str, Callable],
              limits: Limits = Limits()) -> AgentResult:
    """call_model(messages) returns an object with .message (dict to append),
    .text, .tool_calls (items with .id, .name, .args) and .cost_usd."""
    messages = [{"role": "user", "content": task}]
    steps = tool_calls = 0
    cost = 0.0

    def stopped(reason: str) -> AgentResult:
        return AgentResult("stopped", reason, None, steps, tool_calls, cost, messages)

    while True:
        if steps >= limits.max_steps:
            return stopped("max_steps")
        if cost >= limits.max_cost_usd:
            return stopped("max_cost_usd")

        turn = call_model(messages)
        steps += 1
        cost += turn.cost_usd

        if not turn.tool_calls:
            messages.append(turn.message)
            return AgentResult("done", None, turn.text, steps, tool_calls, cost, messages)

        if tool_calls + len(turn.tool_calls) > limits.max_tool_calls:
            return stopped("max_tool_calls")

        messages.append(turn.message)
        for call in turn.tool_calls:
            tool_calls += 1
            try:
                content = str(tools[call.name](**call.args))
            except Exception as exc:  # the model sees the error, the counters still tick
                content = f"tool error: {exc}"
            messages.append({"role": "tool", "tool_call_id": call.id, "content": content})

call_model is your adapter for whichever provider you use. The checks run before each model call, so a task can overshoot the cap by one call: set the cap below what you can actually afford.

A failing tool now burns budget at a fixed rate and then stops. If you spawn sub-agents, hand them the same counters, not fresh ones.

Return a clean "stopped" result to the caller

Hitting a limit is an outcome, not a crash. Return a typed result and let the caller decide what happens next.

result = run_agent("Reconcile invoice 4471 against the ERP", call_model, tools)

if result.status == "stopped":
    log.warning("agent stopped: %s after %d steps, $%.2f",
                result.stop_reason, result.steps, result.cost_usd)
    queue_for_human(task_id, result.messages, reason=result.stop_reason)
else:
    save(result.output)

log, queue_for_human and save stand in for your own plumbing. Two rules: never return partial text as if the task succeeded, and keep the transcript so a person, or a retry with a higher limit, can pick up.

Use the agent max iterations your framework already ships

The Claude Agent SDK has max_turns and max_budget_usd, and the docs list both as unlimited by default. When either trips, the final ResultMessage carries the subtype error_max_turns or error_max_budget_usd.

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
    try:
        async for message in query(
            prompt="Find and fix the bug causing test failures in the auth module",
            options=ClaudeAgentOptions(
                allowed_tools=["Read", "Edit", "Bash", "Glob", "Grep"],
                max_turns=30,
                max_budget_usd=2.00,
            ),
        ):
            if isinstance(message, ResultMessage):
                if message.subtype == "success":
                    print(message.result)
                else:
                    print(f"Stopped: {message.subtype}, session {message.session_id}")
    except Exception as error:
        # a single-shot query() raises after yielding an error result
        print(f"Session ended with an error: {error}")


asyncio.run(main())

The docs warn that spend is checked after each response arrives, so you can overshoot by one response, plus any sub-agents still running. Leave headroom.

The OpenAI Agents SDK raises MaxTurnsExceeded when a run passes max_turns. LangGraph has recursion_limit and raises GraphRecursionError. Those cap turns, not dollars, so check whether yours has a spend cap before you rely on it.

The objection: limits cut off legitimate long tasks

Fair, some tasks need many steps. But the limit is a per-task-type setting, not a global one. A refactoring agent and an invoice lookup need different numbers; set each from the step counts of runs that succeeded.

And a stopped run is not a lost run. With the transcript saved you can resume with a higher limit, which is what the Claude Agent SDK docs suggest for error_max_turns.

The real comparison is "task cut off at step 40" against "task still looping at 3 a.m." A job that truly needs hundreds of steps with no checkpoint is one you want to hear about.

First step: set limits from your own logs

Log steps, tool calls and cost per task if you don't already. As a starting rule of thumb, take the longest successful run for each task type, double it, and ship that as the limit. Alert on the stopped rate, then tighten from there.

In production, an agent without a hard limit is a loop you have agreed to pay for without knowing the price, so cap it before it ships.