Turn off web search once an agent reads private data

A bank's support agent has two tools: one reads customer accounts, the other searches the web. Each is safe alone. Together they can leak: once an account is in the conversation, the model can put the holder's name or balance into a search query, and the query leaves the bank. The rule: once a session has read private data, web search is off for the rest of that session.

The guard is one hook with one flag, kept per session.

Demonstrates: stateful hooks, pre_tool_call and agents.skip, how the model calls a tool, the local runtime, and testing with a scripted model.

Run: put the code blocks of steps 1 to 4 in one file, private_data_guard.py. With the SDK installed and MISTRAL_API_KEY set, run 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

The account system and the search engine are faked by tools that return fixed data, so the example runs with only a Mistral API key. In production, each tool calls the real system.

The guard refuses each search in pre_tool_call, before the search runs, with a reason the model reads. It does not take web_search out of the tools the model is offered, for two reasons:

• Hiding a tool is context management, not safety. What the model sees only steers what it tries. Safety is the check on the call itself: a refused call never runs, so no query leaves the bank, whatever the model tries. The refusal works wherever the session runs.
• Changing the tool list in the middle of a session breaks the KV cache. The tool list is part of every request to the model. When it changes, the cached conversation no longer matches, so the next call processes the whole conversation again instead of reusing the cache.

A shell works the same way. An agent with a bash tool can try to write any file. A file that it may not write is not hidden from it: the write fails with a permission error. The guard is that permission check for web_search. The model still sees the tool after the lookup and may try it; the reason it reads tells it not to try again and to say that it could not use the tool.

The account lookup returns fixed data, and the search engine records each query in SEARCHES, so the run shows what would have left the bank:

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 is a direct tool: the model calls it by name, so a refused call gives the model your reason as the tool's result. The lookup keeps the default: the model calls it from code it writes in run_typescript. Called from code, a refused call rejects in that code with an error whose code is tool_skipped. The guard sees both calls the same way.

The hook keeps its settings and its flag in fields. The SDK saves the fields per session, so the flag set in one turn holds in the next:

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."
            )

The guard sets the flag before the lookup runs, so it errs on the safe side: a lookup that fails still turns search off. It matches on call.tool_id, the tool's full id: your tools have ids of the form client.<name>. List every tool that sends data out in public_tools, such as a tool that sends an email, not only the search. In production, write each refusal to your audit log instead of printing it.

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

The agent cannot start subagents. A child is a session of its own, so its copy of the guard would start with read_private=False, and the model could give it the account data in its task. agents.Builtins(subagents=False) takes the subagent tools away from the model.

The first session reads an account, then asks for two searches. The second session asks for the same search as the first one's second turn, without reading an account:

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

The program prints:

In the first session the guard let the account read through, then refused both searches the model tried for news about the account holder. One turn later, the model did not even try to search for the exchange rate: the refusals told it search was off, and the flag holds for the whole session. The second session had read no account, so its searches ran. The holder's name never reached the search engine.

The flag works the same once you deploy the agent with Studio (self-hosted or Mistral-managed): the session keeps the hook's fields, as agents.Hook explains.

A scripted model replaces the real one, so the test needs no API key and gives the same steps each time. The script calls the lookup from code, with agents.testing.run_typescript, as a real model does. The search is a direct tool, so agents.testing.call calls it by name. Put the test beside 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

The test replaces the module's support_agent with support_agent.override(model=model), so main runs as written, against the script. With pytest and pytest-asyncio installed, run it:

pytest -o asyncio_mode=auto test_private_data_guard.py

pytest -o asyncio_mode=auto test_private_data_guard.py

• Public tools you did not list. The guard refuses only the tools in public_tools. A tool you add later, such as an email tool, an MCP server's tool, a connector's tool, or the sandbox's shell tool when the sandbox can reach the network, runs until you list it. To fail closed instead, give the hook an allowed_tools list of tool ids, such as client.lookup_customer_account, and once read_private is set, refuse every tool that is not in it. List the built-in tools the agent still needs too; print call.tool_id to see each id. run_typescript itself does not reach the hook; the calls made from its code do.
• Private tools you did not list. A tool that reads private data sets the flag only when its id is in private_tools. A second account tool, or a CRM connector, that you leave out lets every search run.
• A running session keeps its lists. private_tools and public_tools are fields, so they are saved with the session like the flag. A tool you add to a list applies to sessions started after the change.
• Private data that does not come from a tool. The flag is set only by a call to a tool in private_tools. Account data the user types into a message does not set it.

Check the ids in private_tools and public_tools. Your tools have ids of the form client.<name>; a tool in a namespace has <namespace>.<name>. A name without its namespace never matches. To see a tool's exact id, print call.tool_id in pre_tool_call.

That is by design: each session starts from the hook you built, with read_private=False. The flag belongs to the session that read the data.

• Guardrails: hooks that limit what the agent may call.
• agents.Hook and agents.skip in the API reference.
• Suspicious payment review: approvals answered by a policy and by a person.