Events

While a turn runs, the session sends events: the turn started, the user's message, each tool call and its result, the model's reply as it writes it, a question the turn waits on, the turn ended. The SDK reads them for you and gives each one as a typed value. Follow them while you wait to show the agent's progress in a chat or a dashboard, or replay them later.

Every event has an id, its position in the session's events, and a turn_id, the turn it happened in, or None between turns or when the event does not name its turn. A message's first TextDelta has the id of its EntryAdded, and after= that id reads after both. Every other event has its own id. New event types come in minor versions, as UnknownEvent until then: ignore the types you do not know.

The session's history comes as entry events: the user's message, right before its TurnStarted, each context message the model got with a turn's input, each tool call, each tool's result, a client tool's too, and each message of the model. Each entry comes once as EntryAdded, when it is added, then once as EntryCompleted, when it is complete, on every host, live and on replay. A read that starts after an entry's EntryAdded gets only its EntryCompleted. Keep the entry of EntryCompleted: only it is final. While the model writes a message, its new text comes as TextDelta events between the two. A context message the model got with a turn's input comes as an agents.ContextMessage entry: see History entries. An entry type your SDK version has no class for, such as the model's reasoning or a message from another agent, comes as an agents.UnknownEntry, with the entry unparsed in raw. The ids below skip numbers: treat them as positions, not counts.

To show each finished tool call while the turn runs, such as a progress bar, follow the EntryCompleted events whose entry is an agents.ToolResultEntry: it holds the tool's name, is_error, and the tool_call_id of its agents.ToolCallEntry, which holds the call's arguments. To push progress to an app that does not follow the session, use a hook: see Live progress.

Pass on_event=, a plain or async function, to session.run(), turn.result() or agent.run(). It gets the turn's events in order. When the turn ends, the last one is TurnEnded, whose state is the record the call returns:

import asyncio

from mistralai import agents

from reviewer import payment_reviewer

def show(event: agents.Event) -> None:
    match event:
        case agents.TurnStarted():
            print(event.id, "turn started")
        case agents.CallbackOpened(callback):
            print(event.id, "waits for:", callback.title)
        case agents.CallbackResolved(callback_id):
            print(event.id, "answered:", callback_id)
        case agents.EntryAdded(entry=entry):
            print(event.id, "entry added:", type(entry).__name__)
        case agents.TextDelta(text=text):
            print(event.id, "text:", repr(text))
        case agents.EntryCompleted(entry=entry):
            print(event.id, "entry completed:", type(entry).__name__)
        case agents.TurnEnded(state):
            print(event.id, "turn ended:", type(state).__name__)
        case agents.SessionUpdated(status):
            print(event.id, "session:", status)
        case agents.UnknownEvent(kind):
            print(event.id, "other event:", kind)

async def main() -> None:
    session = await payment_reviewer.start_session()
    state = await session.run("Review transaction monitoring alert 4711.", on_event=show)

asyncio.run(main())

import asyncio

from mistralai import agents

from reviewer import payment_reviewer

def show(event: agents.Event) -> None:
    match event:
        case agents.TurnStarted():
            print(event.id, "turn started")
        case agents.CallbackOpened(callback):
            print(event.id, "waits for:", callback.title)
        case agents.CallbackResolved(callback_id):
            print(event.id, "answered:", callback_id)
        case agents.EntryAdded(entry=entry):
            print(event.id, "entry added:", type(entry).__name__)
        case agents.TextDelta(text=text):
            print(event.id, "text:", repr(text))
        case agents.EntryCompleted(entry=entry):
            print(event.id, "entry completed:", type(entry).__name__)
        case agents.TurnEnded(state):
            print(event.id, "turn ended:", type(state).__name__)
        case agents.SessionUpdated(status):
            print(event.id, "session:", status)
        case agents.UnknownEvent(kind):
            print(event.id, "other event:", kind)

async def main() -> None:
    session = await payment_reviewer.start_session()
    state = await session.run("Review transaction monitoring alert 4711.", on_event=show)

asyncio.run(main())

The user's message comes first, as an entry, then the turn starts. Each entry of the turn comes as an EntryAdded, then an EntryCompleted: here the model's reasoning, as UnknownEntry, then a tool call and its result. The reply's text comes as TextDelta events between its two. A turn with permission_mode="ask" also sends a CallbackOpened for each approval it waits for, and a CallbackResolved once it has its answer.

Wait with on_event= to show a turn's events while an on_approval= function answers its approvals. Iterating events and waiting in one task blocks on an approval: run them in two tasks. To steer the turn and answer its approvals while you show it, see Steer a turn while you follow it.

Match EntryCompleted to show the conversation: the user's message, each tool call and its result, and the agent's reply. To show the reply while the model writes it, print its TextDelta texts as they come. TextDelta comes only live: a replay, or a read that starts late, misses some or all of them. When the deltas you printed do not add up to the reply, its EntryCompleted gives what is missing: the rest of the text, or the whole text when the first deltas were missed:

shown: dict[str, str] = {}  # the reply text printed so far, by entry id

def show_chat(event: agents.Event) -> None:
    match event:
        case agents.EntryCompleted(entry=agents.MessageEntry(role="user") as message):
            print("user:", message.text)
        case agents.EntryCompleted(entry=agents.ToolCallEntry(name, arguments)):
            print("calling", name, arguments)
        case agents.EntryCompleted(entry=agents.ToolResultEntry(name)):
            print(name, "returned")
        case agents.TextDelta(entry_id=entry_id, text=text):
            if entry_id not in shown:
                print("agent: ", end="")
            shown[entry_id] = shown.get(entry_id, "") + text
            print(text, end="", flush=True)  # the reply, while the model writes it
        case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant") as reply):
            so_far = shown.pop(reply.id, "")
            if so_far and reply.text.startswith(so_far):
                print(reply.text[len(so_far) :])  # the rest, if the last deltas were missed
            else:  # no deltas, or the first ones missed
                print(("\n" if so_far else "") + "agent: " + reply.text)

async def chat() -> None:
    session = await payment_reviewer.start_session()
    await session.run("Review transaction monitoring alert 4711.", on_event=show_chat)
    await session.cancel()

asyncio.run(chat())

shown: dict[str, str] = {}  # the reply text printed so far, by entry id

def show_chat(event: agents.Event) -> None:
    match event:
        case agents.EntryCompleted(entry=agents.MessageEntry(role="user") as message):
            print("user:", message.text)
        case agents.EntryCompleted(entry=agents.ToolCallEntry(name, arguments)):
            print("calling", name, arguments)
        case agents.EntryCompleted(entry=agents.ToolResultEntry(name)):
            print(name, "returned")
        case agents.TextDelta(entry_id=entry_id, text=text):
            if entry_id not in shown:
                print("agent: ", end="")
            shown[entry_id] = shown.get(entry_id, "") + text
            print(text, end="", flush=True)  # the reply, while the model writes it
        case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant") as reply):
            so_far = shown.pop(reply.id, "")
            if so_far and reply.text.startswith(so_far):
                print(reply.text[len(so_far) :])  # the rest, if the last deltas were missed
            else:  # no deltas, or the first ones missed
                print(("\n" if so_far else "") + "agent: " + reply.text)

async def chat() -> None:
    session = await payment_reviewer.start_session()
    await session.run("Review transaction monitoring alert 4711.", on_event=show_chat)
    await session.cancel()

asyncio.run(chat())

turn.iter_events() and session.iter_events() give events without driving the turn: they never return a record and never answer anything. Use them to show a session that another process drives.

turn.iter_events() gives one turn's events, from the previous turn's end, then stops at its end. For a turn that has ended, it replays them. Here session and state are the ones of main() above:

async for event in session.turn(state.turn_id).iter_events():
    show(event)

async for event in session.turn(state.turn_id).iter_events():
    show(event)

The replay has every entry's EntryAdded and EntryCompleted, as the live read did. A replay gives at most a reply's first TextDelta: read the reply's text from its EntryCompleted.

session.iter_events(after=) gives every turn's events, the ones the session has, then the live ones. It does not stop at a turn's end:

await session.start_turn("Which case did you open? Reply with the case id only.")
async for event in session.iter_events(after=state):
    show(event)
    if isinstance(event, agents.TurnEnded):
        print(event.state.text)
        break
await session.cancel()

await session.start_turn("Which case did you open? Reply with the case id only.")
async for event in session.iter_events(after=state):
    show(event)
    if isinstance(event, agents.TurnEnded):
        print(event.state.text)
        break
await session.cancel()

A cursor is a position in the session's events: an event's id, or a record's cursor. Treat it as an opaque string: its format differs between a session in your process and one of an agent you deployed with Studio (self-hosted or Mistral-managed). after= takes a cursor or a record, in every call, and the events start right after that position, none repeated and none missed except TextDelta, which comes only live (see Show the conversation as it comes):

• after=state reads from where the record state was taken. The second turn above starts after the first turn's record.
• after=event.id reads after that event. Store the id of the last event you showed, with the session id. For a deployed agent, a restarted process continues from there with agent.session(session_id).iter_events(after=stored_id). A session in your process stops when the event loop that started it closes: see Local runtime. After a TurnEnded, store its id, not its state.

Without after=, session.iter_events() starts at the session's start, and turn.iter_events() after the previous turn's end.

on_event= gets the events up to the record the call returns, so a next call with after= that record misses no event except TextDelta, and repeats none. In a loop of turn.result(after=state, timeout=...), a call that returns while the model writes a reply leaves out the rest of that reply's TextDelta events: show the reply from its EntryCompleted, which has the full text.

• Turns and the turn record: what each record holds.
• History entries: read a turn's entries, or start a session from an earlier conversation.
• Approvals: answer the approvals that CallbackOpened announces.