Callbacks

Un callback est une question qu'un tour pose à votre programme et dont il attend la réponse. Le tour s'interrompt au callback et reprend lorsqu'il reçoit une réponse. Il existe trois types de questions :

La façon de répondre dépend de la façon dont vous pilotez la session :

• Avec run() : les fonctions que vous passez via on_approval=, on_client_tool= et on_sign_in= répondent pendant que l'exécution attend. Les callbacks qu'elles laissent ouverts sont renvoyés dans agents.Blocked, et vous y répondez par leur identifiant.
• Dans une application de session : l'application suit les événements du tour et répond à chaque agents.CallbackOpened par son identifiant.

Utilisez des fonctions lorsque votre programme peut décider. Si la décision revient à une personne ou à un autre processus, laissez le callback ouvert et répondez-y par son identifiant. Les deux approches se combinent : les fonctions répondent aux questions qu'elles peuvent traiter, et Blocked renvoie les autres.

Les exemples utilisent un assistant doté d'un outil côté client et d'un connecteur. Avec permission_mode="ask", chaque appel d'outil demande aussi une approbation. get_location déclare n'effectuer que des opérations de lecture :

import asyncio

from mistralai import agents


@agents.client_tool(annotations=agents.ToolAnnotations(read_only_hint=True))
def get_location() -> str:
    """The user's current city."""
    raise NotImplementedError  # your program answers it


def read_city() -> str:
    return "Lyon"  # read it from the device


assistant = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_location],
    connectors=[agents.connector("gmail")],
    permission_mode="ask",  # every tool call asks for approval
)

import asyncio

from mistralai import agents


@agents.client_tool(annotations=agents.ToolAnnotations(read_only_hint=True))
def get_location() -> str:
    """The user's current city."""
    raise NotImplementedError  # your program answers it


def read_city() -> str:
    return "Lyon"  # read it from the device


assistant = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_location],
    connectors=[agents.connector("gmail")],
    permission_mode="ask",  # every tool call asks for approval
)

Passez une fonction par type à agent.run(), agent.start_session() ou agent.session(id). Elles répondent pendant que agent.run(), session.run() ou turn.result() attend. Ajoutez le code suivant au fichier de l'agent d'exemple :

def approve_reads(
    ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
) -> agents.AllowApproval | agents.KeepCallbackOpen:
    if request.tool.annotations.read_only_hint:
        return agents.AllowApproval()
    return agents.KeepCallbackOpen()  # a person decides


def refuse_sign_ins(sign_in: agents.OpenSignIn) -> agents.SignInAnswer:
    return agents.SignInAnswer(denied=[request.id for request in sign_in.requests])


async def with_run() -> None:
    state = await assistant.run(
        "Where am I, and what is my latest email about?",
        on_approval=approve_reads,
        on_client_tool=agents.ClientToolAnswers(get_location.answer(read_city)),
        on_sign_in=refuse_sign_ins,
    )
    if isinstance(state, agents.Blocked):
        ...  # a person decides: see Answer by id

def approve_reads(
    ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
) -> agents.AllowApproval | agents.KeepCallbackOpen:
    if request.tool.annotations.read_only_hint:
        return agents.AllowApproval()
    return agents.KeepCallbackOpen()  # a person decides


def refuse_sign_ins(sign_in: agents.OpenSignIn) -> agents.SignInAnswer:
    return agents.SignInAnswer(denied=[request.id for request in sign_in.requests])


async def with_run() -> None:
    state = await assistant.run(
        "Where am I, and what is my latest email about?",
        on_approval=approve_reads,
        on_client_tool=agents.ClientToolAnswers(get_location.answer(read_city)),
        on_sign_in=refuse_sign_ins,
    )
    if isinstance(state, agents.Blocked):
        ...  # a person decides: see Answer by id

approve_reads autorise get_location, ce qui permet à read_city de répondre à son appel. Les outils du connecteur ne déclarent pas se limiter à des opérations de lecture. Leurs demandes d'approbation restent donc ouvertes pour qu'une personne y réponde.

Les fonctions répondent aussi aux callbacks des sous-agents de la session. on_approval= peut laisser une approbation à une personne avec agents.KeepCallbackOpen(). Un callback auquel aucune fonction ne répond reste ouvert, et l'exécution renvoie Blocked.

Blocked répertorie les callbacks ouverts dans state.callbacks. Répondez à chacun avec turn.answer(callback.id, answer), depuis ce processus ou tout autre processus capable d'ouvrir la session. Choisissez la réponse en fonction de la classe du callback :

async def answer_callback(turn: agents.Turn, callback: agents.OpenCallback) -> None:
    """Answer one open callback, whatever its kind."""
    match callback:
        case agents.OpenApproval():
            await turn.answer(callback.id, agents.AllowApproval())  # a person said yes
        case agents.OpenClientToolCall() if get_location.called(callback):
            await turn.answer(callback.id, await callback.run(read_city))
        case agents.OpenSignIn(requests=requests):
            await turn.answer(callback.id, agents.SignInAnswer(denied=[request.id for request in requests]))
        case _:
            pass  # not ours: leave it open for another client

async def answer_callback(turn: agents.Turn, callback: agents.OpenCallback) -> None:
    """Answer one open callback, whatever its kind."""
    match callback:
        case agents.OpenApproval():
            await turn.answer(callback.id, agents.AllowApproval())  # a person said yes
        case agents.OpenClientToolCall() if get_location.called(callback):
            await turn.answer(callback.id, await callback.run(read_city))
        case agents.OpenSignIn(requests=requests):
            await turn.answer(callback.id, agents.SignInAnswer(denied=[request.id for request in requests]))
        case _:
            pass  # not ours: leave it open for another client

Une exécution sans fonctions renvoie Blocked dès que le tour attend une réponse à un callback. Répondez aux callbacks répertoriés, puis attendez à nouveau le tour avec turn.result(after=state), jusqu'à ce qu'il se termine :

async def by_id() -> None:
    state = await assistant.run("Where am I, and what is my latest email about?")
    session = assistant.session(state.session_id)
    while isinstance(state, agents.Blocked):
        turn = session.turn(state.turn_id)
        for callback in state.callbacks:
            await answer_callback(turn, callback)
        state = await turn.result(after=state)
    await session.cancel()

async def by_id() -> None:
    state = await assistant.run("Where am I, and what is my latest email about?")
    session = assistant.session(state.session_id)
    while isinstance(state, agents.Blocked):
        turn = session.turn(state.turn_id)
        for callback in state.callbacks:
            await answer_callback(turn, callback)
        state = await turn.result(after=state)
    await session.cancel()

Parcourir une session étape par étape exécute cette boucle pour les approbations et en confie certaines à une personne.

Une application de session démarre un tour avec send_message() et affiche ses événements. Pour que les fonctions répondent pendant que l'application affiche le tour, passez-les à la session et attendez avec turn.result(on_event=). Cette méthode appelle votre fonction à chaque nouvel événement :

def show(event: agents.Event) -> None:
    match event:
        case agents.CallbackOpened(callback=callback):
            print(f"waiting on: {callback.kind}")  # in an app: a button for a person


async def show_and_answer() -> None:
    session = await assistant.start_session(
        on_approval=approve_reads,
        on_client_tool=agents.ClientToolAnswers(get_location.answer(read_city)),
        on_sign_in=refuse_sign_ins,
    )
    turn = await session.send_message("Where am I, and what is my latest email about?")
    state = await turn.result(on_event=show)
    while isinstance(state, agents.Blocked):
        for callback in state.callbacks:
            await answer_callback(turn, callback)  # in an app: when the person clicks
        state = await turn.result(after=state, on_event=show)
    await session.cancel()

def show(event: agents.Event) -> None:
    match event:
        case agents.CallbackOpened(callback=callback):
            print(f"waiting on: {callback.kind}")  # in an app: a button for a person


async def show_and_answer() -> None:
    session = await assistant.start_session(
        on_approval=approve_reads,
        on_client_tool=agents.ClientToolAnswers(get_location.answer(read_city)),
        on_sign_in=refuse_sign_ins,
    )
    turn = await session.send_message("Where am I, and what is my latest email about?")
    state = await turn.result(on_event=show)
    while isinstance(state, agents.Blocked):
        for callback in state.callbacks:
            await answer_callback(turn, callback)  # in an app: when the person clicks
        state = await turn.result(after=state, on_event=show)
    await session.cancel()

iter_events() ne fait que lire : les fonctions passées à la session ne répondent jamais pendant son exécution. Une application dans laquelle une personne répond à chaque callback suit le tour avec iter_events() et répond à chaque callback par son identifiant à l'arrivée de son événement CallbackOpened, avec la même fonction answer_callback() :

async def from_app() -> None:
    session = await assistant.start_session()
    turn = await session.send_message("Where am I, and what is my latest email about?")
    async for event in turn.iter_events():
        match event:
            case agents.CallbackOpened(callback=callback):
                await answer_callback(turn, callback)  # in an app: when the person clicks
    await session.cancel()

async def from_app() -> None:
    session = await assistant.start_session()
    turn = await session.send_message("Where am I, and what is my latest email about?")
    async for event in turn.iter_events():
        match event:
            case agents.CallbackOpened(callback=callback):
                await answer_callback(turn, callback)  # in an app: when the person clicks
    await session.cancel()

turn.iter_events() se termine avec l'événement TurnEnded du tour. Un événement CallbackResolved indique qu'un callback n'est plus ouvert : un client, celui-ci ou un autre, y a répondu, ou son tour s'est terminé. Construire une application de session affiche chaque approbation ouverte sous forme de bouton et le supprime à la réception de CallbackResolved.

Par défaut, une approbation ou un appel d'outil côté client attend une réponse pendant une heure : consultez Délai d'expiration des approbations.

• run() renvoie Blocked. Un callback est ouvert : aucune fonction n'y a répondu, ou on_approval= a renvoyé KeepCallbackOpen(). Répondez-y par son identifiant ou passez une fonction correspondant à son type.
• Les fonctions ne s'exécutent jamais dans une application de session. Elles ne répondent que pendant l'attente de run() ou de turn.result(), pas dans iter_events(). Attendez avec turn.result(on_event=) ou répondez à chaque CallbackOpened par son identifiant.
• turn.answer() lève une exception agents.SdkError. La réponse ne correspond pas au callback : une réponse d'approbation pour un appel d'outil côté client, ou un SignInAnswer qui omet une source demandée pour la connexion. Rien n'a été envoyé.

• Approbations : quels appels d'outils demandent une approbation et comment y répondre.
• Construire une application de session : afficher les callbacks sous forme de boutons sur lesquels une personne peut cliquer.
• Référence API : OpenCallback, Turn.answer et CallbackOpened.