Home
Studio

Callbacks

A callback is a question a turn asks your program and waits for. The turn stops at the callback and continues once it has an answer. A turn asks three kinds:

CallbackThe turn asks whenYou answer with
Approvalthe agent's permissions or a hook ask before a tool callagents.AllowApproval() or agents.DenyApproval()
Client tool callthe model calls a client tool: a tool answered through a callback, not within the sessionthe tool's result
Sign-inthe agent needs a connector or MCP server the user has not connected, or must reconnectagents.SignInAnswer(connected=[...], denied=[...])

How you answer depends on how you drive the session:

  • With run(): functions you pass as on_approval=, on_client_tool= and on_sign_in= answer while the run waits. The callbacks they leave open come back in agents.Blocked, and you answer them by id.
  • In a session app, with functions: pass them to the session, and wait with turn.result(on_event=) while the app shows the events. Blocked returns what they leave open.
  • In a session app, by hand: follow the turn with iter_events(), and answer each agents.CallbackOpened by id.

Use functions when your program can decide. When a person or another process decides, leave the callback open and answer it by id. The two combine: the functions answer what they can, and Blocked returns the rest.

The example agent

The example agent

The examples use an assistant with a client tool and the gmail connector, made available on Mistral Studio (see Connectors). Its permission policy asks before each get_location call, so that your program decides. get_location declares that it only reads:

import asyncio
from typing import Literal

from pydantic import BaseModel

from mistralai import agents

class Location(BaseModel):
    city: str
    country: str

@agents.client_tool(annotations=agents.ToolAnnotations(read_only_hint=True))
def get_location(precision: Literal["city", "street"] = "city") -> Location:
    """The user's current location."""
    raise NotImplementedError  # answered through a callback

def read_location(precision: Literal["city", "street"] = "city") -> Location:
    return Location(city="Lyon", country="France")  # read it from the device

rules = agents.PermissionPolicy(
    default="allow",
    rules=[agents.PermissionRule(tools=["client.get_location"], decision="ask")],
)

assistant = agents.Agent(
    model="mistral-large-4",
    instructions="Use your tools. Answer in one short sentence.",
    tools=[get_location],
    connectors=[agents.connector("gmail")],
    permission_policy=rules,
)
Answer while a run waits

Answer while a run waits

Pass one function per kind to agent.run(), agent.start_session() or agent.session(id). They answer while agent.run(), session.run() or turn.result() waits. Add to the file of The example agent:

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

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_location)),
        on_sign_in=refuse_sign_ins,
    )
    print(type(state).__name__, state.text)
    await assistant.session(state.session_id).cancel()

asyncio.run(with_run())
Output
Completed You're in Lyon, France.
As for your latest email — I asked you to sign in to Gmail so I could check, but the sign-in was denied, so I wasn't able to read your inbox. If you'd like, I can ask again to enable the Gmail connector and then pull up your most recent email.

approve_reads allowed the get_location call, because the tool declares that it only reads. read_location answered the call, and refuse_sign_ins refused the gmail sign-in.

The functions also answer the callbacks of the session's subagents. on_approval=, on_client_tool= and a client tool's answer can return agents.KeepCallbackOpen() to leave the callback to a person. A callback that no function answers stays open, and the run returns Blocked.

Answer by id

Answer by id

Blocked lists the open callbacks in state.callbacks. Answer each one with turn.answer(callback.id, answer), from this process or, for a deployed agent, from any other process. Match the callback's class to pick the answer. Add to the file:

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_location))
        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

A run without functions returns Blocked as soon as the turn waits on a callback. Answer what it lists, then wait for the turn again with turn.result(after=state), until the turn ends. Add to the file:

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:
            print("answering:", callback.kind)
            await answer_callback(turn, callback)
        state = await turn.result(after=state)
    print(type(state).__name__, state.text)
    await session.cancel()

asyncio.run(by_id())
Output
answering: sign_in
answering: approval
answering: client_tool
Completed You're in Lyon, France.
As for your latest email: I asked to connect your Gmail account, but the sign-in was denied, so I wasn't able to access your inbox. If you'd like, I can ask again to authorize Gmail and then tell you what your latest email is about.

To hand some approvals to a person in this loop, see Step through a session.

Answer from a session app

Answer from a session app

A session app starts a turn with send_message() and shows its events. To let the functions answer while the app shows the turn, pass them to the session and wait with turn.result(on_event=). It calls your function with each event as it comes. Add to the file:

def show(event: agents.Event) -> None:
    match event:
        case agents.CallbackOpened(callback=callback):
            print(f"waiting on: {callback.kind}")  # in an app: show it until its CallbackResolved

async def show_and_answer() -> None:
    session = await assistant.start_session(
        on_approval=approve_reads,
        on_client_tool=agents.ClientToolAnswers(get_location.answer(read_location)),
        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)
    print(type(state).__name__, state.text)
    await session.cancel()

asyncio.run(show_and_answer())
Output
waiting on: sign_in
waiting on: approval
waiting on: client_tool
Completed You're in Lyon, France.
As for your latest email: the Gmail connector isn't signed in, and when I asked to enable it, the sign-in was denied — so I can't read your inbox right now. If you sign in to Gmail, I can check your latest email for you.

iter_events() only reads: functions passed to the session never answer while it runs. An app where a person answers every callback follows the turn with iter_events(), and answers each callback by id when its CallbackOpened event comes, with the same answer_callback(). Add to the file:

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):
                print("answering:", callback.kind)
                await answer_callback(turn, callback)  # in an app: when the person clicks
            case agents.TurnEnded(state=state):
                print(type(state).__name__)
    await session.cancel()

asyncio.run(from_app())
Output
answering: approval
answering: client_tool
answering: sign_in
Completed

turn.iter_events() ends with the turn's TurnEnded event. A CallbackResolved event says a callback is no longer open: a client answered it, or it closed because its turn ended or it timed out. Build a session app shows each open approval as a button, and removes it on CallbackResolved.

An approval or a client tool call waits one hour for an answer by default: see Approval timeout.

Common issues

Common issues

  • run() returns Blocked. A callback is open: no function answered it, or on_approval=, on_client_tool= or a client tool's answer returned KeepCallbackOpen(). Answer it by id, or pass a function for its kind.
  • The functions never run in a session app. They answer only while run() or turn.result() waits, not in iter_events(). Wait with turn.result(on_event=), or answer each CallbackOpened by id.
  • turn.answer() raises agents.SdkError. The answer does not fit the callback: an approval answer for a client tool call, a client tool result that does not match the tool's return type, or a SignInAnswer that does not name each source the sign-in asks for. Nothing was sent.