Hooks de sandbox

Un hook de sandbox exécute votre code à deux étapes du cycle de vie d'une sandbox. on_create s'exécute une seule fois lorsque la sandbox est prête, avant le démarrage de l'agent. on_destroy s'exécute une seule fois à la fermeture de la session. Utilisez-les pour initialiser un espace de travail ou effectuer le téléchargement de ce que l'agent a produit.

Prérequis : Installez le SDK et l'extra bashkit : uv add "mistralai-agents[bashkit]".

Créez une sous-classe de agents.SandboxHook et placez une instance dans le paramètre hooks= de la sandbox, pas dans celui de l'agent. Les deux étapes reçoivent ctx, un objet agents.SandboxCallbackContext. Son attribut ctx.sandbox exécute des commandes avec execute() et lit des fichiers avec read_file(). Les commandes s'exécutent dans l'espace de travail : les fichiers que vous y créez sont donc accessibles aux outils de l'agent :

from pathlib import Path

from mistralai import agents


class SeedWorkspace(agents.SandboxHook):
    async def on_create(self, ctx: agents.SandboxCallbackContext) -> None:
        result = await ctx.sandbox.execute("echo 'Run the tests with pytest.' > NOTES.md")
        if result.exit_code != 0:
            raise RuntimeError(f"seeding failed: {result.stderr}")

    async def on_destroy(self, ctx: agents.SandboxCallbackContext) -> None:
        report = await ctx.sandbox.read_file("NOTES.md")
        Path("saved-notes.md").write_bytes(report)  # in your process; deployed, upload it from the sandbox


agent = agents.Agent(
    model="mistral-large-4",
    environment=agents.BashkitSandbox(hooks=[SeedWorkspace()]),
)

from pathlib import Path

from mistralai import agents


class SeedWorkspace(agents.SandboxHook):
    async def on_create(self, ctx: agents.SandboxCallbackContext) -> None:
        result = await ctx.sandbox.execute("echo 'Run the tests with pytest.' > NOTES.md")
        if result.exit_code != 0:
            raise RuntimeError(f"seeding failed: {result.stderr}")

    async def on_destroy(self, ctx: agents.SandboxCallbackContext) -> None:
        report = await ctx.sandbox.read_file("NOTES.md")
        Path("saved-notes.md").write_bytes(report)  # in your process; deployed, upload it from the sandbox


agent = agents.Agent(
    model="mistral-large-4",
    environment=agents.BashkitSandbox(hooks=[SeedWorkspace()]),
)

Chaque session de cet agent démarre avec un fichier NOTES.md dans son espace de travail et le copie hors de la sandbox à la fermeture de la session. Une session se ferme lors de l'appel à await session.cancel(), lorsqu'une limite de session est atteinte ou, dans votre processus, une fois qu'un appel à agent.run() sans idempotency_key= termine son tour. Dans un bloc async with agents.local_runtime(), elle se ferme à la sortie du bloc. Consultez le fonctionnement de la fin d'une session.

Lorsque vous déployez avec Studio (déploiements Auto-hébergés ou gérés par Mistral), le hook s'exécute dans du code qui est rejoué : effectuez ses opérations via ctx.sandbox, utilisez agents.now() et agents.sleep() pour gérer le temps et veillez à ce que tout téléchargement soit idempotent.

Une commande qui échoue ne lève pas d'exception : vérifiez donc exit_code. Dans une sandbox disposant de git, comme MistralSandbox ou LocalDockerSandbox, une commande lente telle que git clone peut dépasser le délai d'exécution par défaut de la sandbox (default_exec_timeout_seconds sur MistralSandbox et LocalDockerSandbox). Attribuez-lui un délai spécifique :

await ctx.sandbox.execute("git clone https://github.com/acme/api.git", timeout_seconds=240)

await ctx.sandbox.execute("git clone https://github.com/acme/api.git", timeout_seconds=240)

Un hook de sandbox ne conserve aucun état entre ses deux étapes : déduisez ce dont on_destroy a besoin à partir des arguments du constructeur du hook ou récupérez ces informations dans l'espace de travail.

Pour définir l'identité Git associée aux commits dans la sandbox, utilisez le hook intégré agents.pattern.sandbox_hook.GitIdentity : consultez Identité Git. NoSandbox, utilisé par défaut, ne crée jamais de sandbox : aucune des deux étapes ne s'exécute donc.

Les deux étapes déclarent des entrées de session comme le ferait un outil, avec Annotated[T, agents.Depends()]. Utilisez une entrée pour initialiser l'espace de travail avec les éléments sur lesquels porte cette session :

import shlex
from typing import Annotated

from mistralai import agents


class RepositoryContext(agents.SessionInput):
    input_id = "acme.repository"

    provider: str
    repository: str
    revision: str


class RecordRepository(agents.SandboxHook):
    async def on_create(
        self,
        ctx: agents.SandboxCallbackContext,
        *,
        repo: Annotated[RepositoryContext, agents.Depends()],
    ) -> None:
        line = shlex.quote(f"{repo.repository}@{repo.revision}")
        result = await ctx.sandbox.execute(f"echo {line} > REPOSITORY")
        if result.exit_code != 0:
            raise RuntimeError(f"seeding failed: {result.stderr}")


agent = agents.Agent(
    model="mistral-large-4",
    environment=agents.BashkitSandbox(hooks=[RecordRepository()]),
)

import shlex
from typing import Annotated

from mistralai import agents


class RepositoryContext(agents.SessionInput):
    input_id = "acme.repository"

    provider: str
    repository: str
    revision: str


class RecordRepository(agents.SandboxHook):
    async def on_create(
        self,
        ctx: agents.SandboxCallbackContext,
        *,
        repo: Annotated[RepositoryContext, agents.Depends()],
    ) -> None:
        line = shlex.quote(f"{repo.repository}@{repo.revision}")
        result = await ctx.sandbox.execute(f"echo {line} > REPOSITORY")
        if result.exit_code != 0:
            raise RuntimeError(f"seeding failed: {result.stderr}")


agent = agents.Agent(
    model="mistral-large-4",
    environment=agents.BashkitSandbox(hooks=[RecordRepository()]),
)

Transmettez l'entrée au démarrage de la session, dans une fonction asynchrone, en poursuivant l'exemple ci-dessus :

repo = RepositoryContext(provider="github", repository="acme/api", revision="main")
session = await agent.start_session(session_inputs={RepositoryContext.input_id: repo})

repo = RepositoryContext(provider="github", repository="acme/api", revision="main")
session = await agent.start_session(session_inputs={RepositoryContext.input_id: repo})

Protégez les valeurs d'entrée avec shlex.quote() avant de les insérer dans une commande : le shell exécute tout ce que contient la chaîne. L'entrée est vérifiée avant la création de la sandbox.

Si on_create lève une exception, la sandbox est supprimée et la session ne démarre pas. Dans votre processus, vous recevez l'exception du hook. Pour un agent déployé, la session se termine en échec et son message indique uniquement le nom du hook et le type d'exception. Les logs du déploiement contiennent l'erreur complète : n'incluez donc pas d'identifiants dans les messages d'exception. agents serve n'affiche ce log que si AGENTS_SDK_DEBUG=1. Si on_destroy lève une exception, l'erreur est consignée dans les logs et les autres hooks s'exécutent quand même.

Accès réseau : choisissez les ressources auxquelles la sandbox peut accéder.

Voir aussi :

• Références des hooks de sandbox : SandboxHook et ses étapes.