Listing, Client and local runtime

List the sessions of a host, hold your own connection settings in a Client, and scope the local sessions of a program.

Guides: Run modes, Local runtime and Session statuses.

def agents.list_sessions(
    *,
    host: Literal['deployed', 'local'] = 'deployed',
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
    client: Client | None = None,
) -> AsyncIterator[SessionInfo]

def agents.list_sessions(
    *,
    host: Literal['deployed', 'local'] = 'deployed',
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
    client: Client | None = None,
) -> AsyncIterator[SessionInfo]

Iterate over the sessions of a host, newest first, reading pages as they are needed.

scope="roots" (the default) lists root sessions, scope="children", parent=id the direct subagents of a session, and scope="tree", root=id a root session and all its subagents. status= keeps only the sessions in that status. limit= stops after that many sessions. Archived sessions are not listed.

host="deployed" (the default) lists the sessions of your deployed agents that your API key started, through client. Without one, a default agents.Client() is used: the API key and URLs come from the environment, and a missing MISTRAL_API_KEY raises SdkError at the call. list_sessions(client=client) is client.list_sessions(). host="local" lists the sessions of this event loop's local runtime, as agents.local_runtime().list_sessions() does, and takes no client.

A root= or parent= that is not a UUID raises SdkError at the call. One that names no session there raises SessionNotFoundError, a LookupError, whose session_id is that ID, and so does a subagent as the root= of scope="tree", for both host values. Any other wrong argument raises SdkError at the call too.

class SessionInfo(
    *,
    id: str,
    parent_id: str | None,
    root_id: str,
    title: str | None,
    status: Literal['running', 'waiting', 'terminated'] | UnknownValue,
    created_at: datetime,
    updated_at: datetime,
    archived_at: datetime | None,
    termination: Termination | None,
)

class SessionInfo(
    *,
    id: str,
    parent_id: str | None,
    root_id: str,
    title: str | None,
    status: Literal['running', 'waiting', 'terminated'] | UnknownValue,
    created_at: datetime,
    updated_at: datetime,
    archived_at: datetime | None,
    termination: Termination | None,
)

One session found in a listing. The times are in UTC.

type Termination = (
    CompletedTermination
    | CancelledTermination
    | FailedTermination
    | UnknownTermination
)

type Termination = (
    CompletedTermination
    | CancelledTermination
    | FailedTermination
    | UnknownTermination
)

How a session ended: one class for each outcome, and UnknownTermination for an outcome added after your SDK version.

class CompletedTermination(
    *,
    stop_reason: Literal['iteration_limit', 'skipped', 'rejected', 'output_limit'] | UnknownValue | None = None,
)

class CompletedTermination(
    *,
    stop_reason: Literal['iteration_limit', 'skipped', 'rejected', 'output_limit'] | UnknownValue | None = None,
)

The session completed.

class CancelledTermination(*, cause: str | None = None)

class CancelledTermination(*, cause: str | None = None)

The session was cancelled.

class FailedTermination(*, error: ProtocolError)

class FailedTermination(*, error: ProtocolError)

The session failed.

class UnknownTermination(*, outcome: str, raw: dict[str, JsonValue] = {})

class UnknownTermination(*, outcome: str, raw: dict[str, JsonValue] = {})

The session ended with an outcome added after your SDK version, such as "future_outcome".

Leave a case for this record, so your code still runs when outcomes are added.

class Client(
    *,
    api_key: str | None = None,
    server_url: str | None = None,
    agents_url: str | None = None,
    async_client: httpx.AsyncClient | None = None,
)

class Client(
    *,
    api_key: str | None = None,
    server_url: str | None = None,
    agents_url: str | None = None,
    async_client: httpx.AsyncClient | None = None,
)

A connection to the Mistral API with one API key. Every call is async.

api_key defaults to MISTRAL_API_KEY; with no key from either, Client() raises SdkError. server_url is the Mistral API root: MISTRAL_API_URL, else https://api.mistral.ai. agents_url is the URL of the agents endpoints of the Mistral API, by default the API root + /v2/agents, so one URL is enough for another cloud. On an on-premises Mistral platform, its operator may set AGENTS_SDK_DEPLOY_CONTROL_PLANE_URL instead, which applies when server_url is not given.

async_client is your own httpx.AsyncClient for the requests, as in mistralai.Mistral; its timeouts and connection limits apply, and the client never closes it. Without it, the client opens its own connection pool at its first call, and await client.aclose() or the end of async with agents.Client() as client: closes it. Creating a client sends nothing and needs no event loop. Its own pool serves one event loop: a call from a later one (a second asyncio.run) opens a new pool, and a call while another thread's event loop still uses it raises SdkError. aclose closes the pool of the event loop it runs in, so call it in the loop that used the client last. An async_client also serves one event loop. Agents loaded with Agent.load_from(client=client) and their sessions send their requests through the client; once it is closed, they raise SdkError.

def list_sessions(
    *,
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
) -> AsyncIterator[SessionInfo]

def list_sessions(
    *,
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
) -> AsyncIterator[SessionInfo]

Iterate over the deployed sessions of this client's API key, newest first.

The keywords are those of agents.list_sessions. Pages are read through this client's connection pool as you iterate.

async def aclose() -> None

async def aclose() -> None

Close the client's own connection pool. A given async_client stays open.

class UnknownValue(text: str)

class UnknownValue(text: str)

A value added after your SDK version, such as a turn status "paused".

A field that can get new values, such as TurnUpdated.status, is typed as its known values or UnknownValue. text is the value as Mistral sent it, and str(value) gives it too. It is not a str, so a type checker still flags a typo in a known value.

Use it to log or to fall back, never to match on its text: once a later SDK knows the value, it reads as the plain string, and case agents.UnknownValue(text="paused") stops matching with no error. str(status) == "paused" matches before and after.

def agents.local_runtime(
    *,
    program_timeout_seconds: float | None = None,
    journal_factory: Callable[[str], experimental.JsonFileSessionJournal] | None = None,
) -> LocalRuntime

def agents.local_runtime(
    *,
    program_timeout_seconds: float | None = None,
    journal_factory: Callable[[str], experimental.JsonFileSessionJournal] | None = None,
) -> LocalRuntime

Return the local runtime of the running event loop, for the calls on local sessions only.

program_timeout_seconds sets the time budget of each program, the code the model runs with run_typescript, in every local session of the event loop. It defaults to 60 seconds and must be greater than 0 and at most 4 hours (14400 seconds). It applies when the async with block is entered, and the budget is fixed when the loop's runtime starts, so enter the block before the first local session: async with agents.local_runtime(program_timeout_seconds=600):. Entering the block while the runtime runs with another budget raises SdkError. list_sessions() neither sets nor checks the budget.

The outermost block's exit ends and frees the sessions of the runtime: a later call on one raises SessionTerminatedError. Each event loop remembers the newest 10,000 sessions freed this way; a call on an older one raises SessionNotFoundError instead.

class LocalRuntime

class LocalRuntime

The local runtime of the running event loop: the one that agent.run and agent.start_session use for local agents.

Getting it starts nothing. Each call reaches the runtime of the event loop it runs in, so it must run inside one. Two asyncio.run calls have two runtimes, which share no session.

async with agents.local_runtime() as runtime: starts the runtime if it has not started, and its exit stops every session of that runtime: running turns, subagents and sandboxes. The exit frees the stopped sessions: a read or command on one then raises SessionTerminatedError. With journal_factory, a new block with the same journals continues them. The next local call starts a new runtime. In a nested block, only the outermost exit stops them.

While the block is open, no session is closed for being idle: a session that no client holds, such as the session of an agent.run that returned, keeps its sandbox and can be driven again until the block exits or session.cancel(). Outside a block, such a session ends once it is idle, and frees its sandbox.

def list_sessions(
    *,
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
) -> AsyncIterator[SessionInfo]

def list_sessions(
    *,
    scope: Literal['roots', 'children', 'tree'] = 'roots',
    root: str | None = None,
    parent: str | None = None,
    status: Literal['running', 'waiting', 'terminated'] | None = None,
    limit: int | None = None,
) -> AsyncIterator[SessionInfo]

Iterate over the sessions of this runtime, newest first.

The arguments are those of agents.list_sessions. An ended session is listed as "terminated" until the event loop stops. Listing starts nothing: before the runtime starts, it lists no session.