PydanticAI Integration

Lucidic makes it easy to automatically track PydanticAI agent completions as part of your agent’s behavior — no code changes required.

How It Works

When you set:
lai.init(..., providers=["pydantic_ai"])
Lucidic will:
  • Instrument PydanticAI’s model classes using OpenTelemetry
  • Automatically create events for each agent call - no manual event creation needed
  • Attach the call to the currently active Step (or auto-create one if none exists)
  • Support all PydanticAI model types (OpenAI, Anthropic, Gemini, Groq, etc.)
This means:
  • You don’t need to call create_event() or end_event() manually
  • You get full observability into prompts, model, cost, result, and more — out of the box
  • Even if you forget to create a step, Lucidic handles it automatically

What Gets Captured

Each PydanticAI call is tracked with:
  • description: formatted messages sent to the model
  • model: the underlying model name (e.g. claude-3-5-sonnet-20241022, gpt-4)
  • cost: calculated based on token usage
  • result: truncated summary of the response

Why This Matters

PydanticAI agent calls are a core part of most AI workflows — but without visibility, it’s impossible to debug or optimize:
  • Which call caused the failure?
  • Which step was it part of?
  • How much did it cost?
  • What was the actual response?
This integration gives you full transparency, with zero boilerplate.

Example

import lucidicai as lai
from pydantic_ai import Agent
import asyncio

# Just initialize - that's it!
lai.init(providers=["pydantic_ai"])

agent = Agent()  # Uses default model

# Steps and events are automatically created
async def run():
    response = await agent.run("Hello!")
    return response

asyncio.run(run())

# Session auto-ends when script exits

Streaming Example

import lucidicai as lai
from pydantic_ai import Agent
import asyncio

lai.init(providers=["pydantic_ai"])

agent = Agent()

async def stream_example():
    # Streaming responses are also tracked automatically
    async with agent.run_stream("Tell me a story") as response:
        async for text in response.stream():
            print(text, end="")

asyncio.run(stream_example())

Explicit Step Management

import lucidicai as lai
from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel
import asyncio

lai.init(
    session_name="pydantic_ai_research",
    providers=["pydantic_ai"]  # API key and agent ID from env vars
)

model = AnthropicModel("claude-3-5-sonnet-20241022")
agent = Agent(model=model, system_prompt="You are a research assistant.")

lai.create_step(state="Research", goal="Generate research ideas")

# All agent calls are automatically tracked as events within this step
async def research():
    response1 = await agent.run(
        "What are 5 research ideas about climate change?"
    )
    
    response2 = await agent.run(
        "What methodologies could be used for each idea?"
    )
    
    return response1, response2

responses = asyncio.run(research())

lai.end_step()

Notes

  • All PydanticAI model types are instrumented (OpenAI, Anthropic, Gemini, Groq, etc.)
  • Both sync and async agent runs are supported
  • If no step exists when an agent call is made, Lucidic automatically creates one
  • Streaming responses are tracked automatically
  • You can always add custom events using create_event() for additional context