pydantic-ai README

The project's README from github.com/pydantic/pydantic-ai. As one markdown file: README.md.

Pydantic AI

How Python does AI

CI Coverage PyPI versions license Join Slack

Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.


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 pydantic.dev/docs/ai.

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 returned type.

Build this → Agents, Function Tools, and Structured Output

Durable workflow

Attach TemporalDurability and the same agent runs inside a Temporal workflow under durable execution: 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

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(session.send_audio(microphone_chunks()))  # your microphone → the model
    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

Image generation

Generate an image with a dedicated image model, no agent run required:

uv add pydantic-ai
from pathlib import Path

from pydantic_ai import ImageGenerator

generator = ImageGenerator('openai:gpt-image-2')
result = generator.generate_sync('A minimalist logo for a coffee shop called Extract.')
Path('logo.png').write_bytes(result.image.data)

That standalone image API is for when your application decides; when an agent run decides, there is provider-native generation with output_type=BinaryImage for a typed image output, and the ImageGeneration capability with its fallbacks for models that generate no images of their own.

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:

pydantic-ai (pydantic/pydantic-ai)

Searching Database…

这是什么?Pydantic AI 是用于构建智能体(agent)的 Python AI SDK:一个带类型、可扩展的智能体循环,换一个字符串就能换模型。同一个智能体可以跑在网页前端、终端、语音通话里,也可以作为一个普通对象直接调用 run()。
它是框架还是库——装完之后我得到的是什么?它是一个可以导入并编写代码的 Python SDK:智能体、实时语音、图像生成和向量嵌入(embeddings)都在同一个包里。可选的 Harness 提供即插即用的能力,智能体甚至可以用 YAML/JSON 规范来定义,而不写代码。
怎么让它跑起来——需要先准备 API 密钥吗?不需要——内置了一个 'test' 模型,没有 API 密钥也能上手。安装这个包,用几行 Python 定义一个智能体即可;如果想在不写任何代码的情况下先试用智能体,可以通过 uvx 运行自带的 Coder 智能体,也就是 Pydantic AI 的命令行工具 clai。
可以接入哪些模型和服务商?几乎所有服务商——OpenAI、Anthropic、Google、Bedrock、Azure AI Foundry、Groq、Mistral、xAI、Ollama 等几十家——改一个字符串就能切换。Pydantic AI Gateway 提供一把通吃的密钥,还内置了故障转移和成本监控。
有没有实验性或还不够成熟的部分需要留意?只有一个部分被标注为实验性:ACP,它把你的智能体作为编辑器智能体对外提供服务。其余部分——CLI、网页聊天、实时语音前端——都是按正式标准功能推出的。
'端到端类型安全' 在实际使用中到底是什么意思?结构化输出、带类型的依赖注入和带类型的工具:你的 IDE、类型检查器和 LLM 对智能体返回什么达成一致,于是整类错误从运行时提前到了编写时。工具参数会在你的代码运行前先做校验,Pydantic Graph 把同样的类型体系延伸到了基于图的工作流。
Harness 是什么,'能力(capability)' 又指什么?Harness 是面向复杂、长时间运行任务的扩展层:记忆、子智能体、上下文管理,甚至一个完整的编码智能体,都以能力(capability)的形式即插即用。一个 capability 把工具、指令、钩子和模型设置打包成一个可复用的单元——核心自带 MCP 和网页搜索,而 Coder、Researcher 这类现成智能体就是由这些能力组合而成的。
同一个智能体也能做实时语音和图像生成吗?可以。语音只是另一种前端:模型在持续说话的同时调用你的工具,支持 OpenAI Realtime、Gemini Live、Azure 和 xAI Grok Voice。图像方面有独立的 API、通过类型化输出调用服务商原生的生成能力,还有一个 ImageGeneration capability,对自身不能生成图像的模型提供回退方案。
怎么调试、监控和测试我的智能体在做什么?可观测性原生支持 OpenTelemetry,可对接任何 OTel 后端;一行代码即可启用 Pydantic Logfire,实现实时调试、追踪和成本统计。Pydantic Evals 用 pytest 测试代码的方式测试智能体行为,而 Temporal 这类持久化执行引擎能让长时间任务在重启和故障后继续运行。
Creator
pydantic
Size
313.1 MB
Stars
20.7K
Age
2y
Commits · 90 Days