Agent Guardrails: Preventing Runaway Loops and Cost Overruns

The 3 AM Bill That Changes How You Build Agents

Somewhere in a Slack channel right now, an engineer is staring at a cloud bill that jumped from $40 to $4,000 overnight. The culprit is almost never a traffic spike. It's an agent that got stuck. A tool call failed, the agent retried, the retry failed the same way, and it retried again — for six hours, unattended, each loop burning a few thousand tokens against a frontier model API. Nobody was watching because nobody thought they needed to.

This is the part of agent engineering that doesn't show up in demos. Demos are short, scripted, and supervised. Production agents run unattended, chain multiple tool calls together, and operate in environments where a single malformed API response can send them into a loop that looks like productive work right up until someone checks the invoice. The fix isn't a smarter model. It's guardrails — explicit, testable limits on iteration count, token spend, wall-clock time, and repeated behavior, enforced in code rather than hoped for in a prompt.

This article walks through the concrete mechanics: how to build a max-iteration cap, how to track and cap token/cost spend mid-run, how to detect an agent going in circles before it burns your budget, and how to combine all three into a guardrail layer you can drop into any agent loop. Every example is runnable Python you can adapt directly.

Why Agents Loop in the First Place

Before writing guardrails, it helps to understand the failure modes they're guarding against. Agent loops aren't random — they cluster into a few repeatable patterns.

• Tool error retry storms. A tool call throws an exception (rate limit, timeout, malformed input), the agent's reasoning concludes "let me try that again," and the model has no built-in sense of "I already tried this three times." Each retry looks locally rational.
• Goal ambiguity oscillation. The agent alternates between two or three strategies because the task's success criteria are underspecified. It writes a file, decides it's wrong, deletes it, rewrites something structurally identical, and repeats.
• Context window drift. As the transcript grows, earlier decisions scroll out of the effective context. The agent "forgets" it already tried an approach and re-derives the same dead end.
• Recursive delegation. In multi-agent systems, agent A spawns agent B to handle a subtask, B decides it needs help and spawns a variant of A, and the delegation graph never terminates because no one node tracks total depth or total spend across the whole tree.
• Silent tool no-ops. A tool call "succeeds" (returns HTTP 200) but does nothing useful — an empty search result, a no-op API call — and the agent reasonably tries variations, none of which change the underlying condition.

None of these require a buggy model. They're structural. An agent that reasons step-by-step, one plausible decision at a time, has no global view of "this session has now cost $80 and accomplished nothing." That global view has to be bolted on from outside the reasoning loop, which is exactly what guardrails are.

Guardrail 1: Hard Iteration Caps

The simplest and most important guardrail is a hard ceiling on the number of agent steps (each model call plus tool execution counts as one iteration). This is non-negotiable — every agent loop should have one, even if you never expect to hit it.

class MaxIterationsExceeded(Exception):
    """Raised when an agent loop exceeds its configured step budget."""
    def __init__(self, iterations, limit):
        self.iterations = iterations
        self.limit = limit
        super().__init__(
            f"Agent exceeded max iterations: {iterations}/{limit}"
        )


class AgentLoopGuard:
    def __init__(self, max_iterations=25):
        self.max_iterations = max_iterations
        self.iteration_count = 0

    def step(self):
        """Call once at the top of every loop iteration."""
        self.iteration_count += 1
        if self.iteration_count > self.max_iterations:
            raise MaxIterationsExceeded(
                self.iteration_count, self.max_iterations
            )
        return self.iteration_count


def run_agent(task, tools, model_client, max_iterations=25):
    guard = AgentLoopGuard(max_iterations=max_iterations)
    transcript = [{"role": "user", "content": task}]

    while True:
        step_number = guard.step()
        response = model_client.generate(transcript, tools=tools)

        if response.is_final_answer:
            return response.content

        tool_result = execute_tool(response.tool_call, tools)
        transcript.append({"role": "assistant", "content": response})
        transcript.append({"role": "tool", "content": tool_result})

        print(f"[step {step_number}/{max_iterations}] "
              f"tool={response.tool_call.name}")

The number you pick matters less than the fact that a number exists. Twenty-five is a reasonable default for a single-agent task with tool use — enough headroom for legitimate multi-step research or debugging, low enough that a stuck loop dies in minutes rather than hours. Tune it per task type: a simple data-lookup agent might cap at 8, a coding agent that needs to run tests and iterate might need 40.

Critically, MaxIterationsExceeded should be a distinct exception type, not a generic error, so your calling code can catch it specifically and decide what to do — alert a human, fall back to a simpler non-agentic path, or return a partial result with an explicit "I hit my step limit" message rather than silently truncating.

Guardrail 2: Token and Cost Budgets

Iteration caps catch loops. They don't catch the case where an agent takes only 6 steps but each step stuffs 50,000 tokens of context into the prompt because it keeps re-reading the same large file. You need a second, independent guardrail tracking actual spend.

class BudgetExceeded(Exception):
    def __init__(self, spent, limit, currency="USD"):
        self.spent = spent
        self.limit = limit
        super().__init__(
            f"Cost budget exceeded: {currency} {spent:.4f} / {limit:.4f}"
        )


# Per-million-token prices; keep this table close to your billing config
# so it doesn't silently drift out of sync with what you're actually charged.
MODEL_PRICING = {
    "claude-sonnet": {"input": 3.00, "output": 15.00},
    "claude-haiku":  {"input": 0.80, "output": 4.00},
}


class CostGuard:
    def __init__(self, max_cost_usd=2.00, model="claude-sonnet"):
        self.max_cost_usd = max_cost_usd
        self.model = model
        self.total_cost = 0.0
        self.total_input_tokens = 0
        self.total_output_tokens = 0

    def record_usage(self, input_tokens, output_tokens):
        pricing = MODEL_PRICING[self.model]
        cost = (
            (input_tokens / 1_000_000) * pricing["input"]
            + (output_tokens / 1_000_000) * pricing["output"]
        )
        self.total_cost += cost
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens

        if self.total_cost > self.max_cost_usd:
            raise BudgetExceeded(self.total_cost, self.max_cost_usd)

        return self.total_cost

    def remaining_budget(self):
        return max(0.0, self.max_cost_usd - self.total_cost)

    def summary(self):
        return {
            "total_cost_usd": round(self.total_cost, 4),
            "input_tokens": self.total_input_tokens,
            "output_tokens": self.total_output_tokens,
        }

Wire this into the same loop as the iteration guard so both limits are checked on every step, not just one:

def run_agent_with_budget(task, tools, model_client,
                            max_iterations=25, max_cost_usd=2.00):
    iter_guard = AgentLoopGuard(max_iterations=max_iterations)
    cost_guard = CostGuard(max_cost_usd=max_cost_usd)
    transcript = [{"role": "user", "content": task}]

    try:
        while True:
            iter_guard.step()
            response = model_client.generate(transcript, tools=tools)

            cost_guard.record_usage(
                response.usage.input_tokens,
                response.usage.output_tokens,
            )

            if response.is_final_answer:
                return response.content, cost_guard.summary()

            tool_result = execute_tool(response.tool_call, tools)
            transcript.append({"role": "assistant", "content": response})
            transcript.append({"role": "tool", "content": tool_result})

    except (MaxIterationsExceeded, BudgetExceeded) as e:
        return f"[stopped: {e}]", cost_guard.summary()

Two things worth calling out. First, record_usage checks the limit *after* accumulating cost from the current step, which means a single step can slightly overshoot the budget — that's fine, and preferable to trying to predict cost before a call completes. Second, always return cost_guard.summary() alongside the result, even on failure. You want that data in your logs whether the agent succeeded, hit a cap, or crashed — it's what lets you tune the limits later instead of guessing.

For agents that call external paid APIs (search, code execution sandboxes, data providers) in addition to the model itself, extend record_usage to accept an arbitrary additional_cost parameter and call it from your tool-execution wrapper too. Model tokens are usually the majority of spend, but not always.

Guardrail 3: Detecting Loops Before You Hit the Cap

Iteration and cost caps are backstops — they guarantee the agent stops eventually. But by the time you hit either, you've already spent the whole budget on nothing. A loop *detector* catches repetition early, often within 3-4 repeated actions, well before either hard cap triggers.

The simplest reliable signal is a hash of the "shape" of each action — tool name plus normalized arguments — compared against recent history.

import hashlib
import json
from collections import deque


class LoopDetector:
    def __init__(self, window_size=6, repeat_threshold=3):
        self.window_size = window_size
        self.repeat_threshold = repeat_threshold
        self.recent_actions = deque(maxlen=window_size)

    def _fingerprint(self, tool_name, tool_args):
        normalized = json.dumps(tool_args, sort_keys=True, default=str)
        raw = f"{tool_name}:{normalized}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def check(self, tool_name, tool_args):
        """Returns True if this action is a repeat that should trip the guard."""
        fingerprint = self._fingerprint(tool_name, tool_args)
        occurrences = sum(1 for f in self.recent_actions if f == fingerprint)
        self.recent_actions.append(fingerprint)
        return occurrences + 1 >= self.repeat_threshold


class LoopDetected(Exception):
    def __init__(self, tool_name, tool_args, occurrences):
        self.tool_name = tool_name
        self.tool_args = tool_args
        super().__init__(
            f"Loop detected: '{tool_name}' called {occurrences}x "
            f"with identical arguments"
        )

A useful refinement: don't just flag *identical* calls, flag *identical failures*. An agent retrying the same search query with slightly different phrasing three times in a row is functionally the same loop as calling it with identical arguments — it's not learning anything from the failure. Track the fingerprint of (tool_name, error_type) on failed calls separately from successful-call fingerprints:

class FailureLoopDetector(LoopDetector):
    def check_failure(self, tool_name, error_type):
        fingerprint = f"fail:{tool_name}:{error_type}"
        occurrences = sum(1 for f in self.recent_actions if f == fingerprint)
        self.recent_actions.append(fingerprint)
        if occurrences + 1 >= self.repeat_threshold:
            raise LoopDetected(tool_name, {"error_type": error_type}, occurrences + 1)

Wire the detector into the main loop right after a tool executes, before the result goes back into the transcript:

def run_agent_full_guard(task, tools, model_client,
                           max_iterations=25, max_cost_usd=2.00):
    iter_guard = AgentLoopGuard(max_iterations=max_iterations)
    cost_guard = CostGuard(max_cost_usd=max_cost_usd)
    loop_detector = FailureLoopDetector(window_size=6, repeat_threshold=3)
    transcript = [{"role": "user", "content": task}]

    try:
        while True:
            iter_guard.step()
            response = model_client.generate(transcript, tools=tools)
            cost_guard.record_usage(
                response.usage.input_tokens, response.usage.output_tokens
            )

            if response.is_final_answer:
                return response.content, cost_guard.summary()

            call = response.tool_call
            if loop_detector.check(call.name, call.arguments):
                raise LoopDetected(call.name, call.arguments, 3)

            try:
                tool_result = execute_tool(call, tools)
            except Exception as e:
                loop_detector.check_failure(call.name, type(e).__name__)
                tool_result = f"error: {e}"

            transcript.append({"role": "assistant", "content": response})
            transcript.append({"role": "tool", "content": tool_result})

    except (MaxIterationsExceeded, BudgetExceeded, LoopDetected) as e:
        return f"[stopped: {e}]", cost_guard.summary()

This is the arrangement that matters most in practice: loop detection catches the common case cheaply (three repeats, stop early), the cost guard catches the case where each individual step is unique but expensive, and the iteration cap is the guaranteed backstop that fires no matter what else slips through. Each one is independent and each one alone would eventually stop a runaway agent — but they trip at very different costs, and you want the cheapest one to fire first.

Timeouts: The Guardrail That Isn't About Tokens

Cost and iteration guardrails assume the failure mode is "too much work." Sometimes the failure mode is "no forward progress but not technically an error" — a tool call that hangs against a flaky network, a subprocess the agent spawned that never exits, a wait-for-condition loop with no timeout. These don't necessarily burn tokens quickly, but they burn wall-clock time, which is its own budget when agents run in CI pipelines or scheduled jobs with billing tied to compute time rather than API calls.

import signal
from contextlib import contextmanager


class AgentTimeout(Exception):
    pass


@contextmanager
def step_timeout(seconds):
    def _handler(signum, frame):
        raise AgentTimeout(f"Step exceeded {seconds}s timeout")

    old_handler = signal.signal(signal.SIGALRM, _handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)


def execute_tool_with_timeout(call, tools, timeout_seconds=30):
    try:
        with step_timeout(timeout_seconds):
            return execute_tool(call, tools)
    except AgentTimeout as e:
        return f"error: tool call timed out after {timeout_seconds}s"

signal.alarm is Unix-only and single-threaded, which is fine for most agent runners but won't work inside async event loops or on Windows — if that's your environment, use asyncio.wait_for around an async tool call instead, or run the tool in a subprocess with subprocess.run(..., timeout=timeout_seconds). The principle carries over regardless of implementation: every tool call gets a hard wall-clock ceiling, and a timeout is treated as a tool failure that feeds into the same failure-loop detector described above, not as a silent retry trigger.

Also set a *session-level* timeout independent of per-step timeouts — the total wall-clock budget for the whole agent run, checked alongside the iteration count. A session that takes 3 minutes per step and never technically errors can still run for hours if nothing else bounds total elapsed time.

Putting It Together: A Guardrail Config Object

In a real codebase, these guards shouldn't be scattered as magic numbers across your agent loop. Centralize them into a single config object that's explicit about defaults, and make it mandatory to pass one in — no guardrail-free code path should exist, even for "quick internal scripts," because those are exactly the ones that get copy-pasted into production later.

from dataclasses import dataclass


@dataclass
class GuardrailConfig:
    max_iterations: int = 25
    max_cost_usd: float = 2.00
    max_session_seconds: int = 600
    step_timeout_seconds: int = 30
    loop_window_size: int = 6
    loop_repeat_threshold: int = 3
    model: str = "claude-sonnet"

    def build_guards(self):
        return {
            "iterations": AgentLoopGuard(self.max_iterations),
            "cost": CostGuard(self.max_cost_usd, self.model),
            "loops": FailureLoopDetector(
                self.loop_window_size, self.loop_repeat_threshold
            ),
        }


# Different task types warrant different ceilings — a one-shot
# classification agent and a multi-file refactor agent should not
# share a config.
RESEARCH_AGENT_CONFIG = GuardrailConfig(
    max_iterations=40, max_cost_usd=5.00, max_session_seconds=1200
)
QUICK_LOOKUP_CONFIG = GuardrailConfig(
    max_iterations=8, max_cost_usd=0.50, max_session_seconds=120
)

This also makes guardrails testable in isolation — you can write a unit test that feeds AgentLoopGuard twenty-six fake steps and asserts it raises on the twenty-sixth, without spinning up a real model call. That test suite is worth having; guardrail code is exactly the kind of thing that gets "temporarily" disabled during a debugging session and never re-enabled.

Observability: Log Every Guard Trip

A guardrail that fires silently is only half a guardrail. When MaxIterationsExceeded, BudgetExceeded, or LoopDetected triggers, that event should be logged with enough structure to answer "why did this happen" without re-running the agent.

import logging
import json

logger = logging.getLogger("agent.guardrails")

def log_guard_trip(guard_type, task_id, details):
    logger.warning(json.dumps({
        "event": "guardrail_triggered",
        "guard_type": guard_type,
        "task_id": task_id,
        "details": details,
    }))

Call this from the except block in your main loop before returning, and feed the logs into whatever dashboard or alert channel your team already watches. In practice, the guard-trip log is what tells you whether your defaults are wrong — if the iteration cap is firing on 15% of runs, it's probably set too low for that task type; if it never fires but the cost guard fires constantly, your per-step token usage is the actual problem, not step count.

What Happens When a Guard Trips: Fail Loud, Not Silent

A guardrail that stops an agent is only half the job. The other half is deciding what the caller sees when it stops, and that decision is easy to get wrong in a way that's worse than having no guardrail at all. The tempting shortcut is to catch the guard exception, swallow it, and return whatever partial answer the agent had assembled so far as if it were a normal successful result. Don't do this. A partial, silently-truncated answer that looks identical to a complete one is how guardrail trips turn into a trust problem downstream — a user or an upstream system acts on a half-finished result believing it's whole.

The better pattern is a distinct response shape for guard-tripped runs, so calling code can branch on it explicitly instead of guessing from string content.

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class AgentResult:
    success: bool
    content: Optional[str]
    stopped_reason: Optional[str] = None
    usage: dict = field(default_factory=dict)

    @property
    def needs_human_review(self):
        return not self.success


def run_agent_safely(task, tools, model_client, config: GuardrailConfig):
    guards = config.build_guards()
    transcript = [{"role": "user", "content": task}]

    try:
        while True:
            guards["iterations"].step()
            response = model_client.generate(transcript, tools=tools)
            guards["cost"].record_usage(
                response.usage.input_tokens, response.usage.output_tokens
            )

            if response.is_final_answer:
                return AgentResult(
                    success=True,
                    content=response.content,
                    usage=guards["cost"].summary(),
                )

            call = response.tool_call
            if guards["loops"].check(call.name, call.arguments):
                raise LoopDetected(call.name, call.arguments, 3)

            tool_result = execute_tool(call, tools)
            transcript.append({"role": "assistant", "content": response})
            transcript.append({"role": "tool", "content": tool_result})

    except (MaxIterationsExceeded, BudgetExceeded, LoopDetected) as e:
        log_guard_trip(type(e).__name__, task_id=id(task), details=str(e))
        return AgentResult(
            success=False,
            content=None,
            stopped_reason=str(e),
            usage=guards["cost"].summary(),
        )

With this shape, a caller can implement an actual escalation policy instead of hoping the text of the response hints at what went wrong: retry once with a fresh transcript and a tighter task description, fall back to a simpler non-agentic code path, or route the task to a human queue with the stopped_reason attached so whoever picks it up knows exactly which guard fired and doesn't have to re-diagnose it from scratch. In a support-ticket triage agent, for instance, a LoopDetected on a specific tool is a strong signal that the tool itself is misbehaving or its schema is confusing the model — that's an actionable bug report, not just a dead task. Treat every guardrail trip as a data point about where your agent's design is weak, not merely as an outage to route around.

Guardrails Are a Design Decision, Not an Afterthought

The pattern across all three guardrails is the same: none of them require the model to be smarter, more careful, or better-prompted. They work because they live outside the model's reasoning, as plain deterministic code that counts, sums, and compares — the one thing language models are structurally bad at doing reliably across a long context. You can ask a model nicely to "stop if you've tried this before," and it will sometimes work. A hash comparison in a while loop works every time.

That reliability is the whole point. Guardrails aren't a defensive afterthought bolted onto a finished agent — they're part of the interface contract for any agent that runs without a human watching every step. An agent without an iteration cap, a cost cap, and a loop detector isn't a finished agent; it's a demo that hasn't met production traffic yet.

If you're building agents that call tools, spawn subagents, or run as MCP servers in a larger pipeline, this same guardrail layer needs to travel with the agent wherever it's deployed — which is exactly the kind of integration detail covered in our Building & Integrating MCP Servers course, where we walk through wiring cost and iteration limits directly into MCP tool boundaries so every agent that connects to a server inherits the same safety rails by default.