Pydantic AI

Pydantic AI is the Pydantic team’s typed agent framework. Every input and every output is validated against a Pydantic model. Tool schemas come from Python type hints. The framework’s identity is “Pydantic discipline applied to LLM calls” — strong types in, strong types out.

Flux’s built-in agent() task (in flux.tasks.ai.agent) offers structured output through a response_format parameter that takes a Pydantic BaseModel. The two overlap, but the depth of typing differs. This page covers running Pydantic AI agents inside Flux and when to pick which.

Prerequisites

pip install pydantic-ai flux-core

Pydantic AI bundles support for OpenAI, Anthropic, Gemini, and a handful of others. Pick one model string when constructing the Agent.

Wrapping a Pydantic AI agent

The integration is the same shape as every other framework on this list: build the agent, call agent.run() inside a @task, return the result.

from pydantic import BaseModel
from pydantic_ai import Agent
from flux import ExecutionContext, task, workflow


class TicketTriage(BaseModel):
    category: str
    severity: str
    summary: str
    suggested_owner: str


triage_agent = Agent(
    "openai:gpt-4o-mini",
    output_type=TicketTriage,
    system_prompt=(
        "Triage support tickets. Return category, severity (low/medium/high/critical), "
        "a one-sentence summary, and the suggested owning team."
    ),
)


@task.with_options(
    retry_max_attempts=3,
    retry_delay=2,
    timeout=60,
    secret_requests=["openai_api_key"],
)
async def triage(ticket_text: str, secrets: dict) -> TicketTriage:
    import os
    os.environ["OPENAI_API_KEY"] = secrets["openai_api_key"]
    result = await triage_agent.run(ticket_text)
    return result.output


@workflow
async def triage_workflow(ctx: ExecutionContext[dict]):
    ticket = (ctx.input or {}).get("ticket")
    if not ticket:
        return {"error": "ticket required"}
    decision = await triage(ticket)
    return {"triage": decision.model_dump(), "execution_id": ctx.execution_id}

Pydantic AI validates the LLM response against TicketTriage before result.output is returned. If the LLM produces malformed JSON or a value that fails Pydantic validation, the framework retries internally (its retries parameter on Agent) before raising. A failure that reaches Flux means Pydantic AI’s own retries are exhausted; Flux retries the whole task from there.

Tool calls

Pydantic AI’s @agent.tool decorator registers Python functions as tools the LLM can call. Tool schemas come from the function signature:

@triage_agent.tool
async def lookup_customer(ctx, customer_id: str) -> dict:
    """Look up a customer by ID. Returns name, tier, and account age."""
    # ... real lookup
    return {"name": "...", "tier": "enterprise", "account_age_days": 412}

Tool calls run inside agent.run(), which means they run inside the Flux task. If the tool itself needs durability — say, it writes to a database and you want that write to survive a crash — pull the work out of the tool and into a separate Flux task that the agent’s caller invokes.

Flux’s built-in agent() vs Pydantic AI

Flux ships flux.tasks.ai.agent, which returns a callable @task. It supports structured output via response_format=PydanticModel, tool calls via Flux @tasks passed as tools=[...], working memory, long-term memory, planning, sub-agent delegation, and human-in-the-loop approval through approval_mode.

The structured-output story is the cleanest difference. Pydantic AI validates inputs and outputs at every step — every tool argument, every intermediate response, every final output. Flux’s agent() validates only the final response against response_format. If your domain demands that every LLM-produced value pass through Pydantic, Pydantic AI is the tighter fit.

The orchestration story is the cleanest difference in the other direction. Flux’s agent() returns a @task, so every tool call is a Flux task event, every retry is a Flux retry, and the agent fits into the workflow’s event log alongside other tasks. A Pydantic AI agent is opaque to Flux — Flux sees one task call that takes 30 seconds and returns a value.

When to pick which

Pick Pydantic AI when:

• Every input and output needs schema validation. You are integrating an LLM into a system that already uses Pydantic for data validation everywhere else.
• The agent is one self-contained unit with its own tools that do not need to be durable on their own.
• You want strong typing on tool arguments enforced by the framework, not by convention.

Pick Flux’s built-in agent() when:

• You want each tool call to be a durable, retryable Flux task — visible in the event log and individually traceable.
• The agent uses durable patterns like working memory across pauses, long-term memory, or sub-agent delegation.
• The agent runs alongside other tasks (database writes, file uploads, external API calls) in the same workflow.

Mixing is fine. A workflow can call Pydantic AI for the strictly-typed extraction step and Flux’s agent() for the conversational step:

@workflow
async def support_pipeline(ctx: ExecutionContext[dict]):
    ticket = ctx.input["ticket"]
    triage_result = await triage(ticket)            # Pydantic AI
    response = await support_agent(ticket, triage_result)  # flux.agent()
    return {"triage": triage_result.model_dump(), "response": response}

What each side handles

Error handling

Pydantic AI raises UnexpectedModelBehavior when the LLM repeatedly fails output validation. Catch it and re-raise with context:

from pydantic_ai.exceptions import UnexpectedModelBehavior


@task.with_options(retry_max_attempts=2, retry_delay=5, timeout=60)
async def triage(ticket: str) -> TicketTriage:
    try:
        result = await triage_agent.run(ticket)
        return result.output
    except UnexpectedModelBehavior as e:
        raise RuntimeError(
            f"Pydantic AI could not validate the LLM response: {e}. "
            "Consider relaxing the output type or improving the system prompt."
        ) from e

Derived against Pydantic AI 0.x and Flux 0.56.0, 2026-07.