The demo works every time: ask the model to return JSON matching a schema, json.loads() the response, move on. Then it ships, and three weeks in, someone finds a stack trace where the model wrapped its JSON in a markdown code fence, or added a friendly “Here’s the extracted data:” sentence before the actual object, or — worse — returned syntactically valid JSON that’s missing a field your downstream code assumed would always be there. None of this is the model being unreliable in some mysterious way. It’s the predictable result of treating “ask nicely for JSON” as equivalent to “get a validated JSON object,” when those are two different levels of guarantee that happen to look the same in a happy-path demo.
The fix isn’t a smarter prompt. It’s a pipeline that assumes the model will occasionally do exactly what you didn’t ask for, and has a defined, tested path for what happens when it does — a schema that’s actually enforced rather than just described, a retry loop that hands the model its own validation error, and a fallback for the case where none of that works and you get prose back anyway. This post builds that pipeline in Python.
You’ll learn:
- Why prompting for JSON and validating JSON are different problems with different failure modes
- How to define an LLM output contract with Pydantic and get a validated instance directly
- The real difference between constrained decoding and “please respond in JSON” prompting
- How to build a retry loop that feeds validation errors back to the model instead of just trying again blind
- What to do when the model returns prose despite every guardrail
- How to handle streamed, partial JSON without your parser choking on an incomplete object
- A production checklist for a structured-output pipeline you can actually trust
Table of Contents
- Two Different Problems Wearing One Name
- Pydantic Schemas as the Contract
- Constrained Decoding vs Prompted JSON
- Retry on Validation Failure
- When the Model Returns Prose Anyway
- Handling Partial and Streamed JSON
- Nested and Union Schemas
- Structured Outputs Inside an Agent Loop
- Production Checklist
- Wrapping Up
Two Different Problems Wearing One Name
“Get structured output from an LLM” is usually shorthand for two separate guarantees that don’t automatically come together. The first is shape: is the response syntactically valid JSON at all, versus prose, a code fence, or JSON with a trailing sentence appended. The second is schema: given valid JSON, does it actually have the fields you need, with the right types, satisfying whatever constraints your downstream code assumes.
A prompt that just says “respond in JSON” addresses neither guarantee reliably — it’s a request, not a constraint, and the model is free to be a few tokens off from what you asked. The techniques in this post address each guarantee at a different layer: constrained decoding (or a schema-aware API mode) addresses shape structurally, so the format literally cannot come out wrong; Pydantic validation addresses schema, catching an output that’s valid JSON but the wrong JSON — a string where you needed an int, a missing required field, an enum value that doesn’t exist. Treating these as one problem is how teams end up with a single try/except json.JSONDecodeError that silently lets through a perfectly parseable object missing half the fields the rest of the code depends on.
Pydantic Schemas as the Contract
Define the output shape once, as a Pydantic model, and let it serve three purposes at once — the schema that documentation would otherwise describe, the validator that runs on every response, and the type your downstream code actually gets to work with.
from pydantic import BaseModel, Field
from typing import Literal
class TicketClassification(BaseModel):
category: Literal["billing", "technical", "account", "other"]
urgency: Literal["low", "medium", "high"]
summary: str = Field(description="One-sentence summary of the customer's issue")
requires_escalation: bool
With the Anthropic SDK, client.messages.parse() takes this model directly and returns a validated instance rather than a content block you have to parse yourself:
import anthropic
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-5",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"Classify this support ticket:\n\n{ticket_text}",
}],
output_format=TicketClassification,
)
ticket = response.parsed_output
print(ticket.category, ticket.urgency) # typed, validated attribute access — not dict lookups
response.parsed_output is a real TicketClassification instance, not a dict you’re hoping has the right keys — a typo like ticket.catgory fails at development time with a normal Python AttributeError, instead of failing silently at runtime with a KeyError three layers deeper in your call stack. That’s a small thing that adds up: the schema stops being documentation you hope stayed accurate and becomes something your type checker and your IDE both actually enforce.
For cases where you need the raw JSON schema rather than a Pydantic-driven parse — talking to a system that hands you a schema dynamically, for instance — the same guarantee is available at the request level via output_config:
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": f"Classify this ticket:\n\n{ticket_text}"}],
output_config={
"format": {
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"category": {"type": "string", "enum": ["billing", "technical", "account", "other"]},
"urgency": {"type": "string", "enum": ["low", "medium", "high"]},
"summary": {"type": "string"},
"requires_escalation": {"type": "boolean"},
},
"required": ["category", "urgency", "summary", "requires_escalation"],
"additionalProperties": False,
},
}
},
)
output_config.format guarantees the response content is text containing valid JSON matching that schema — the shape guarantee — but note it’s the raw dict at that point; you’d still call TicketClassification.model_validate_json(...) yourself to get the schema guarantee (type coercion, enum checking) that .parse() gives you automatically. In practice, prefer .parse() with a Pydantic model whenever you control the schema in code — it’s strictly less to maintain than defining the same constraints twice, once as a Pydantic model and once as a raw JSON schema dict.
Constrained Decoding vs Prompted JSON
It’s worth being precise about what changed here relative to older approaches, because “ask the model to return JSON in the system prompt” is still common advice and it’s meaningfully weaker than a schema-aware output mode. Prompted JSON is a request the model is free to satisfy imperfectly — it can add a preamble, forget a closing brace under unusual sampling conditions, or use a plausible-looking key you didn’t specify instead of the one you did. It usually works, right up until the input is unusual enough that it doesn’t.
output_config.format (and the .parse() helper built on it) constrains the response at generation time so the shape guarantee is structural rather than requested — the model is prevented from emitting content outside the schema, not just asked nicely to stay inside it. This is the actual reason to reach for a schema-aware API mode over a well-crafted prompt: it converts “the model usually gets this right” into “the model cannot get the shape wrong,” which is a categorically different reliability guarantee, even though it still leaves schema-level validation (the second guarantee from the section above) as something your own code has to check.
Note one interaction worth knowing about: output_config.format is not compatible with document citations (citations: {enabled: true}) — if your pipeline needs both structured output and citation tracking against a source document, that’s a real design constraint to account for, not an edge case to discover in production.
Retry on Validation Failure
Even with a schema-constrained response, Pydantic validation can still fail — an enum-adjacent value the schema allowed as a generic string but your Literal type doesn’t accept, a numeric field the model populated with a value outside a range you enforce in a custom validator. The single most effective thing you can do here is not “try again with the same prompt” but “try again with the model’s own error message included,” because a validation error is specific, actionable feedback the model can usually act on directly.
from pydantic import ValidationError
def classify_with_retry(ticket_text: str, max_attempts: int = 3) -> TicketClassification:
messages = [{"role": "user", "content": f"Classify this support ticket:\n\n{ticket_text}"}]
for attempt in range(max_attempts):
response = client.messages.create(
model="claude-opus-5",
max_tokens=1024,
messages=messages,
output_config={"format": {"type": "json_schema", "schema": TicketClassification.model_json_schema()}},
)
text = next(b.text for b in response.content if b.type == "text")
try:
return TicketClassification.model_validate_json(text)
except ValidationError as e:
if attempt == max_attempts - 1:
raise
messages.append({"role": "assistant", "content": text})
messages.append({
"role": "user",
"content": f"That response failed validation with this error:\n{e}\n"
f"Return corrected JSON matching the schema exactly.",
})
raise RuntimeError("unreachable")
Two details make this loop actually effective rather than just slower. First, the failed attempt and the error both go back into the conversation — the model sees exactly what it got wrong, not a generic “try again.” Second, max_attempts is a real ceiling with a real exception at the end, not an infinite retry that turns one bad request into a silent cost multiplier; three attempts catches the overwhelming majority of validation failures in practice, and a request that still fails after three is telling you something about the input or the schema, not just needing one more try.
When the Model Returns Prose Anyway
Even with a schema-constrained mode, plan for the response occasionally not being what you expect — a refusal, an unusual stop_reason, a client library edge case, or a model that decided the honest answer to “extract this data” is “there is no data to extract” and said so in a sentence. Defensive parsing here is cheap insurance against a class of production incident that otherwise looks like a random, unreproducible crash.
import json
import re
def extract_json_defensively(text: str) -> dict | None:
# Try a direct parse first — the common case.
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# Strip a markdown code fence if the model wrapped its output in one.
fence_match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
if fence_match:
try:
return json.loads(fence_match.group(1))
except json.JSONDecodeError:
pass
# Last resort: grab the first {...} block and hope it's balanced.
brace_match = re.search(r"\{.*\}", text, re.DOTALL)
if brace_match:
try:
return json.loads(brace_match.group(0))
except json.JSONDecodeError:
pass
return None
Wire this in as a fallback behind the constrained-mode path, not a replacement for it — reach for regex extraction only after the schema-aware call fails outright (a stop_reason you didn’t expect, an API error) rather than routing every response through best-effort string surgery by default. And whatever this returns, still push it through TicketClassification.model_validate(...) before trusting it; recovering valid-looking JSON from a messy string is not the same thing as recovering JSON that satisfies your schema; those are the same two guarantees from the top of this post, and skipping the second one here defeats the entire point of the exercise.
Handling Partial and Streamed JSON
Streaming a structured response introduces a version of the shape problem you don’t get with a single blocking call: at any point mid-stream, the accumulated text is by definition incomplete JSON, and parsing it before the object is closed will throw on every single chunk until the last one. Don’t parse every chunk — buffer and parse defensively, or use a streaming-JSON parser built for exactly this if you need partial results rendered incrementally (a progressively-filling form in a UI, for instance).
def stream_classification(ticket_text: str):
buffer = ""
with client.messages.stream(
model="claude-opus-5",
max_tokens=1024,
messages=[{"role": "user", "content": f"Classify this ticket:\n\n{ticket_text}"}],
output_config={"format": {"type": "json_schema", "schema": TicketClassification.model_json_schema()}},
) as stream:
for text_delta in stream.text_stream:
buffer += text_delta
# Only attempt a full parse once the stream is done, unless you have
# a genuine incremental-rendering need that justifies a partial-JSON parser.
return TicketClassification.model_validate_json(buffer)
If you do need incremental rendering, treat partial parses as strictly best-effort display state — never validate a partial object against your full Pydantic schema, since required fields that simply haven’t streamed in yet will fail validation in a way that has nothing to do with the model doing anything wrong.
Nested and Union Schemas
Real extraction tasks are rarely one flat object. Pydantic’s nested models and unions work through the same output_format / output_config path without extra plumbing, which is one of the stronger arguments for defining schemas in Pydantic rather than hand-writing JSON Schema dicts as complexity grows.
class LineItem(BaseModel):
description: str
quantity: int
unit_price: float
class RefundRequest(BaseModel):
order_id: str
items: list[LineItem]
reason: Literal["damaged", "wrong_item", "changed_mind", "other"]
response = client.messages.parse(
model="claude-opus-5", max_tokens=1024,
messages=[{"role": "user", "content": f"Extract the refund request:\n\n{email_text}"}],
output_format=RefundRequest,
)
refund = response.parsed_output
total = sum(item.quantity * item.unit_price for item in refund.items)
The failure mode worth testing deliberately here is an empty or single-element list where your prompt implied there’d usually be several — items: [] is valid against the schema above but is very likely a sign the model didn’t find any line items rather than the email genuinely having zero, and that ambiguity is exactly the kind of thing a schema can’t distinguish for you. Add an explicit min_length=1 constraint if an empty list is never actually valid for your use case, so that ambiguity becomes a validation failure your retry loop can catch instead of a 0 silently flowing downstream.
Structured Outputs Inside an Agent Loop
Everything above assumes a single call in isolation, but structured output validation matters even more once it’s sitting inside a node of a larger agent rather than standing alone — a bad classification there doesn’t just return a wrong answer, it routes the rest of the agent’s execution down the wrong path. If you’re building an agent complex enough to need explicit branching and cycles rather than a single linear call, that’s the point where a plain chain stops being enough and something like LangGraph earns its complexity — and a TicketClassification-style node is exactly the kind of thing that ends up gating a conditional edge.
def classify_node(state: AgentState) -> dict:
response = client.messages.parse(
model="claude-opus-5", max_tokens=1024,
messages=[{"role": "user", "content": f"Classify this ticket:\n\n{state['ticket_text']}"}],
output_format=TicketClassification,
)
return {"classification": response.parsed_output}
def route_on_category(state: AgentState) -> str:
return "escalate" if state["classification"].requires_escalation else "auto_respond"
The retry-on-validation-failure loop from earlier belongs inside classify_node, not wrapped around the whole graph invocation — a validation failure on one node’s output shouldn’t force replaying every node that already ran successfully. Keeping the retry local to the node that produces the structured output is what makes this compose cleanly with the rest of a graph’s error handling instead of fighting it.
Production Checklist
- Use a schema-aware output mode (
.parse()oroutput_config.format), not a prompt-only request for JSON, for anything downstream code depends on parsing successfully. - Validate through Pydantic even when the API guarantees shape — shape and schema are different guarantees, and only one of them is enforced at generation time.
- Retry with the validation error included, capped at a small, explicit attempt count with a real exception at the end.
- Keep a defensive extraction fallback for the case a constrained call still doesn’t come back as expected — but validate whatever it recovers before trusting it.
- Never validate a partial streamed buffer against the full schema — treat partial state as display-only.
- Add explicit constraints (
min_length, value ranges) for anything a merely-valid-but-empty or merely-valid-but-implausible value would break downstream. - Log validation failures with the raw model output attached — a growing bucket of retry failures on the same field is a signal your schema or prompt needs to change, not just that retries are absorbing the problem.
A related detail worth flagging for anyone coming from an older codebase: some SDKs still expose a top-level output_format parameter directly on messages.create() from before the current .parse() helper and output_config.format shape existed. That parameter is deprecated — new code should use output_config: {format: {...}} on messages.create() for the raw-schema path, or .parse() with a Pydantic model for the typed path shown throughout this post. If you’re maintaining an existing integration and see the old parameter name, it’s a sign the code predates the current structured-output surface and is worth migrating rather than leaving as-is, since deprecated parameters are exactly the kind of thing that quietly stops being supported in a future SDK version with no warning at call time.
Wrapping Up
The gap between a structured-output demo and a structured-output pipeline you can trust in production is entirely in how each layer handles the model not doing exactly what you asked. A Pydantic model gives you a real, enforceable contract instead of a comment describing one; a schema-aware output mode makes the shape of the response structural rather than requested; a retry loop that hands the model its own validation error recovers from the schema failures that still get through; and a defensive fallback catches the rest without pretending recovered text is automatically valid. None of these individually is exotic engineering — the reliability comes from using all four together, deliberately, instead of discovering in production which one you skipped.
