Pydantic AI Expert (Type-Safe Python Agent Engineering)

English | Bahasa Indonesia

---

<a name="english"></a>

English

Orchestration & Integration

• Python Language & Runtime: python-programming-expert, structured-output-contracts
• Agent Frameworks & Swarms: multi-agent-orchestration, ai-llm-integration-expert, frontier-ai-models-expert
• MCP & Tool Protocols: mcp-server-architect, parallel-tool-orchestrator
• Persistence & Databases: database-orm-expert, vector-db-rag-expert
• Evaluation & Testing: autonomous-tdd-debugger, property-mutation-testing-expert, anti-slop
• Lifecycle & Governance: zero-to-prod-orchestrator, session-memory-manager

Description

Production-grade guide for building enterprise AI agents using Pydantic AI (from the creators of Pydantic). Unlike untyped or fragile agent libraries, Pydantic AI provides strict static typing (Agent[DepsType, ReturnType]), first-class dependency injection (RunContext[Deps]), automatic tool parameter validation via Pydantic v2 schemas, model-agnostic provider switching (Anthropic, OpenAI, Gemini, Groq, Ollama), dynamic system prompts, and deterministic structured outputs.

Swarm Synergy: Within the Backend & APIs Swarm, this skill serves as the Lead Python Agent Engineer. It executes critical business operations, runs data pipelines, and integrates with SQL databases with compile-time type safety in Phase 4.

Trigger Conditions

• Engineering Python-based autonomous agents requiring strict type safety and schema validation.
• Implementing dependency injection (passing database connections, API clients, or user sessions) into agent tools.
• Extracting guaranteed structured outputs from frontier reasoning models without JSON parsing errors.
• Designing multi-agent delegation or agent handoffs in Python microservices.
• Running unit tests and deterministic evals on agent logic using mock models (TestModel).

Pydantic AI Architecture

┌─────────────────────────────────────────────────────────────┐
│                    AGENT DEFINITION                         │
│  Agent[DatabaseDeps, AuditReport](                          │
│    model='anthropic:claude-3-7-sonnet-latest',              │
│    result_type=AuditReport,                                 │
│  )                                                          │
└──────────────────────────────┬──────────────────────────────┘
                               │
       ┌───────────────────────┴───────────────────────┐
       ▼                                               ▼
┌──────────────────────────────┐       ┌──────────────────────────────┐
│     DEPENDENCY INJECTION     │       │     TOOL EXECUTION LOOP      │
│  RunContext[DatabaseDeps]    │       │  @agent.tool                 │
│  • Async DB connection pool  │       │  • Auto Pydantic validation  │
│  • Tenant & session auth     │       │  • Structured error return   │
└──────────────────────────────┘       └──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│             DETERMINISTIC STRUCTURED RESULT                 │
│  result.data -> Pydantic BaseModel instance (Validated)     │
└─────────────────────────────────────────────────────────────┘

Core Implementation Guidelines

1. Type-Safe Agent with Dependency Injection

Pass database pools or HTTP clients directly into tools without global state:

from dataclasses import dataclass
import asyncpg
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

@dataclass
class DatabaseDeps:
    pool: asyncpg.Pool
    tenant_id: str

class UserAnomalyReport(BaseModel):
    user_id: str
    risk_score: float = Field(ge=0.0, le=1.0)
    anomalous_events: list[str]
    action_recommended: str

# Typed agent: Agent[DependencyType, ResultType]
security_agent = Agent[DatabaseDeps, UserAnomalyReport](
    model='google-gla:gemini-2.5-flash',
    result_type=UserAnomalyReport,
    system_prompt=(
        "You are an automated SecOps auditor. Analyze the user access logs provided "
        "and generate a strictly structured anomaly assessment."
    ),
)

@security_agent.system_prompt
async def add_tenant_context(ctx: RunContext[DatabaseDeps]) -> str:
    return f"Active Tenant ID: {ctx.deps.tenant_id}. Only audit records matching this tenant."

@security_agent.tool
async def query_audit_logs(ctx: RunContext[DatabaseDeps], user_id: str, limit: int = 50) -> list[dict]:
    """Retrieve raw authentication events for a given user from the audit store."""
    async with ctx.deps.pool.acquire() as conn:
        rows = await conn.fetch(
            "SELECT event_time, ip_address, action, status FROM audit_logs "
            "WHERE tenant_id = $1 AND user_id = $2 ORDER BY event_time DESC LIMIT $3",
            ctx.deps.tenant_id, user_id, limit
        )
        return [dict(r) for r in rows]

# Execution:
async def run_audit(pool: asyncpg.Pool, tenant_id: str, target_user: str) -> UserAnomalyReport:
    deps = DatabaseDeps(pool=pool, tenant_id=tenant_id)
    result = await security_agent.run(f"Audit user activity for {target_user}", deps=deps)
    return result.data  # Guaranteed instance of UserAnomalyReport

2. Model-Agnostic Switching & Local SLM Fallback

Pydantic AI allows switching between cloud APIs and local Ollama models effortlessly:

import os
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.ollama import OllamaModel

def get_agent_model():
    if os.getenv("ENV") == "production":
        return AnthropicModel('claude-3-7-sonnet-20250219')
    else:
        # Zero cost local development with Ollama
        return OllamaModel(model_name='qwen2.5-coder:7b')

3. Deterministic Testing with TestModel

Write unit tests that verify tool invocations and structured output parsing without calling live LLMs:

import pytest
from pydantic_ai.models.test import TestModel

@pytest.mark.asyncio
async def test_security_agent_tool_dispatch():
    # TestModel can simulate deterministic responses or tool calls
    test_model = TestModel(call_tools=['query_audit_logs'])
    
    with security_agent.override(model=test_model):
        deps = DatabaseDeps(pool=mock_pool, tenant_id="tenant_123")
        result = await security_agent.run("Audit user 456", deps=deps)
        assert result.data is not None

4. Streaming Structured Output Validation (run_stream)

Stream typed Pydantic models with real-time field-level validation as tokens arrive:

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class IncidentTriage(BaseModel):
    severity: str = Field(description="Low, Medium, High, or Critical")
    affected_components: list[str] = Field(default_factory=list)
    immediate_actions: list[str] = Field(default_factory=list)
    root_cause_hypothesis: str

triage_agent = Agent[None, IncidentTriage](
    model='google-gla:gemini-2.5-flash',
    result_type=IncidentTriage,
    system_prompt="You are a real-time site reliability incident responder."
)

async def stream_incident_triage(incident_log: str):
    async with triage_agent.run_stream(f"Triage incident: {incident_log}") as response:
        async for partial_triage in response.stream_structured():
            print(f"[Streaming Partial Validation] Severity: {partial_triage.severity}, Actions: {len(partial_triage.immediate_actions)}")
        
        final_result = await response.get_data()
        return final_result

5. Multi-Agent Delegation & Handoffs via RunContext

Sub-agents invoked as specialized tools with context isolation:

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class ResearchContext:
    api_key: str
    deep_mode: bool

class FactReport(BaseModel):
    statement: str
    verified: bool
    source_citation: str

class SynthesisReport(BaseModel):
    executive_summary: str
    verified_facts: list[FactReport]

# Specialist Sub-Agent
verifier_agent = Agent[ResearchContext, FactReport](
    model='openai:gpt-4o-mini',
    result_type=FactReport,
    system_prompt="Fact-check individual technical claims against primary specifications."
)

# Root Orchestrator Agent
director_agent = Agent[ResearchContext, SynthesisReport](
    model='anthropic:claude-3-7-sonnet-20250219',
    result_type=SynthesisReport,
    system_prompt="Analyze claims and delegate individual verification to the verifier tool."
)

@director_agent.tool
async def verify_claim(ctx: RunContext[ResearchContext], claim: str) -> FactReport:
    """Delegates verification of a specific claim to the specialized verifier agent."""
    sub_result = await verifier_agent.run(claim, deps=ctx.deps)
    return sub_result.data

async def conduct_synthesis(claim_text: str, context: ResearchContext) -> SynthesisReport:
    result = await director_agent.run(claim_text, deps=context)
    return result.data

6. Observability & OpenTelemetry Instrumentation with Pydantic Logfire

Instrument agents with zero boilerplate for token tracking, tool latency, and structured traces:

import logfire
from pydantic_ai import Agent

# 1. Initialize Logfire with automatic OpenTelemetry export
logfire.configure(service_name="secops-pydantic-agent", send_to_logfire=True)
logfire.instrument_pydantic_ai()

# 2. Agent operations are automatically traced with token counts & tool timings
monitor_agent = Agent(
    model='anthropic:claude-3-7-sonnet-20250219',
    system_prompt="Autonomous database optimization auditor."
)

@monitor_agent.tool_plain
def calculate_index_efficiency(table_name: str, scan_count: int) -> float:
    with logfire.span("evaluate_index_metrics", table=table_name):
        return round(1.0 - (1.0 / (scan_count + 1)), 4)

---

<a name="bahasa-indonesia"></a>

Bahasa Indonesia

Integrasi Orkestrasi

• Bahasa Python & Runtime: python-programming-expert, structured-output-contracts
• Framework Agen & Swarm: multi-agent-orchestration, ai-llm-integration-expert, frontier-ai-models-expert
• Protokol MCP & Tools: mcp-server-architect, parallel-tool-orchestrator
• Persistensi & Database: database-orm-expert, vector-db-rag-expert
• Evaluasi & Pengujian: autonomous-tdd-debugger, property-mutation-testing-expert, anti-slop
• Siklus Hidup & Tata Kelola: zero-to-prod-orchestrator, session-memory-manager

Deskripsi

Panduan produksi untuk membangun agen AI kelas perusahaan menggunakan Pydantic AI (dari pembuat library Pydantic). Tidak seperti library agen yang tidak bertipe statis atau mudah mengalami galat runtime, Pydantic AI menyediakan sistem tipe statis yang ketat (Agent[DepsType, ReturnType]), dependency injection bawaan kelas satu (RunContext[Deps]), validasi parameter tool otomatis melalui skema Pydantic v2, fleksibilitas pergantian model (model-agnostic), dan hasil output terstruktur yang terjamin validitasnya.

Sinergi Swarm: Di dalam Backend & APIs Swarm, skill ini bertindak sebagai Insinyur Agen AI Python Utama. Bertanggung jawab atas eksekusi logika bisnis kritis, orkestrasi pipeline data, dan query database SQL dengan keamanan tipe kompilasi pada Fase 4.

Kondisi Pemicu

• Membangun agen otonom Python yang menuntut keamanan tipe statis dan validasi skema data ketat.
• Menerapkan injeksi dependensi (koneksi database, klien HTTP, session user) ke dalam tool agen tanpa variabel global.
• Mengekstrak data terstruktur yang dijamin valid dari model penalaran tanpa kegagalan parsing JSON.
• Menjalankan unit test dan evaluasi deterministik pada agen menggunakan model tiruan (TestModel).

---

Implementasi Pola Kunci (Python 3.13+)

1. Agen Type-Safe dengan Injeksi Dependensi Database

from dataclasses import dataclass
import asyncpg
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

@dataclass
class DependensiDatabase:
    pool: asyncpg.Pool
    tenant_id: str

class LaporanAnomali(BaseModel):
    user_id: str
    skor_risiko: float = Field(ge=0.0, le=1.0)
    peristiwa_anomali: list[str]
    tindakan_disarankan: str

# Agen dengan tipe statis ketat: Agent[TipeDependensi, TipeHasil]
agen_keamanan = Agent[DependensiDatabase, LaporanAnomali](
    model='google-gla:gemini-2.5-flash',
    result_type=LaporanAnomali,
    system_prompt=(
        "Anda adalah auditor keamanan SecOps otomatis. Analisis log aktivitas "
        "dan hasilkan evaluasi anomali terstruktur."
    ),
)

@agen_keamanan.system_prompt
async def tambahkan_konteks_tenant(ctx: RunContext[DependensiDatabase]) -> str:
    return f"Tenant ID Aktif: {ctx.deps.tenant_id}. Hanya audit catatan milik tenant ini."

@agen_keamanan.tool
async def query_log_audit(ctx: RunContext[DependensiDatabase], user_id: str, batas: int = 50) -> list[dict]:
    """Mengambil riwayat autentikasi pengguna dari database audit."""
    async with ctx.deps.pool.acquire() as conn:
        baris = await conn.fetch(
            "SELECT event_time, ip_address, action, status FROM audit_logs "
            "WHERE tenant_id = $1 AND user_id = $2 ORDER BY event_time DESC LIMIT $3",
            ctx.deps.tenant_id, user_id, batas
        )
        return [dict(b) for b in baris]

# Eksekusi agen secara aman:
async def jalankan_audit(pool: asyncpg.Pool, tenant_id: str, target_user: str) -> LaporanAnomali:
    deps = DependensiDatabase(pool=pool, tenant_id=tenant_id)
    hasil = await agen_keamanan.run(f"Audit aktivitas user {target_user}", deps=deps)
    return hasil.data  # Dijamin merupakan instance tervalidasi dari LaporanAnomali

2. Pergantian Model Dinamis & Fallback SLM Lokal (Ollama)

Pydantic AI memungkinkan pergantian tanpa friksi antara model cloud frontier dan model open-weights lokal:

import os
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.ollama import OllamaModel

def ambil_model_agen():
    if os.getenv("ENV") == "production":
        return AnthropicModel('claude-3-7-sonnet-20250219')
    else:
        # Pengembangan lokal zero-cost menggunakan Ollama
        return OllamaModel(model_name='qwen2.5-coder:7b')

3. Pengujian Unit Deterministik Menggunakan TestModel

Tulis pengujian unit yang memvalidasi pemanggilan tool dan parsing output terstruktur tanpa biaya API live:

import pytest
from pydantic_ai.models.test import TestModel

@pytest.mark.asyncio
async def test_eksekusi_tool_agen():
    model_tes = TestModel(call_tools=['query_log_audit'])
    with agen_keamanan.override(model=model_tes):
        deps = DependensiDatabase(pool=mock_pool, tenant_id="tenant_123")
        hasil = await agen_keamanan.run("Audit user 456", deps=deps)
        assert hasil.data is not None

4. Validasi Output Terstruktur Real-Time (run_stream())

Stream model Pydantic dengan validasi level-properti saat token diterima secara bertahap:

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class TriaseInsiden(BaseModel):
    tingkat_keparahan: str = Field(description="Rendah, Sedang, Tinggi, atau Kritis")
    komponen_terdampak: list[str] = Field(default_factory=list)
    tindakan_segera: list[str] = Field(default_factory=list)
    hipotesis_akar_masalah: str

agen_triase = Agent[None, TriaseInsiden](
    model='google-gla:gemini-2.5-flash',
    result_type=TriaseInsiden,
    system_prompt="Anda adalah responder insiden keandalan infrastruktur real-time."
)

async def stream_triase_insiden(log_insiden: str):
    async with agen_triase.run_stream(f"Triase insiden: {log_insiden}") as respons:
        async for triase_parsial in respons.stream_structured():
            print(f"[Streaming Validasi Parsial] Keparahan: {triase_parsial.tingkat_keparahan}, Tindakan: {len(triase_parsial.tindakan_segera)}")
        
        hasil_akhir = await respons.get_data()
        return hasil_akhir

5. Delegasi & Handoff Multi-Agen Berbasis RunContext

Panggil sub-agen sebagai tool terspesialisasi dengan isolasi konteks dependensi:

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class KonteksRiset:
    kunci_api: str
    mode_mendalam: bool

class LaporanFakta(BaseModel):
    pernyataan: str
    terverifikasi: bool
    sitasi_sumber: str

class LaporanSintesis(BaseModel):
    ringkasan_eksekutif: str
    fakta_terverifikasi: list[LaporanFakta]

# Sub-Agen Spesialis Verifikasi
agen_verifikator = Agent[KonteksRiset, LaporanFakta](
    model='openai:gpt-4o-mini',
    result_type=LaporanFakta,
    system_prompt="Verifikasi kebenaran klaim teknis berdasarkan dokumen spesifikasi primer."
)

# Agen Utama / Direktur Sintesis
agen_direktur = Agent[KonteksRiset, LaporanSintesis](
    model='anthropic:claude-3-7-sonnet-20250219',
    result_type=LaporanSintesis,
    system_prompt="Analisis laporan dan delegasikan verifikasi klaim ke tool verifikator."
)

@agen_direktur.tool
async def verifikasi_klaim(ctx: RunContext[KonteksRiset], klaim: str) -> LaporanFakta:
    """Mendelegasikan verifikasi klaim tertentu ke sub-agen spesialis."""
    sub_hasil = await agen_verifikator.run(klaim, deps=ctx.deps)
    return sub_hasil.data

async def jalankan_sintesis(teks_klaim: str, konteks: KonteksRiset) -> LaporanSintesis:
    hasil = await agen_direktur.run(teks_klaim, deps=konteks)
    return hasil.data

6. Observabilitas & Tracing OpenTelemetry dengan Pydantic Logfire

Instrumentasi performa agen, pelacakan konsumsi token, dan latensi tool secara otomatis:

import logfire
from pydantic_ai import Agent

# 1. Konfigurasi Logfire dengan ekspor OpenTelemetry otomatis
logfire.configure(service_name="secops-pydantic-agent", send_to_logfire=True)
logfire.instrument_pydantic_ai()

# 2. Operasi agen terlacak secara otomatis dalam dasbor telemetri
agen_pemantau = Agent(
    model='anthropic:claude-3-7-sonnet-20250219',
    system_prompt="Auditor pengoptimalan database otonom."
)

@agen_pemantau.tool_plain
def hitung_efisiensi_indeks(nama_tabel: str, jumlah_scan: int) -> float:
    with logfire.span("evaluasi_metrik_indeks", tabel=nama_tabel):
        return round(1.0 - (1.0 / (jumlah_scan + 1)), 4)