Validate structured output with a hook

agents.pattern.hook.StructuredOutputHook validates the final response of each turn with your Pydantic model. If the response is not valid, the hook sends the error to the agent, and the agent corrects its response.

The hook does not guarantee a valid response. It stops asking for corrections after max_retries, or when the turn has no iteration left under max_iterations. Then it accepts the last response, or fails the turn if you set on_limit="reject".

Demonstrates: a hook that checks the final answer of each turn and asks the agent to correct it, output_type=, and an application that switches the hook off.

Run: put the code block below in one file, structured_output_hook.py, and run python structured_output_hook.py (setup).

On the "mistral" and "openai" providers, output_type= already constrains the response to JSON that matches your model's schema. Use the hook when:

• Your model has rules that a JSON Schema cannot express, such as a @model_validator.
• You validate without output_type=: on the "anthropic" provider, a custom completion provider, or a subagent.

The hook does not read output_type=. You can use it with or without it.

Name your Pydantic model in brackets, and pass the other parameters to the constructor: agents.pattern.hook.StructuredOutputHook[City](max_retries=2).

The StructuredOutputHook reference gives its full signature.

With output_type=, set schema_in_prompt=False: the provider already gives the model the schema. With the hook alone, keep it True, or set it to False and put the schema in your own instructions.

With "accept", the run returns the last response, and your application decides what to do with it. With output_type=, agent.run() also validates that response. When it does not pass, the Completed record's output is an agents.InvalidOutput with the response and the Pydantic error.

With "reject", the turn fails: agent.run() returns agents.Failed with the code capability_execution_failed, with or without output_type=. The message says which check failed, and gives the Pydantic error and the first 1,000 characters of the last response. The session stays usable for the next turn.

Once you deploy the agent with Studio (self-hosted or Mistral-managed), the session history also shows each response the hook did not accept, as an assistant message that looks like any other.

Create one hook object and keep a reference to it. Your application can switch it off and on with enabled. In-process, the change applies to the next agent.run() or session: a session keeps the values it started with. A deployed agent uses the value from when it was deployed.

The example uses on_limit="reject", so a response that is still not a valid City comes back as agents.Failed:

"""Validate structured output with a hook that the application can switch off.

Set MISTRAL_API_KEY, then run: python structured_output_hook.py
"""

import asyncio

from pydantic import BaseModel

from mistralai import agents


class City(BaseModel):
    name: str
    country: str


@agents.tool
async def lookup_landmark(name: str) -> dict[str, str]:
    """Find where a landmark is."""
    return {"landmark": name, "location": "Champ de Mars, 7th arrondissement, Paris, France"}


async def main() -> None:
    # The hook makes the agent correct a response that is not a valid City. output_type= below
    # already gives the model the schema, so the hook does not add it to the prompt.
    check_city = agents.pattern.hook.StructuredOutputHook[City](
        max_retries=2, on_limit="reject", schema_in_prompt=False
    )
    agent = agents.Agent(
        model="mistral-medium-latest",
        name="structured-output-hook-example",
        instructions="Use lookup_landmark to find where a landmark is.",
        max_iterations=6,
        tools=[lookup_landmark],
        hooks=[check_city],
    )

    # output_type=City parses the final response into Completed.output.
    match await agent.run("Which city and country contain the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city.model_dump())
        case agents.Failed(error=error) if error.code == "capability_execution_failed":
            # on_limit="reject": still not a valid City after max_retries corrections.
            raise RuntimeError(f"No valid City: {error.message}")
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")

    # The application switches the hook off for a free-text answer.
    check_city.enabled = False
    match await agent.run("Describe the Eiffel Tower in one sentence."):
        case agents.Completed(text=text) if text:
            print(text)
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")


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

"""Validate structured output with a hook that the application can switch off.

Set MISTRAL_API_KEY, then run: python structured_output_hook.py
"""

import asyncio

from pydantic import BaseModel

from mistralai import agents


class City(BaseModel):
    name: str
    country: str


@agents.tool
async def lookup_landmark(name: str) -> dict[str, str]:
    """Find where a landmark is."""
    return {"landmark": name, "location": "Champ de Mars, 7th arrondissement, Paris, France"}


async def main() -> None:
    # The hook makes the agent correct a response that is not a valid City. output_type= below
    # already gives the model the schema, so the hook does not add it to the prompt.
    check_city = agents.pattern.hook.StructuredOutputHook[City](
        max_retries=2, on_limit="reject", schema_in_prompt=False
    )
    agent = agents.Agent(
        model="mistral-medium-latest",
        name="structured-output-hook-example",
        instructions="Use lookup_landmark to find where a landmark is.",
        max_iterations=6,
        tools=[lookup_landmark],
        hooks=[check_city],
    )

    # output_type=City parses the final response into Completed.output.
    match await agent.run("Which city and country contain the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city.model_dump())
        case agents.Failed(error=error) if error.code == "capability_execution_failed":
            # on_limit="reject": still not a valid City after max_retries corrections.
            raise RuntimeError(f"No valid City: {error.message}")
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")

    # The application switches the hook off for a free-text answer.
    check_city.enabled = False
    match await agent.run("Describe the Eiffel Tower in one sentence."):
        case agents.Completed(text=text) if text:
            print(text)
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")


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

The first run returned a valid City. With the hook off, the second returned free text.

• Structured output: the output_type= argument.
• Hooks API: hook points, what each can return, and state.