Approbations

Avant qu'un appel d'outil s'exécute, le tour peut demander l'autorisation : cette question est une approbation.

• La politique d'autorisation de l'Agent décide quels appels s'exécutent, lesquels demandent une approbation et lesquels sont refusés.
• Une fonction on_approval= dans votre programme répond à certaines approbations, comme tous les appels en lecture seule.
• Pour les autres, run() renvoie agents.Blocked jusqu'à ce qu'une personne y réponde par identifiant.

Les exemples utilisent le relecteur de paiements de Ce qu'est un agent.

Une PermissionPolicy sur l'Agent décide, par outil et par argument, si un appel s'exécute, demande une approbation ou est refusé. Ici, le relecteur lit une alerte sans demander d'approbation et en demande une avant d'ouvrir un dossier :

import asyncio
import uuid

from mistralai import agents

from reviewer import get_alert, open_case, payment_reviewer

rules = agents.PermissionPolicy(
    default="deny",
    rules=[
        agents.PermissionRule(tools=["client.get_alert"], decision="allow"),
        agents.PermissionRule(tools=["client.open_case"], decision="ask"),  # opening a case asks
    ],
)

reviewer = agents.Agent(
    model="mistral-large-4",
    name="payment-reviewer",
    instructions=payment_reviewer.instructions,
    tools=[get_alert, open_case],
    permission_policy=rules,
)


async def policy() -> None:
    state = await reviewer.run("Review alert 4711.")
    print(type(state).__name__, [callback.title for callback in state.callbacks])
    await reviewer.session(state.session_id).cancel()


asyncio.run(policy())

import asyncio
import uuid

from mistralai import agents

from reviewer import get_alert, open_case, payment_reviewer

rules = agents.PermissionPolicy(
    default="deny",
    rules=[
        agents.PermissionRule(tools=["client.get_alert"], decision="allow"),
        agents.PermissionRule(tools=["client.open_case"], decision="ask"),  # opening a case asks
    ],
)

reviewer = agents.Agent(
    model="mistral-large-4",
    name="payment-reviewer",
    instructions=payment_reviewer.instructions,
    tools=[get_alert, open_case],
    permission_policy=rules,
)


async def policy() -> None:
    state = await reviewer.run("Review alert 4711.")
    print(type(state).__name__, [callback.title for callback in state.callbacks])
    await reviewer.session(state.session_id).cancel()


asyncio.run(policy())

Blocked ['Allow tools.client.open_case?']

Blocked ['Allow tools.client.open_case?']

La lecture de l'alerte s'est exécutée. L'ouverture d'un dossier a demandé une approbation, et rien n'y a répondu, donc l'exécution a renvoyé Blocked.

Les motifs du champ tools d'une règle correspondent à l'identifiant de l'outil appelé, comme client.open_case, et non au nom que voit le modèle : les identifiants de vos outils commencent par leur espace de noms, client par défaut. La première règle qui correspond décide, et un appel auquel aucune règle ne correspond reçoit la valeur default de la politique. PermissionRule liste les autres champs sur lesquels une règle peut se baser.

Sans règles, permission_mode= sur l'Agent définit une même décision pour tous les appels :

Une permission_policy nécessite permission_mode="auto" ; avec un autre mode, l'Agent lève une exception. Les sous-agents utilisent la politique de l'Agent racine : voir Ce dont un sous-agent hérite.

Les exemples ci-dessous utilisent le même relecteur avec permission_mode="ask" au lieu de la politique, si bien que chaque appel d'outil demande une approbation. get_alert déclare être en lecture seule :

asking_reviewer = agents.Agent(
    model="mistral-large-4",
    name="payment-reviewer",
    instructions=payment_reviewer.instructions,
    tools=[get_alert, open_case],
    permission_mode="ask",  # every tool call asks for approval
)

asking_reviewer = agents.Agent(
    model="mistral-large-4",
    name="payment-reviewer",
    instructions=payment_reviewer.instructions,
    tools=[get_alert, open_case],
    permission_mode="ask",  # every tool call asks for approval
)

Passez une fonction en on_approval= à agent.start_session(), agent.session(id) ou agent.run(). Elle reçoit le contexte et la requête de l'approbation, et renvoie agents.AllowApproval(), agents.DenyApproval() ou agents.KeepCallbackOpen() pour laisser l'approbation à une personne :

def approve_reads(
    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: a person decides


async def review() -> None:
    session = await asking_reviewer.start_session(on_approval=approve_reads)
    state = await session.run("Review alert 4711.")
    print(type(state).__name__)
    if isinstance(state, agents.Blocked):
        for callback in state.callbacks:
            print(callback.title, callback.request.input["alert_id"])
    await session.cancel()


asyncio.run(review())

def approve_reads(
    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: a person decides


async def review() -> None:
    session = await asking_reviewer.start_session(on_approval=approve_reads)
    state = await session.run("Review alert 4711.")
    print(type(state).__name__)
    if isinstance(state, agents.Blocked):
        for callback in state.callbacks:
            print(callback.title, callback.request.input["alert_id"])
    await session.cancel()


asyncio.run(review())

Blocked
Allow tools.client.open_case? 4711

Blocked
Allow tools.client.open_case? 4711

La fonction a autorisé get_alert. Pour open_case, elle a renvoyé KeepCallbackOpen(), donc session.run() a renvoyé Blocked, avec l'approbation ouverte dans callbacks. Une personne y répond par identifiant : avec DenyApproval(), l'appel ne s'exécute pas, et le modèle lit l'erreur d'outil Tool call was rejected during approval. Ensuite, votre programme attend de nouveau le tour : Parcourir les approbations pas à pas montre cette boucle.

La fonction s'exécute dans votre programme, et seulement pendant que agent.run(), session.run() ou turn.result() attend. Elle répond aussi aux approbations des sous-agents de la session. Décidez selon request.tool.id, et non selon le nom que le modèle a utilisé pour l'appeler. ApprovalHandler liste ce qu'il peut renvoyer.

turn.answer(callback_id, answer) répond à une approbation ouverte de la session ou de l'un de ses sous-agents, depuis n'importe quel tour, et depuis n'importe quel processus capable d'ouvrir la session. Dans un autre processus, ouvrez le tour par ses identifiants : agent.session(session_id).turn(turn_id).answer(callback_id, answer), comme dans Ouvrir une session par son identifiant.

Comparez la réponse qui a fait foi avec la vôtre :

async def answer_once() -> None:
    session = await asking_reviewer.start_session(on_approval=approve_reads)
    state = await session.run("Review alert 4711.")
    assert isinstance(state, agents.Blocked)
    turn = session.turn(state.turn_id)
    callback_id = state.callbacks[0].id

    mine = agents.DenyApproval()
    match await turn.answer(callback_id, mine):
        case agents.Answered(answer=applied) if applied != mine:
            print("another client answered first:", applied)
        case agents.Answered() as answered:
            print(answered)
        case agents.AlreadySettled() | agents.NotFound() as other:
            print(other)

    print(await turn.answer(callback_id, agents.AllowApproval()))
    print(await turn.answer(str(uuid.uuid4()), agents.AllowApproval()))
    await session.cancel()


asyncio.run(answer_once())

async def answer_once() -> None:
    session = await asking_reviewer.start_session(on_approval=approve_reads)
    state = await session.run("Review alert 4711.")
    assert isinstance(state, agents.Blocked)
    turn = session.turn(state.turn_id)
    callback_id = state.callbacks[0].id

    mine = agents.DenyApproval()
    match await turn.answer(callback_id, mine):
        case agents.Answered(answer=applied) if applied != mine:
            print("another client answered first:", applied)
        case agents.Answered() as answered:
            print(answered)
        case agents.AlreadySettled() | agents.NotFound() as other:
            print(other)

    print(await turn.answer(callback_id, agents.AllowApproval()))
    print(await turn.answer(str(uuid.uuid4()), agents.AllowApproval()))
    await session.cancel()


asyncio.run(answer_once())

Answered(session_id='87ea6b10-...', callback_id='756f3ba3-...', answer=DenyApproval(decision='deny'))
AlreadySettled(session_id='87ea6b10-...', callback_id='756f3ba3-...')
NotFound(session_id='87ea6b10-...', callback_id='c9525f8a-...')

Answered(session_id='87ea6b10-...', callback_id='756f3ba3-...', answer=DenyApproval(decision='deny'))
AlreadySettled(session_id='87ea6b10-...', callback_id='756f3ba3-...')
NotFound(session_id='87ea6b10-...', callback_id='c9525f8a-...')

Answered signifie que l'approbation était ouverte et a maintenant reçu une réponse. AlreadySettled signifie qu'elle n'est plus ouverte : quelqu'un a répondu en premier, elle a expiré, ou son tour ou sa session a pris fin. NotFound signifie que la session ne l'a jamais ouverte, ou qu'il s'agissait de l'approbation d'un sous-agent, qui n'est plus ouverte. Turn.answer liste les erreurs qu'il lève et comment répondre à l'approbation d'un sous-agent.

Une approbation attend pendant une heure. Quand personne ne répond, l'appel ne s'exécute pas, le modèle lit que personne ne l'a approuvé, et le tour continue.

Une approbation est un type de callback : une question qu'un tour pose à votre programme et dont il attend la réponse. Un tour attend aussi les appels d'outils client, traités avec on_client_tool=, et les connexions aux connecteurs, traitées avec on_sign_in=.

• run() renvoie Blocked pour une approbation. L'agent demande une approbation, via permission_mode="ask" ou une règle qui demande, et aucune fonction on_approval= n'y a répondu. Passez une fonction, ou répondez par identifiant.
• La fonction on_approval= ne s'exécute jamais. Elle ne s'exécute que pendant qu'un appel attend, sur un handle auquel elle a été passée. Passez-la à agent.session(id, on_approval=) quand vous rouvrez la session.
• Aucune règle ne correspond. Les règles correspondent à l'identifiant stable de l'outil, comme client.open_case, et non au nom que voit le modèle.
• AllowApproval(scope="session") redemande une approbation. scope="session" autorise aussi les appels ultérieurs de la session au même outil, avec les mêmes arguments : un appel avec d'autres arguments redemande une approbation.

• Parcourir une session pas à pas : répondez aux approbations en boucle, tour après tour.
• Construire une application de session : affichez les approbations sous forme de boutons sur lesquels une personne clique.
• Référence API : Approbations et connexions : tous les types de réponse et de résultat.