Custom hooks

A hook runs your code inside the agent loop, at six points the runtime opens up: before a turn, before and after every model call, before and after every tool call, and after the turn. Each point can observe the operation and return its point-specific decision.

Subclass agents.Hook, override the points you care about, and put the instance in the harness's hooks=[...] slot:

from pydantic import BaseModel

from mistralai.agents import agents


class AuditTools(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        print(f"{ctx.session.id}: {call.tool_call.call}")
        return agents.hook.PreToolCallHookContinue(effective_arguments=call.tool_call.call.arguments)


class WeatherArgs(BaseModel):
    city: str


@agents.tool(name="get_weather", description="Weather for a city.", input_schema=WeatherArgs)
async def get_weather(args: WeatherArgs) -> dict[str, str]:
    return {"weather": f"sunny in {args.city}"}


harness = agents.Harness(tools=[get_weather], hooks=[AuditTools()])
agent = agents.Agent(model="mistral-medium-3-5", instructions="Use the tool.", harness=harness)

from pydantic import BaseModel

from mistralai.agents import agents


class AuditTools(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        print(f"{ctx.session.id}: {call.tool_call.call}")
        return agents.hook.PreToolCallHookContinue(effective_arguments=call.tool_call.call.arguments)


class WeatherArgs(BaseModel):
    city: str


@agents.tool(name="get_weather", description="Weather for a city.", input_schema=WeatherArgs)
async def get_weather(args: WeatherArgs) -> dict[str, str]:
    return {"weather": f"sunny in {args.city}"}


harness = agents.Harness(tools=[get_weather], hooks=[AuditTools()])
agent = agents.Agent(model="mistral-medium-3-5", instructions="Use the tool.", harness=harness)

The result types live under agents.hook. What you return is what continues: the user_content, effective_arguments or tool_result you hand back replaces the original, so returning the value you were given is the no-op.

The model-call points run once per completion, so a Turn that invokes tools can pass through them multiple times. pre_llm_call identifies the pending action and Turn. post_llm_call also carries the candidate message, finish reason, and token usage.

pre_tool_call sees the arguments the model produced and decides what the tool actually receives:

from mistralai.agents import agents

class ForceSafeMode(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        target = call.tool_call.call
        arguments = dict(target.arguments)
        if target.type == "provided" and target.tool_name == "deploy":
            arguments["dry_run"] = True
        return agents.hook.PreToolCallHookContinue(effective_arguments=arguments)

from mistralai.agents import agents

class ForceSafeMode(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        target = call.tool_call.call
        arguments = dict(target.arguments)
        if target.type == "provided" and target.tool_name == "deploy":
            arguments["dry_run"] = True
        return agents.hook.PreToolCallHookContinue(effective_arguments=arguments)

Returning a Skip instead means the tool never runs. It is delivered to the model in place of the result, so the turn continues and the model can explain itself or try something else.

from mistralai.agents import agents
from mistralai.vibe.harness.core import protocol


class BusinessHoursOnly(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        target = call.tool_call.call
        if target.type == "provided" and target.tool_name == "send_email":
            return agents.hook.PreToolCallHookSkip(
                reason=[protocol.TextContentBlock(text="Outside business hours. Tell the user to retry tomorrow.")]
            )
        return agents.hook.PreToolCallHookContinue(effective_arguments=target.arguments)

from mistralai.agents import agents
from mistralai.vibe.harness.core import protocol


class BusinessHoursOnly(agents.Hook):
    async def pre_tool_call(self, ctx, call):
        target = call.tool_call.call
        if target.type == "provided" and target.tool_name == "send_email":
            return agents.hook.PreToolCallHookSkip(
                reason=[protocol.TextContentBlock(text="Outside business hours. Tell the user to retry tomorrow.")]
            )
        return agents.hook.PreToolCallHookContinue(effective_arguments=target.arguments)

A hook binding has no tool filter, so pre_tool_call and post_tool_call fire for every call. call.tool_call.call is either a ProvidedToolCall (type="provided", with group_name and tool_name) or a sandbox built-in (type="runtime_builtin", with name) - match on type first, as above.

Permission is built this way: permission_mode="ask" and "deny" install their own pre_tool_call hook, ahead of yours.

Every point receives ctx, which carries the session id (ctx.session.id, ctx.session.turn_id), the compiled harness (ctx.harness), and the current session sandbox (ctx.sandbox). Hooks run in the order you list them in hooks=[...], each one seeing what the previous returned.

ctx.sandbox.is_executable tells you whether the hook can use the sandbox. When it can, call await ctx.sandbox.execute("pwd") to run a command; the result has exit_code, stdout, and stderr. ctx.sandbox.workspace is the same workspace used by built-in tools. A remote hook has this access with MistralSandbox. Local in-process hooks use the sandbox resource directly. Other remote sandbox providers currently have no activity-side connection and report is_executable=False.

A hook can also carry tools of its own, which land in the same catalog as the harness's:

agents.Harness(hooks=[AuditTools(tools=[write_audit_entry])])

agents.Harness(hooks=[AuditTools(tools=[write_audit_entry])])

:::caution A hook's effects are not in the session history session.list_history() reports the turn as if no hook had run: a post_tool_call rewrite of content leaves the entry showing what the tool originally returned, and a pre_tool_call skip writes no tool_call and no tool_result entry at all. Enforcement is unaffected — the skipped tool really does not run — but if you need a record of what a hook changed or refused, write it yourself. See Sessions. :::

A hook method declares Session inputs like a tool does, with Annotated[T, agents.Depends(T)]. It receives only the inputs it declares:

from typing import Annotated

from mistralai.agents import agents
from mistralai.vibe.harness.core import protocol


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

    repository: str


class AuditRepository(agents.Hook):
    async def pre_tool_call(
        self,
        ctx,
        call,
        repo: Annotated[RepositoryContext, agents.Depends(RepositoryContext)],
    ):
        target = call.tool_call.call
        if repo.repository.startswith("acme/"):
            return agents.hook.PreToolCallHookContinue(effective_arguments=target.arguments)
        return agents.hook.PreToolCallHookSkip(
            reason=[protocol.TextContentBlock(text="This repository is not allowed.")]
        )

from typing import Annotated

from mistralai.agents import agents
from mistralai.vibe.harness.core import protocol


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

    repository: str


class AuditRepository(agents.Hook):
    async def pre_tool_call(
        self,
        ctx,
        call,
        repo: Annotated[RepositoryContext, agents.Depends(RepositoryContext)],
    ):
        target = call.tool_call.call
        if repo.repository.startswith("acme/"):
            return agents.hook.PreToolCallHookContinue(effective_arguments=target.arguments)
        return agents.hook.PreToolCallHookSkip(
            reason=[protocol.TextContentBlock(text="This repository is not allowed.")]
        )

A Session that omits the input fails to start.

agents.SandboxHook is a separate, smaller hook for the sandbox's own lifecycle. It goes on the sandbox rather than on the harness, and fires once when the sandbox is provisioned and once when the session closes — a place to seed a workspace or upload what the agent produced.

from mistralai.agents import agents

class SeedWorkspace(agents.SandboxHook):
    async def on_create(self, ctx) -> None:
        ...

    async def on_destroy(self, ctx) -> None:
        ...


agents.Harness(sandbox=agents.BashkitSandbox(hooks=[SeedWorkspace()]))

from mistralai.agents import agents

class SeedWorkspace(agents.SandboxHook):
    async def on_create(self, ctx) -> None:
        ...

    async def on_destroy(self, ctx) -> None:
        ...


agents.Harness(sandbox=agents.BashkitSandbox(hooks=[SeedWorkspace()]))

NoSandbox never provisions anything, so neither point fires on the default harness. See Sandbox.

To get durability, a hook must be serializable into a State object.

It means that if your hook is not stateless (i.e. receives argument in __init__), you must implement to_unified_state() and from_unified_state(). A Hook that does not is rebuilt with its __init__ defaults and no error, so the state set at construction is quietly gone.

What state is for

State is a construction recipe, not storage. serialize_unified() produces {"type": "<module>:<ClassName>", "state": {...}}. The rebuilder looks the class up by that type key and calls from_unified_state(state).

Use it for the arguments you passed to __init__:

class BlockTools(agents.Hook):
    def __init__(self, *, blocked: list[str]) -> None:
        super().__init__()
        self._blocked = blocked

    def to_unified_state(self) -> dict[str, object]:
        return {"blocked": self._blocked}

    @classmethod
    def from_unified_state(cls, state: dict[str, object]) -> "BlockTools":
        return cls(blocked=[str(name) for name in state["blocked"]])

class BlockTools(agents.Hook):
    def __init__(self, *, blocked: list[str]) -> None:
        super().__init__()
        self._blocked = blocked

    def to_unified_state(self) -> dict[str, object]:
        return {"blocked": self._blocked}

    @classmethod
    def from_unified_state(cls, state: dict[str, object]) -> "BlockTools":
        return cls(blocked=[str(name) for name in state["blocked"]])

Two constraints. The value must be JSON: it passes through TypeAdapter(dict[str, JsonValue]). And the default from_unified_state calls cls() with no arguments, so a hook with a required constructor argument fails at rebuild unless you override the method.

What you cannot do with it

Do not accumulate per-turn data in it. The snapshot is taken once, at compile time, in _register_authored_hook. It is then closed over and only replayed at each rebuild. Nothing reads your object afterwards, so a counter you increment is discarded.

For per-session data, use a store outside the hook and key it by ctx.session.id.

Behaviour depends on the Run mode

• In-process, Harness(hooks=[...]): no serialization occurs. assemble_agent binds your instance directly, so self works and to_unified_state() is never called.
• File-defined agent: copy_for_isolation() does a deepcopy per session. Attributes set in __init__ survive. Mutations do not cross sessions.
• Worker or deployment: the rebuild path above runs. The reason is isolation and replay, not transport — the worker imports your code and constructs the hook locally, then rebuilds it per activity so each one starts identical.

state and ctx are unrelated

ctx answers "which session and turn am I in". State answers "how do I build this hook/tool/skill object again".

• The Harness - the hooks=[...] slot and everything beside it.
• Tools - what pre_tool_call and post_tool_call intercept.
• Permission - the hook the SDK ships for you.
• Sandbox - where a SandboxHook attaches.
• Sessions - the history a hook's effects do not reach.