Trace Pydantic AI agents with OpenTelemetry

Send Pydantic AI's built-in OpenTelemetry spans to Maple so each conversation shows up as one Agent Session.

Pydantic AI already emits OpenTelemetry spans for runs, model calls and tool calls. You export them to Maple and pass a conversation id on every run. Without the id, each message becomes its own session.

Quick setup with a coding agent

Copy this prompt into a coding agent that can run shell commands, such as Claude Code, Codex or Cursor. It installs the maple-agent-tracing-pydantic-ai skill and follows it.

Set up Maple agent tracing for Pydantic AI in this project.

Install the skill with `npx skills add MapleTechLabs/maple/skills --skill maple-agent-tracing-pydantic-ai -y`, then follow it.

My Maple ingest key is maple_pk_... and my organization is in the US region.

Your ingest key is in Settings → Ingestion. If your organization is in the EU region, change US to EU in the prompt.

Export spans to Maple

Install the OpenTelemetry SDK and the OTLP/HTTP exporter next to Pydantic AI. Swap [openai] for the extras of the providers you use.

pip install "pydantic-ai-slim[openai]>=2.51" "opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45"

uv add "pydantic-ai-slim[openai]>=2.51" "opentelemetry-sdk>=1.45" "opentelemetry-exporter-otlp-proto-http>=1.45"

Point the exporter at Maple. For an EU organization, use https://ingest.eu.maple.dev.

export OTEL_EXPORTER_OTLP_ENDPOINT="https://ingest.maple.dev"
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer YOUR_INGEST_KEY"
export OTEL_EXPORTER_OTLP_PROTOCOL="http/protobuf"

Set up tracing once, when your process starts:

# tracing.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from pydantic_ai import Agent, InstrumentationSettings

provider = TracerProvider(
    resource=Resource.create(
        {"service.name": "support-agent", "deployment.environment.name": "production"}
    )
)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

Agent.instrument_all(
    InstrumentationSettings(
        tracer_provider=provider,
        include_content=True,
        include_binary_content=False,
    )
)

Import tracing at the top of your entry point (main.py, the FastAPI app module, the worker), before the first agent.run().

If your app already has a TracerProvider (from opentelemetry-instrument, Sentry or your own setup), add the BatchSpanProcessor to it instead of creating a second one, and call Agent.instrument_all(InstrumentationSettings(include_content=True, include_binary_content=False)) without tracer_provider.

If you already use Logfire

Skip the pip install above, because it conflicts with Logfire’s OpenTelemetry pins. Keep the three environment variables. Logfire exports to Maple whenever OTEL_EXPORTER_OTLP_ENDPOINT is set.

import logfire

logfire.configure(service_name="support-agent", environment="production", send_to_logfire=False)
logfire.instrument_pydantic_ai()

Logfire scrubs tool arguments and results by default. If they arrive as [Scrubbed due to ...], see Troubleshooting.

Pass the conversation id on every run

Pass the chat or thread id your app already has as conversation_id= on every run(), run_stream() and iter():

from pydantic_ai import Agent

support = Agent("openai:gpt-4o-mini", name="support")


async def handle_message(chat_id: str, text: str, history: list) -> str:
    result = await support.run(text, conversation_id=chat_id, message_history=history)
    return result.output

The id must stay the same for the whole conversation and differ between conversations.

When streaming, keep the async with block open until the stream is finished. The run’s span ends when the block exits.

async def stream_reply(chat_id: str, text: str, history: list):
    async with support.run_stream(text, conversation_id=chat_id, message_history=history) as run:
        async for delta in run.stream_text(delta=True):
            yield delta

Sub-agents

When a tool runs another agent, pass conversation_id and usage from the tool’s context. Give every agent a name= so each one gets its own lane.

from pydantic_ai import Agent, RunContext

weather_worker = Agent("openai:gpt-4o-mini", name="weather_worker", tools=[get_weather])
orchestrator = Agent("openai:gpt-4o-mini", name="orchestrator")


@orchestrator.tool
async def research_weather(ctx: RunContext[None], city: str) -> str:
    """Delegate to the weather worker."""
    result = await weather_worker.run(
        f"What is the current weather in {city}?",
        usage=ctx.usage,
        conversation_id=ctx.conversation_id,
    )
    return result.output

Flush short-lived processes

A Lambda, a killed worker or a notebook kernel doesn’t exit normally, so flush explicitly:

import asyncio

from opentelemetry import trace


def handler(event, context):
    try:
        return asyncio.run(handle_message(event["chat_id"], event["text"], []))
    finally:
        trace.get_tracer_provider().force_flush()

In a script or CLI, call provider.shutdown() at the end. With Logfire, use logfire.force_flush() or logfire.shutdown().

Check that it works

If Pydantic AI prints an observability: off banner on the first run, tracing.py didn’t run before your first agent.run().

Run a conversation with two messages and a tool call, then open Agent Sessions. You should see one session with your conversation id and framework Pydantic AI, one turn per run(), a transcript with prompts, replies and tool calls, and token counts and cost on every model call.

Troubleshooting

Every message is its own session. Pass conversation_id= on every run(), run_stream() and iter().
A multi-agent run is split into several turns or sessions. Pass conversation_id=ctx.conversation_id to every nested run().
Tool arguments or results read [Scrubbed due to ...]. Logfire’s scrubbing matched a word like session or auth. Pass scrubbing=logfire.ScrubbingOptions(callback=...) that keeps gen_ai.tool.call.arguments and gen_ai.tool.call.result, or scrubbing=False.
A failed tool shows as successful. The tool returned an error value. Raise ToolFailed("...") so the call is marked failed and the model still sees the message.
Spans show up twice. Another instrumentor (Logfire’s instrument_openai(), OpenInference, OpenLLMetry) also traces the model client. Remove it and keep Pydantic AI’s.