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())With output_type=City, a Completed record's output is City | None | agents.InvalidOutput:
- a
Citywhen the answer validates; - an
agents.InvalidOutputwhen it does not:rawis the text anderrorsays why; Nonewhen the turn ended without a final answer, withstop_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
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())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
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
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 you pass it | Which turns it covers | Does the model get the schema? | Is output parsed? |
|---|---|---|---|
agents.Agent(..., output_type=M) | every turn of every session, deployed too | yes | yes, on handles of this Agent |
agent.run(..., output_type=M), in-process | that one run, overriding the Agent's | yes | yes |
agent.run(..., output_type=M), deployed | that run's record only | no: the deployed Agent's output type applies | yes |
session.run or turn.result with output_type=M | that call's record only | no: the Agent's output type applies | yes |
Common issues
| You see | Cause | Fix |
|---|---|---|
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 None | The 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 model | No 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.
Next
Environment: give the agent a computer to run commands on.
See also: