Désactiver la recherche sur le Web dès qu'un agent lit des données privées

L'agent d'assistance d'une banque dispose de deux outils : l'un lit les comptes des clients, l'autre effectue des recherches sur le Web. Chacun est sûr lorsqu'il est utilisé seul. Ensemble, ils peuvent provoquer une fuite : dès qu'un compte figure dans la conversation, le modèle peut inclure le nom du titulaire ou son solde dans une requête de recherche, qui quitte alors la banque. La règle : dès qu'une session a lu des données privées, la recherche sur le Web est désactivée pour le reste de cette session.

La protection repose sur un seul hook et un seul indicateur, conservé par session.

Cet exemple illustre : les hooks avec état, pre_tool_call et agents.skip, la façon dont le modèle appelle un outil, l'environnement d'exécution local, et les tests avec un modèle scripté.

Exécution : placez les blocs de code des étapes 1 à 4 dans un seul fichier, private_data_guard.py. Une fois le SDK installé et MISTRAL_API_KEY défini, exécutez python private_data_guard.py.

session 1: lookup_customer_account → the guard sets read_private, the read runs
           web_search              → the guard refuses it: the model gets the reason, no query leaves
           (next turn) web_search  → refused again: the flag holds for the whole session
session 2: web_search              → runs: a new session starts with read_private = False

session 1: lookup_customer_account → the guard sets read_private, the read runs
           web_search              → the guard refuses it: the model gets the reason, no query leaves
           (next turn) web_search  → refused again: the flag holds for the whole session
session 2: web_search              → runs: a new session starts with read_private = False

Le système de gestion des comptes et le moteur de recherche sont simulés par des outils qui renvoient des données fixes. L'exemple s'exécute donc avec une simple clé API Mistral. En production, chaque outil appelle le système réel.

Le mécanisme de protection refuse chaque recherche dans pre_tool_call, avant son exécution, en donnant un motif que le modèle peut lire. Il ne retire pas web_search des outils proposés au modèle, pour deux raisons :

• Masquer un outil relève de la gestion du contexte, pas de la sécurité. Ce que le modèle voit influence seulement ce qu'il tente de faire. La sécurité repose sur la vérification de l'appel lui-même : un appel refusé ne s'exécute jamais, donc aucune requête ne quitte la banque, quelle que soit la tentative du modèle. Le refus fonctionne quel que soit l'environnement d'exécution de la session.
• Modifier la liste des outils en cours de session invalide le cache KV. La liste des outils fait partie de chaque requête envoyée au modèle. Lorsqu'elle change, la conversation en cache ne correspond plus à la requête. L'appel suivant traite donc à nouveau toute la conversation au lieu de réutiliser le cache.

Un shell fonctionne de la même façon. Un agent doté d'un outil bash peut tenter d'écrire dans n'importe quel fichier. Un fichier dans lequel il n'a pas le droit d'écrire ne lui est pas masqué : l'écriture échoue avec une erreur d'autorisation. Le mécanisme de protection joue ce rôle de vérification des autorisations pour web_search. Le modèle voit toujours l'outil après la consultation du compte et peut tenter de l'utiliser ; le motif qu'il lit lui indique de ne pas réessayer et de préciser qu'il n'a pas pu utiliser l'outil.

La consultation du compte renvoie des données fixes, et le moteur de recherche enregistre chaque requête dans SEARCHES. L'exécution montre ainsi ce qui aurait quitté la banque :

import asyncio

from mistralai import agents

# --- The bank's account system and a search engine, faked ---------------------------------------
# In production, lookup_customer_account calls the core banking system, and web_search a real
# search engine. Here both return fixed data, and SEARCHES records every query that would have
# left the bank.

_ACCOUNTS = {
    "ACC-0042": {
        "holder": "Camille Durand",
        "employer": "Durand Menuiserie SARL",
        "balance": "12,480.17 EUR",
        "last_payment": {"merchant": "Ferronnerie Morel", "amount": "1,250.00 EUR", "status": "pending"},
    }
}
SEARCHES: list[str] = []

@agents.tool
def lookup_customer_account(account_id: str) -> dict[str, object]:
    """Core banking: the account holder, the balance and the last payment of a customer account."""
    print(f"core banking: read {account_id}")
    if account_id not in _ACCOUNTS:
        raise agents.ToolError(f"No account {account_id}.")
    return _ACCOUNTS[account_id]

@agents.tool(model_access="direct")  # in the model's tool list: the model calls it by name
def web_search(query: str) -> list[dict[str, str]]:
    """Search the public web. Returns the top results."""
    print(f"web search: {query!r}")
    SEARCHES.append(query)
    return [{"title": "Euro to US dollar", "url": "https://example.com/eur-usd", "snippet": "1 EUR = 1.16 USD"}]

import asyncio

from mistralai import agents

# --- The bank's account system and a search engine, faked ---------------------------------------
# In production, lookup_customer_account calls the core banking system, and web_search a real
# search engine. Here both return fixed data, and SEARCHES records every query that would have
# left the bank.

_ACCOUNTS = {
    "ACC-0042": {
        "holder": "Camille Durand",
        "employer": "Durand Menuiserie SARL",
        "balance": "12,480.17 EUR",
        "last_payment": {"merchant": "Ferronnerie Morel", "amount": "1,250.00 EUR", "status": "pending"},
    }
}
SEARCHES: list[str] = []

@agents.tool
def lookup_customer_account(account_id: str) -> dict[str, object]:
    """Core banking: the account holder, the balance and the last payment of a customer account."""
    print(f"core banking: read {account_id}")
    if account_id not in _ACCOUNTS:
        raise agents.ToolError(f"No account {account_id}.")
    return _ACCOUNTS[account_id]

@agents.tool(model_access="direct")  # in the model's tool list: the model calls it by name
def web_search(query: str) -> list[dict[str, str]]:
    """Search the public web. Returns the top results."""
    print(f"web search: {query!r}")
    SEARCHES.append(query)
    return [{"title": "Euro to US dollar", "url": "https://example.com/eur-usd", "snippet": "1 EUR = 1.16 USD"}]

web_search est un outil direct : le modèle l'appelle par son nom. Un appel refusé transmet donc votre motif au modèle en tant que résultat de l'outil. La consultation du compte conserve le comportement par défaut : le modèle l'appelle depuis le code qu'il écrit dans run_typescript. Lorsqu'un appel effectué depuis du code est refusé, il échoue dans ce code avec une erreur dont le code est tool_skipped. Le mécanisme de protection traite les deux appels de la même façon.

Le hook conserve ses paramètres et son indicateur dans des champs. Le SDK enregistre ces champs pour chaque session : l'indicateur défini lors d'un tour reste donc actif au suivant.

class NoSearchAfterPrivateData(agents.Hook):
    """Once a session has called a private tool, refuse its public tools for the rest of the session."""

    private_tools: list[str]  # tool ids that read private data
    public_tools: list[str]  # tool ids that send data out of the bank
    read_private: bool = False  # True once this session called a private tool

    async def pre_tool_call(self, ctx, call):
        if call.tool_id in self.private_tools:
            self.read_private = True
        elif self.read_private and call.tool_id in self.public_tools:
            print(f"guard: refused {call.tool_id}")
            return agents.skip(
                f"{call.tool_name} is turned off for the rest of this session, because the session has "
                "read private customer data. Do not try it again. Answer without it, and say that you "
                "could not use it."
            )

class NoSearchAfterPrivateData(agents.Hook):
    """Once a session has called a private tool, refuse its public tools for the rest of the session."""

    private_tools: list[str]  # tool ids that read private data
    public_tools: list[str]  # tool ids that send data out of the bank
    read_private: bool = False  # True once this session called a private tool

    async def pre_tool_call(self, ctx, call):
        if call.tool_id in self.private_tools:
            self.read_private = True
        elif self.read_private and call.tool_id in self.public_tools:
            print(f"guard: refused {call.tool_id}")
            return agents.skip(
                f"{call.tool_name} is turned off for the rest of this session, because the session has "
                "read private customer data. Do not try it again. Answer without it, and say that you "
                "could not use it."
            )

Le mécanisme de protection définit l'indicateur avant l'exécution de la consultation du compte, par précaution : même si cette consultation échoue, la recherche est désactivée. Il vérifie call.tool_id, l'identifiant complet de l'outil : vos outils ont des identifiants de la forme client.<name>. Indiquez dans public_tools tous les outils qui envoient des données à l'extérieur, par exemple un outil qui envoie un e-mail, et pas seulement la recherche. En production, consignez chaque refus dans votre log d'audit au lieu de l'afficher.

support_agent = agents.Agent(
    model="mistral-large-4",
    name="bank-support",
    instructions=(
        "You help the bank's support staff. Use lookup_customer_account for customer accounts and "
        "web_search for public information. Answer in at most two sentences."
    ),
    tools=[lookup_customer_account, web_search],
    builtins=agents.Builtins(subagents=False),  # a child session would start with read_private=False
    hooks=[
        NoSearchAfterPrivateData(
            private_tools=["client.lookup_customer_account"],
            public_tools=["client.web_search"],
        )
    ],
)

support_agent = agents.Agent(
    model="mistral-large-4",
    name="bank-support",
    instructions=(
        "You help the bank's support staff. Use lookup_customer_account for customer accounts and "
        "web_search for public information. Answer in at most two sentences."
    ),
    tools=[lookup_customer_account, web_search],
    builtins=agents.Builtins(subagents=False),  # a child session would start with read_private=False
    hooks=[
        NoSearchAfterPrivateData(
            private_tools=["client.lookup_customer_account"],
            public_tools=["client.web_search"],
        )
    ],
)

L'agent ne peut pas démarrer de sous-agents. Un sous-agent possède sa propre session : sa copie du mécanisme de protection démarrerait donc avec read_private=False, et le modèle pourrait lui transmettre les données du compte dans sa tâche. agents.Builtins(subagents=False) retire les outils de sous-agents au modèle.

La première session lit un compte, puis demande deux recherches. La seconde demande la même recherche que le deuxième tour de la première session, sans lire de compte :

async def main() -> None:
    async with agents.local_runtime():  # at exit: every local session of the event loop stops
        print("--- session 1")
        session = await support_agent.start_session()
        state = await session.run(
            "Customer ACC-0042 asks about a pending payment. Look up the account, then search the web "
            "for news about the account holder."
        )
        print(f"agent: {state.text}")
        state = await session.run("Search the web for today's EUR to USD exchange rate.")
        print(f"agent: {state.text}")
        await session.cancel()

        print("--- session 2")
        other = await support_agent.start_session()
        state = await other.run("Search the web for today's EUR to USD exchange rate.")
        print(f"agent: {state.text}")
        await other.cancel()

    print(f"queries that left the bank: {SEARCHES}")

if __name__ == "__main__":
    asyncio.run(main())

async def main() -> None:
    async with agents.local_runtime():  # at exit: every local session of the event loop stops
        print("--- session 1")
        session = await support_agent.start_session()
        state = await session.run(
            "Customer ACC-0042 asks about a pending payment. Look up the account, then search the web "
            "for news about the account holder."
        )
        print(f"agent: {state.text}")
        state = await session.run("Search the web for today's EUR to USD exchange rate.")
        print(f"agent: {state.text}")
        await session.cancel()

        print("--- session 2")
        other = await support_agent.start_session()
        state = await other.run("Search the web for today's EUR to USD exchange rate.")
        print(f"agent: {state.text}")
        await other.cancel()

    print(f"queries that left the bank: {SEARCHES}")

if __name__ == "__main__":
    asyncio.run(main())

Le programme affiche :

Dans la première session, le mécanisme de protection autorise la lecture du compte, puis refuse les deux recherches que le modèle tente d'effectuer sur les actualités concernant le titulaire. Au tour suivant, le modèle ne tente même pas de rechercher le taux de change : les refus lui ont indiqué que la recherche était désactivée, et l'indicateur reste actif pendant toute la session. La seconde session n'a lu aucun compte, ses recherches s'exécutent donc. Le nom du titulaire n'a jamais été transmis au moteur de recherche.

L'indicateur fonctionne de la même façon lorsque vous déployez l'agent avec Studio (auto-hébergé ou géré par Mistral) : la session conserve les champs du hook, comme l'explique agents.Hook.

Un modèle scripté remplace le modèle réel. Le test ne nécessite donc aucune clé API et suit les mêmes étapes à chaque exécution. Le script appelle la consultation du compte depuis du code, avec agents.testing.run_typescript, comme le ferait un modèle réel. La recherche est un outil direct : agents.testing.call l'appelle donc par son nom. Placez le test à côté de private_data_guard.py :

import private_data_guard as guard
import pytest

from mistralai import agents

async def test_the_guard_refuses_search_for_the_rest_of_the_session(
    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    guard.SEARCHES.clear()
    model = agents.testing.ScriptedModel(
        [
            # Session 1, turn 1. The lookup is called from code, as a real model does.
            agents.testing.run_typescript(
                'return await tools.client.lookup_customer_account({"account_id": "ACC-0042"});'
            ),
            agents.testing.call("web_search", query="Camille Durand news"),  # a direct tool: called by name
            agents.testing.reply("Camille Durand's payment is pending; I could not search."),
            # Session 1, turn 2.
            agents.testing.call("web_search", query="EUR to USD today"),
            agents.testing.reply("I could not search."),
            # Session 2.
            agents.testing.call("web_search", query="EUR to USD today"),
            agents.testing.reply("1 EUR = 1.16 USD."),
        ]
    )
    monkeypatch.setattr(guard, "support_agent", guard.support_agent.override(model=model))

    await guard.main()

    model.assert_consumed()
    assert guard.SEARCHES == ["EUR to USD today"]  # only the second session searched
    out = capsys.readouterr().out
    assert "core banking: read ACC-0042" in out
    assert out.count("guard: refused client.web_search") == 2

import private_data_guard as guard
import pytest

from mistralai import agents

async def test_the_guard_refuses_search_for_the_rest_of_the_session(
    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    guard.SEARCHES.clear()
    model = agents.testing.ScriptedModel(
        [
            # Session 1, turn 1. The lookup is called from code, as a real model does.
            agents.testing.run_typescript(
                'return await tools.client.lookup_customer_account({"account_id": "ACC-0042"});'
            ),
            agents.testing.call("web_search", query="Camille Durand news"),  # a direct tool: called by name
            agents.testing.reply("Camille Durand's payment is pending; I could not search."),
            # Session 1, turn 2.
            agents.testing.call("web_search", query="EUR to USD today"),
            agents.testing.reply("I could not search."),
            # Session 2.
            agents.testing.call("web_search", query="EUR to USD today"),
            agents.testing.reply("1 EUR = 1.16 USD."),
        ]
    )
    monkeypatch.setattr(guard, "support_agent", guard.support_agent.override(model=model))

    await guard.main()

    model.assert_consumed()
    assert guard.SEARCHES == ["EUR to USD today"]  # only the second session searched
    out = capsys.readouterr().out
    assert "core banking: read ACC-0042" in out
    assert out.count("guard: refused client.web_search") == 2

Le test remplace le support_agent du module par support_agent.override(model=model). Ainsi, main s'exécute sans modification, avec le script. Une fois pytest et pytest-asyncio installés, lancez le test :

pytest -o asyncio_mode=auto test_private_data_guard.py

pytest -o asyncio_mode=auto test_private_data_guard.py

• Les outils publics absents de la liste. Le mécanisme de protection ne refuse que les outils figurant dans public_tools. Un outil que vous ajoutez plus tard, par exemple un outil d'envoi d'e-mails, un outil d'un serveur MCP, un outil d'un connecteur ou l'outil shell de la sandbox lorsque celle-ci a accès au réseau, peut s'exécuter tant que vous ne l'avez pas ajouté à la liste. Pour bloquer par défaut, donnez plutôt au hook une liste allowed_tools d'identifiants d'outils, par exemple client.lookup_customer_account. Dès que read_private est défini, refusez tout outil qui n'y figure pas. Ajoutez aussi les outils intégrés dont l'agent a encore besoin ; affichez call.tool_id pour voir chaque identifiant. run_typescript lui-même ne passe pas par le hook ; les appels effectués depuis son code, si.
• Les outils privés absents de la liste. Un outil qui lit des données privées ne définit l'indicateur que si son identifiant figure dans private_tools. Si vous omettez un second outil d'accès aux comptes ou un connecteur CRM, toutes les recherches restent autorisées.
• Une session en cours conserve ses listes. private_tools et public_tools sont des champs : ils sont donc enregistrés avec la session, comme l'indicateur. Un outil ajouté à une liste n'est pris en compte que dans les sessions démarrées après cette modification.
• Les données privées qui ne proviennent pas d'un outil. Seul un appel à un outil figurant dans private_tools définit l'indicateur. Les données de compte que l'utilisateur saisit dans un message ne le définissent pas.

Vérifiez les identifiants dans private_tools et public_tools. Vos outils ont des identifiants de la forme client.<name> ; un outil appartenant à un espace de noms utilise <namespace>.<name>. Un nom sans son espace de noms ne correspond jamais. Pour connaître l'identifiant exact d'un outil, affichez call.tool_id dans pre_tool_call.

C'est le comportement prévu : chaque session démarre avec le hook que vous avez créé, avec read_private=False. L'indicateur appartient à la session qui a lu les données.

• Les mécanismes de contrôle : des hooks qui limitent les appels autorisés pour l'agent.
• agents.Hook et agents.skip dans la Référence API.
• Examen des paiements suspects : des demandes d'approbation traitées par une politique et par une personne.