Notebooks · In-depth guides · Guide 08

Structured output

Name: SHIPIT Agent
Author: SHIPIT

Ask for a shape, get it back validated. Pass a Pydantic model (or a JSON Schema) and the answer arrives as a typed object. An answer that doesn't fit goes back to the model with the validation error, in the same…

Open in Colab View on GitHub 12 code cells · 11 with saved output

Run end to end on shipit-agent 3.0 with real model calls (Haiku 5.5 on Anthropic, Gemma 4 31B on Bedrock). Every output below is from that run.

1 · Models

In [1]

python

import os

from shipit_agent import Agent
from shipit_agent.llms import build_llm_from_env
from shipit_agent.llms.factory import build_llm_from_settings

os.environ["SHIPIT_ANTHROPIC_MODEL"] = "claude-haiku-5-5"
haiku = build_llm_from_env("anthropic")
gemma = build_llm_from_settings({"model": "google.gemma-4-31b"}, provider="bedrock")

2 · A typed answer from a run

Pass the model class as output_schema. The typed object is result.parsed; result.output keeps the text.

In [2]

python

from pydantic import BaseModel

class Ticket(BaseModel):
    category: str
    priority: str
    customer_email: str

result = Agent(llm=haiku).run(
    "From: [email] — Checkout charges me twice since this morning!!", output_schema=Ticket)
print(repr(result.parsed))
print(type(result.parsed).__name__, "| priority:", result.parsed.priority)

Out

Ticket(category='billing', priority='high', customer_email='[email]')
Ticket | priority: high

3 · Nested models, lists, enums and constraints

Everything Pydantic validates is enforced: here an enum for severity, a list of findings, and a bounded score.

In [3]

python

from enum import Enum

from pydantic import Field

class Severity(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"

class Finding(BaseModel):
    line: int
    problem: str
    severity: Severity

class Review(BaseModel):
    summary: str = Field(max_length=200)
    findings: list[Finding]
    score: int = Field(ge=0, le=10, description="Overall quality, 0 to 10")

SNIPPET = '''
1  def charge(amount, card):
2      fee = amount * 0.029 + 0.30
3      total = round(amount + fee, 2)
4      if card.balance > total:
5          card.balance -= total
6      return total
'''
review = Agent(llm=haiku).run(f"Review this payment code:\n{SNIPPET}", output_schema=Review).parsed
print(f"score {review.score}/10 — {review.summary}")
for f in review.findings:
    print(f"  line {f.line:2} [{f.severity.value:6}] {f.problem}")

Out

score 3/10 — Charge logic uses floating-point money math, has no input validation, a check-then-act race on balance, and reports declines the same way as successes.
  line  2 [medium] Money is computed with binary floats (0.029 and amount * 0.029), which can produce rounding errors. Use Decimal and define the rounding mode explicitly.
  line  3 [medium] round() uses banker's rounding by default and rounds only the sum, so the fee is not rounded to cents on its own. The charged amount may not match the fee the business expects.
  line  4 [high  ] The check-then-act on card.balance is not atomic. Concurrent charges can both pass the check and overdraw the balance.
  line  4 [high  ] No validation of amount: a negative or zero amount passes the check, so a negative charge can credit the card. Non-numeric or None inputs also fail unpredictably.
  line  4 [low   ] The comparison is strict (>). A balance exactly equal to total is declined, which may or may not be intended. Confirm the boundary.
  line  6 [high  ] On insufficient funds the function still returns total, identical to success. Callers cannot tell a decline from a completed charge. Raise an exception or return an explicit status.
  line  5 [medium] No transaction record or idempotency key is created, so a retried call can double-charge and there is no audit trail.

4 · Tools first, then a typed answer

The agent can use its tools along the way; only the final answer has to fit the schema.

In [4]

python

import json

from shipit_agent.tools import FunctionTool

ACCOUNTS = {"acme": {"plan": "Team", "seats": 42, "mrr": 1218, "open_tickets": 7},
            "globex": {"plan": "Starter", "seats": 5, "mrr": 60, "open_tickets": 0}}

def account(name: str) -> str:
    """Look up a customer account by name."""
    return json.dumps(ACCOUNTS.get(name.lower(), {}))

class Health(BaseModel):
    account: str
    plan: str
    at_risk: bool
    reason: str

agent = Agent(llm=haiku, tools=[FunctionTool.from_callable(account, read_only=True)])
result = agent.run("Is Acme at risk? Look it up.", output_schema=Health)
print([r.name for r in result.tool_results])
print(result.parsed)

Out

['account']
account='Acme' plan='Team' at_risk=False reason='The account lookup returned no health or risk field, so this is not a confirmed assessment. Acme has a Team plan with 42 seats, $1,218 MRR, and 7 open tickets. Nothing in that data shows churn signals, but the record lacks usage, renewal, or sentiment data, so the account could still be at risk.'

5 · Validation retries

If the answer doesn't validate, the error goes back to the model and it tries again, up to max_validation_retries (2 by default). A custom check, an exact word count, is enforced the same way; here the first answer already fit.

In [5]

python

class Slogan(BaseModel):
    text: str = Field(description="Exactly five words")

    def model_post_init(self, _):
        if len(self.text.split()) != 5:
            raise ValueError(f"text must be exactly 5 words, got {len(self.text.split())}")

result = Agent(llm=haiku).run("A slogan for a sandboxed AI agent runtime.", output_schema=Slogan)
print(result.parsed)
print("model turns:", sum(1 for m in result.messages if m.role == "assistant"))

Out

text='Agents act freely, safely contained'
model turns: 1

6 · When retries run out

An impossible constraint: once the retries are used up, run() raises, with the last validation error and the model's own text, so your code can decide what to do.

In [6]

python

class Impossible(BaseModel):
    n: int = Field(gt=10, lt=5)

try:
    Agent(llm=haiku).run("Give me a number.", output_schema=Impossible, max_validation_retries=1)
except Exception as error:
    print(type(error).__name__)
    print(str(error)[:400])

Out

output_schema parsing failed after 2 attempts. Last error: Pydantic validation failed: 1 validation error for Impossible
n
  Field required [type=missing, input_value={'error': 'The schema is ...o valid object exists.'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing — raw: No integer satisfies this schema. It requires `n > 10` and `n < 5` at the same time, which is impossible, so I can't return a valid object.

If you meant a different range, adjust the bounds. For example, `exclusiveMinimum: 4` and `exclusiveMaximum: 6` would allow only `5`.

7 · No agent: `StructuredOutput`

For extraction or classification with no tools, call the model directly. The result has the value, the attempts it took, the raw_text, and the failed attempts in history.

In [7]

python

from shipit_agent import StructuredOutput

extract = StructuredOutput(llm=haiku, schema=Ticket, max_retries=2)
emails = [
    "From: [email]. Checkout charges me twice since this morning!!",
    "[email] here — could you add dark mode at some point? No rush.",
    "From [email]: the export button 500s for every report. Board meeting in an hour.",
]
for email in emails:
    r = extract.run(email)
    print(f"{r.value.priority:7} {r.value.category:16} {r.value.customer_email:16} attempts={r.attempts}")

Out

high    billing          [email]     attempts=1

low     feature_request  [email]      attempts=1

urgent  bug              [email]    attempts=1

8 · Stream the object as it fills in

stream() yields the object as it's written; each yield has at least what the one before had, and the last is the validated object.

In [8]

python

for partial in extract.stream("From [email]: the app crashes when I upload a photo."):
    print(partial)

Out

{'category': 'bug', 'priority': 'high', 'customer_email': '[email]'}
category='bug' priority='high' customer_email='[email]'

9 · Partial JSON from any text

parse_partial_json reads whatever valid prefix a stream has so far, and skips chatter around the JSON.

In [9]

python

from shipit_agent.parsers import parse_partial_json

for text in ['{"name": "Al', '{"items": [1, 2, ', 'Sure! Here: {"a": 1}', '{"a": {"b": [true, nu']:
    print(f"{text!r:28} -> {parse_partial_json(text)}")

Out

'{"name": "Al'               -> {'name': 'Al'}
'{"items": [1, 2, '          -> {'items': [1, 2]}
'Sure! Here: {"a": 1}'       -> {'a': 1}
'{"a": {"b": [true, nu'      -> {'a': {'b': [True]}}

10 · JSON Schema instead of Pydantic

A plain JSON Schema dict works too; the parsed value is a dict.

In [10]

python

schema = {
    "type": "object",
    "properties": {
        "language": {"type": "string"},
        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
    },
    "required": ["language", "confidence"],
}
result = Agent(llm=haiku).run("Which language is this: 'Wo ist der Bahnhof?'", output_schema=schema)
print(result.parsed, type(result.parsed).__name__)

Out

{'language': 'German', 'confidence': 0.99} dict

11 · The building blocks

StructuredOutput and output_schema= are made of three helpers you can use with your own model calls.

In [11]

python

from shipit_agent import build_schema_prompt, parse_structured_output, schema_to_response_format

print(build_schema_prompt(Ticket)[:300], "…\n")
print(json.dumps(schema_to_response_format(Ticket))[:200], "…\n")
print(parse_structured_output('Here you go: {"category": "billing", "priority": "high", "customer_email": "[email]"}', Ticket))
try:
    parse_structured_output('{"category": "billing"}', Ticket)
except Exception as error:
    print(type(error).__name__, "->", str(error)[:120])

Out

Respond with a JSON object matching this schema:
```json
{
  "properties": {
    "category": {
      "title": "Category",
      "type": "string"
    },
    "priority": {
      "title": "Priority",
      "type": "string"
    },
    "customer_email": {
      "title": "Customer Email",
      "type":  …

{"type": "json_schema", "json_schema": {"name": "Ticket", "schema": {"properties": {"category": {"title": "Category", "type": "string"}, "priority": {"title": "Priority", "type": "string"}, "customer_ …

category='billing' priority='high' customer_email='[email]'
ParseError -> Pydantic validation failed: 2 validation errors for Ticket
priority
  Field required [type=missing, input_value={'catego

12 · The same schemas on Gemma 4

In [12]

python

review = Agent(llm=gemma).run(f"Review this payment code:\n{SNIPPET}", output_schema=Review).parsed
print(f"score {review.score}/10 — {review.summary}")
for f in review.findings:
    print(f"  line {f.line:2} [{f.severity.value:6}] {f.problem}")

r = StructuredOutput(llm=gemma, schema=Ticket).run(emails[2])
print("\n", r.value, "| attempts:", r.attempts)

Out

score 3/10 — The code lacks critical error handling for insufficient funds and uses floating-point arithmetic for currency, which leads to precision errors.
  line  2 [high  ] Using floats for currency calculations can cause precision errors. Decimal should be used instead.
  line  4 [high  ] The function does not handle the case where card.balance <= total; it simply returns the total without charging the card or raising an error.
  line  6 [medium] The function returns the total regardless of whether the transaction actually succeeded or failed.

 category='Bug' priority='High' customer_email='[email]' | attempts: 1

13 · Summary

run(..., output_schema=Model) returns the typed answer in result.parsed, after any tool use.
Pydantic constraints are enforced; failures go back to the model up to max_validation_retries.
When retries run out, run() raises with the last error and the model's text.
StructuredOutput does single calls, with stream() for partial objects.
JSON Schema dicts work too; build_schema_prompt, schema_to_response_format and parse_structured_output are the parts underneath.

Next: Guide 09 — permissions, hooks and the sandbox.

Notebooks · In-depth guides · Guide 08

Structured output

Open in Colab View on GitHub 12 code cells · 11 with saved output

Run end to end on shipit-agent 3.0 with real model calls (Haiku 5.5 on Anthropic, Gemma 4 31B on Bedrock). Every output below is from that run.

1 · Models

In [1]

python

import os

from shipit_agent import Agent
from shipit_agent.llms import build_llm_from_env
from shipit_agent.llms.factory import build_llm_from_settings

os.environ["SHIPIT_ANTHROPIC_MODEL"] = "claude-haiku-5-5"
haiku = build_llm_from_env("anthropic")
gemma = build_llm_from_settings({"model": "google.gemma-4-31b"}, provider="bedrock")

2 · A typed answer from a run

Pass the model class as output_schema. The typed object is result.parsed; result.output keeps the text.

In [2]

python

from pydantic import BaseModel

class Ticket(BaseModel):
    category: str
    priority: str
    customer_email: str

result = Agent(llm=haiku).run(
    "From: [email] — Checkout charges me twice since this morning!!", output_schema=Ticket)
print(repr(result.parsed))
print(type(result.parsed).__name__, "| priority:", result.parsed.priority)

Out

Ticket(category='billing', priority='high', customer_email='[email]')
Ticket | priority: high

3 · Nested models, lists, enums and constraints

Everything Pydantic validates is enforced: here an enum for severity, a list of findings, and a bounded score.

In [3]

python

from enum import Enum

from pydantic import Field

class Severity(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"

class Finding(BaseModel):
    line: int
    problem: str
    severity: Severity

class Review(BaseModel):
    summary: str = Field(max_length=200)
    findings: list[Finding]
    score: int = Field(ge=0, le=10, description="Overall quality, 0 to 10")

SNIPPET = '''
1  def charge(amount, card):
2      fee = amount * 0.029 + 0.30
3      total = round(amount + fee, 2)
4      if card.balance > total:
5          card.balance -= total
6      return total
'''
review = Agent(llm=haiku).run(f"Review this payment code:\n{SNIPPET}", output_schema=Review).parsed
print(f"score {review.score}/10 — {review.summary}")
for f in review.findings:
    print(f"  line {f.line:2} [{f.severity.value:6}] {f.problem}")

Out

score 3/10 — Charge logic uses floating-point money math, has no input validation, a check-then-act race on balance, and reports declines the same way as successes.
  line  2 [medium] Money is computed with binary floats (0.029 and amount * 0.029), which can produce rounding errors. Use Decimal and define the rounding mode explicitly.
  line  3 [medium] round() uses banker's rounding by default and rounds only the sum, so the fee is not rounded to cents on its own. The charged amount may not match the fee the business expects.
  line  4 [high  ] The check-then-act on card.balance is not atomic. Concurrent charges can both pass the check and overdraw the balance.
  line  4 [high  ] No validation of amount: a negative or zero amount passes the check, so a negative charge can credit the card. Non-numeric or None inputs also fail unpredictably.
  line  4 [low   ] The comparison is strict (>). A balance exactly equal to total is declined, which may or may not be intended. Confirm the boundary.
  line  6 [high  ] On insufficient funds the function still returns total, identical to success. Callers cannot tell a decline from a completed charge. Raise an exception or return an explicit status.
  line  5 [medium] No transaction record or idempotency key is created, so a retried call can double-charge and there is no audit trail.

4 · Tools first, then a typed answer

The agent can use its tools along the way; only the final answer has to fit the schema.

In [4]

python

import json

from shipit_agent.tools import FunctionTool

ACCOUNTS = {"acme": {"plan": "Team", "seats": 42, "mrr": 1218, "open_tickets": 7},
            "globex": {"plan": "Starter", "seats": 5, "mrr": 60, "open_tickets": 0}}

def account(name: str) -> str:
    """Look up a customer account by name."""
    return json.dumps(ACCOUNTS.get(name.lower(), {}))

class Health(BaseModel):
    account: str
    plan: str
    at_risk: bool
    reason: str

agent = Agent(llm=haiku, tools=[FunctionTool.from_callable(account, read_only=True)])
result = agent.run("Is Acme at risk? Look it up.", output_schema=Health)
print([r.name for r in result.tool_results])
print(result.parsed)

Out

['account']
account='Acme' plan='Team' at_risk=False reason='The account lookup returned no health or risk field, so this is not a confirmed assessment. Acme has a Team plan with 42 seats, $1,218 MRR, and 7 open tickets. Nothing in that data shows churn signals, but the record lacks usage, renewal, or sentiment data, so the account could still be at risk.'

5 · Validation retries

In [5]

python

class Slogan(BaseModel):
    text: str = Field(description="Exactly five words")

    def model_post_init(self, _):
        if len(self.text.split()) != 5:
            raise ValueError(f"text must be exactly 5 words, got {len(self.text.split())}")

result = Agent(llm=haiku).run("A slogan for a sandboxed AI agent runtime.", output_schema=Slogan)
print(result.parsed)
print("model turns:", sum(1 for m in result.messages if m.role == "assistant"))

Out

text='Agents act freely, safely contained'
model turns: 1

6 · When retries run out

An impossible constraint: once the retries are used up, run() raises, with the last validation error and the model's own text, so your code can decide what to do.

In [6]

python

class Impossible(BaseModel):
    n: int = Field(gt=10, lt=5)

try:
    Agent(llm=haiku).run("Give me a number.", output_schema=Impossible, max_validation_retries=1)
except Exception as error:
    print(type(error).__name__)
    print(str(error)[:400])

Out

output_schema parsing failed after 2 attempts. Last error: Pydantic validation failed: 1 validation error for Impossible
n
  Field required [type=missing, input_value={'error': 'The schema is ...o valid object exists.'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.13/v/missing — raw: No integer satisfies this schema. It requires `n > 10` and `n < 5` at the same time, which is impossible, so I can't return a valid object.

If you meant a different range, adjust the bounds. For example, `exclusiveMinimum: 4` and `exclusiveMaximum: 6` would allow only `5`.

7 · No agent: `StructuredOutput`

For extraction or classification with no tools, call the model directly. The result has the value, the attempts it took, the raw_text, and the failed attempts in history.

In [7]

python

from shipit_agent import StructuredOutput

extract = StructuredOutput(llm=haiku, schema=Ticket, max_retries=2)
emails = [
    "From: [email]. Checkout charges me twice since this morning!!",
    "[email] here — could you add dark mode at some point? No rush.",
    "From [email]: the export button 500s for every report. Board meeting in an hour.",
]
for email in emails:
    r = extract.run(email)
    print(f"{r.value.priority:7} {r.value.category:16} {r.value.customer_email:16} attempts={r.attempts}")

Out

high    billing          [email]     attempts=1

low     feature_request  [email]      attempts=1

urgent  bug              [email]    attempts=1

8 · Stream the object as it fills in

stream() yields the object as it's written; each yield has at least what the one before had, and the last is the validated object.

In [8]

python

for partial in extract.stream("From [email]: the app crashes when I upload a photo."):
    print(partial)

Out

{'category': 'bug', 'priority': 'high', 'customer_email': '[email]'}
category='bug' priority='high' customer_email='[email]'

9 · Partial JSON from any text

parse_partial_json reads whatever valid prefix a stream has so far, and skips chatter around the JSON.

In [9]

python

from shipit_agent.parsers import parse_partial_json

for text in ['{"name": "Al', '{"items": [1, 2, ', 'Sure! Here: {"a": 1}', '{"a": {"b": [true, nu']:
    print(f"{text!r:28} -> {parse_partial_json(text)}")

Out

'{"name": "Al'               -> {'name': 'Al'}
'{"items": [1, 2, '          -> {'items': [1, 2]}
'Sure! Here: {"a": 1}'       -> {'a': 1}
'{"a": {"b": [true, nu'      -> {'a': {'b': [True]}}

10 · JSON Schema instead of Pydantic

A plain JSON Schema dict works too; the parsed value is a dict.

In [10]

python

schema = {
    "type": "object",
    "properties": {
        "language": {"type": "string"},
        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
    },
    "required": ["language", "confidence"],
}
result = Agent(llm=haiku).run("Which language is this: 'Wo ist der Bahnhof?'", output_schema=schema)
print(result.parsed, type(result.parsed).__name__)

Out

{'language': 'German', 'confidence': 0.99} dict

11 · The building blocks

StructuredOutput and output_schema= are made of three helpers you can use with your own model calls.

In [11]

python

from shipit_agent import build_schema_prompt, parse_structured_output, schema_to_response_format

print(build_schema_prompt(Ticket)[:300], "…\n")
print(json.dumps(schema_to_response_format(Ticket))[:200], "…\n")
print(parse_structured_output('Here you go: {"category": "billing", "priority": "high", "customer_email": "[email]"}', Ticket))
try:
    parse_structured_output('{"category": "billing"}', Ticket)
except Exception as error:
    print(type(error).__name__, "->", str(error)[:120])

Out

Respond with a JSON object matching this schema:
```json
{
  "properties": {
    "category": {
      "title": "Category",
      "type": "string"
    },
    "priority": {
      "title": "Priority",
      "type": "string"
    },
    "customer_email": {
      "title": "Customer Email",
      "type":  …

{"type": "json_schema", "json_schema": {"name": "Ticket", "schema": {"properties": {"category": {"title": "Category", "type": "string"}, "priority": {"title": "Priority", "type": "string"}, "customer_ …

category='billing' priority='high' customer_email='[email]'
ParseError -> Pydantic validation failed: 2 validation errors for Ticket
priority
  Field required [type=missing, input_value={'catego

12 · The same schemas on Gemma 4

In [12]

python

review = Agent(llm=gemma).run(f"Review this payment code:\n{SNIPPET}", output_schema=Review).parsed
print(f"score {review.score}/10 — {review.summary}")
for f in review.findings:
    print(f"  line {f.line:2} [{f.severity.value:6}] {f.problem}")

r = StructuredOutput(llm=gemma, schema=Ticket).run(emails[2])
print("\n", r.value, "| attempts:", r.attempts)

Out

score 3/10 — The code lacks critical error handling for insufficient funds and uses floating-point arithmetic for currency, which leads to precision errors.
  line  2 [high  ] Using floats for currency calculations can cause precision errors. Decimal should be used instead.
  line  4 [high  ] The function does not handle the case where card.balance <= total; it simply returns the total without charging the card or raising an error.
  line  6 [medium] The function returns the total regardless of whether the transaction actually succeeded or failed.

 category='Bug' priority='High' customer_email='[email]' | attempts: 1

13 · Summary

run(..., output_schema=Model) returns the typed answer in result.parsed, after any tool use.
Pydantic constraints are enforced; failures go back to the model up to max_validation_retries.
When retries run out, run() raises with the last error and the model's text.
StructuredOutput does single calls, with stream() for partial objects.
JSON Schema dicts work too; build_schema_prompt, schema_to_response_format and parse_structured_output are the parts underneath.

Next: Guide 09 — permissions, hooks and the sandbox.

Structured output

1 · Models

2 · A typed answer from a run

3 · Nested models, lists, enums and constraints

4 · Tools first, then a typed answer

5 · Validation retries

6 · When retries run out

7 · No agent: StructuredOutput

8 · Stream the object as it fills in

9 · Partial JSON from any text

10 · JSON Schema instead of Pydantic

11 · The building blocks

12 · The same schemas on Gemma 4

13 · Summary

Structured output

1 · Models

2 · A typed answer from a run

3 · Nested models, lists, enums and constraints

4 · Tools first, then a typed answer

5 · Validation retries

6 · When retries run out

7 · No agent: StructuredOutput

8 · Stream the object as it fills in

9 · Partial JSON from any text

10 · JSON Schema instead of Pydantic

11 · The building blocks

12 · The same schemas on Gemma 4

13 · Summary

7 · No agent: `StructuredOutput`

7 · No agent: `StructuredOutput`