Structured output.
Constrain an agent turn to a pydantic model with run_object and stream_object.
Structured output constrains the agent's response to a JSON Schema, server-side (grammar-constrained decoding), and validates it client-side against your pydantic model. Pass a pydantic model class (or JSON-schema dict), receive a validated object — no prompt engineering, no brittle JSON extraction.
runner.stream_object(agent, prompt, schema, **options)
Streams exactly like runner.stream() — you receive every AgentEvent (including tool-loop and usage events) — plus one final event with the parsed, schema-validated object. Options are the run options.
from pydantic import BaseModel
class Issue(BaseModel):
severity: str # 'low' | 'medium' | 'high'
description: str
class Review(BaseModel):
summary: str
issues: list[Issue]
async def main():
reviewer = Agent(name="reviewer", instructions="Review the code the user provides.")
async for event in runner.stream_object(reviewer, "Review this function: ...", Review):
if event.type == "response.output_text.delta":
print(event.delta, end="", flush=True)
if event.type == "object":
review = event.object # a validated Review instance
print(f"\nValidated: {review.summary}")The final event:
@dataclass(frozen=True)
class ObjectEvent[T]:
object: T # your pydantic model
# type = "object"runner.run_object(agent, prompt, schema, **options)
Non-streaming convenience. Like runner.run(), it honours callbacks, throw_on_stream_error, retries, and the tool loop — and returns an AgentObjectResult: the full AgentRunResult plus the validated object.
from pydantic import BaseModel
class Invoice(BaseModel):
total: float
currency: str
async def main():
result = await runner.run_object(
extractor,
"Invoice total $1,250.00 AUD due 30 Sep.",
Invoice,
)
print(result.object.total) # 1250.0 (float)
print(result.object.currency) # "AUD" (str)
print(result.final_output) # raw JSON text as returned
print(result.usage) # token usage, as usualAgentObjectResult
Extends AgentRunResult with one additional field:
| Field | type | description |
|---|---|---|
| object | T | None | The response text parsed as JSON and validated against your schema. |
Convenience functions
Shared-default-runner-backed, like the other top-level helpers:
from matilda_agent_sdk import run_object, stream_object
result = await run_object(agent, prompt, schema, ...)
async for event in stream_object(agent, prompt, schema, ...):
...Raw JSON Schema via response_schema
The run options (and therefore Session turns) accept a stringified JSON Schema directly on any run or stream:
import json
result = await runner.run(
agent,
"List three Australian birds.",
response_schema=json.dumps({
"type": "object",
"properties": {"birds": {"type": "array", "items": {"type": "string"}}},
"required": ["birds"],
"additionalProperties": False,
}),
)
data = json.loads(result.final_output) # guaranteed valid, schema-conforming JSONWith response_schema set, final_output is guaranteed to be valid JSON conforming to the schema — but parsing and validation are up to you.
Safety replace and structured output. Like run_text() / stream_text(), the object helpers raise SafetyReplaceError when the server replaces the output mid-stream — str(err) holds the replacement text and .categories the triggering categories. Token deltas already yielded to your consumer are not rolled back; run_object() is unaffected at the value level, since it raises before returning a result.
Truncation throws. If the stream is truncated before the JSON completes, both helpers raise MatildaObjectParseError with the partial text in .raw. See Error handling.
Stream errors throw. If the server emits an error event mid-stream, stream_object() raises MatildaStreamError with the server's error code and message, and run_object() raises MatildaAgentStreamError with the partial result attached — matching the behaviour of the text helpers.