← プロジェクト一覧に戻る

Pydantic AI

Pydantic開発チームによる型安全なエージェント実装

公式MIT
スター
19.3k
フォーク
2.5k
オープンIssue
687
最終コミット
2026年8月15日

概要

Python開発者が既に信頼しているバリデーションモデルをエージェントの出力に適用します。不正な形式のレスポンスは、3層下流ではなく目に見える場所で失敗します。オールインワン型のフレームワークより意図的にスコープを絞っており、それ自体が設計思想です。コードベース全体を型付けしている場合に特によく噛み合います。

Pydantic AIで何ができますか?

  • 出力型の宣言による検証Agentにoutput_type=Sentimentと宣言しておけば、実行結果は常にそのPydanticモデルとして返ります。@agent.toolを付けた関数はシグネチャとdocstringがそのままツールスキーマになり、引数は自分のコードに渡る前に検証されます。
  • モデル指定は文字列1つopenai:gpt-5.6-solのような文字列を差し替えるだけでOpenAI、Anthropic、Google、Bedrock、Groq、Mistral、xAI、Ollamaなどを行き来できます。Pydantic AI Gateway経由にすれば、キー1本にフェイルオーバーとコスト集計をまとめられます。
  • 同じ定義を複数の入口で使うagent.to_cli_sync()でターミナル用のCLIになり、agent.realtime(...).session()ではツール定義を保ったまま音声セッションに載ります。AG-UIやVercel AIのイベントストリーム経由で自作フロントエンドにも接続できます。
  • 長時間実行の中断耐性TemporalDurability()を加えるとモデル呼び出しとツール呼び出しがTemporalのアクティビティになり、再起動や長い待ち時間をまたいで処理が続きます。DBOSやPrefectも同じ差し込み方です。
  • 機能をcapability単位で束ねるCapabilityはツール、instructions、フック、モデル設定を再利用可能な1単位にまとめます。MCP(url)やWebSearch()は本体に同梱されており、defer_loading=Trueを付けたものは会話の途中でモデルが必要とした時点で読み込まれます。

ドキュメント

pydantic/pydantic-ai のREADMEより転載(MIT)。 原文を読む ↗


Pydantic AI is the Python AI SDK: a typed, extensible agent loop with every model a string swap away. The same agent runs everywhere you need it: behind a web frontend, in the terminal, on a voice call, on a durable background queue, or as a plain object you call run() on. Image generation and embeddings come in the same box.

Pydantic AI Harness has everything an agent needs for complex, long-running work, snapped on as capabilities, from memory, sub-agents, and context management to a complete coding agent.

View the complete documentation at ai.pydantic.dev.

What are you building?

From simple typed data extraction to complex, long-running multi-agent collaboration, Pydantic AI and Pydantic AI Harness have got you covered.

Coding agent

A complete coding agent in your terminal: workspace-rooted file access, allowlisted shell, repo orientation, planning, and context management that survives long sessions. Here with web search and a second-opinion advisor snapped on alongside:

uv add pydantic-ai pydantic-ai-harness
from pydantic_ai import Agent
from pydantic_ai.capabilities import WebSearch
from pydantic_ai_harness import Advisor, Coder

agent = Agent(
    'anthropic:claude-fable-5',
    capabilities=[
        Coder(),  # files, shell, repo context, planning, sub-agents, context management
        WebSearch(),  # look up docs and error messages on the web
        Advisor('openai:gpt-5.6-sol'),  # a second opinion from another model when stuck
    ],
)
agent.to_cli_sync()

Coder is a regular combined capability, not a black box: use it whole, or use the blocks it bundles directly; the two are equivalent:

capabilities = [
    FileSystem('.'), Shell(cwd='.'), RepoContext(), Planning(), SubAgents(...),
    ClearToolResults(), WarnNearLimits(), ToolOutputLimits(),
]

Run the file and you’re chatting with the agent in your terminal. To try it before writing any code, run the exported coder_agent with clai (the Pydantic AI CLI), via uvx:

uvx --with pydantic-ai-harness clai -a pydantic_ai_harness.coder:coder_agent -m anthropic:claude-fable-5

Build this → Coder, from the Harness

Data extraction

Give the agent an output type and tools, and every run comes back validated and typed:

uv add pydantic-ai
from typing import Literal

from pydantic import BaseModel, Field

from pydantic_ai import Agent, RunContext


class Sentiment(BaseModel):
    label: Literal['positive', 'negative', 'neutral']
    score: float = Field(ge=-1, le=1)


agent = Agent('openai:gpt-5.6-sol', output_type=Sentiment)


@agent.tool
def recent_reviews(ctx: RunContext[None], product: str) -> list[str]:
    """Fetch recent review snippets for a product."""
    return ['The new release fixed everything I complained about!']


result = agent.run_sync('How are people feeling about the Extract app?')
print(result.output)
#> label='positive' score=0.9

The @agent.tool function receives a RunContext that carries your dependencies in; the rest of its signature and its docstring become the tool schema, arguments are validated before your code runs, and the run is guaranteed to return a Sentiment, so your IDE, type checker, and the LLM all agree on the shape.

Build this → Agents, Function Tools, and Structured Output

Realtime voice

Put the same agent on a live voice session, tools and capabilities included:

uv add "pydantic-ai[openai-realtime]"
import asyncio

from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP

agent = Agent(
    instructions='You are a helpful voice assistant.',
    capabilities=[MCP('https://internal.example.com/mcp')],  # capabilities work in voice too
)

@agent.tool_plain
def order_status(order_id: str) -> str:
    """Look up the status of an order."""
    return f'Order {order_id}: shipped, arriving Thursday.'

async with agent.realtime('openai:gpt-realtime-2.1').session() as session:
    microphone = asyncio.create_task(stream_microphone(session))  # chunks → session.send_audio()
    speaker = asyncio.create_task(play_audio(session.stream_audio()))  # model audio → your speaker
    async for part in session.stream_transcripts():
        print(f'{part.speaker}: {part.transcript}')

The model calls your tools mid-conversation while it keeps talking, and every session is instrumented; voice is just another frontend, on OpenAI Realtime, Gemini Live, Azure, and xAI Grok Voice.

Build this → Realtime Voice

Durable background agent

Attach TemporalDurability and the same agent runs inside a Temporal workflow: every model and tool call becomes a durable activity, so a run working through a background queue survives restarts, failures, and long waits:

uv add "pydantic-ai[temporal]"
from temporalio import workflow

from pydantic_ai import Agent
from pydantic_ai.capabilities import WebFetch, WebSearch
from pydantic_ai.durable_exec.temporal import PydanticAIWorkflow, TemporalDurability

agent = Agent(
    'openai:gpt-5.6-sol',
    instructions='Research the topic and write a structured brief.',
    name='researcher',
    capabilities=[WebSearch(), WebFetch(), TemporalDurability()],
)


@workflow.defn
class ResearchWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [agent]

    @workflow.run
    async def run(self, topic: str) -> str:
        result = await agent.run(f'Write a brief on: {topic}')
        return result.output

DBOS and Prefect attach the same way, first-party and co-maintained, with Restate, Kitaru, and Airflow integrations besides.

Build this → Durable Execution

Image generation

Ask for an image and make it the run’s typed output:

uv add pydantic-ai
from pathlib import Path

from pydantic_ai import Agent, BinaryImage

agent = Agent('openai:gpt-5.6-sol', output_type=BinaryImage)
result = agent.run_sync('Generate a minimalist logo for a coffee shop called Extract.')
Path('logo.png').write_bytes(result.output.data)

Provider-native generation on models that support it (like this one), a subagent fallback you can configure for the rest, and a standalone image API on the way.

Build this → Image Generation

Why Pydantic AI

Built by the Pydantic team: Pydantic Validation is the validation layer of the OpenAI SDK, the Anthropic SDK, the Google ADK, LangChain, and most of the AI ecosystem (and the foundation FastAPI was built on). Pydantic AI brings that same feeling to agents.

Putting it together: a bank support agent

A typed support agent showing several features working together: dependency injection, function tools, structured output, a reusable capability bundling the customer context, and an on-demand capability the model loads only when the conversation calls for it:

from dataclasses import dataclass

from pydantic import BaseModel, Field

from pydantic_ai import Agent, Capability, RunContext

from bank_database import DatabaseConn


@dataclass
class SupportDependencies:  # inject any client: DB pools, HTTP APIs, user info
    customer_id: int
    db: DatabaseConn


class SupportOutput(BaseModel):
    support_advice: str = Field(description='Advice returned to the customer')
    block_card: bool = Field(description="Whether to block the customer's card")
    risk: int = Field(description='Risk level of query', ge=0, le=10)


customer_context = Capability[SupportDependencies](  # a reusable unit of tools + instructions
    id='customer-context',
    description="Who the customer is and what's on their account.",
)


@customer_context.instructions
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {customer_name!r}"


@customer_context.tool  # signature and docstring become the tool schema the LLM sees
async def customer_balance(
    ctx: RunContext[SupportDependencies], include_pending: bool
) -> float:
    """Returns the customer's current account balance."""
    return await ctx.deps.db.customer_balance(
        id=ctx.deps.customer_id,
        include_pending=include_pending,
    )


refunds = Capability[SupportDependencies](  # deferred: loads on demand, like a skill
    id='refunds',
    description='Refund eligibility and refund status.',
    defer_loading=True,
)


@refunds.tool
async def refund_status(ctx: RunContext[SupportDependencies]) -> str:
    """Look up the refund status for the customer's most recent charge."""
    return await ctx.deps.db.refund_status(id=ctx.deps.customer_id)


support_agent = Agent(
    'openai:gpt-5.6-sol',
    deps_type=SupportDependencies,
    output_type=SupportOutput,  # the run returns a validated SupportOutput, typed as such
    instructions=(
        'You are a support agent in our bank, give the '
        'customer support and judge the risk level of their query.'
    ),
    capabilities=[customer_context, refunds],
)


...  # in a real use case: more tools, longer instructions


async def main():
    deps = SupportDependencies(customer_id=123, db=DatabaseConn())
    result = await support_agent.run('What is my balance?', deps=deps)
    print(result.output)
    """
    support_advice='Hello John, your current account balance, including pending transactions, is $123.45.' block_card=False risk=1
    """

    result = await support_agent.run('I just lost my card!', deps=deps)
    print(result.output)
    """
    support_advice="I'm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions." block_card=True risk=8
    """

    result = await support_agent.run(  # the model loads `refunds` on demand, then answers
        'Was I refunded for the duplicate charge on my last statement?', deps=deps
    )
    print(result.output)
    """
    support_advice='Good news, John: the duplicate charge on your last statement was refunded on 2026-05-01.' block_card=False risk=1
    """

For the annotated walkthrough and Logfire tracing, see the same example in the docs.

Next Steps

Part of the Pydantic Stack

Everything you need to ship production-grade AI agents: