Stop an agent that repeats itself

Some agents get stuck calling the same tool again and again. Ask a parcel tracking agent to "keep checking until the parcel is delivered", and it calls the carrier over and over, even though the answer will not change before it replies.

This cookbook adds a hook that counts how many times each tool is called in a turn:

The first two calls to a tool run normally.
From the third call, the hook blocks the call and tells the model to stop and answer.
If the model is refused three times, the hook ends the turn.

An agent's max_iterations also caps a turn, but it counts every step and ends the turn with stop_reason="iteration_limit". This hook caps one tool at a time and tells the model why, so the model can still answer with what it has.

The hook keeps its counts in fields. The SDK saves fields between calls, so the counts are not lost.

Demonstrates: stateful hooks, pre_tool_call and agents.skip, and ending a turn from pre_llm_call.

Run: put the code blocks of steps 1 to 4 in one file, loop_breaker.py. With the SDK installed and MISTRAL_API_KEY set, run python loop_breaker.py.

How it works

turn 1: track_parcel → runs                      calls = 1
        track_parcel → runs                      calls = 2
        track_parcel → blocked, with a message   refusals = 1
        if the model keeps calling:
        ... two more blocked calls ...           refusals = 3
        next model call → blocked: the turn ends with the hook's message
turn 2: track_parcel → runs: each turn starts a new count

turn 1: track_parcel → runs                      calls = 1
        track_parcel → runs                      calls = 2
        track_parcel → blocked, with a message   refusals = 1
        if the model keeps calling:
        ... two more blocked calls ...           refusals = 3
        next model call → blocked: the turn ends with the hook's message
turn 2: track_parcel → runs: each turn starts a new count

1. Fake the carrier

import asyncio

from pydantic import Field

from mistralai import agents

# --- The carrier, faked -------------------------------------------------------------------------
# In production, track_parcel calls the carrier's API. Here the parcel stays in transit, and
# TRACKED records every call that reached the carrier.

TRACKED: list[str] = []

@agents.tool
def track_parcel(tracking_number: str) -> dict[str, str]:
    """The carrier's latest scan of a parcel."""
    print(f"carrier: tracked {tracking_number}")
    TRACKED.append(tracking_number)
    return {"tracking_number": tracking_number, "status": "in transit", "last_scan": "Lyon hub, 06:12"}

import asyncio

from pydantic import Field

from mistralai import agents

# --- The carrier, faked -------------------------------------------------------------------------
# In production, track_parcel calls the carrier's API. Here the parcel stays in transit, and
# TRACKED records every call that reached the carrier.

TRACKED: list[str] = []

@agents.tool
def track_parcel(tracking_number: str) -> dict[str, str]:
    """The carrier's latest scan of a parcel."""
    print(f"carrier: tracked {tracking_number}")
    TRACKED.append(tracking_number)
    return {"tracking_number": tracking_number, "status": "in transit", "last_scan": "Lyon hub, 06:12"}

2. Write the loop breaker

class StopRepeatedCalls(agents.Hook):
    """Refuse a tool once the turn has called it too often; end the turn if the model keeps trying."""

    max_calls: int = 2  # calls allowed to each tool in one turn
    max_refusals: int = 3  # refusals in one turn before the hook ends it
    calls: dict[str, int] = Field(default_factory=dict)  # this turn: tool id -> times called
    refusals: int = 0  # this turn: calls refused

    async def pre_agent_turn(self, ctx, request):
        self.calls = {}
        self.refusals = 0

    async def pre_tool_call(self, ctx, call):
        self.calls[call.tool_id] = self.calls.get(call.tool_id, 0) + 1
        if self.calls[call.tool_id] <= self.max_calls:
            return None
        self.refusals += 1
        print(f"loop breaker: refused {call.tool_id}, call {self.calls[call.tool_id]} of this turn")
        return agents.skip(
            f"You already called {call.tool_name} {self.max_calls} times in this turn, and its result will not "
            "change before the turn ends. Do not call it again: answer with what you know."
        )

    async def pre_llm_call(self, ctx, invocation):
        if self.refusals < self.max_refusals:
            return None
        print("loop breaker: ended the turn")
        return agents.skip("I stopped checking: the status did not change. Ask me again later.")

class StopRepeatedCalls(agents.Hook):
    """Refuse a tool once the turn has called it too often; end the turn if the model keeps trying."""

    max_calls: int = 2  # calls allowed to each tool in one turn
    max_refusals: int = 3  # refusals in one turn before the hook ends it
    calls: dict[str, int] = Field(default_factory=dict)  # this turn: tool id -> times called
    refusals: int = 0  # this turn: calls refused

    async def pre_agent_turn(self, ctx, request):
        self.calls = {}
        self.refusals = 0

    async def pre_tool_call(self, ctx, call):
        self.calls[call.tool_id] = self.calls.get(call.tool_id, 0) + 1
        if self.calls[call.tool_id] <= self.max_calls:
            return None
        self.refusals += 1
        print(f"loop breaker: refused {call.tool_id}, call {self.calls[call.tool_id]} of this turn")
        return agents.skip(
            f"You already called {call.tool_name} {self.max_calls} times in this turn, and its result will not "
            "change before the turn ends. Do not call it again: answer with what you know."
        )

    async def pre_llm_call(self, ctx, invocation):
        if self.refusals < self.max_refusals:
            return None
        print("loop breaker: ended the turn")
        return agents.skip("I stopped checking: the status did not change. Ask me again later.")

pre_agent_turn runs when a turn starts. It resets the counts, because a new question may get a new answer.
pre_tool_call runs before each tool call. It counts the call and blocks it once the tool has been called max_calls times. The model gets the message instead of a result.
pre_llm_call runs before each model call. Once max_refusals calls have been refused, it ends the turn. The run returns Completed with stop_reason="skipped".

The counts are fields, not private attributes: each call of the hook runs on a copy, in your process too, and the SDK carries only fields from one call to the next, per session. When the agent is deployed, the next call can also run in another process.

3. Define the agent

tracker = agents.Agent(
    model="mistral-large-4",
    name="parcel-tracker",
    instructions=(
        "You track parcels with track_parcel. Answer in at most two sentences. You cannot check again later: say so."
    ),
    tools=[track_parcel],
    hooks=[StopRepeatedCalls()],
)

tracker = agents.Agent(
    model="mistral-large-4",
    name="parcel-tracker",
    instructions=(
        "You track parcels with track_parcel. Answer in at most two sentences. You cannot check again later: say so."
    ),
    tools=[track_parcel],
    hooks=[StopRepeatedCalls()],
)

4. Run two turns

def show(state: agents.TurnState[object]) -> None:
    if isinstance(state, agents.Completed):
        print(f"agent ({state.stop_reason or 'answered'}): {state.text}")
    else:
        print(f"agent: {type(state).__name__}")

async def main() -> None:
    async with agents.local_runtime():  # at exit: every local session of the event loop stops
        session = await tracker.start_session()
        print("--- turn 1")
        show(await session.run("Parcel FR-7731 is due today. Keep checking until it is delivered, then tell me."))
        print("--- turn 2")
        show(await session.run("Check FR-7731 once more."))
        await session.cancel()

    print(f"calls that reached the carrier: {len(TRACKED)}")

if __name__ == "__main__":
    asyncio.run(main())

def show(state: agents.TurnState[object]) -> None:
    if isinstance(state, agents.Completed):
        print(f"agent ({state.stop_reason or 'answered'}): {state.text}")
    else:
        print(f"agent: {type(state).__name__}")

async def main() -> None:
    async with agents.local_runtime():  # at exit: every local session of the event loop stops
        session = await tracker.start_session()
        print("--- turn 1")
        show(await session.run("Parcel FR-7731 is due today. Keep checking until it is delivered, then tell me."))
        print("--- turn 2")
        show(await session.run("Check FR-7731 once more."))
        await session.cancel()

    print(f"calls that reached the carrier: {len(TRACKED)}")

if __name__ == "__main__":
    asyncio.run(main())

python loop_breaker.py prints each call that reaches the carrier, the hook's refusal and the agent's answer to each turn:

--- turn 1
carrier: tracked FR-7731
carrier: tracked FR-7731
loop breaker: refused client.track_parcel, call 3 of this turn
agent (answered): I can't keep checking — the tracking result for this turn won't change and I have no way to check again later. As of now, parcel FR-7731 is **in transit**, with its last scan at the Lyon hub at 06:12; it has not yet been delivered.
--- turn 2
carrier: tracked FR-7731
agent (answered): Parcel FR-7731 is still **in transit** — last scan at the Lyon hub at 06:12, not yet delivered.
calls that reached the carrier: 3

--- turn 1
carrier: tracked FR-7731
carrier: tracked FR-7731
loop breaker: refused client.track_parcel, call 3 of this turn
agent (answered): I can't keep checking — the tracking result for this turn won't change and I have no way to check again later. As of now, parcel FR-7731 is **in transit**, with its last scan at the Lyon hub at 06:12; it has not yet been delivered.
--- turn 2
carrier: tracked FR-7731
agent (answered): Parcel FR-7731 is still **in transit** — last scan at the Lyon hub at 06:12, not yet delivered.
calls that reached the carrier: 3

The hook blocked the third call, and the model answered with what it had. The second turn started a new count, so its call ran.

Here the model stopped after one message. A model that keeps calling is refused three times; then pre_llm_call ends the turn before the model reads the third message, and show prints agent (skipped): I stopped checking: the status did not change. Ask me again later.

Next steps

agents.Hook in the API reference: fields, private attributes and the limits on saved values.
Live progress: a hook that pushes each step to your app.