hubagenticai

Tutorials Intermediate · 1/12

Structured outputs for agents: make models emit JSON you can actually trust

Agents break where prose meets code. Build the validate-and-retry loop that turns flaky model JSON into a reliable contract — in plain Python, runnable without an API key.

updated 2026-10-08 ⏱ 35 min

Every step where one component consumes another’s output is a place your agent can silently rot: the planner hands subtasks to the dispatcher, a tool result feeds a decision, a triage agent files a note. If those handoffs are prose, you’re parsing vibes. Structured outputs make each handoff a contract — and this tutorial builds the enforcement machinery, because the contract is worthless without it.

Step 1 — Know the three failure modes

Models break JSON in exactly three ways, and your pipeline needs an answer for each:

  1. Wrapping — valid JSON buried in prose or markdown fences (“Sure! Here’s your JSON: …”). Answer: extract before parsing.
  2. Malformed syntax — single quotes, trailing commas, truncation. Answer: parse, and feed the error back.
  3. Schema violations — valid JSON, wrong shape: missing keys, invented enum values, a string where a number belongs. Answer: validate, and feed the violation back.

The pattern for all three is the same: detect precisely, tell the model exactly what was wrong, retry with a bounded budget.

Step 2 — Build the pipeline

Create structured.py. The mock model fails like real ones do — first a syntax error, then a schema violation, then success — so you can watch every recovery path fire:

import json

REQUIRED = {"category": str, "summary": str}
ALLOWED_CATEGORIES = {"task", "reference", "archive"}

def extract_json(text: str) -> str:
    """Failure mode 1: strip prose/fences around the first {...} block."""
    start, end = text.find("{"), text.rfind("}")
    return text[start:end + 1] if start != -1 and end > start else text

def validate(raw: str):
    """Failure modes 2 and 3: returns (data, None) or (None, precise_error)."""
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:
        return None, f"invalid JSON: {e}"
    for key, typ in REQUIRED.items():
        if key not in data:
            return None, f"missing required key: {key!r}"
        if not isinstance(data[key], typ):
            return None, f"{key!r} must be a {typ.__name__}"
    if data["category"] not in ALLOWED_CATEGORIES:
        return None, (f"category must be one of {sorted(ALLOWED_CATEGORIES)}, "
                      f"got {data['category']!r}")
    return data, None

class FlakyMockModel:
    """Fails the way real models fail, deterministically, for testing."""
    def __init__(self):
        self.calls = 0
    def complete(self, prompt: str) -> str:
        self.calls += 1
        if self.calls == 1:   # single quotes -> JSONDecodeError
            return "{'category': 'task', 'summary': 'Renew passport'}"
        if self.calls == 2:   # invented enum value -> schema violation
            return '{"category": "todo", "summary": "Renew passport"}'
        return '{"category": "task", "summary": "Renew passport"}'

def structured_call(model, prompt: str, max_attempts: int = 3) -> dict:
    error = None
    for attempt in range(1, max_attempts + 1):
        ask = prompt if error is None else (
            f"{prompt}\n\nYour previous reply failed validation: {error}\n"
            f"Reply with corrected JSON only — no prose, no fences.")
        data, error = validate(extract_json(model.complete(ask)))
        if data is not None:
            print(f"valid on attempt {attempt}")
            return data
        print(f"attempt {attempt} rejected: {error}")
    raise ValueError(f"no valid output after {max_attempts} attempts: {error}")

if __name__ == "__main__":
    result = structured_call(
        FlakyMockModel(),
        'Classify this note as JSON {"category": ..., "summary": ...}: '
        "'Renew passport by Friday'")
    print("result:", result)

Run it:

python structured.py

You’ll watch attempt 1 die on syntax, attempt 2 die on the invented "todo" category, and attempt 3 succeed, because each retry told the model precisely what to fix. Vague retries (“that was wrong, try again”) give the model nothing to act on. The exact error message is the part that makes a retry converge, so log your retry rate per step and watch it (see Troubleshooting).

attempt 1 rejected: invalid JSON: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)
attempt 2 rejected: category must be one of ['archive', 'reference', 'task'], got 'todo'
valid on attempt 3
result: {'category': 'task', 'summary': 'Renew passport'}

Step 3 — Use native structured outputs where you can

Every major API can now enforce a JSON Schema while the model generates (constrained decoding), so the output cannot have a syntax error or a missing required key. The parameter is named differently on each provider. Here is the same contract, written once as JSON Schema and then sent to four backends. Save it as native.py next to structured.py:

import json, sys

NOTE_SCHEMA = {
    "type": "object",
    "properties": {
        "category": {"type": "string", "enum": ["task", "reference", "archive"]},
        "summary": {"type": "string"},
    },
    "required": ["category", "summary"],
    "additionalProperties": False,
}
PROMPT = "Classify this note and summarize it in five words or fewer: 'Renew passport by Friday'"

def claude(prompt: str) -> str:
    import anthropic                      # pip install anthropic
    client = anthropic.Anthropic()        # reads ANTHROPIC_API_KEY
    resp = client.messages.create(
        model="claude-haiku-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
        output_config={"format": {"type": "json_schema", "schema": NOTE_SCHEMA}},
    )
    # Take the text block. If thinking is on, content[0] may not be text.
    return next(b.text for b in resp.content if b.type == "text")

def openai_responses(prompt: str) -> str:
    from openai import OpenAI             # pip install openai
    client = OpenAI()                     # reads OPENAI_API_KEY
    resp = client.responses.create(
        model="gpt-6-luna",
        input=prompt,
        text={"format": {"type": "json_schema", "name": "note",
                         "schema": NOTE_SCHEMA, "strict": True}},
    )
    return resp.output_text

def gemini(prompt: str) -> str:
    from google import genai              # pip install google-genai
    client = genai.Client()               # reads GEMINI_API_KEY
    interaction = client.interactions.create(
        model="gemini-3.8-flash",
        input=prompt,
        response_format={"type": "text", "mime_type": "application/json",
                         "schema": NOTE_SCHEMA},
    )
    return interaction.output_text

def llama_cpp(prompt: str, base_url="http://127.0.0.1:8080") -> str:
    import urllib.request                 # stdlib only; llama-server must be running
    body = {
        "messages": [{"role": "user", "content": prompt}],
        "response_format": {"type": "json_schema", "json_schema": {"schema": NOTE_SCHEMA}},
    }
    req = urllib.request.Request(f"{base_url}/v1/chat/completions",
                                 data=json.dumps(body).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)["choices"][0]["message"]["content"]

if __name__ == "__main__":
    fn = {"claude": claude, "openai": openai_responses,
          "gemini": gemini, "llama": llama_cpp}[sys.argv[1]]
    print(json.loads(fn(PROMPT, *sys.argv[2:])))
python native.py claude      # or: openai, gemini, llama [base_url]
ProviderWhere the schema goesNotes
Anthropic (Claude)output_config.format = {"type": "json_schema", "schema": ...}The SDK helper client.messages.parse(output_format=YourPydanticModel) builds this for you. For tool calls, set "strict": true on the tool definition. Objects must set additionalProperties: false. The docs list numeric and length constraints (minimum, maxLength) as unsupported. In our test a maxLength was accepted without an error, so don’t count on it being enforced: check it in validate().
OpenAI (Responses API)text.format = {"type": "json_schema", "name": ..., "schema": ..., "strict": true}On Chat Completions the same idea is response_format={"type": "json_schema", "json_schema": {...}}. client.responses.parse(text_format=Model) accepts a Pydantic model.
Google (Gemini)response_format = {"type": "text", "mime_type": "application/json", "schema": ...} on the Interactions APIOlder generate_content code passes response_mime_type="application/json" plus response_json_schema in the config.
llama.cpp (llama-server)response_format = {"type": "json_schema", "json_schema": {"schema": ...}} on /v1/chat/completionsThe server converts the schema into a GBNF grammar. You can also pass a raw grammar with --grammar-file, or json_schema on the native /completion endpoint.

What we actually ran on 2026-10-08: the claude path live against claude-haiku-5-5, and the llama path against a local llama-server (set one up with run agents locally) (build b11514) running gpt-oss-20b. Three runs out of three returned schema-valid JSON, for example {'category': 'task', 'summary': 'Renew passport Friday'}. We had no OpenAI or Gemini key on the test machine, so for those two we captured the HTTP request each SDK builds and checked it against the provider’s docs. The request shapes are verified; the live responses are not.

Keep the validation loop anyway. Native modes guarantee shape, not sense. A schema-valid refund of €0.00 for order “0000” still needs your semantic checks. And every system has handoffs (sub-agent to sub-agent, cached outputs, providers or models without schema support) where native enforcement isn’t available. Run the native output through the same validate() from Step 2, and you get one contract with two layers of defense.

Step 4 — Design contracts that models can hit

Reliability is half enforcement, half contract design:

  • Flat beats nested. Every nesting level multiplies failure modes.
  • Enums beat free text — "category": "task" is checkable; "category": "a to-do item" is a judgment call.
  • Forbid extras (additionalProperties: false in JSON Schema) so hallucinated fields fail loudly instead of flowing downstream.
  • One schema, one place — share the contract between your prompt, your validator, and your golden tasks, or they will drift apart. The tool-schema template is the reusable starting point.

Troubleshooting

Claude returns 400: “For ‘object’ type, ‘additionalProperties’ must be explicitly set to false”

This is the exact error we got when we removed “additionalProperties”: False from NOTE_SCHEMA. Anthropic’s structured outputs require it on every object in the schema, nested ones included. Add it. It is good contract design anyway (Step 4).

The model wraps JSON in markdown fences no matter what I say

Stop fighting it in the prompt — handle it in code. extract_json exists precisely because “JSON only, no fences” is a suggestion the model will eventually ignore. Extraction + validation is cheaper than prompt whack-a-mole.

Validation passes but downstream code still breaks

Your schema is looser than your consumer. Every assumption the consuming code makes (non-empty strings? positive numbers? known keys only?) must exist in the validator — additionalProperties: false and enums close most of the gap.

Retries succeed but latency tripled

Count retries per call in your metrics. A rising retry rate usually means the schema grew too complex for the model tier — flatten the schema or upgrade the model for that step, don’t just raise max_attempts.

Where to go next

Schemas are what make the seams between deterministic code and model steps safe. Workflow vs. agent shows where those seams belong in a larger system. To check that your contracts hold across model upgrades, put them in a golden-task eval harness.

newsletter

One practical agentic-AI guide in your inbox. No news, no hype.

Tutorials and decision frameworks as they ship. Unsubscribe anytime.

Quick Jump: