Mistral AI Agents API Cost Control: Tool Call Loops, Thread Accumulation, Delegation Cascades, and Code Interpreter Spirals

Mistral's Agents API arrived as one of the more thoughtfully designed stateful agent platforms in the current wave — offering persistent conversation threads, a built-in code interpreter, native web search, document retrieval, and first-class multi-agent handoff support. The handoff system in particular is architecturally interesting: agents can delegate to other agents as if they were tool calls, enabling composable multi-agent pipelines without external orchestration frameworks. For teams building customer-support agents, data-analysis pipelines, and multi-step research tasks, the Mistral Agents API offers a compelling combination of model quality and platform primitives in a single SDK.

That same architecture introduces four cost failure modes that are easy to miss until a weekend run produces an unexpected bill. The tool call loop runs until the model decides to terminate — without an explicit step budget, a broad research task can drive 30–60 sequential tool calls before the model surfaces a final answer. Persistent conversation threads inject the full message history into every completion, growing token costs with every turn even when prior context is irrelevant to the current query. Multi-agent handoffs multiply spend: a top-level agent that delegates to three sub-agents, each running their own tool loops, can generate 10–15× the call volume of a single-agent design. The code interpreter closes the loop literally — when code fails, the error is injected back into the conversation and the model generates a corrected version, triggering another execution, which can fail again, adding a full execution trace to the context window on every iteration.

Four failure modes specific to the Mistral AI Agents API:

1. Tool call loop without step budget — client.beta.agents.run() iterates tool calls until the model returns a final text response. With no max_steps parameter enforced at the application layer, a research agent given a broad question will call tools sequentially until it decides it has enough information — which can take 40–60 iterations on under-specified tasks.
2. Conversation thread history accumulation — Mistral Agents persist all messages on a thread_id across runs. Every subsequent run() call on the same thread includes the full prior message history in the context window. A thread with 30 prior turns can add 10,000–20,000 tokens of history overhead before the user's new query appears, compounding costs quadratically as the thread grows.
3. Multi-agent delegation cascade — When a top-level agent uses agent handoffs, each sub-agent runs its own full tool call loop on the delegated sub-task. Delegation trees that aren't bounded by sub-task scope limits or concurrent agent caps turn a single user request into a tree of independently-looping agents, each billing independently for completions and tool calls.
4. Code interpreter retry spiral — The built-in code interpreter executes Python in a sandboxed environment and returns stdout, stderr, and exit codes back into the conversation as tool results. When execution fails, the model reads the error trace, generates a corrected version, which the interpreter runs again. An environment issue (missing library, wrong file path) that isn't correctable from the model's perspective produces an indefinite retry spiral, with each attempt injecting a full execution trace into the context.

Failure mode 1: Tool call loop without step budget

The Mistral Agents API uses a function-calling loop internally: the model returns either a final text response or a tool_call message, the SDK executes the referenced tool, injects the result as a tool role message, and calls the model again. This continues until the model returns a final response. The platform does not enforce a maximum number of tool call iterations at the API level — the contract is that the model will eventually stop, which is a contract that broad, under-specified tasks routinely violate.

A market research agent asked "give me a comprehensive analysis of the top 10 competitors in the enterprise document processing space, including their pricing, API availability, recent funding, and technical limitations" will legitimately call a web search tool 30–50 times before assembling enough information to answer. The model treats thoroughness as its termination condition. Without a step budget, you're paying for however thorough the model decides to be.

The default usage pattern from the Mistral documentation:

from mistralai import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# Creates an agent with web_search tool
agent = client.beta.agents.create(
    model="mistral-large-latest",
    name="research-agent",
    tools=[{"type": "web_search"}],
    instructions="You are a thorough research agent. Investigate topics comprehensively.",
)

thread = client.beta.threads.create()

client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze the top 10 competitors in enterprise document processing — pricing, APIs, funding, limitations.",
)

# This loops until the model decides it's done — could be 40-60 tool calls
run = client.beta.agents.run(
    agent_id=agent.id,
    thread_id=thread.id,
)

The run blocks until completion or a network timeout. During that time, every tool call round-trip bills a full completion at the mistral-large-latest rate. Forty iterations of a 4,000-token context at Mistral Large pricing produces roughly $0.80–$1.20 from a single user query — before accounting for the growing context size as tool results accumulate.

The fix is a step counter that the agent host enforces at the application layer by catching the streaming events or by wrapping the run in a polling loop that terminates after a configurable maximum:

import os
from mistralai import Mistral
from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class MistralStepBudget:
    max_steps: int = 20
    alert_at_steps: int = 15
    steps_taken: int = field(default=0, init=False)
    alert_fired: bool = field(default=False, init=False)

    def record_step(self, agent_id: str, thread_id: str) -> bool:
        """Returns False when budget is exhausted — caller should abort the run."""
        self.steps_taken += 1
        if not self.alert_fired and self.steps_taken >= self.alert_at_steps:
            self.alert_fired = True
            print(
                f"[RunGuard] ALERT — agent {agent_id} on thread {thread_id} "
                f"has taken {self.steps_taken} tool call steps (alert threshold: {self.alert_at_steps})"
            )
        if self.steps_taken >= self.max_steps:
            print(
                f"[RunGuard] CIRCUIT BREAKER — agent {agent_id} on thread {thread_id} "
                f"exhausted {self.max_steps}-step budget. Halting run."
            )
            return False
        return True


def run_with_step_budget(
    client: Mistral,
    agent_id: str,
    thread_id: str,
    budget: Optional[MistralStepBudget] = None,
) -> str:
    if budget is None:
        budget = MistralStepBudget()

    # Poll the run, counting tool call steps
    run = client.beta.agents.run(
        agent_id=agent_id,
        thread_id=thread_id,
        stream=False,
    )

    # For streaming, count requires_action events
    # For synchronous, inspect the run steps after completion
    steps = client.beta.threads.runs.steps.list(
        thread_id=thread_id,
        run_id=run.id,
    )
    for step in steps.data:
        if step.type == "tool_calls":
            if not budget.record_step(agent_id, thread_id):
                # Cancel the run if somehow still running
                try:
                    client.beta.threads.runs.cancel(
                        thread_id=thread_id,
                        run_id=run.id,
                    )
                except Exception:
                    pass
                return f"[RunGuard] Run halted after {budget.steps_taken} tool call steps."

    # Extract the final assistant message
    messages = client.beta.threads.messages.list(thread_id=thread_id)
    for msg in messages.data:
        if msg.role == "assistant":
            return msg.content[0].text.value if msg.content else ""
    return ""


# Usage
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
budget = MistralStepBudget(max_steps=20, alert_at_steps=15)

thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Analyze the top 10 competitors in enterprise document processing.",
)

result = run_with_step_budget(
    client=client,
    agent_id=agent.id,
    thread_id=thread.id,
    budget=budget,
)
print(result)

Failure mode 2: Conversation thread history accumulation

Mistral Agents persist every message — user queries, assistant responses, tool calls, and tool results — on the thread. This is the feature that makes stateful agents useful: you can pick up a conversation days later and the agent has full context of everything said before. The cost implication is that every new run() call on that thread re-sends the entire history to the model as input tokens.

A thread starts cheap: turn 1 might be 500 input tokens. By turn 20, the accumulated history adds 8,000–15,000 tokens of prior context before the new user message appears. By turn 50 — common for long-running support tickets or multi-session research projects — the history overhead alone can exceed 30,000 tokens per completion, which at Mistral Large pricing is $0.045–$0.06 per turn, purely from history that isn't needed for the current query.

The pattern compounds with tool call loops: each tool result also gets stored on the thread, so a 20-step tool loop adds 20 additional messages to the thread, all of which appear in the context of every subsequent run. A thread that has seen ten 20-step research sessions will accumulate 200 historical tool call messages that prepend every future completion.

from mistralai import Mistral
from dataclasses import dataclass

@dataclass
class ThreadTokenEstimate:
    message_count: int
    estimated_input_tokens: int
    alert_threshold_tokens: int

    @property
    def over_threshold(self) -> bool:
        return self.estimated_input_tokens > self.alert_threshold_tokens


def estimate_thread_tokens(
    client: Mistral,
    thread_id: str,
    alert_threshold_tokens: int = 8_000,
    chars_per_token: float = 3.8,  # conservative estimate for English + code
) -> ThreadTokenEstimate:
    """
    Estimate token overhead from thread history before starting a new run.
    Prompt to summarize or start a new thread if over threshold.
    """
    messages = client.beta.threads.messages.list(thread_id=thread_id)
    total_chars = sum(
        len(part.text.value) if hasattr(part, "text") else 0
        for msg in messages.data
        for part in (msg.content if msg.content else [])
    )
    estimated_tokens = int(total_chars / chars_per_token)
    return ThreadTokenEstimate(
        message_count=len(messages.data),
        estimated_input_tokens=estimated_tokens,
        alert_threshold_tokens=alert_threshold_tokens,
    )


def run_with_thread_guard(
    client: Mistral,
    agent_id: str,
    thread_id: str,
    user_message: str,
    max_thread_tokens: int = 8_000,
    summarize_threshold_tokens: int = 6_000,
) -> tuple[str, str]:
    """
    Returns (result, thread_id) — thread_id may change if a new thread was created.
    """
    estimate = estimate_thread_tokens(client, thread_id, alert_threshold_tokens=summarize_threshold_tokens)

    if estimate.over_threshold:
        if estimate.estimated_input_tokens > max_thread_tokens:
            # Start a fresh thread — inject a one-line context summary
            print(
                f"[RunGuard] Thread {thread_id} has ~{estimate.estimated_input_tokens} tokens of history "
                f"(limit {max_thread_tokens}). Starting fresh thread."
            )
            thread = client.beta.threads.create()
            thread_id = thread.id
            # Optionally prepend a summary as the first system message
            # This is application-specific — summarize or discard per use case
        else:
            print(
                f"[RunGuard] ALERT — Thread {thread_id} approaching token limit: "
                f"~{estimate.estimated_input_tokens} tokens across {estimate.message_count} messages."
            )

    client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_message,
    )
    run = client.beta.agents.run(agent_id=agent_id, thread_id=thread_id)
    messages = client.beta.threads.messages.list(thread_id=thread_id)
    result = ""
    for msg in messages.data:
        if msg.role == "assistant":
            result = msg.content[0].text.value if msg.content else ""
            break
    return result, thread_id

For use cases where conversation continuity matters but token costs are a concern, the right pattern is summarization rather than truncation. After every N turns (or when the token estimate crosses a threshold), ask the model to produce a concise summary of the conversation state, start a new thread, and inject the summary as a system message. The new thread starts at near-zero token overhead while preserving semantic continuity.

Failure mode 3: Multi-agent delegation cascade

The Mistral Agents API supports agent handoffs: an agent can list other agents by ID in its tool configuration, and the platform routes the call to the target agent when selected. From the top-level agent's perspective, calling a sub-agent looks identical to calling a function tool — it sends a request and receives a response. The critical difference is that each sub-agent call is itself a full agent run with its own tool call loop, its own thread, and its own billing.

The failure mode occurs in three common patterns. First, an over-delegating top-level agent: the top-level agent's instructions say "use the research-agent for background, the analysis-agent for data, and the writing-agent to compose the response" — even for simple queries that don't need all three. Second, sub-agents that delegate again: a research-agent that can call a web-search-agent and a summarization-agent creates a three-level delegation tree, where the leaf agents each run their own loops. Third, parallel delegation without concurrency caps: if the top-level agent can call multiple sub-agents simultaneously (via parallel tool calls), a single user request can trigger five or six independent agent runs simultaneously, each on their own billing meter.

import threading
from dataclasses import dataclass, field
from typing import Optional
from mistralai import Mistral

# Thread-local delegation depth tracker
_depth = threading.local()

@dataclass
class DelegationPolicy:
    max_depth: int = 3
    max_concurrent_delegates: int = 2
    alert_at_depth: int = 2

    def get_current_depth(self) -> int:
        return getattr(_depth, "value", 0)

    def enter_delegation(self, parent_agent_id: str, child_agent_id: str) -> bool:
        """Returns False if delegation should be blocked."""
        current = self.get_current_depth()
        if current >= self.max_depth:
            print(
                f"[RunGuard] CIRCUIT BREAKER — delegation depth {current} exceeds "
                f"max_depth {self.max_depth}. Blocking {parent_agent_id} → {child_agent_id}."
            )
            return False
        if current >= self.alert_at_depth:
            print(
                f"[RunGuard] ALERT — delegation depth {current + 1} "
                f"({parent_agent_id} → {child_agent_id})"
            )
        _depth.value = current + 1
        return True

    def exit_delegation(self):
        _depth.value = max(0, self.get_current_depth() - 1)


# Semaphore for concurrent delegation cap
_delegation_semaphore: Optional[threading.Semaphore] = None

def get_delegation_semaphore(policy: DelegationPolicy) -> threading.Semaphore:
    global _delegation_semaphore
    if _delegation_semaphore is None:
        _delegation_semaphore = threading.Semaphore(policy.max_concurrent_delegates)
    return _delegation_semaphore


def guarded_agent_run(
    client: Mistral,
    parent_agent_id: str,
    child_agent_id: str,
    thread_id: str,
    policy: DelegationPolicy,
) -> Optional[str]:
    """Wraps a sub-agent run with depth and concurrency guards."""
    sem = get_delegation_semaphore(policy)

    if not policy.enter_delegation(parent_agent_id, child_agent_id):
        return "[RunGuard] Sub-agent call blocked: delegation depth limit reached."

    acquired = sem.acquire(blocking=False)
    if not acquired:
        policy.exit_delegation()
        print(
            f"[RunGuard] CIRCUIT BREAKER — concurrent delegation limit "
            f"({policy.max_concurrent_delegates}) reached. Blocking call to {child_agent_id}."
        )
        return "[RunGuard] Sub-agent call blocked: concurrent delegation limit reached."

    try:
        run = client.beta.agents.run(agent_id=child_agent_id, thread_id=thread_id)
        messages = client.beta.threads.messages.list(thread_id=thread_id)
        for msg in messages.data:
            if msg.role == "assistant":
                return msg.content[0].text.value if msg.content else ""
        return ""
    finally:
        sem.release()
        policy.exit_delegation()


# Example: instrument a top-level agent's tool dispatch
policy = DelegationPolicy(max_depth=3, max_concurrent_delegates=2, alert_at_depth=2)
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

Failure mode 4: Code interpreter retry spiral

Mistral's built-in code interpreter executes Python in a sandboxed environment and returns the output — stdout, stderr, the return value, and any generated files — as tool results back into the conversation. This is designed for data analysis, computation, and file transformation tasks where the agent needs to run code to make progress. It works well when the execution environment is stable. It fails badly when the environment has a problem the model can't fix.

The retry spiral unfolds like this: the model generates Python code to accomplish a task, the interpreter runs it and returns an error (import error for a missing library, file not found, permission denied, memory limit exceeded), the error trace is injected into the conversation as a tool result, the model reads the trace, generates a corrected version of the code (sometimes fixing the actual problem, sometimes generating a semantically-identical script that will fail the same way), the interpreter runs it again. Each iteration adds: one tool call message, one code block, one execution result (which can be hundreds of lines of traceback), all added to the context window for the next completion call.

A missing package that can't be installed in the sandbox produces this loop indefinitely. After 10 iterations, the context window contains 5,000+ tokens of repeated tracebacks and near-identical code attempts, each costing a full completion call at the Mistral Large rate.

import re
from dataclasses import dataclass, field
from collections import deque
from typing import Optional

@dataclass
class CodeInterpreterGuard:
    max_execution_attempts: int = 5
    alert_at_attempts: int = 3
    error_similarity_threshold: float = 0.85
    attempts: int = field(default=0, init=False)
    recent_errors: deque = field(default_factory=lambda: deque(maxlen=5), init=False)

    def _error_signature(self, error_text: str) -> str:
        """Extract error type and first frame — ignores line numbers for similarity check."""
        lines = error_text.strip().splitlines()
        error_type = ""
        for line in lines:
            m = re.match(r"(\w+(?:Error|Exception|Warning))", line)
            if m:
                error_type = m.group(1)
                break
        # Take first 3 non-whitespace lines as a signature
        sig_lines = [l.strip() for l in lines if l.strip()][:3]
        return error_type + "|" + "|".join(sig_lines)

    def _is_repeated_error(self, error_text: str) -> bool:
        sig = self._error_signature(error_text)
        if sig in self.recent_errors:
            return True
        self.recent_errors.append(sig)
        return False

    def record_execution(
        self,
        agent_id: str,
        thread_id: str,
        stderr: str,
        exit_code: int,
    ) -> bool:
        """
        Returns False when the guard should block further code execution attempts.
        Call this after each interpreter tool result is received.
        """
        if exit_code != 0:
            self.attempts += 1
            if self.attempts >= self.alert_at_attempts:
                print(
                    f"[RunGuard] ALERT — agent {agent_id} on thread {thread_id} "
                    f"has failed {self.attempts} code interpreter attempts."
                )
            if self._is_repeated_error(stderr):
                print(
                    f"[RunGuard] CIRCUIT BREAKER — repeated error pattern detected "
                    f"after {self.attempts} attempts. Halting code interpreter loop."
                )
                return False
            if self.attempts >= self.max_execution_attempts:
                print(
                    f"[RunGuard] CIRCUIT BREAKER — {self.attempts} code interpreter "
                    f"attempts exhausted max ({self.max_execution_attempts}). Halting."
                )
                return False
        else:
            # Successful execution resets the attempt counter
            self.attempts = 0
        return True


# Integration: check after each tool result arrives
def handle_tool_result(
    agent_id: str,
    thread_id: str,
    tool_name: str,
    result: dict,
    guard: CodeInterpreterGuard,
) -> Optional[str]:
    """Returns a halt message if the guard trips, otherwise None (continue)."""
    if tool_name != "code_interpreter":
        return None

    exit_code = result.get("exit_code", 0)
    stderr = result.get("stderr", "")

    if not guard.record_execution(
        agent_id=agent_id,
        thread_id=thread_id,
        stderr=stderr,
        exit_code=exit_code,
    ):
        return (
            "The code interpreter encountered a persistent error that cannot be resolved "
            "through further attempts. Please check the execution environment or reformulate "
            "the task without requiring code execution."
        )
    return None

The error similarity check is the key component: many retry spirals produce near-identical error messages on every attempt, even if the model rewrites the code. Detecting repeated error signatures breaks the loop before the context window fills with redundant tracebacks. Combined with a raw attempt counter for errors that vary slightly each time, this covers the two common spiral patterns — stuck errors and slow convergence.

Composite policy: MistralCostPolicy

In production, all four failure modes can occur simultaneously and interact. A multi-agent research task delegates to a sub-agent that runs a long tool call loop while accumulating history on a shared thread, and one of the tools calls the code interpreter which spirals on a broken dependency. Guard each failure mode independently but apply them from a single configuration object so policy tuning happens in one place:

from dataclasses import dataclass, field
from typing import Optional
from mistralai import Mistral

@dataclass
class MistralCostPolicy:
    # Step budget per agent run
    max_tool_steps: int = 20
    alert_at_tool_steps: int = 15

    # Thread history threshold
    max_thread_tokens: int = 12_000
    summarize_threshold_tokens: int = 8_000

    # Multi-agent delegation
    max_delegation_depth: int = 3
    max_concurrent_delegates: int = 2

    # Code interpreter
    max_code_attempts: int = 5
    alert_at_code_attempts: int = 3

    # Runtime state
    _step_budgets: dict = field(default_factory=dict, init=False, repr=False)
    _code_guards: dict = field(default_factory=dict, init=False, repr=False)
    _delegation_policy: Optional[DelegationPolicy] = field(default=None, init=False, repr=False)

    def get_step_budget(self, run_key: str) -> MistralStepBudget:
        if run_key not in self._step_budgets:
            self._step_budgets[run_key] = MistralStepBudget(
                max_steps=self.max_tool_steps,
                alert_at_steps=self.alert_at_tool_steps,
            )
        return self._step_budgets[run_key]

    def get_code_guard(self, thread_id: str) -> CodeInterpreterGuard:
        if thread_id not in self._code_guards:
            self._code_guards[thread_id] = CodeInterpreterGuard(
                max_execution_attempts=self.max_code_attempts,
                alert_at_attempts=self.alert_at_code_attempts,
            )
        return self._code_guards[thread_id]

    def get_delegation_policy(self) -> DelegationPolicy:
        if self._delegation_policy is None:
            self._delegation_policy = DelegationPolicy(
                max_depth=self.max_delegation_depth,
                max_concurrent_delegates=self.max_concurrent_delegates,
            )
        return self._delegation_policy


# Single config object per application
policy = MistralCostPolicy(
    max_tool_steps=20,
    alert_at_tool_steps=15,
    max_thread_tokens=12_000,
    summarize_threshold_tokens=8_000,
    max_delegation_depth=3,
    max_concurrent_delegates=2,
    max_code_attempts=5,
    alert_at_code_attempts=3,
)

Cost impact: guarded vs unguarded

Failure mode	Unguarded cost (typical bad run)	Guarded cost (with circuit breaker)	Reduction
Tool call loop (50 steps @ Mistral Large)	~$1.40–$2.20 per run	~$0.30–$0.55 (20-step cap)	~70–75%
Thread history accumulation (50-turn thread)	~$0.045–$0.06 per turn in history overhead	~$0.005–$0.01 per turn (fresh thread after 30 turns)	~80–85%
Delegation cascade (3-level tree, 3 concurrent)	~$3.50–$6.00 per top-level request	~$0.80–$1.40 (depth limit + concurrency cap)	~75–80%
Code interpreter spiral (10 retry attempts)	~$0.60–$1.00 in completion + trace overhead	~$0.15–$0.25 (5-attempt cap)	~70–75%

The table above reflects Mistral Large pricing as of mid-2026 ($2/M input tokens, $6/M output tokens) on realistic context sizes for each failure pattern. Smaller models (Mistral Medium, Mistral Small) reduce the per-token rate but don't eliminate the structural failure modes — a loop on Mistral Small is still 3–4× more expensive than an intended single-pass completion on Mistral Large.

What about Mistral's built-in safeguards?

Mistral does expose a max_tokens parameter on individual completions, which caps the output length of a single response but has no effect on the number of tool call iterations in a run. The platform also enforces a timeout on runs that have been active for an extended period, but the timeout is long enough (many minutes) that the cost damage from a loop or spiral is done before it triggers. There is no native max_steps parameter on agents.run() at the time of writing — step budget enforcement is an application-layer concern.

The code interpreter has an execution timeout per cell, which prevents a single infinite-loop script from running forever, but resets on every retry attempt — so a 30-second timeout per attempt, 10 attempts, still produces 5 minutes of billable execution and 10 completion calls.

Thread history has no built-in summarization or truncation in the Agents API. The API will return a context-length error if a thread grows past the model's context window, but that error itself doesn't reduce costs — it terminates the run without producing output, meaning you pay for the context window used to discover the limit.