Construire une application de session

Une application de session montre ce que fait un agent et envoie les commandes d'une personne à sa session. Les événements viennent de la session, chacun avec un identifiant, dans l'ordre, ce qui permet à une vue de continuer à partir du dernier qu'elle a lu. Les commandes vont à la session : un message, une réponse d'approbation, un steer ou une interruption. Chaque commande revient une fois que la session l'a acceptée, pas quand le tour se termine. Observer ne change jamais rien : fermer une vue laisse le travail s'exécuter.

L'agent de l'exemple

L'application utilise un alert_reviewer avec les outils de Parcourir une session pas à pas, lookup_payment et open_case, qui répond en une seule phrase. L'agent a permission_mode="ask" et la session n'a pas de on_approval=, donc chaque appel d'outil attend un clic :

import asyncio

from mistralai import agents

# lookup_payment and open_case: as in Step through a session

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, in one sentence."
    ),
    tools=[lookup_payment, open_case],
    permission_mode="ask",
)

import asyncio

from mistralai import agents

# lookup_payment and open_case: as in Step through a session

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, in one sentence."
    ),
    tools=[lookup_payment, open_case],
    permission_mode="ask",
)

Observer la session

Une vue conserve ce qu'elle affiche à l'écran : les approbations ouvertes sous forme de boutons, les messages sous forme de lignes, le nombre de tours terminés et le cursor, l'identifiant du dernier événement lu :

class Screen:
    """What one view shows: the open approvals, the conversation, and where it stopped reading."""

    def __init__(self) -> None:
        self.cursor: str | None = None
        self.buttons: dict[str, agents.OpenCallback] = {}
        self.lines: list[str] = []  # one line per message
        self.ended = 0  # the turns that ended
        self.changed = asyncio.Condition()

    async def update(self) -> None:
        async with self.changed:
            self.changed.notify_all()

class Screen:
    """What one view shows: the open approvals, the conversation, and where it stopped reading."""

    def __init__(self) -> None:
        self.cursor: str | None = None
        self.buttons: dict[str, agents.OpenCallback] = {}
        self.lines: list[str] = []  # one line per message
        self.ended = 0  # the turns that ended
        self.changed = asyncio.Condition()

    async def update(self) -> None:
        async with self.changed:
            self.changed.notify_all()

La vue lit les événements de la session avec iter_events(), à partir de son curseur, et dessine chacun d'eux :

async def view(session: agents.Session, screen: Screen) -> None:
    """Watch the session and draw it. A view that opens again continues from screen.cursor."""
    async for event in session.iter_events(after=screen.cursor):
        match event:
            case agents.CallbackOpened(callback=callback):
                screen.buttons[callback.id] = callback
                print(f"[Allow] {callback.title}")
            case agents.CallbackResolved(callback_id=callback_id):
                screen.buttons.pop(callback_id, None)
            case agents.EntryCompleted(entry=agents.MessageEntry(role="user" | "assistant") as message):
                screen.lines.append(f"{message.role}: {message.text}")
                print(screen.lines[-1])
            case agents.TurnEnded(state=state):
                screen.ended += 1
                print(type(state).__name__)
        screen.cursor = event.id
        await screen.update()

async def view(session: agents.Session, screen: Screen) -> None:
    """Watch the session and draw it. A view that opens again continues from screen.cursor."""
    async for event in session.iter_events(after=screen.cursor):
        match event:
            case agents.CallbackOpened(callback=callback):
                screen.buttons[callback.id] = callback
                print(f"[Allow] {callback.title}")
            case agents.CallbackResolved(callback_id=callback_id):
                screen.buttons.pop(callback_id, None)
            case agents.EntryCompleted(entry=agents.MessageEntry(role="user" | "assistant") as message):
                screen.lines.append(f"{message.role}: {message.text}")
                print(screen.lines[-1])
            case agents.TurnEnded(state=state):
                screen.ended += 1
                print(type(state).__name__)
        screen.cursor = event.id
        await screen.update()

CallbackOpened porte une approbation ouverte et la vue affiche un bouton pour elle. CallbackResolved indique qu'une approbation n'est plus ouverte : un client y a répondu, cette vue ou une autre, ou son tour s'est terminé. La vue retire le bouton. EntryCompleted porte chaque entrée de l'historique de la session une fois qu'elle est complète : la vue dessine les messages de l'utilisateur et ceux de l'agent. TurnEnded porte les données du tour. La vue avance son curseur au-delà de chaque événement, aussi bien ceux qu'elle dessine que ceux qu'elle ignore. iter_events() suit la session jusqu'à ce que la session se termine.

Envoyer des commandes

Le clic d'une personne et le message d'une personne sont des commandes. Le clic répond à l'approbation du tour qui l'a ouverte :

async def click_allow(session: agents.Session, callback: agents.OpenCallback) -> str:
    mine = agents.AllowApproval()
    match await session.turn(callback.turn_id).answer(callback.id, mine):
        case agents.Answered(answer=applied) if applied != mine:
            return "the other view answered first"
        case agents.Answered():
            return "allowed"
        case agents.AlreadySettled():
            return "no longer open"
        case agents.NotFound():
            return "no such approval"


async def on_typed(session: agents.Session, text: str) -> agents.Turn:
    return await session.send_message(text)  # starts a turn, or adds the message to the running one


async def on_steer(turn: agents.Turn, text: str) -> str:
    match await turn.steer(text):
        case agents.TurnConflict():
            return "that turn has ended: send a new message"
        case agents.Turn():
            return "steered"

async def click_allow(session: agents.Session, callback: agents.OpenCallback) -> str:
    mine = agents.AllowApproval()
    match await session.turn(callback.turn_id).answer(callback.id, mine):
        case agents.Answered(answer=applied) if applied != mine:
            return "the other view answered first"
        case agents.Answered():
            return "allowed"
        case agents.AlreadySettled():
            return "no longer open"
        case agents.NotFound():
            return "no such approval"


async def on_typed(session: agents.Session, text: str) -> agents.Turn:
    return await session.send_message(text)  # starts a turn, or adds the message to the running one


async def on_steer(turn: agents.Turn, text: str) -> str:
    match await turn.steer(text):
        case agents.TurnConflict():
            return "that turn has ended: send a new message"
        case agents.Turn():
            return "steered"

Les branches couvrent les trois résultats de answer() : Answered, AlreadySettled et NotFound. Quand deux vues répondent à la même approbation en même temps, les deux reçoivent Answered, et answer est la réponse qui a compté : comparez-la avec la vôtre. AlreadySettled signifie que l'approbation n'est plus ouverte : quelqu'un y a répondu en premier, ou son tour ou sa session s'est terminé. send_message() lance un tour quand aucun ne s'exécute, et ajoute sinon le message au tour en cours. turn.steer() ajoute un message à ce tour précis, et renvoie TurnConflict une fois le tour terminé. Approbations couvre les fonctions on_approval=.

Exécuter l'application

La personne de ce script attend que l'écran change, puis clique sur le premier bouton. Le bouton disparaît dès qu'on a cliqué dessus, donc l'attente suivante ne le retrouve pas. click_all clique jusqu'à ce qu'un tour de plus se soit terminé, donc une approbation que le modèle demande plus tard reçoit aussi un clic :

async def next_click(screen: Screen, ended: int) -> agents.OpenCallback | None:
    """Wait for a button to click, or None once more than `ended` turns have ended."""
    async with screen.changed:
        await screen.changed.wait_for(lambda: screen.buttons or screen.ended > ended)
    return next(iter(screen.buttons.values()), None)


async def click_all(session: agents.Session, screen: Screen, ended: int) -> None:
    """Click Allow on each button, until more than `ended` turns have ended."""
    while (callback := await next_click(screen, ended)) is not None:
        screen.buttons.pop(callback.id)  # the button goes once clicked
        print(await click_allow(session, callback))

async def next_click(screen: Screen, ended: int) -> agents.OpenCallback | None:
    """Wait for a button to click, or None once more than `ended` turns have ended."""
    async with screen.changed:
        await screen.changed.wait_for(lambda: screen.buttons or screen.ended > ended)
    return next(iter(screen.buttons.values()), None)


async def click_all(session: agents.Session, screen: Screen, ended: int) -> None:
    """Click Allow on each button, until more than `ended` turns have ended."""
    while (callback := await next_click(screen, ended)) is not None:
        screen.buttons.pop(callback.id)  # the button goes once clicked
        print(await click_allow(session, callback))

main() exécute la vue comme une tâche et envoie les commandes de la personne depuis la tâche principale :

async def main() -> None:
    session = await alert_reviewer.start_session()
    screen = Screen()
    watcher = asyncio.create_task(view(session, screen))

    turn = await on_typed(session, "Review alert 4711.")
    print(f"sent: turn started={turn.started}")
    await click_all(session, screen, ended=0)

    watcher.cancel()  # the view closes: the session keeps running
    turn = await on_typed(session, "Which case did you open?")
    await asyncio.sleep(5)  # stand-in for the time the app stays closed: the turn keeps running
    watcher = asyncio.create_task(view(session, screen))  # a new view, from the cursor
    await click_all(session, screen, ended=1)

    turn = await on_typed(session, "Now review alerts 4712 to 4790, one at a time.")
    print(await on_steer(turn, "Skip alert 4713."))  # the turn runs: the message joins it
    await turn.interrupt()  # stops this turn; the session stays open
    await click_all(session, screen, ended=2)
    print(await on_steer(turn, "Skip alert 4714."))  # the turn has ended

    await session.cancel()  # ends the session and its running turns
    await watcher  # the view's loop ends with the session


asyncio.run(main())

async def main() -> None:
    session = await alert_reviewer.start_session()
    screen = Screen()
    watcher = asyncio.create_task(view(session, screen))

    turn = await on_typed(session, "Review alert 4711.")
    print(f"sent: turn started={turn.started}")
    await click_all(session, screen, ended=0)

    watcher.cancel()  # the view closes: the session keeps running
    turn = await on_typed(session, "Which case did you open?")
    await asyncio.sleep(5)  # stand-in for the time the app stays closed: the turn keeps running
    watcher = asyncio.create_task(view(session, screen))  # a new view, from the cursor
    await click_all(session, screen, ended=1)

    turn = await on_typed(session, "Now review alerts 4712 to 4790, one at a time.")
    print(await on_steer(turn, "Skip alert 4713."))  # the turn runs: the message joins it
    await turn.interrupt()  # stops this turn; the session stays open
    await click_all(session, screen, ended=2)
    print(await on_steer(turn, "Skip alert 4714."))  # the turn has ended

    await session.cancel()  # ends the session and its running turns
    await watcher  # the view's loop ends with the session


asyncio.run(main())

Cela affiche :

user: Review alert 4711.
sent: turn started=True
[Allow] Allow tools.client.lookup_payment?
allowed
[Allow] Allow tools.client.open_case?
allowed
assistant: Alert 4711 concerns a suspicious 9,800 EUR payment to a new payee (NL91 4421) at 03:12, so I opened case file CASE-4711.
Completed
user: Which case did you open?
assistant: I opened case file CASE-4711 for alert 4711.
Completed
user: Now review alerts 4712 to 4790, one at a time.
user: Skip alert 4713.
steered
Interrupted
that turn has ended: send a new message

user: Review alert 4711.
sent: turn started=True
[Allow] Allow tools.client.lookup_payment?
allowed
[Allow] Allow tools.client.open_case?
allowed
assistant: Alert 4711 concerns a suspicious 9,800 EUR payment to a new payee (NL91 4421) at 03:12, so I opened case file CASE-4711.
Completed
user: Which case did you open?
assistant: I opened case file CASE-4711 for alert 4711.
Completed
user: Now review alerts 4712 to 4790, one at a time.
user: Skip alert 4713.
steered
Interrupted
that turn has ended: send a new message

La sortie montre trois règles :

Les approbations parviennent à la vue sous forme d'événements. Chaque appel d'outil a ouvert une approbation, la vue a affiché un bouton et le clic a permis au tour de continuer.
Quitter l'application n'arrête jamais le travail. asyncio.sleep(5) représente le temps pendant lequel l'application est fermée : le tour s'est exécuté et s'est terminé sans vue ouverte. La nouvelle vue a repris à partir du curseur enregistré, donc elle n'a lu que la question et la réponse qu'elle avait manquées.
Un steer rejoint un tour en cours ; seuls interrupt() et cancel() arrêtent le travail. Le premier steer a atteint le tour en cours : son message est dans la conversation. Le tour interrompu s'est ensuite terminé avec le statut Interrupted, et la session prend le message suivant. Un steer sur ce tour terminé a renvoyé TurnConflict : envoyez plutôt un nouveau message.

Envoyer un steer tout en suivant le tour

Une application de chat peut aussi afficher un tour dans l'appel qui l'attend : turn.result(on_event=show) appelle show avec chaque événement du tour à mesure qu'il arrive. Pendant ce temps, une tâche UI distincte envoie les commandes de la personne : un steer, un clic sur Autoriser, un clic sur Arrêter. La session n'a pas de fonction on_approval=, donc la personne répond à chaque approbation.

follow() affiche la réponse à mesure que le modèle l'écrit, et met chaque approbation dans la file buttons de l'UI :

async def follow(turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]) -> None:
    """Show the turn while it runs: the reply as the model writes it, and each approval as a button."""

    async def show(event: agents.Event) -> None:
        match event:
            case agents.EntryCompleted(entry=agents.MessageEntry(role="user") as message):
                print(f"user: {message.text}")
            case agents.EntryAdded(entry=agents.MessageEntry(role="assistant")):
                print("assistant: ", end="")
            case agents.TextDelta(text=text):
                print(text, end="", flush=True)  # the reply, a few words at a time
            case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant")):
                print()
            case agents.CallbackOpened(callback=callback):
                print(f"[Allow] {callback.title}")
                buttons.put_nowait(callback)  # the UI shows the button

    state = await turn.result(on_event=show)
    while isinstance(state, agents.Blocked):  # an approval waits for the person: keep following
        state = await turn.result(after=state, on_event=show)
    print(type(state).__name__)

async def follow(turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]) -> None:
    """Show the turn while it runs: the reply as the model writes it, and each approval as a button."""

    async def show(event: agents.Event) -> None:
        match event:
            case agents.EntryCompleted(entry=agents.MessageEntry(role="user") as message):
                print(f"user: {message.text}")
            case agents.EntryAdded(entry=agents.MessageEntry(role="assistant")):
                print("assistant: ", end="")
            case agents.TextDelta(text=text):
                print(text, end="", flush=True)  # the reply, a few words at a time
            case agents.EntryCompleted(entry=agents.MessageEntry(role="assistant")):
                print()
            case agents.CallbackOpened(callback=callback):
                print(f"[Allow] {callback.title}")
                buttons.put_nowait(callback)  # the UI shows the button

    state = await turn.result(on_event=show)
    while isinstance(state, agents.Blocked):  # an approval waits for the person: keep following
        state = await turn.result(after=state, on_event=show)
    print(type(state).__name__)

EntryAdded commence la ligne de la réponse, chaque TextDelta ajoute les mots suivants, et EntryCompleted termine la ligne. Quand une approbation s'ouvre, turn.result() renvoie Blocked. follow() attend de nouveau avec after=state, et continue ainsi à suivre au-delà de cette approbation pendant que la personne y répond, sans afficher deux fois le même événement.

Les commandes de la personne viennent de la tâche UI. Deux doublures pour une personne : l'une envoie un steer au premier bouton, puis clique sur Autoriser à chaque bouton ; l'autre clique sur Arrêter :

async def steer_then_allow(
    session: agents.Session, turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]
) -> None:
    """The person at the UI: types a steer when the first button shows, then clicks Allow on each button."""
    callback = await buttons.get()
    print(await on_steer(turn, "Reply in French, please."))
    while True:
        print(await click_allow(session, callback))
        callback = await buttons.get()


async def stop(turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]) -> None:
    """The person at the UI: clicks Stop when the first button shows."""
    await buttons.get()
    print("clicked Stop")
    await turn.interrupt()  # stops this turn; the session stays open

async def steer_then_allow(
    session: agents.Session, turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]
) -> None:
    """The person at the UI: types a steer when the first button shows, then clicks Allow on each button."""
    callback = await buttons.get()
    print(await on_steer(turn, "Reply in French, please."))
    while True:
        print(await click_allow(session, callback))
        callback = await buttons.get()


async def stop(turn: agents.Turn, buttons: asyncio.Queue[agents.OpenCallback]) -> None:
    """The person at the UI: clicks Stop when the first button shows."""
    await buttons.get()
    print("clicked Stop")
    await turn.interrupt()  # stops this turn; the session stays open

follow_and_steer() démarre une tâche UI à côté de chaque attente :

async def follow_and_steer() -> None:
    session = await alert_reviewer.start_session()
    buttons: asyncio.Queue[agents.OpenCallback] = asyncio.Queue()

    turn = await on_typed(session, "Review alert 4711.")
    ui = asyncio.create_task(steer_then_allow(session, turn, buttons))  # the UI sends commands meanwhile
    await follow(turn, buttons)
    ui.cancel()

    turn = await on_typed(session, "Now review alerts 4712 to 4790, one at a time.")
    ui = asyncio.create_task(stop(turn, buttons))
    await follow(turn, buttons)
    await ui
    await session.cancel()


asyncio.run(follow_and_steer())

async def follow_and_steer() -> None:
    session = await alert_reviewer.start_session()
    buttons: asyncio.Queue[agents.OpenCallback] = asyncio.Queue()

    turn = await on_typed(session, "Review alert 4711.")
    ui = asyncio.create_task(steer_then_allow(session, turn, buttons))  # the UI sends commands meanwhile
    await follow(turn, buttons)
    ui.cancel()

    turn = await on_typed(session, "Now review alerts 4712 to 4790, one at a time.")
    ui = asyncio.create_task(stop(turn, buttons))
    await follow(turn, buttons)
    await ui
    await session.cancel()


asyncio.run(follow_and_steer())

Cela affiche :

user: Review alert 4711.
[Allow] Allow tools.client.lookup_payment?
steered
allowed
user: Reply in French, please.
[Allow] Allow tools.client.open_case?
allowed
assistant: Alerte 4711 examinée : le paiement de 9 800 EUR vers un nouveau bénéficiaire (NL91 4421) à 03h12 est suspect, j'ai donc ouvert le dossier CASE-4711.
Completed
user: Now review alerts 4712 to 4790, one at a time.
[Allow] Allow tools.client.lookup_payment?
clicked Stop
Interrupted

user: Review alert 4711.
[Allow] Allow tools.client.lookup_payment?
steered
allowed
user: Reply in French, please.
[Allow] Allow tools.client.open_case?
allowed
assistant: Alerte 4711 examinée : le paiement de 9 800 EUR vers un nouveau bénéficiaire (NL91 4421) à 03h12 est suspect, j'ai donc ouvert le dossier CASE-4711.
Completed
user: Now review alerts 4712 to 4790, one at a time.
[Allow] Allow tools.client.lookup_payment?
clicked Stop
Interrupted

Le steer a rejoint le tour en cours. Son message est réapparu dans les événements, et la réponse qui a suivi est en French.
Chaque approbation a attendu un clic et les événements ont continué d'arriver. La tâche UI a répondu pendant que follow() attendait.
Arrêter a mis fin au tour, pas à la session. interrupt() a terminé le deuxième tour avec le statut Interrupted à sa première approbation ; la session prend le message suivant.

Envoyez les commandes depuis leur propre tâche : show s'exécute dans l'attente, donc un show qui attend la personne retient les événements suivants.

Rouvrir la vue depuis son curseur

Le curseur d'une vue est tout ce dont elle a besoin pour continuer. Stockez screen.cursor avec l'identifiant de la session. Une vue qui s'ouvre à nouveau, dans ce processus ou dans un autre, se lie à la session par son identifiant et lit à partir de là. Elle ne manque aucun événement, sauf les TextDelta d'une réponse, et n'en répète aucun :

session = alert_reviewer.session(session_id)  # sends no request
async for event in session.iter_events(after=stored_cursor):
    ...

session = alert_reviewer.session(session_id)  # sends no request
async for event in session.iter_events(after=stored_cursor):
    ...

Un autre processus peut se lier à une session de cette façon quand vous avez déployé l'agent avec Studio (auto-hébergé ou géré par Mistral) : voir Modes d'exécution. Il peut aussi répondre à une approbation par ses identifiants, avec session.turn(turn_id).answer(callback_id, ...).

Problèmes courants

Le tour reste Blocked : les boutons restent ouverts. Une vue ne répond à rien d'elle-même. Quelque chose doit appeler turn.answer() : un clic, comme ici, ou une fonction on_approval= donnée à start_session(), qui répond pendant qu'un appel tel que session.run() attend le tour.
Aucune approbation ne s'ouvre jamais. Le permission_mode de l'agent est "auto", la valeur par défaut, et aucune règle permission_policy ne le demande. Définissez permission_mode="ask", ou ajoutez une règle avec decision="ask".
La même tâche observe et pilote. Une tâche qui boucle sur iter_events() ne revient jamais pour envoyer un clic. Exécutez la vue dans sa propre tâche, comme le fait main(), et envoyez les commandes depuis une autre.

Et après

Événements : les types d'événements et les curseurs.
Modes d'exécution : exécuter la même application avec un agent déployé.
Référence API de Session et Turn : Session.iter_events, Turn.steer, Turn.interrupt et Turn.answer.