Toute l’API sur une seule page

from mistralai import agents

agent = agents.Agent(model="mistral-large-4", instructions="You review pull requests.")

session = await agent.start_session(on_approval=policy, idempotency_key=key)
session = await agent.start_session(agent={"model": "mistral-large-4"})   # or an agents.SessionAgentCustomization
session = await agent.start_session(session_inputs={"acme.ticket": ticket})  # agents.SessionInput instances
session = await agent.start_session(history=[agents.HistoryImportMessage(role="user", content=blocks)])  # local only
session = agent.session(session_id, on_approval=policy)   # bind an existing session: no request
turn = session.turn(turn_id)                              # bind an existing turn: no request
agent = agents.Agent.load_from(agent_name="triage")       # a deployed agent, by name or agent_id=

turn = await session.send_message("also check CI", idempotency_key=key)   # a new turn, or steers the running one
turn = await session.start_turn("Deploy to staging", idempotency_key=key) # a Turn, or agents.Busy: never steers

state = await turn.result(timeout=600)                    # waits for the end of the turn, or Blocked
state = await turn.result(after=state)                    # waits past what state showed
state = await session.run("Deploy to staging")            # start_turn, then result
state = await session.run("Draft a plan", on_event=show)  # show(event) for each event
reviewer = agents.Agent(model="mistral-medium-latest", instructions="You review pull requests.")
state = await reviewer.run("Review this diff", output_type=Review)   # start_session, then run

state = await turn.get()
async for event in turn.iter_events(): ...                # this turn only; stops at its end
async for event in session.iter_events(after=state.cursor): ...   # the whole session; waits for new events
children = await session.subagents()
async for info in agents.list_sessions(status="running"): ...     # deployed sessions by default
async for info in agents.list_sessions(host="local", scope="tree", root=session.id, limit=100): ...
async with agents.Client(api_key=api_key) as client:      # your key, URLs and connection pool
    async for info in client.list_sessions(): ...         # or agents.list_sessions(client=client)

await turn.steer("use the staging cluster")               # Turn | TurnConflict
await turn.answer(callback_id, agents.AllowApproval())    # Answered | AlreadySettled | NotFound
await turn.interrupt()                                    # Turn | TurnConflict
await session.cancel()                                    # ends the session and its subagents

class Agent:
    def __init__(self, *, model, instructions=None, name="default_agent", description=None,
                 tools=None, environment=None, connectors=None, mcps=None, skills=None,
                 hooks=None, guardrails=None, subagents=(), subagent_policy=None,
                 permission_mode="auto", permission_policy=None, output_type=None,
                 tool_errors="hide", secrets=None, redaction=None, builtins=None,
                 provider="mistral", temperature=None, reasoning_effort=None,
                 max_iterations=None, first_output_timeout_seconds=None,
                 auto_compact_threshold=None): ...
    def override(self, *, model: testing.ScriptedModel) -> Agent: ...   # the same agent on a scripted model

environment=agents.MistralSandbox(network=agents.Network.allow(hosts=["api.github.com"]))
    # every sandbox is an agents.Sandbox: NoSandbox, MistralSandbox, LocalDockerSandbox,
    # BashkitSandbox, DangerousLocalSandbox, experimental.CustomSandbox
permission_policy=agents.PermissionPolicy(default="ask", rules=[agents.PermissionRule(...)])
subagent_policy=agents.SubagentPolicy(max_depth=2)   # on the root agent

class Agent:
    async def start_session(self, *, history=None, session_inputs=None, agent=None,
                            idempotency_key=None, on_approval=None, on_sign_in=None,
                            on_client_tool=None) -> Session: ...
    def session(self, session_id: str, *, on_approval=None, on_sign_in=None,
                on_client_tool=None) -> Session: ...
    async def run(self, content, *, output_type=None, on_approval=None,
                  on_sign_in=None, on_client_tool=None, on_event=None,
                  idempotency_key=None, timeout=None) -> TurnState[T]: ...
                  # with idempotency_key=: TurnState[T] | Busy
    # @classmethod Agent.load_from(agent_name="triage", client=client) -> Agent: a deployed agent
    #   agent_id=: names it by ID instead; pass one of the two
    #   client=, or server_url= and api_key= to build an agents.Client

class Session:
    id: str
    def turn(self, turn_id: str) -> Turn: ...
    async def send_message(self, content, *, idempotency_key=None) -> Turn: ...
    async def start_turn(self, content, *, idempotency_key=None) -> Turn | Busy: ...
    async def run(self, content, *, output_type=None, idempotency_key=None,
                  timeout=None, on_event=None) -> TurnState[T] | Busy: ...
    def iter_events(self, *, after=None) -> AsyncIterator[Event]: ...
    async def subagents(self) -> list[Session]: ...
    async def cancel(self) -> None: ...

class Turn:
    id: str
    started: bool | None      # from send_message: True new turn, False steered; None when bound by id
    async def get(self) -> TurnState[T]: ...
    async def result(self, *, output_type=None, after=None, timeout=None,
                     on_event=None) -> TurnState[T]: ...   # after=: any record, or a cursor
    def iter_events(self, *, after=None) -> AsyncIterator[Event]: ...
    async def answer(self, callback_id: str, answer: AllowApproval | DenyApproval
                     | SignInAnswer | object) -> Answered | AlreadySettled | NotFound: ...
                     # any open callback: an approval, a sign-in, or a client tool call's result
    async def steer(self, content) -> Turn | TurnConflict: ...
    async def interrupt(self) -> Turn | TurnConflict: ...

def list_sessions(*, host="deployed", scope="roots", root=None, parent=None,
                  status=None, limit=None, client=None) -> AsyncIterator[SessionInfo]: ...
    # host: "deployed" | "local"; scope: "roots" | "children" | "tree"
    # status: "running" | "waiting" | "terminated"

class Client:                 # async only; with no client, the SDK builds one from the environment
    def __init__(self, *, api_key=None, server_url=None, agents_url=None, async_client=None): ...
    def list_sessions(self, **filters) -> AsyncIterator[SessionInfo]: ...
    async def aclose(self) -> None: ...                # or: async with agents.Client(...) as client

async with agents.local_runtime(program_timeout_seconds=None, journal_factory=None) as runtime:
    # a LocalRuntime; journal_factory=: experimental, warns agents.ExperimentalWarning
    ...                       # the local work stops when the block exits
    runtime.list_sessions()   # runtime.program_timeout_seconds: read-only

class SessionInfo:            # frozen
    id; parent_id; root_id; title
    status: Literal["running", "waiting", "terminated"] | UnknownValue
    created_at; updated_at; archived_at
    termination: Termination | None     # None while the session lives
type Termination = CompletedTermination | CancelledTermination | FailedTermination | UnknownTermination
class CompletedTermination: stop_reason: Literal["iteration_limit", "skipped", "rejected", "output_limit"] | UnknownValue | None
class CancelledTermination: cause: str | None    # "lifetime_limit" | "idle_limit" | None
class FailedTermination: error: ProtocolError
class UnknownTermination: outcome: str; raw: dict  # an outcome added after your SDK version
class UnknownValue: text  # a status, kind or level added after your SDK version; str(value) gives it

type TurnState[T] = Running | Blocked | Completed[T] | Interrupted | Failed | UnknownOutcome

class Running: ...
class Blocked: callbacks: tuple[OpenCallback, ...]    # what the turn waits on
class Completed[T]: output: T | InvalidOutput | None; stop_reason: str | None; tool_calls: tuple[ToolCall, ...]
class Interrupted: cause: str     # "interrupted" | "cancelled" | "lifetime_limit" | "idle_limit"
class Failed: error: ProtocolError
class UnknownOutcome: outcome: str; raw: dict    # an outcome added after your SDK version

class ProtocolError: message: str; code: str | None; retryable: bool; details: dict | None
class InvalidOutput: raw: Any; error: str   # a reply that does not match output_type=
class ToolCall: name: str; arguments    # one tool call the model made, in Completed.tool_calls

class ApprovalHandler(Protocol):   # a plain or async function; any parameter names
    def __call__(self, ctx: ApprovalCallbackContext, request: ApprovalRequest, /)
        -> AllowApproval | DenyApproval | KeepCallbackOpen   # or an awaitable of one
        # KeepCallbackOpen() leaves it open; any other return, None included, raises SdkError
type SignInHandler = Callable[[OpenSignIn], SignInAnswer | Awaitable[SignInAnswer]]
                              # any other return, KeepCallbackOpen() included, raises SdkError

class ApprovalRequest: tool: ApprovalTool; hook; title; input
class ApprovalTool:
    id: str; name: str
    kind: Literal["builtin", "client", "mcp", "connector", "plugin"] | UnknownValue
    description; input_schema; output_schema
    annotations: ToolAnnotations     # the hints the tool declares; MCP's defaults for the rest
    definition: PythonTool | ClientTool | ToolDefinition | None    # your tool object; for an MCP or connector tool, the tool as its server listed it; .display_title on each
class ToolAnnotations: read_only_hint; destructive_hint; idempotent_hint; open_world_hint
class AllowApproval: scope: Literal["once", "session"] = "once"   # "session": this same call again
class DenyApproval: ...
class SignInAnswer: connected: tuple[str, ...]; denied: tuple[str, ...]   # source ids
class KeepCallbackOpen: ...      # an approval or client tool handler's answer that leaves it open

class OpenCallback: id; session_id; turn_id; kind; title
class OpenApproval(OpenCallback): request: ApprovalRequest       # the request on_approval gets
class OpenClientToolCall(OpenCallback): call: ClientToolCall     # a client tool call to answer
class OpenSignIn(OpenCallback): requests: tuple[SignInRequest, ...]   # sources to connect
class SignInRequest: source: Literal["connector", "mcp"] | UnknownValue; id; name
    mode: Literal["connect", "reconnect"] | UnknownValue
class UnknownOpenCallback(OpenCallback): raw: dict               # any other kind

@agents.tool                       # or @agents.tool(name=, namespace=, title=, description=, input_schema=,
def search(ctx: agents.ToolCallbackContext, /, query: str) -> list[str]: ...   # output_schema=, annotations=, model_access=)
                                   # ctx: optional, positional-only, first
search.tool_definition             # ToolDefinition: name, description, input_schema, output_schema
                                   # search is an agents.PythonTool
                                   # model_access=: an agents.ToolModelAccess, "programmatic" (default) | "direct" | "both"
agent = agents.Agent(tools=[search, plain_function])    # a plain function is a tool too

@agents.client_tool                # a stub: your program answers it
def get_location(precision: str = "city") -> Location: raise NotImplementedError
                                   # get_location is an agents.ClientTool
@get_location.answer               # typed against the stub; returns an agents.ClientToolAnswer
async def answer_location(precision: str = "city") -> Location: ...
answers = agents.ClientToolAnswers(answer_location, default=handle)   # default: every other client tool call
state = await agent.run("...", on_client_tool=answers)   # also on start_session() and agent.session(id)
async def handle(call: agents.ClientToolCall) -> object: ...

@agents.tool(namespace="billing")  # the tool's id is billing.refund; the default namespace is "client"
def refund(order_id: str) -> agents.ToolResult: ...
raise agents.ToolError("order not found")                 # the model sees this text
return agents.ToolResult.error("order not found")         # the same, as a value
agent = agents.Agent(tools=[refund], tool_errors="hide")  # other exceptions: "hide" (default) | "show" | "raise"

class ToolResult: content; structured_content; is_error; meta
    # content: TextContentBlock, ImageContentBlock, AudioContentBlock,
    #          ResourceLinkContentBlock, EmbeddedResourceContentBlock,
    #          UnknownContentBlock (original_type: a block type added after your SDK version)
    # a block's annotations: ContentAnnotations(audience, priority, last_modified)
class ToolError(Exception): message; structured_content; meta   # the model never sees meta

class CallbackContext: session: CallbackSession; sandbox   # the shared base
class CallbackSession: id; turn_id
# ctx.sandbox: workspace
#   await ctx.sandbox.execute(command, cwd=, timeout_seconds=, env=) -> SandboxExecutionResult
#   await ctx.sandbox.read_file(path, max_bytes=) -> bytes
#   await ctx.sandbox.write_file(name, content: bytes) -> str
class SandboxExecutionResult: exit_code: int; stdout: str; stderr: str
class HookCallbackContext(CallbackContext): ...    # hook methods
class ToolCallbackContext(CallbackContext): tool_call_id   # tool bodies and client tool answers
class ApprovalCallbackContext(CallbackContext): callback_id   # on_approval; its sandbox is unbound: calling it raises
class SandboxCallbackContext: ...                  # a sandbox hook's on_create and on_destroy

class Hook:                        # pass subclasses to hooks= or guardrails=
    async def pre_agent_turn(self, ctx: HookCallbackContext, request: hook.PreAgentTurn)
        # -> hook.PreAgentTurnHookContinue | hook.PreAgentTurnHookSkip | skip
    async def post_agent_turn(self, ctx, result: hook.PostAgentTurn)    # -> hook.PostAgentTurnHookAccept
    async def pre_llm_call(self, ctx, invocation: hook.PreLlmCall)
        # -> hook.PreLlmCallHookContinue | hook.PreLlmCallHookSkip | skip
    async def post_llm_call(self, ctx, invocation: hook.PostLlmCall)
        # -> hook.PostLlmCallHookAccept | hook.PostLlmCallHookRetry | hook.PostLlmCallHookReject | retry | reject
    async def pre_tool_call(self, ctx, call: hook.PreToolCall)
        # -> hook.PreToolCallHookContinue | hook.PreToolCallHookSkip | skip
    async def post_tool_call(self, ctx, result: hook.PostToolCall)     # -> hook.PostToolCallHookContinue

# what each method gets, in agents.hook; every one has turn_id
class PreAgentTurn: user_content: list[ContentBlock]
class PreLlmCall: ...
class PostLlmCall: content; text; tool_calls: tuple[ToolCall, ...]; reasoning; finish_reason; usage: TokenUsage | None
class PostAgentTurn: ...           # the same fields as PostLlmCall, for the turn's final answer
class PreToolCall: tool_name; tool_id; tool_call_id; arguments
class PostToolCall(PreToolCall): tool_result: ToolResult   # arguments: as the tool ran with them
class TokenUsage: input_tokens; output_tokens; total_tokens; cached_input_tokens; cache_write_tokens; reasoning_tokens
hook.PostLlmCallHookAccept(acceptance=hook.ReplaceAssistantContent(content=[agents.TextContentBlock(text="...")]))

agents.skip(reason) -> HookDecision        # pre_tool_call: the model reads reason as the result; pre_agent_turn, pre_llm_call: the turn ends "skipped"
agents.retry(feedback) -> HookDecision     # call the model again with this feedback
agents.reject(reason) -> HookDecision      # end the turn
class HookDecision: kind; message

class SandboxHook:                 # pass to a sandbox's hooks=
    async def on_create(self, ctx: SandboxCallbackContext) -> None: ...
    async def on_destroy(self, ctx: SandboxCallbackContext) -> None: ...

agents.pattern.hook.StructuredOutputHook[City](max_retries=2)              # a ready-made hook
agents.pattern.sandbox_hook.GitIdentity(name="Acme Bot", email="bot@acme.com")  # a ready-made sandbox hook

class Busy: session_id; active_turn_id: str; turn: Turn    # the running turn
class TurnConflict: session_id; expected_turn_id; active_turn_id
class Answered: session_id; callback_id; answer: AllowApproval | DenyApproval | None   # the answer that counted
    # None for a client tool call or a sign-in, or when the answer that counted could not be read in time
class AlreadySettled: session_id; callback_id
class NotFound: session_id; callback_id

# history entries: frozen
class MessageEntry: id; turn_id; role: Literal["user", "assistant", "system", "context"]; content; text
class ContextMessage(MessageEntry): kind: Literal["input", "capability_source"]; source: str | None   # role="context"
class ToolCallEntry: id; turn_id; tool_call_id; name; arguments
class ToolResultEntry: id; turn_id; tool_call_id; name; content; structured_content; is_error; meta
class UnknownEntry: id; turn_id; type; raw          # an entry of a type added after your SDK version, or one read only as raw data, such as a checkpoint

class TurnStarted(Event): ...
class TurnUpdated(Event): status: Literal["running", "blocked", "terminated"] | UnknownValue
class TurnEnded(Event): state: Completed | Interrupted | Failed | UnknownOutcome
class CallbackOpened(Event): callback: OpenCallback
class CallbackResolved(Event): callback_id: str
class SessionUpdated(Event): status: Literal["running", "waiting", "terminated"] | UnknownValue
    title: str | None; archived: bool
class EntryAdded(Event): entry: MessageEntry | ToolCallEntry | ToolResultEntry | UnknownEntry       # started, may be partial
class EntryCompleted(Event): entry: MessageEntry | ToolCallEntry | ToolResultEntry | UnknownEntry   # the final entry
class TextDelta(Event): entry_id: str; text: str      # a piece of a message, live only
class Notice(Event): message; source; details   # for you; the model does not get it
    level: Literal["info", "warning", "error"] | UnknownValue
class UnknownEvent(Event): type: str; raw: dict    # an event type added after your SDK version

class AgentsError(Exception)
class SdkError(AgentsError, ValueError)                     # an invalid call or setting
class UnsupportedAgentFeatureError(SdkError, RuntimeError)  # a feature this agent's runtime cannot run
class LegacySessionError(UnsupportedAgentFeatureError)      # a session an older version of the platform created: start a new one
class SessionNotFoundError(AgentsError, LookupError)
class SessionIdempotencyConflictError(AgentsError, RuntimeError)   # the same idempotency_key with other options
class TurnNotFoundError(AgentsError, LookupError)
class InvalidCursorError(SdkError)                          # after= is not a position of the session
class SessionBusyError(AgentsError, RuntimeError)           # nothing was applied: retry the same request later
class SessionTerminatedError(AgentsError, RuntimeError)     # the session has ended and takes no more calls
class RemoteAgentDeploymentNotFoundError(AgentsError, RuntimeError)   # the deployment behind the handle is gone
class ObservationLost(AgentsError, RuntimeError)            # the event stream was lost
class OutcomeUnknownError(AgentsError, RuntimeError)        # a request may or may not have landed
class SessionStartupTimeoutError(OutcomeUnknownError)
class SessionInUseError(AgentsError, RuntimeError)          # another process holds a saved session
class SessionConfigMismatchError(AgentsError, RuntimeError) # a saved session, another agent setup
class ClientToolAnswerError(AgentsError, RuntimeError)      # a client tool answer could not be sent
class MistralConnectorNotFoundError(AgentsError, LookupError)   # not raised for connectors=: the session sends a Notice
class UnsupportedSessionProcedureError(AgentsError, RuntimeError)   # this session cannot run the call, such as an upload
class APIStatusError(AgentsError, httpx.HTTPStatusError)
    # AuthenticationError, PermissionDeniedError, RateLimitError, InternalServerError
class APIConnectionError(AgentsError, httpx.TransportError)
class APITimeoutError(APIConnectionError, httpx.TimeoutException)

# a warning, not an error
class ExperimentalWarning(UserWarning)                      # see The experimental layer below