Pydantic AI en Python 2026 :
agents type-safe et structured output

# Package complet (recommandé pour débuter)
pip install pydantic-ai

# Slim + provider spécifique (prod, image Docker plus légère)
pip install pydantic-ai-slim
pip install 'pydantic-ai-slim[openai]'    # + openai SDK
pip install 'pydantic-ai-slim[anthropic]' # + anthropic SDK
pip install 'pydantic-ai-slim[google]'    # + google-genai SDK
pip install 'pydantic-ai-slim[groq,mistral]' # multi-providers

# Python requis : >= 3.10 (union types X | Y natifs)

from pydantic import BaseModel
from pydantic_ai import Agent
from typing import Literal

# Structured output défini comme modèle Pydantic
class SentimentAnalysis(BaseModel):
    sentiment: Literal['positive', 'negative', 'neutral']
    confidence: float           # 0.0 – 1.0
    key_points: list[str]

# Agent avec output structuré, sans dépendances (NoneType par défaut)
agent = Agent(
    'anthropic:claude-opus-4-6',
    output_type=SentimentAnalysis,
    system_prompt="Tu es un analyste de texte expert. Analyse le sentiment du texte fourni.",
    retries=2,
)

# Exécution synchrone
result = agent.run_sync("Ce framework est incroyablement bien conçu !")
print(result.output)
# SentimentAnalysis(sentiment='positive', confidence=0.97, key_points=['framework', 'bien conçu'])

# Métriques d'usage
usage = result.usage()
print(f"Tokens: {usage.input_tokens} → {usage.output_tokens}")

result = await agent.run("Analyse ce texte")

result.output          # SentimentAnalysis — résultat final validé par Pydantic
result.usage()         # Usage(requests=1, input_tokens=245, output_tokens=42, tool_calls=1)
result.all_messages()  # list[ModelMessage] — historique complet incluant tool calls
result.new_messages()  # list[ModelMessage] — messages de CE run seulement (pour multi-turn)
result.run_id          # UUID unique par appel agent.run()
result.conversation_id # UUID partagé entre runs liés (nouveau en v1.89.0)

from dataclasses import dataclass
import asyncpg
import httpx
from pydantic_ai import Agent, RunContext

# 1. Déclarer les dépendances comme dataclass
@dataclass
class AppDeps:
    db: asyncpg.Pool           # Pool de connexions PostgreSQL
    http: httpx.AsyncClient    # Client HTTP async réutilisable
    user_id: str               # Contexte utilisateur courant

# 2. Agent paramétré sur AppDeps
agent = Agent('openai:gpt-4o', deps_type=AppDeps)

# 3. System prompt dynamique avec accès aux deps
@agent.system_prompt
async def get_context(ctx: RunContext[AppDeps]) -> str:
    username = await ctx.deps.db.fetchval(
        "SELECT name FROM users WHERE id = $1", ctx.deps.user_id
    )
    return f"Tu assistes {username}. Date : {datetime.now().strftime('%Y-%m-%d')}"

# 4. Tool avec accès aux deux deps
@agent.tool
async def fetch_article(ctx: RunContext[AppDeps], url: str) -> str:
    """Récupère le contenu d'un article depuis son URL.

    Args:
        url: L'URL complète de l'article à analyser.
    """
    resp = await ctx.deps.http.get(url)
    return resp.text[:5000]

# 5. Exécution avec injection des deps
async def main():
    async with asyncpg.create_pool("postgresql://...") as pool:
        async with httpx.AsyncClient() as client:
            deps = AppDeps(db=pool, http=client, user_id="usr_42")
            result = await agent.run("Analyse l'article https://...", deps=deps)

# Tool AVEC contexte (accès aux deps)
@agent.tool
async def search_documents(
    ctx: RunContext[AppDeps],
    query: str,
    limit: int = 5,
    min_score: float = 0.7
) -> list[dict]:
    """Recherche des documents dans la base de connaissances.

    Args:
        query: La requête de recherche en langage naturel.
        limit: Nombre maximum de résultats (1-20).
        min_score: Score de similarité minimum (0.0-1.0).
    """
    return await ctx.deps.db.fetch(
        "SELECT * FROM docs WHERE similarity(embedding, $1) > $2 LIMIT $3",
        embed(query), min_score, limit
    )

# Tool SANS contexte (@tool_plain)
@agent.tool_plain
def format_currency(amount: float, currency: str = "EUR") -> str:
    """Formate un montant monétaire.

    Args:
        amount: Le montant à formater.
        currency: Code ISO 4217 de la devise (EUR, USD, GBP…).
    """
    return f"{amount:,.2f} {currency}"

# Tool déclaré via constructeur (réutilisable entre agents)
from pydantic_ai import Tool

agent = Agent('openai:gpt-4o', tools=[Tool(search_documents), Tool(format_currency)])

@agent.tool(requires_approval=True)
async def delete_record(ctx: RunContext[AppDeps], record_id: str) -> str:
    """Supprime définitivement un enregistrement. Nécessite approbation humaine."""
    await ctx.deps.db.execute("DELETE FROM records WHERE id = $1", record_id)
    return "Supprimé"

# Le run se met en pause et retourne DeferredToolRequests
result = await agent.run("Supprime l'enregistrement rec_123", deps=deps)

if isinstance(result, DeferredToolRequests):
    for approval_req in result.approvals:
        print(f"Approuver : {approval_req.tool_name}({approval_req.args})")
        user_ok = input("Confirmer ? [y/n] ") == 'y'
        # Construire DeferredToolResults avec ToolApproved / ToolDenied

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class ExtractedEntity(BaseModel):
    name: str
    type: Literal['person', 'org', 'location', 'date']
    confidence: float = Field(ge=0.0, le=1.0)
    context: str

class ExtractionResult(BaseModel):
    entities: list[ExtractedEntity]
    language: str
    ambiguities: list[str]

# Mode Tool Output — comportement par défaut
ner_agent = Agent(
    'anthropic:claude-opus-4-6',
    output_type=ExtractionResult,
    output_retries=3,  # Réessaie jusqu'à 3 fois si la validation Pydantic échoue
)

# Validation post-parsing : peut lever ModelRetry
@ner_agent.output_validator
async def validate_confidence(ctx, result: ExtractionResult) -> ExtractionResult:
    low = [e.name for e in result.entities if e.confidence < 0.5]
    if low:
        from pydantic_ai import ModelRetry
        raise ModelRetry(f"Confidence trop faible pour : {low}. Reconsidère ces entités.")
    return result

from pydantic_ai.output import NativeOutput

agent = Agent(
    'openai:gpt-4o',
    output_type=NativeOutput(ExtractionResult),
    # Certains providers/modèles exposent un mode structuré natif
    # — vérifier la compatibilité exacte côté provider avant d'utiliser ce mode
)

from pydantic_ai import Agent
from pydantic_ai.models.ollama import OllamaModel
from pydantic_ai.providers.ollama import OllamaProvider
from pydantic_ai.output import PromptedOutput

# Via OllamaModel + OllamaProvider (approche recommandée)
model = OllamaModel(
    'qwen2.5:14b',
    provider=OllamaProvider(base_url='http://localhost:11434/v1'),
)

agent = Agent(
    model,
    output_type=PromptedOutput(ExtractionResult),
)

class Success(BaseModel):
    data: dict
    execution_time_ms: int

class Failure(BaseModel):
    error_code: str
    message: str
    recoverable: bool

# Chaque type devient un tool séparé pour le LLM
# → réduit la complexité du schéma JSON unique
agent = Agent('openai:gpt-4o', output_type=Success | Failure)

result = await agent.run("Execute this task...")
match result.output:
    case Success(data=d): print(f"OK: {d}")
    case Failure(message=m): print(f"Erreur: {m}")

# Hiérarchie des types
# ModelMessage = ModelRequest | ModelResponse
# ModelRequest → parts: SystemPromptPart | UserPromptPart | ToolReturnPart | RetryPromptPart
# ModelResponse → parts: TextPart | ToolCallPart | ThinkingPart (extended thinking)

agent = Agent('anthropic:claude-opus-4-6')

# Tour 1
r1 = await agent.run("Qu'est-ce que le PagedAttention ?")
print(r1.output)

# Tour 2 — passer new_messages() pour continuer la conversation
r2 = await agent.run(
    "Comment ça se compare à Flash Attention ?",
    message_history=r1.new_messages(),  # Messages du tour 1
)

# Persistance : sérialisation/désérialisation JSON
from pydantic_ai import ModelMessagesTypeAdapter
from pydantic_core import to_json

messages = r2.all_messages()
json_bytes = to_json(messages)  # → stocker en Redis, PostgreSQL, etc.
restored = ModelMessagesTypeAdapter.validate_json(json_bytes)  # → restaurer

# Limiter l'historique envoyé au LLM à 10 derniers messages
def trim_history(messages, ctx):
    return messages[-10:]

agent = Agent(
    'openai:gpt-4o',
    history_processors=[trim_history],
    # Ou : résumer les anciens messages via un sous-agent
)

# Agents spécialisés
code_agent = Agent('openai:gpt-4o', system_prompt="Tu es un expert Python senior.")
review_agent = Agent('anthropic:claude-opus-4-6', system_prompt="Tu es un code reviewer.")

# Agent orchestrateur
orchestrator = Agent('openai:gpt-4o', deps_type=AppDeps)

@orchestrator.tool
async def generate_code(ctx: RunContext[AppDeps], spec: str) -> str:
    """Génère du code Python depuis une spécification."""
    result = await code_agent.run(
        spec,
        deps=ctx.deps,
        usage=ctx.usage,  # CRITIQUE : agrège les tokens dans le run parent
    )
    return result.output

@orchestrator.tool
async def review_code(ctx: RunContext[AppDeps], code: str) -> str:
    """Effectue une revue de code et retourne les suggestions."""
    result = await review_agent.run(
        f"Revue ce code Python :\n{code}",
        usage=ctx.usage,
    )
    return result.output

# Différents agents peuvent utiliser différents providers LLM
result = await orchestrator.run(
    "Crée une fonction de tri avec complexité O(n log n) et fais-la réviser",
    deps=deps
)

# Flux interne :
# UserPromptNode → ModelRequestNode → CallToolsNode → (boucle ou End)

# Itération manuelle nœud par nœud via agent.iter() (API publique)
async with agent.iter("Quelle est la capitale de la France ?") as run:
    async for node in run:
        print(f"Nœud actif : {type(node).__name__}")
    print(f"Résultat : {run.result.output}")

from dataclasses import dataclass
from pydantic_graph import BaseNode, Graph, End, GraphRunContext

@dataclass
class State:
    attempts: int = 0
    result: str = ""

# Les arêtes sortantes sont déclarées dans les annotations de retour
@dataclass
class GenerateNode(BaseNode[State, None, str]):
    prompt: str

    async def run(self, ctx: GraphRunContext[State, None]) -> 'ValidateNode | End[str]':
        ctx.state.attempts += 1
        response = await llm_call(self.prompt)
        ctx.state.result = response
        if ctx.state.attempts > 3:
            return End(response)  # Forcer la terminaison
        return ValidateNode(response)

@dataclass
class ValidateNode(BaseNode[State, None, str]):
    content: str

    async def run(self, ctx: GraphRunContext) -> 'GenerateNode | End[str]':
        if is_valid(self.content):
            return End(self.content)
        return GenerateNode(f"Améliore cette réponse : {self.content}")

graph = Graph(nodes=[GenerateNode, ValidateNode])
result, final_state = await graph.run(GenerateNode("Explique la rétropropagation"), state=State())

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio, MCPServerStreamableHTTP

# Transport 1 : subprocess via stdin/stdout
python_server = MCPServerStdio(
    'uv',
    args=['run', 'mcp-run-python', 'stdio'],
    timeout=10,
    tool_prefix='python',  # Évite les conflits de noms de tools
)

# Transport 2 : HTTP Streamable (production)
db_server = MCPServerStreamableHTTP(
    'http://mcp-db-service:8080/mcp',
    read_timeout=300,  # 5 min — défaut pour les tools longs
    cache_tools=True,   # Cache la liste des tools entre appels
)

# Passer les serveurs MCP comme toolsets
agent = Agent('openai:gpt-4o', toolsets=[python_server, db_server])

async def main():
    async with python_server, db_server:
        result = await agent.run(
            "Calcule la moyenne des ventes du mois dernier depuis la DB et affiche un graphique"
        )

# Chargement depuis fichier de config (style Claude Desktop)
from pydantic_ai.mcp import load_mcp_servers
servers = load_mcp_servers('mcp_config.json')  # Supporte ${VAR:-default}

from pydantic_ai.models.test import TestModel
from pydantic_ai import capture_run_messages
import pytest

# Bloquer les appels réels en test — à placer dans conftest.py, pas dans les tests
# import pydantic_ai.models as _m; _m.ALLOW_MODEL_REQUESTS = False
# Préférer agent.override(model=TestModel()) qui est scoped au contexte

# TestModel : appelle TOUS les tools et génère des données valides
# C'est du Python procédural pur — aucun ML impliqué
@pytest.mark.asyncio
async def test_agent_calls_tools():
    with my_agent.override(model=TestModel()):
        with capture_run_messages() as messages:
            result = await application_code("Analyse les données")

        # Vérifier que les tools ont été appelés
        tool_calls = [m for m in messages if hasattr(m, 'parts')]
        assert any(p.tool_name == 'fetch_data' for m in tool_calls for p in m.parts)

# FunctionModel : logique de réponse entièrement contrôlée
from pydantic_ai.models.function import FunctionModel, AgentInfo, ModelFuncContext

def mock_model(messages, info: AgentInfo):
    # Décider des tool calls et réponses en Python pur
    if "capital" in str(messages[-1]).lower():
        return "Paris est la capitale de la France."
    return "Je ne sais pas."

async def test_specific_behavior():
    with my_agent.override(model=FunctionModel(mock_model)):
        result = await my_agent.run("Quelle est la capitale de la France ?")
        assert "Paris" in result.output

import logfire

# Option 1 : Logfire SaaS (une ligne)
logfire.configure()
logfire.instrument_pydantic_ai()

# Option 2 : OpenTelemetry générique (Jaeger, Tempo, Datadog...)
from pydantic_ai import Agent
Agent.instrument_all()  # Instrumente tous les agents du process

# Option 3 : par agent
agent = Agent('openai:gpt-4o', instrument=True)

# Ce qui est tracé automatiquement :
# - Span par run d'agent (avec run_id, conversation_id)
# - Span par requête modèle (latence, model_name)
# - Span par tool call (nom, arguments, résultat, durée)
# - Token usage par provider
# - Cost tracking agrégé cross-agents

from pydantic_evals import Dataset, Case
from pydantic_evals.evaluators import EqualsExpected, IsInstance, LLMJudge

# Définir les cas de test
dataset = Dataset(
    cases=[
        Case(
            name="sentiment_positif",
            inputs={"text": "Ce produit est fantastique !"},
            expected_output=SentimentAnalysis(sentiment="positive", confidence=0.9, key_points=[]),
        ),
        Case(
            name="sentiment_negatif",
            inputs={"text": "Service client déplorable."},
            expected_output=SentimentAnalysis(sentiment="negative", confidence=0.85, key_points=[]),
        ),
    ],
    evaluators=[
        EqualsExpected(fields=["sentiment"]),  # Vérifie le champ sentiment uniquement
        IsInstance(SentimentAnalysis),           # Vérifie le type de retour
        LLMJudge(                                   # Évaluateur LLM pour qualité subjective
            prompt="Les key_points sont-ils pertinents ? Score 0-1.",
            model='openai:gpt-4o-mini',
        ),
    ]
)

# Fonction à évaluer
async def run_sentiment(inputs: dict) -> SentimentAnalysis:
    result = await ner_agent.run(inputs["text"])
    return result.output

# Lancer l'évaluation
report = await dataset.evaluate(run_sentiment)
print(report.summary())
# → accuracy: 2/2, avg_llm_score: 0.87, duration_ms: 1240

from pydantic_ai.capabilities import Thinking, WebSearch, WebFetch

# Built-in capabilities — stables
agent = Agent(
    'anthropic:claude-opus-4-6',
    capabilities=[
        Thinking(effort='high'),  # Raisonnement étendu cross-provider
        WebSearch(),              # Recherche web native ou DuckDuckGo fallback
        WebFetch(),               # Lecture de pages web
    ]
)

# Les capabilities custom (AbstractCapability) permettent d'encapsuler
# tools + lifecycle hooks + instructions. L'API des hooks custom évolue
# entre les mineures — consulter la doc de votre version exacte.