One call: agent.run()

agent.run() does a whole task in one call. It starts a session, runs one turn in it, waits until the turn ends or needs you, and returns the turn's record: a value that says where the turn is. You write the message, then a match on the record.

Use agent.run() for one task with no follow-up, and a session for more: see Choose agent.run() or a session.

Ask one question:

import asyncio

from mistralai import agents

agent = agents.Agent(model="mistral-large-4")

async def hello() -> None:
    result = await agent.run("Say hello in one short sentence, without emoji.")
    print(result.text)

asyncio.run(hello())

import asyncio

from mistralai import agents

agent = agents.Agent(model="mistral-large-4")

async def hello() -> None:
    result = await agent.run("Say hello in one short sentence, without emoji.")
    print(result.text)

asyncio.run(hello())

run() started a session, ran one turn and returned its record. Every record has the turn's text, its session_id and turn_id, and a cursor.

A turn can end in more than one way, and run() returns a different record for each. None of them raises, so read the record with match. Add this to the file of Run a task:

reviewer = agents.Agent(
    model="mistral-large-4",
    instructions="You review payment alerts. Say how suspicious each one looks, in one sentence of at most 20 words.",
)

async def review() -> None:
    state = await reviewer.run("Review alert 4711: 9,800 EUR to a new payee at 03:12.")
    match state:
        case agents.Completed(text=text):
            print(text)
        case agents.Failed(error=error):
            print(f"failed: {error.code}: {error.message}")
        case _:
            print(f"the turn is {type(state).__name__}")

asyncio.run(review())

reviewer = agents.Agent(
    model="mistral-large-4",
    instructions="You review payment alerts. Say how suspicious each one looks, in one sentence of at most 20 words.",
)

async def review() -> None:
    state = await reviewer.run("Review alert 4711: 9,800 EUR to a new payee at 03:12.")
    match state:
        case agents.Completed(text=text):
            print(text)
        case agents.Failed(error=error):
            print(f"failed: {error.code}: {error.message}")
        case _:
            print(f"the turn is {type(state).__name__}")

asyncio.run(review())

run() returns the same records as session.run(): see Turns and the turn record. From run(), Blocked means the turn waits for an approval, a connector sign-in or a client tool call that no function answered, and Running means timeout= passed while the turn keeps running.

Pass a Pydantic model as output_type=, and a Completed record's output is the reply parsed into that model. In the same file:

from pydantic import BaseModel

class RiskScore(BaseModel):
    score: int
    reason: str

scorer = agents.Agent(
    model="mistral-medium-latest",
    instructions="You score payment alerts for fraud risk. Give the reason in one sentence of at most 20 words.",
)

async def score_alert() -> None:
    state = await scorer.run(
        "Score alert 4711 from 0 to 100: 9,800 EUR to a new payee at 03:12.",
        output_type=RiskScore,
    )
    match state:
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            print(f"the reply is not a RiskScore: {error}")
        case agents.Completed(output=RiskScore() as score):
            print(score.score, score.reason)
        case _:
            print(f"no score: {state!r}")

asyncio.run(score_alert())

from pydantic import BaseModel

class RiskScore(BaseModel):
    score: int
    reason: str

scorer = agents.Agent(
    model="mistral-medium-latest",
    instructions="You score payment alerts for fraud risk. Give the reason in one sentence of at most 20 words.",
)

async def score_alert() -> None:
    state = await scorer.run(
        "Score alert 4711 from 0 to 100: 9,800 EUR to a new payee at 03:12.",
        output_type=RiskScore,
    )
    match state:
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            print(f"the reply is not a RiskScore: {error}")
        case agents.Completed(output=RiskScore() as score):
            print(score.score, score.reason)
        case _:
            print(f"no score: {state!r}")

asyncio.run(score_alert())

When the reply does not validate, output is an agents.InvalidOutput that holds the raw reply and the validation error, and the turn is still Completed: match it first. See Structured output.

A client tool is a tool answered through a callback, not within the session, such as one that reads the user's device. Declare it on a stub, bind the function that answers it with .answer, and pass it to run() in on_client_tool=agents.ClientToolAnswers(...). In the same file:

from typing import Literal

class Location(BaseModel):
    city: str
    country: str

@agents.client_tool
def get_location(precision: Literal["city", "street"] = "city") -> Location:
    """The user's current location."""
    raise NotImplementedError  # answered through a callback

@get_location.answer
async def answer_location(precision: Literal["city", "street"] = "city") -> Location:
    return Location(city="Lyon", country="France")  # read it from the device

locator = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_location],
)

async def where_am_i() -> None:
    state = await locator.run("Where am I?", on_client_tool=agents.ClientToolAnswers(answer_location))
    print(type(state).__name__, state.text)

asyncio.run(where_am_i())

from typing import Literal

class Location(BaseModel):
    city: str
    country: str

@agents.client_tool
def get_location(precision: Literal["city", "street"] = "city") -> Location:
    """The user's current location."""
    raise NotImplementedError  # answered through a callback

@get_location.answer
async def answer_location(precision: Literal["city", "street"] = "city") -> Location:
    return Location(city="Lyon", country="France")  # read it from the device

locator = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_location],
)

async def where_am_i() -> None:
    state = await locator.run("Where am I?", on_client_tool=agents.ClientToolAnswers(answer_location))
    print(type(state).__name__, state.text)

asyncio.run(where_am_i())

The model called client.get_location, and run() ran answer_location in your program and sent its result to the turn. Without on_client_tool=, run() returns Blocked on the call. To answer it by id, or from a session app, see Callbacks. API reference: agents.client_tool, ClientToolAnswers and OpenClientToolCall.

A failed turn returns agents.Failed, wherever the agent runs. Its error has a code, a message, retryable and details. When the model provider refused the request, the message ends with the provider's own reason. In the same file:

async def failed() -> None:
    agent = agents.Agent(model="mistral-nope-9")
    match await agent.run("Say hello"):
        case agents.Failed(error=error):
            print(error.code, error.retryable)
            print(error.message)

asyncio.run(failed())

async def failed() -> None:
    agent = agents.Agent(model="mistral-nope-9")
    match await agent.run("Say hello"):
        case agents.Failed(error=error):
            print(error.code, error.retryable)
            print(error.message)

asyncio.run(failed())

When error.retryable is True, such as for a rate limit (completion_rate_limited) or a provider outage (completion_unavailable), the same request may succeed if you send it again. A rejected request (completion_rejected) or a refused API key (completion_unauthorized) is not retryable: fix it first. ProtocolError lists every code.

Errors that are not about the turn, such as an empty message or a wrong argument, raise exceptions before a turn starts: see Errors.

Use agent.run() for one task with no follow-up. Start a session when you send more than one message, or when another process watches or answers the turn: see Step through a session. Agent.run says when the session that run() starts closes.

To answer approvals or follow the turn as it runs, pass on_approval= or on_event=. Callbacks covers every question a turn can ask your program.

• state.output is None with output_type=. The turn ended without a final answer: state.stop_reason says why, such as "iteration_limit" at max_iterations, or "output_limit" when the model's last reply stopped at the output token limit. Read state.text for the last thing the model said.
• run() returns Blocked. The turn waits for an approval, a sign-in or a client tool call, and no on_approval=, on_sign_in= or on_client_tool= answered it. Pass a function for it, or answer it by id: see Callbacks.
• run() returns Running. timeout= passed, and the turn keeps running: reach it with agent.session(state.session_id).turn(state.turn_id).

• Step through a session: send several messages to one session, and step a turn through its approvals.
• Build a session app: watch a session and send it commands.
• Callbacks: answer approvals, client tool calls and sign-ins.