Structured output

The SDK can return the agent's final answer as an instance of your own Pydantic model. You pass the model as output_type=, and the output of the agents.Completed record holds the parsed answer, typed as your model. A reply that does not match comes back as an agents.InvalidOutput value, not as an exception.

Use it when your code acts on the answer: pull the fields of an invoice into a database row, sort a support ticket into a category with a priority, or get an approve or refuse decision with its reason.

Set output_type= on the Agent, or on an in-process agent.run as below, and the model is asked for that format. On session.run, turn.result and an agent.run of a deployed agent, it only parses the reply. Set it on the Agent when every answer has the same shape.

import asyncio

from pydantic import BaseModel

from mistralai import agents

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

agent = agents.Agent(model="mistral-medium-latest")

async def main() -> None:
    match await agent.run("Where is the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city)
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            print(f"not a City: {error}")
        case state:
            print(f"no answer: {state}")

asyncio.run(main())
Output
name='Paris' country='France'

With output_type=City, a Completed record's output is City | None | agents.InvalidOutput:

  • a City when the answer validates;
  • an agents.InvalidOutput when it does not: raw is the text and error says why;
  • None when the turn ended without a final answer, with stop_reason "iteration_limit", "skipped", "rejected" or "output_limit".

When no output type applies, on the call or on the Agent, output is the reply text, or "" when the turn has none, whatever its stop_reason.

text carries the raw reply in every case. The other records, such as Blocked or Failed, have no output and come back as they are.

On the Agent

On the Agent

Set output_type= on the Agent to constrain the final answer of every turn, in every session the agent opens. This is the way to give a deployed agent an output type:

import asyncio

from pydantic import BaseModel

from mistralai import agents

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

agent = agents.Agent(model="mistral-medium-latest", output_type=City)

async def main() -> None:
    state = await agent.run("Where is the Eiffel Tower?")
    if isinstance(state, agents.Completed):
        print(repr(state.output))
        print(state.text)

asyncio.run(main())
Output
City(name='Paris', country='France')
{"name": "Paris", "country": "France"}

agent.run(..., output_type=SomeModel) in-process overrides the Agent's output type for that run.

Without output_type= on the call, the static type of output is Any. Pass the model to run() as well when you want your type checker to see City.

agents serve, agents deploy and agent.to_workflow() use the Agent's output type. A subagent never inherits it.

On session handles

On session handles

A handle from agent.start_session() or agent.session(session_id) keeps the Agent's output_type, and so do its turns: session.run() and turn.result() parse with it when you pass none, and turn.get() parses with it.

session.run() and turn.result() also take output_type= (see Session.run), and it replaces the Agent's for that call. Use it when one turn of a session needs a structured answer and the others do not.

The model does not see this output_type: the session was set up when it started. Ask for JSON in the prompt instead. When the turn ends, the SDK tries to parse the reply as your model. If it validates, output holds an instance of it. If not, output is an agents.InvalidOutput, and text keeps the reply:

import asyncio

from pydantic import BaseModel

from mistralai import agents

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

agent = agents.Agent(model="mistral-medium-latest")

async def main() -> None:
    session = await agent.start_session()
    state = await session.run(
        "Where is Big Ben? Answer only with a JSON object with name and country.", output_type=City
    )
    if isinstance(state, agents.Completed) and isinstance(state.output, City):
        print(f"{state.output.name} is in {state.output.country}.")
    elif isinstance(state, agents.Completed):
        print(f"No city in the answer: {state.text}")
    else:
        print(f"No answer: {state}")
    await session.cancel()

asyncio.run(main())

A model that is not constrained can still wrap its JSON in a code fence, and the parse then fails, so keep the fallback. When every turn needs the same shape, set output_type on the Agent: the model is then asked for it.

Tools and the final answer

Tools and the final answer

An output type does not stop the agent from using its tools. The model can call tools first, then give its final answer in your output type, built from what the tools returned. For example, an agent with a get_weather tool and output_type=Forecast calls the tool, then answers with a Forecast.

This applies to an output type set on the Agent or on an in-process agent.run: the SDK sends it with every model call of the turn, along with the tools.

Where structured output applies

Where structured output applies

Where you pass itWhich turns it coversDoes the model get the schema?Is output parsed?
agents.Agent(..., output_type=M)every turn of every session, deployed tooyesyes, on handles of this Agent
agent.run(..., output_type=M), in-processthat one run, overriding the Agent'syesyes
agent.run(..., output_type=M), deployedthat run's record onlyno: the deployed Agent's output type appliesyes
session.run or turn.result with output_type=Mthat call's record onlyno: the Agent's output type appliesyes
Common issues

Common issues

You seeCauseFix
Completed(output=InvalidOutput(raw='Hello there friend!', error='1 validation error for City\n Invalid JSON: ...'))The model was not constrained, so it answered in free text.Set output_type= on the Agent, or on an in-process agent.run.
output is NoneThe turn ended without a final answer: stop_reason is "iteration_limit", "skipped", "rejected" or "output_limit".Raise max_iterations, check the hook that skipped the turn or rejected the reply, or, on "output_limit", ask for a shorter answer or raise the model's output limit.
output holds a string, not your modelNo output type applies to the record: a subagent's handle, or an Agent from load_from.Pass output_type= to session.run or turn.result.

To make the agent correct an answer that does not validate, add agents.pattern.hook.StructuredOutputHook[YourModel]() to the agent's hooks=[...]: it checks the final JSON and asks the model to try again.

Environment: give the agent a computer to run commands on.

See also:

  • Agent run: the records that agent.run returns.
  • Turns: read a turn with turn.result.