Home
Studio

Step through a session

A session keeps the agent's work, turn after turn. session.run() sends a message, waits for the turn it starts, and returns the turn's record. You write the loop around it: read the record, then decide the next step, such as answering an approval or sending the next message. That loop is a run stepper.

The example agent

The example agent

alert_reviewer is a smaller agent than the payment reviewer, with two tools: lookup_payment only reads, and open_case writes, so a person decides. With permission_mode="ask", every tool call asks for an approval. review_policy, an on_approval= function, allows the reads and leaves the rest open for a person: Approvals covers on_approval= functions.

import asyncio

from mistralai import agents

@agents.tool(annotations=agents.ToolAnnotations(read_only_hint=True))
def lookup_payment(alert_id: str) -> dict[str, str]:
    """Read the payment behind a transaction monitoring alert."""
    return {"alert_id": alert_id, "amount": "9,800 EUR", "payee": "new payee NL91 4421", "time": "03:12"}

@agents.tool
def open_case(alert_id: str, summary: str) -> str:
    """Open a case file on a suspicious payment. Returns the case id."""
    return f"CASE-{alert_id}"

alert_reviewer = agents.Agent(
    model="mistral-large-4",
    instructions=(
        "Review the alert: look up its payment first. If the payment looks suspicious, open a case file. "
        "Answer in plain text, with no markdown and no dashes, in at most two sentences."
    ),
    tools=[lookup_payment, open_case],
    permission_mode="ask",
)
def review_policy(
    ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
) -> agents.AllowApproval | agents.DenyApproval | agents.KeepCallbackOpen:
    if request.tool.annotations.read_only_hint:
        return agents.AllowApproval()
    return agents.KeepCallbackOpen()  # not mine: the turn is Blocked until a person answers
Step through approvals

Step through approvals

When review_policy returns KeepCallbackOpen(), session.run() returns Blocked, with the open approvals in callbacks. The stepper answers each one and waits for the turn again, until the turn ends:

async def stepper() -> None:
    session = await alert_reviewer.start_session(on_approval=review_policy)
    try:
        state = await session.run("Review alert 4711.")
        while isinstance(state, agents.Blocked):
            turn = session.turn(state.turn_id)
            for callback in state.callbacks:
                print(f"waiting on: {callback.title}")
                await turn.answer(callback.id, agents.AllowApproval())  # an analyst said yes
            state = await turn.result(after=state)
        match state:
            case agents.Completed(text=text, tool_calls=calls):
                print([call.name for call in calls])
                print(text)
            case _:
                print(f"the review stopped: {state!r}")
    finally:
        await session.cancel()

asyncio.run(stepper())
Output
waiting on: Allow tools.client.open_case?
['client.lookup_payment', 'client.open_case']
Alert 4711 covers a 9,800 EUR payment at 03:12 to a new payee (NL91 4421), which looks
suspicious due to the large amount, unfamiliar beneficiary, and unusual hour. I opened case file
CASE-4711 for it.

review_policy allowed lookup_payment and left open_case open, so the turn was Blocked once. The loop has three steps:

  1. session.run() returns Blocked, with the open approvals in state.callbacks.
  2. turn.answer(callback.id, ...) sends a decision for each one.
  3. turn.result(after=state) waits for the turn again, after this record, and returns its next record: Completed, or Blocked again when the turn asks for another approval.

A tool you wrote shows as client.<name>: see Tool namespaces.

Hand an approval off

Hand an approval off

The stepper can also hand an approval to someone else and wait. Here it puts each open approval in an analyst's inbox, and turn.result(after=state) returns once the analyst's answer lets the turn move on:

async def analyst(session: agents.Session, inbox: asyncio.Queue[agents.OpenCallback]) -> None:
    """Stand-in for an analyst's review screen: it answers each approval it gets, by its ids."""
    while True:
        callback = await inbox.get()
        print(f"analyst: {callback.title}")
        await session.turn(callback.turn_id).answer(callback.id, agents.AllowApproval())

async def hand_off() -> None:
    session = await alert_reviewer.start_session(on_approval=review_policy)
    inbox: asyncio.Queue[agents.OpenCallback] = asyncio.Queue()
    desk = asyncio.create_task(analyst(session, inbox))
    try:
        state = await session.run("Review alert 4711.")
        while isinstance(state, agents.Blocked):
            for callback in state.callbacks:
                await inbox.put(callback)  # hand off: the analyst answers, not the stepper
            state = await session.turn(state.turn_id).result(after=state)  # returns once the turn moves on
        print(type(state).__name__, state.text)
    finally:
        desk.cancel()
        await session.cancel()

asyncio.run(hand_off())
Output
analyst: Allow tools.client.open_case?
Completed Alert 4711 covers a 9,800 EUR payment at 03:12 to a brand new payee (NL91 4421), which is a
classic fraud pattern, so I opened case CASE-4711 on it.

When you deployed the agent with Studio (self-hosted or Mistral-managed), the analyst can be another process: it binds the session by its id and answers by the approval's ids, as in Build a session app.

To do something between starting a turn and waiting for it, or to send from a chat box while a turn runs, use session.start_turn() or session.send_message(): see When a turn already runs.

Send several messages in one session

Send several messages in one session

Start a session, then call session.run() once per message:

async def conversation() -> None:
    session = await alert_reviewer.start_session(on_approval=review_policy)
    try:
        state = await session.run("Look up the payment of alert 4711. Do not open a case.")
        if isinstance(state, agents.Completed):
            print(state.text)
        state = await session.run("What time was that payment made?")
        if isinstance(state, agents.Completed):
            print(state.text)
    finally:
        await session.cancel()  # returning does not end the session

asyncio.run(conversation())
Output
The payment behind alert 4711 is 9,800 EUR sent at 03:12 to a new payee with account NL91 4421. As
requested, no case was opened.
The payment was made at 03:12.

Each session.run() is one turn. Returning from the function leaves the session open: session.cancel() ends it.

Common issues

Common issues

  • session.run() returns Busy. Another turn is running in the session, so no turn started. Follow busy.turn, or use send_message() to add the message to the running turn.
  • turn.result(after=state) never returns. A callback in state.callbacks is still unanswered, and the turn waits for it. Answer each one before you wait again, or let an on_approval= function answer the calls it knows. With timeout=, the wait returns at the deadline: see Wait with a timeout.
  • The session stays open after your function returns. Leaving a call never stops work: turn.interrupt() stops a turn, and session.cancel() ends the session.