PydanticAI agent loop prevention: circuit breaker and cost cap

PydanticAI, released by the Pydantic team in late 2024, brings Pydantic’s type-safety philosophy to AI agents: tools are typed Python functions, dependencies are injected via a typed RunContext, and model responses are validated against Pydantic models before reaching your application logic. That type safety catches a whole class of bugs at the schema boundary. It does not prevent the LLM from generating semantically valid but practically useless tool calls in a loop. When a PydanticAI agent’s tool returns an empty list, an error model, or a “not found” response that validates successfully against the return type, the LLM has no way to distinguish that from a useful partial result — it generates a follow-up tool call, which produces another empty/error result, and the loop continues until the model’s internal context limit is reached or you notice the API bill. This page shows how to break that loop early and add a dollar cap to any PydanticAI agent without touching your tool definitions or agent configuration.

PydanticAI’s tool validation and the loop failure mode

PydanticAI validates tool inputs and outputs at the type level. If your tool returns list[Document] and the return value is [], that is a valid list[Document] — no validation error, no exception, no interruption. The LLM receives the empty list as a legitimate tool result and must decide what to do next.

In practice, well-prompted agents handle empty results gracefully. The failure mode occurs in three scenarios:

Query rephrasing loops. The agent rephrases the query and calls the same tool again, hoping the re-phrased query returns something. If the data store simply has no relevant documents, every rephrase produces the same empty result. Each rephrase is an LLM call (to generate the rephrased query) plus a tool call.
Validation retry loops. PydanticAI retries model calls when the structured output fails validation (the model returned malformed JSON or a schema mismatch). If the model consistently generates the same invalid output for a given input, the retry loop fires on every iteration. This is bounded by max_retries on the model, but contributes to cost and is worth monitoring.
Multi-step tool dependency loops. When tool A returns an ID that tool B needs, and tool B fails (because the ID is stale), the agent may re-call tool A to get a fresh ID, which is still stale, starting the cycle again. This type of loop crosses two tools and is not caught by single-tool fingerprinting alone.

Wrapping PydanticAI tools with RunGuard

PydanticAI tools can be registered as plain functions or as methods decorated with @agent.tool. RunGuard’s guard() wraps the underlying function and returns a callable with the same signature — compatible with both registration styles.

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from runguard import guard, BudgetTracker, LoopDetectedError, BudgetExceededError

class SearchDeps(BaseModel):
    document_store: any

agent = Agent(
    model="anthropic:claude-sonnet-4-6",
    deps_type=SearchDeps,
    result_type=str,
)

# Define the raw tool function first
def _search_docs(ctx: RunContext[SearchDeps], query: str) -> list[str]:
    return ctx.deps.document_store.search(query)

# Wrap with RunGuard before registration
# guard() preserves the function signature for PydanticAI's introspection
guarded_search = guard(
    _search_docs,
    loop_window=8,
    loop_threshold=2,
    budget=BudgetTracker(max_usd=1.00),
)

# Register the guarded function as the tool
agent.tool(guarded_search)

async def run_query(user_query: str) -> str:
    deps = SearchDeps(document_store=my_store)
    try:
        result = await agent.run(user_query, deps=deps)
        return result.data
    except LoopDetectedError as e:
        return f"Search loop detected ({e.tool_name} repeated {e.count}x). Please try a more specific query."
    except BudgetExceededError as e:
        return f"Query cost limit reached (${e.accumulated_usd:.2f}). Please refine your query."

Using the @agent.tool decorator style

If you prefer PydanticAI’s decorator style, apply RunGuard’s decorator before @agent.tool. Decorator order matters: RunGuard must be the innermost decorator so it sees the actual tool function, not the PydanticAI wrapper.

from runguard import loop_guard  # decorator form of guard()

agent = Agent(model="openai:gpt-4.1", deps_type=SearchDeps)

# Apply decorators bottom-up: loop_guard fires first, then agent.tool registers
@agent.tool
@loop_guard(loop_window=6, loop_threshold=2)
async def fetch_record(ctx: RunContext[SearchDeps], record_id: str) -> dict:
    # Returns {} if not found — valid type, but useless result
    return ctx.deps.document_store.get(record_id) or {}

@agent.tool
@loop_guard(loop_window=6, loop_threshold=2)
async def list_records(ctx: RunContext[SearchDeps], filter_key: str) -> list[dict]:
    return ctx.deps.document_store.list(filter_key)

Detecting cross-tool dependency loops

The query-rephrase loop is caught by fingerprinting a single tool. The cross-tool dependency loop (tool A returns an ID, tool B uses it and fails, tool A is called again) requires RunGuard’s session-level loop detector, which tracks the sequence of tool calls, not just individual tool repeat patterns.

from runguard import SessionGuard

# SessionGuard tracks cross-tool patterns within one agent.run() call
session_guard = SessionGuard(
    sequence_window=6,     # look at last 6 tool calls as a sequence
    sequence_threshold=2,  # trip if same 2-tool sequence repeats twice
)

# Inject into each tool via a shared context object
async def run_with_session_guard(query: str):
    async with session_guard.session() as s:
        deps = SearchDeps(document_store=my_store, run_guard=s)
        return await agent.run(query, deps=deps)

PydanticAI native vs RunGuard: loop prevention comparison

Scenario	PydanticAI native	RunGuard
Single-tool empty-result loop	No detection	LoopDetectedError on 2nd identical fingerprint
Structured output validation retry loop	max_retries (count only)	BudgetTracker fires before retries consume budget
Cross-tool dependency loop (A→B→A)	No detection	SessionGuard detects repeated tool-call sequences
Per-run dollar cap	No	BudgetTracker raises BudgetExceededError
Type-safe tool inputs/outputs	Full Pydantic validation	Fingerprint includes validated value, not raw LLM output
Async support	Full async/await	guard() wraps both sync and async tools transparently

Running PydanticAI in streaming mode

PydanticAI supports streaming responses via agent.run_stream(). Loop detection works the same way in streaming mode: the guard triggers before the LLM call, not after, so the breaker fires before a new streaming response starts rather than interrupting a stream in progress. This is the safest behavior for streaming consumers — they never receive a partial response that cuts off mid-sentence due to a budget trip.

Add loop prevention to your PydanticAI agent today

RunGuard’s Python SDK installs with pip install runguard. Wrap your PydanticAI tool functions with guard() or the @loop_guard decorator, attach a BudgetTracker, and catch LoopDetectedError and BudgetExceededError in your run loop. Full compatibility with PydanticAI’s typed dependency injection and async execution model.

Get started with RunGuard — or see the same pattern for Phidata / Agno, Haystack, and Python agents generally.