PydanticAI structured outputs across gateway-routed models

Routing LLM calls through an inference gateway changes how you configure client libraries. This guide shows how to get PydanticAI structured outputs gateway routed models working end to end, using an OpenAI-compatible endpoint that sits in front of multiple providers and handles failover transparently.

Step 1: Point PydanticAI at the gateway

PydanticAI’s OpenAI provider is just a thin wrapper over the official openai SDK. To route through a gateway, construct an AsyncOpenAI client with the gateway’s base URL and pass it to OpenAIModel.

import os
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from openai import AsyncOpenAI

GATEWAY_BASE = os.environ["GATEWAY_BASE_URL"]  # e.g. https://gateway.example.com/v1
GATEWAY_KEY = os.environ["GATEWAY_API_KEY"]

client = AsyncOpenAI(
    base_url=GATEWAY_BASE,
    api_key=GATEWAY_KEY,
    # gateway honors client routing directives via model name or headers
    default_headers={"x-gateway-tag": "prod-agents"},
)

# model string is whatever the gateway expects: "provider/model" or just "model"
model = OpenAIModel("anthropic/claude-3.5-sonnet", openai_client=client)

A gateway such as n4n.ai exposes one OpenAI-compatible endpoint for 240+ models and automatically falls back when a provider is rate-limited, so the same code works without hard-coding retries.

Step 2: Define strict output schemas

Structured outputs only work if the model returns JSON that matches your Pydantic model. Keep schemas flat, avoid unions with ambiguous discriminators, and use Field constraints to push the model toward valid values.

from pydantic import BaseModel, Field

class SupportTicket(BaseModel):
    title: str = Field(max_length=120)
    priority: int = Field(ge=1, le=5)
    tags: list[str] = Field(default_factory=list, max_length=5)
    requires_oncall: bool

class TicketBatch(BaseModel):
    tickets: list[SupportTicket]

PydanticAI serializes this to a JSON schema and sends it in the request. Some gateway-routed models enforce strict schema mode; others approximate. Test each model you route to.

Step 3: Build the agent with routing intent

Create the agent with output_type set to your schema. The model name you passed to OpenAIModel is the primary routing directive. If you want the gateway to pick a fallback, omit a specific model and send a header the gateway understands, or rely on its default policy.

agent = Agent(
    model,
    output_type=TicketBatch,
    system_prompt=(
        "Parse the incoming support channel text into structured tickets. "
        "Set requires_oncall=true if priority >= 4 or the text mentions prod/sev1."
    ),
)

If your gateway supports per-request model hints, pass them through model_settings or a custom client header:

# example: force a specific model for this run
run_model = OpenAIModel("openai/gpt-4o-mini", openai_client=client)
agent_alt = Agent(run_model, output_type=TicketBatch)

Step 4: Execute and validate

Run the agent inside an async context. PydanticAI raises ValidationError if the gateway-routed model returns malformed JSON. Catch it and retry with a different model or a simplified schema.

import asyncio
from pydantic import ValidationError

async def main():
    text = """
    Prod DB is down (sev1). Page oncall. Also: typo in docs link, low priority.
    """
    try:
        result = await agent.run(text)
        batch = result.output
        for t in batch.tickets:
            print(f"{t.priority} | {t.title} | oncall={t.requires_oncall}")
    except ValidationError as e:
        print("Schema mismatch:", e)

asyncio.run(main())

When the gateway performs automatic fallback, the response still arrives as valid OpenAI-compatible chat completion JSON, so PydanticAI never knows the underlying provider changed.

Step 5: Read usage and cache hints

Gateways meter per-token usage and forward provider cache-control hints. PydanticAI exposes the raw response via result.usage (or result._response in older versions). Inspect it to confirm you’re billed correctly and that cache hits are recorded.

# after a successful run
usage = result.usage()
print(usage.request_tokens, usage.response_tokens, usage.total_tokens)
# gateway may include vendor-specific fields in response headers
resp = result._response
if resp and "x-cache" in resp.headers:
    print("cache status:", resp.headers["x-cache"])

If you send cache-control hints (e.g., {"cache_control": {"type": "ephemeral"}} on system messages where the provider supports it), the gateway forwards them. Not all routed models honor them; verify with the provider docs.

Step 6: Verify success

You have a working integration when:

1. The agent returns a TicketBatch instance without raising ValidationError.
2. usage.total_tokens is non-zero and matches expected prompt+completion size.
3. Logs from the gateway show the request hit the intended model or a documented fallback.
4. Running the same input against two different model strings yields schema-compatible objects.

A quick smoke test:

python -m pytest test_agent.py -q

where test_agent.py asserts isinstance(result.output, TicketBatch) and len(result.output.tickets) >= 1.

Pitfalls when mixing models

Schema strictness varies. OpenAI’s structured output mode is strict; Anthropic via gateway may be best-effort. Always validate client-side even if the gateway claims schema enforcement.

Enum and regex constraints. Some models ignore pattern or enum silently. Replace enums with post-hoc mapping in Python:

PRIORITY_MAP = {"p1": 5, "p2": 4, "p3": 3, "p4": 2, "p5": 1}

Streaming conflicts. PydanticAI structured outputs are not streamed token-by-token in most gateway configurations. Disable stream unless your gateway explicitly supports partial JSON validation.

Latency from fallback. Automatic fallback adds tail latency. Set a timeout on the AsyncOpenAI client:

client = AsyncOpenAI(base_url=GATEWAY_BASE, api_key=GATEWAY_KEY, timeout=30.0)

Model name drift. Gateway model catalogs change. Pin a versioned model ID (claude-3.5-sonnet-20241022) instead of a floating alias when production stability matters.

Step 7: Extend to multi-agent routing

Once the base pattern works, spawn multiple agents with different output_type schemas and let the gateway load-balance. Because PydanticAI structured outputs gateway routed models all share the same OpenAI-compatible contract, you can swap the OpenAIModel instance without touching business logic.

def make_agent(output_cls, model_id: str) -> Agent:
    m = OpenAIModel(model_id, openai_client=client)
    return Agent(m, output_type=output_cls)

triage_agent = make_agent(TicketBatch, "anthropic/claude-3.5-sonnet")
summary_agent = make_agent(Summary, "openai/gpt-4o-mini")

Keep the schemas in a separate module so both agents import the same source of truth. That prevents drift when the gateway routes to a model with stricter validation.

Closing checks

Run the full pipeline against at least three routed models. Confirm token counts differ (proving routing happened) but output shapes stay identical. If a model repeatedly fails validation, drop it from the gateway routing list or add a pre-parsing prompt step. The payoff is one code path, many providers, and no bespoke retry logic.