Liste des sessions, Client et environnement d'exécution local

Listez les sessions d'un hôte, conservez vos paramètres de connexion dans un Client et délimitez la portée des sessions locales d'un programme.

Guides : Modes d'exécution, Environnement d'exécution local et Statuts des sessions.

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]

Itérer sur les sessions d'un hôte, de la plus récente à la plus ancienne, en lisant les pages au fur et à mesure des besoins.

scope="roots" (par défaut) liste les sessions racines, scope="children", parent=id les sous-agents directs d'une session, et scope="tree", root=id une session racine et tous ses sous-agents. status= ne conserve que les sessions ayant ce statut. limit= arrête la lecture après ce nombre de sessions. Les sessions archivées ne sont pas listées.

host="deployed" (par défaut) liste, via client, les sessions de vos agents déployés que votre clé API a démarrées. Sans client, un agents.Client() par défaut est utilisé : la clé API et les URL proviennent de l'environnement, et l'absence de MISTRAL_API_KEY lève SdkError lors de l'appel. list_sessions(client=client) équivaut à client.list_sessions(). host="local" liste les sessions de l'environnement d'exécution local de cette boucle d'événements, comme agents.local_runtime().list_sessions(), et n'accepte pas de client.

Une valeur de root= ou de parent= qui n'est pas un UUID lève SdkError lors de l'appel. Si elle ne désigne aucune session dans cet environnement, elle lève SessionNotFoundError, une LookupError dont session_id correspond à cet identifiant. Il en va de même si un sous-agent est utilisé comme root= avec scope="tree", pour les deux valeurs de host. Tout autre argument incorrect lève également SdkError lors de l'appel.

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

Une session renvoyée dans une liste. Les dates et heures sont en UTC.

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

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

La manière dont une session s'est terminée : une classe par résultat, et UnknownTermination pour un résultat ajouté après votre version du SDK.

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

La session s'est terminée.

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

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

La session a été annulée.

class FailedTermination(*, error: ProtocolError)

class FailedTermination(*, error: ProtocolError)

La session a échoué.

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

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

La session s'est terminée avec un résultat ajouté après votre version du SDK, comme "future_outcome".

Prévoyez un case pour cet enregistrement afin que votre code continue de fonctionner lorsque de nouveaux résultats sont ajoutés.

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

Une connexion à l'API Mistral avec une clé API. Tous les appels sont asynchrones.

api_key utilise MISTRAL_API_KEY par défaut ; si aucune des deux valeurs ne fournit de clé, Client() lève SdkError. server_url est l'URL racine de l'API Mistral : MISTRAL_API_URL, ou à défaut https://api.mistral.ai. agents_url est l'URL des endpoints d'agents de l'API Mistral, par défaut l'URL racine de l'API + /v2/agents. Une seule URL suffit donc pour un autre cloud. Sur une plateforme Mistral sur site, l'opérateur peut définir AGENTS_SDK_DEPLOY_CONTROL_PLANE_URL à la place. Cette variable s'applique lorsque server_url n'est pas fourni.

async_client est votre propre httpx.AsyncClient pour les requêtes, comme dans mistralai.Mistral ; ses délais d'expiration et ses limites de connexion s'appliquent, et le client ne le ferme jamais. Sans lui, le client ouvre son propre pool de connexions lors du premier appel. await client.aclose() ou la sortie de async with agents.Client() as client: le ferme. Créer un client n'envoie rien et ne nécessite aucune boucle d'événements. Son propre pool sert une seule boucle d'événements : un appel depuis une boucle ultérieure (un deuxième asyncio.run) ouvre un nouveau pool. Un appel effectué alors que la boucle d'événements d'un autre thread l'utilise encore lève SdkError. aclose ferme le pool de la boucle d'événements dans laquelle il s'exécute ; appelez-le donc dans la boucle qui a utilisé le client en dernier. Un async_client sert également une seule boucle d'événements. Les agents chargés avec Agent.load_from(client=client) et leurs sessions envoient leurs requêtes via le client ; une fois celui-ci fermé, ils lèvent 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]

Itérer sur les sessions déployées associées à la clé API de ce client, de la plus récente à la plus ancienne.

Les arguments nommés sont ceux de agents.list_sessions. Les pages sont lues via le pool de connexions de ce client au fur et à mesure de l'itération.

async def aclose() -> None

async def aclose() -> None

Ferme le propre pool de connexions du client. Un async_client fourni reste ouvert.

class UnknownValue(text: str)

class UnknownValue(text: str)

Une valeur ajoutée après votre version du SDK, comme le statut d'un tour "paused".

Un champ pouvant recevoir de nouvelles valeurs, comme TurnUpdated.status, a pour type l'union de ses valeurs connues et de UnknownValue. text contient la valeur telle que Mistral l'a envoyée, et str(value) la renvoie également. Ce n'est pas un str : un vérificateur de types signale donc toujours une faute de frappe dans une valeur connue.

Utilisez cette valeur pour l'écrire dans un log ou appliquer un comportement de repli, jamais pour filtrer selon son texte : dès qu'un SDK ultérieur reconnaît la valeur, elle est lue comme une simple chaîne, et case agents.UnknownValue(text="paused") ne correspond plus, sans générer d'erreur. str(status) == "paused" fonctionne avant comme après.

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

Renvoie l'environnement d'exécution local de la boucle d'événements en cours, uniquement pour les appels sur les sessions locales.

program_timeout_seconds définit le temps alloué à chaque programme, c'est-à-dire au code que le modèle exécute avec run_typescript, dans toutes les sessions locales de la boucle d'événements. La valeur par défaut est de 60 secondes et elle doit être supérieure à zéro et ne pas dépasser 4 heures (14 400 secondes). Elle s'applique à l'entrée dans le bloc async with. Le temps alloué est fixé au démarrage de l'environnement d'exécution de la boucle ; entrez donc dans le bloc avant la première session locale : async with agents.local_runtime(program_timeout_seconds=600):. Entrer dans le bloc alors que l'environnement d'exécution fonctionne avec un autre temps alloué lève SdkError. list_sessions() ne définit ni ne vérifie le temps alloué.

La sortie du bloc le plus externe termine et libère les sessions de l'environnement d'exécution : tout appel ultérieur sur l'une d'elles lève SessionTerminatedError. Chaque boucle d'événements conserve les 10 000 dernières sessions libérées de cette manière ; un appel sur une session plus ancienne lève à la place SessionNotFoundError.

class LocalRuntime

class LocalRuntime

L'environnement d'exécution local de la boucle d'événements en cours : celui que agent.run et agent.start_session utilisent pour les agents locaux.

L'obtenir ne démarre rien. Chaque appel atteint le runtime de la boucle d'événements dans laquelle il s'exécute ; il doit donc s'exécuter dans une telle boucle. Deux appels à asyncio.run ont deux runtimes, qui ne partagent aucune session.

async with agents.local_runtime() as runtime: démarre le runtime s'il n'est pas démarré, et sa sortie arrête toutes les sessions de ce runtime : les tours en cours, les sous-agents et les sandbox. La sortie libère les sessions arrêtées : une lecture ou une commande sur l'une d'elles lève alors SessionTerminatedError. Avec journal_factory, un nouveau bloc avec les mêmes journaux les reprend. L'appel local suivant démarre un nouveau runtime. Dans un bloc imbriqué, seule la sortie du bloc le plus externe les arrête.

Tant que le bloc est ouvert, aucune session n'est fermée pour inactivité : une session qu'aucun client ne détient, comme celle d'un agent.run qui a retourné, garde sa sandbox et peut être pilotée à nouveau jusqu'à la sortie du bloc ou session.cancel(). En dehors d'un bloc, une telle session se termine dès qu'elle est inactive et libère sa 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]

Itérer sur les sessions de ce runtime, les plus récentes en premier.

Les arguments sont ceux de agents.list_sessions. Une session terminée est listée comme "terminated" jusqu'à l'arrêt de la boucle d'événements. Lister ne démarre rien : avant le démarrage du runtime, aucune session n'est listée.