Structured JSON from an LLM: validate it with Pydantic
A practical Python pipeline for treating model output as untrusted text, validating it with Pydantic, and limiting repair to one attempt.
Contents

“Return JSON only” is a formatting request, not a data contract. A model can still add Markdown, omit a required field, return a string where a number is expected, or invent an enum value. Treat every LLM response as untrusted text until application code validates it.
The boundary should be explicit:
LLM -> raw response -> Pydantic validation -> typed object
| validation error
v
one repair attempt -> success or explicit failure
Downstream code receives either a known type or a separate failure result. It never receives an almost-correct dictionary.
Define one Pydantic contract
This example asks the model to classify a support ticket. The schema rejects extra keys and constrains every field that matters to the consumer.
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class SupportTicket(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=120)
priority: Literal["low", "medium", "high"]
affected_services: list[str] = Field(min_length=1, max_length=5)
needs_human_review: bool
Generate the prompt schema from the same model with SupportTicket.model_json_schema(). That avoids maintaining a prompt schema and a validator separately. A schema in the prompt can guide the model, but only the local validator decides whether the result is acceptable.
Keep the raw response before parsing
The following client uses BetterToken’s documented OpenAI-compatible Chat Completions interface. The API key and full model ID stay in environment variables.
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["BETTERTOKEN_API_KEY"],
base_url="https://www.bettertoken.ai/v1",
)
def ask_model(user_text: str, schema_text: str) -> str:
response = client.chat.completions.create(
model=os.environ["BETTERTOKEN_MODEL_ID"],
messages=[
{
"role": "system",
"content": (
"Return one JSON object without Markdown. "
"It must match this JSON Schema:\n" + schema_text
),
},
{"role": "user", "content": user_text},
],
)
return response.choices[0].message.content or ""
Keep that returned string until validation finishes. If diagnostics must be retained, use protected storage with limited access and retention. Do not put complete prompts, personal data, API keys, or sensitive model output into a shared log. An HTTP 200 only means that the API returned a response; it does not prove that the text matches the business schema.
Validate the JSON string directly
Pydantic v2 can parse and validate in one step:
from pydantic import ValidationError
def validate_ticket(raw: str) -> SupportTicket:
return SupportTicket.model_validate_json(raw, strict=True)
With strict=True, a value such as the string "true" is not silently converted to a boolean. A ValidationError can represent invalid JSON syntax, a missing field, a wrong type, an unsupported enum value, or an extra field. Do not hide these cases with defaults.
For a repair request, extract a compact list containing each error location, type, and message. Use exc.errors(include_url=False, include_input=False) so the diagnostic object does not repeat the rejected input.
Allow one repair attempt
Repair is a new request. Send the original response, the compact validation errors, and the same schema; instruct the model not to add facts. Then run the result through exactly the same validator.
Use a two-pass loop: the original response and one repair. If the second response is invalid, return a result such as value=None, both raw responses, and the final error list. The caller can route it to manual review or an error queue. Never execute a tool call, payment, or other external action before validation succeeds.
Handle transport failures separately. A timeout says nothing about JSON validity, while an unlimited repair loop can consume quota and resend sensitive text indefinitely.
Test the contract without an API call
Start with fixed fixtures: one valid object and one object containing an invalid enum, an empty list, a string instead of a boolean, and an extra field. Assert that the valid fixture is accepted and that the invalid one raises the expected Pydantic error types. Also simulate two invalid responses and confirm that the result is an explicit failure.
For this article, those local checks passed with Python and Pydantic 2.12.5. No live model request was made, so this is not a claim that a particular model always returns valid JSON.
Before production, version the Pydantic model with its consumer, limit raw response size and repair count, separate transport/API/schema metrics, redact sensitive logs, and add fixtures whenever the schema changes.
Check the current request format in BetterToken Chat Completions. Pydantic remains the local trust boundary: the API generates text, and your application decides whether that text is data.
Complete executable reference
These blocks preserve the exact executable reference for schema generation, compact validation errors, one repair attempt, and fixed fixtures.
import json
schema = SupportTicket.model_json_schema()
schema_text = json.dumps(schema, ensure_ascii=False)
def error_summary(exc: ValidationError) -> list[dict[str, object]]:
return [
{
"path": ".".join(str(part) for part in item["loc"]),
"type": item["type"],
"message": item["msg"],
}
for item in exc.errors(include_url=False, include_input=False)
]
import json
from dataclasses import dataclass
@dataclass
class ParseResult:
value: SupportTicket | None
raw_responses: list[str]
errors: list[dict[str, object]]
def repair_model(
raw: str,
errors: list[dict[str, object]],
schema_text: str,
) -> str:
response = client.chat.completions.create(
model=os.environ["BETTERTOKEN_MODEL_ID"],
messages=[
{
"role": "system",
"content": (
"Repair the JSON. Return one JSON object without Markdown. "
"Do not add facts. The object must match this schema:\n"
+ schema_text
),
},
{
"role": "user",
"content": json.dumps(
{"raw": raw, "validation_errors": errors},
ensure_ascii=False,
),
},
],
)
return response.choices[0].message.content or ""
def parse_with_one_repair(user_text: str) -> ParseResult:
raw_responses: list[str] = []
raw = ask_model(user_text, schema_text)
for attempt in range(2):
raw_responses.append(raw)
try:
value = validate_ticket(raw)
return ParseResult(value=value, raw_responses=raw_responses, errors=[])
except ValidationError as exc:
errors = error_summary(exc)
if attempt == 1:
return ParseResult(
value=None,
raw_responses=raw_responses,
errors=errors,
)
raw = repair_model(raw, errors, schema_text)
raise AssertionError("unreachable")
valid_raw = """{
"title": "Login fails after token refresh",
"priority": "high",
"affected_services": ["auth-api"],
"needs_human_review": true
}"""
invalid_raw = """{
"title": "Login fails",
"priority": "urgent",
"affected_services": [],
"needs_human_review": "yes",
"confidence": 0.98
}"""
ticket = validate_ticket(valid_raw)
assert ticket.priority == "high"
try:
validate_ticket(invalid_raw)
except ValidationError as exc:
assert {item["type"] for item in exc.errors()} >= {
"literal_error",
"too_short",
"bool_type",
"extra_forbidden",
}
else:
raise AssertionError("invalid fixture was accepted")