Pydantic AI

Trace Pydantic AI agents in Muster via the framework's built-in instrumentation.

Pydantic AI is a Python agent framework for building production-grade generative AI applications. Muster captures Pydantic AI agent runs through the framework's built-in OpenTelemetry instrumentation.

Setup

1. Install dependencies

pip install langfuse pydantic-ai

2. Configure credentials

import os

os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_BASE_URL"] = "https://app.getmuster.io"
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

from langfuse import get_client

langfuse = get_client()

if langfuse.auth_check():
    print("Muster client is authenticated and ready!")

3. Instrument all agents

from pydantic_ai import Agent

Agent.instrument_all()

4. Build an agent with `instrument=True`

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class RouletteDeps:
    winning_number: int

roulette_agent = Agent(
    "openai:gpt-4o",
    deps_type=RouletteDeps,
    output_type=bool,
    system_prompt="Decide if the player has won given the number they bet on.",
    instrument=True,
)

@roulette_agent.tool
async def roulette_wheel(ctx: RunContext[RouletteDeps], square: int) -> str:
    """Check if the square is a winner."""
    return "winner" if square == ctx.deps.winning_number else "loser"

result = roulette_agent.run_sync("Put my money on square 18", deps=RouletteDeps(winning_number=18))
print(result.output)

Every run becomes a trace in Muster with sub-spans for each LLM call and tool invocation.

Add user/session attributes

from langfuse import propagate_attributes

with propagate_attributes(user_id="user_123", session_id="sess_abc", tags=["pydantic-ai"]):
    roulette_agent.run_sync(...)

Troubleshooting

Enable debug logging: export LANGFUSE_DEBUG="True".
Call langfuse.flush() before exit in short-lived scripts.