Maple agent tracing: Pydantic AI

Goal: every conversation = one Maple Agent Session. Each agent.run() = one turn (one trace) with transcript, chat spans with tokens, execute_tool spans with args/results, failed tools marked failed, sub-agents in their own lanes.

Mechanism: Pydantic AI's native OTel instrumentation (scope pydantic-ai, GenAI semconv on span attributes). No extra instrumentation package. Maple reads gen_ai.conversation.id as the session key for this framework.

Step 0: Detect

1. Pydantic AI version: python -c "import pydantic_ai; print(pydantic_ai.__version__)" (or read pyproject.toml / uv.lock / requirements*.txt).
   • Need 2.x (tested 2.51.0). 1.x has no conversation_id=: tell the user to upgrade; do not work around it.
   • ToolFailed needs >= 2.16.
2. Existing OTel setup. Search for TracerProvider(, set_tracer_provider, logfire.configure, opentelemetry-instrument, sentry_sdk.init, instrument_all, Instrumentation(, .instrument =.
   • Logfire already configured → use the Logfire path (Step 2b).
   • Another TracerProvider exists → add a BatchSpanProcessor(OTLPSpanExporter()) to it; do NOT create a second provider.
   • Nothing → Step 2a.
3. Find: every agent.run( / run_sync( / run_stream( / iter( call, where the chat/thread id lives in the request, every Agent( construction, and every tool that calls another agent's run().
4. Other instrumentors on the same model client (logfire.instrument_openai, OpenAIInstrumentor, OpenLLMetry Traceloop.init, a global logfire.instrument_httpx()) → they double-trace model calls. Keep Pydantic AI's; ask before removing the others if they serve something else. logfire.instrument_httpx(client) on a non-model client (tool HTTP calls) is fine to keep.

Step 1: Key and region

• US: https://ingest.maple.dev. EU: https://ingest.eu.maple.dev.
• Header: Authorization=Bearer <key>.
• Key given in the prompt → use it.
• No key → use the literal MAPLE_TEST (ingest accepts and discards it) and tell the user to replace it with their key from Settings → Ingestion.
• Never put a private maple_sk_ key in browser code.
• Follow the repo's secret/env convention (.env, settings module, secret manager) if it has one. Otherwise inline is acceptable: ingest keys are write-only.
• Key read from a secret env var in code: when it is unset, log one warning (MAPLE_INGEST_KEY is not set; Maple telemetry export is disabled) and skip the Maple exporter so the app runs normally. Never raise or exit over the key; no bare KeyError on import, no header without a key.
• A 401 ingest_unauthorized ("Invalid ingest key") with a key you trust usually means the key belongs to the other region (keys are region-bound): try the other endpoint.

Env vars (the exporter reads them when it is built; it appends /v1/traces). If the app loads .env (load_dotenv()), call it at the top of the tracing module, before the exporter or logfire.configure(); otherwise the exporter silently targets localhost:4318 with no key.

OTEL_EXPORTER_OTLP_ENDPOINT=https://ingest.maple.dev
OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer <key>"
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf

Step 2a: Install + init (plain OpenTelemetry, default)

Add with the repo's package manager (uv/poetry/pip):

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

Keep the project's existing pydantic-ai extras; only add the two OTel packages if pydantic-ai is already installed.

Create tracing.py (adapt service name / environment to the project):

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 it first in the entry point (app module, main.py, worker). It must run before the first agent.run(); agents constructed earlier are still covered.
• Existing provider: skip creating one; add the processor to it and call Agent.instrument_all(InstrumentationSettings(include_content=True, include_binary_content=False)) (uses the global provider).
• Do not pass version=. Default 5 is the tested format; 2-4 are deprecated.
• Set a real service.name (never leave unknown_service).

Step 2b: Logfire path (only if the project already uses Logfire)

Keep the Step 1 env vars; they must be in os.environ before logfire.configure() runs. Logfire then adds OTLP span, metric and log exporters for this endpoint; pass metrics=False to logfire.configure() if only traces are wanted.

Do not install the Step 2a OTel packages: logfire already depends on the SDK and the OTLP/HTTP exporter, and logfire 5.1.x pins opentelemetry-sdk<1.45, so adding opentelemetry-sdk>=1.45 makes the install unresolvable. Tested with logfire 5.1.1 (OTel SDK 1.44.0).

import logfire

TOOL_CONTENT = {"gen_ai.tool.call.arguments", "gen_ai.tool.call.result", "final_result"}


def keep_tool_content(match: logfire.ScrubMatch):
    if len(match.path) > 1 and match.path[1] in TOOL_CONTENT:
        return match.value


logfire.configure(
    service_name="support-agent",
    environment="production",  # adapt to the project
    send_to_logfire=False,  # keep the project's existing value as is (incl. 'if-token-present'); False only if unset
    scrubbing=logfire.ScrubbingOptions(callback=keep_tool_content),
)
logfire.instrument_pydantic_ai()

• Default scrubbing replaces tool args/results containing session, auth, password, cookie, secret, api key... with [Scrubbed due to '<word>']. gen_ai.conversation.id and message attributes are exempt. If the project already has a scrubbing config, merge the callback into it instead of replacing it.
• Do not also call Agent.instrument_all(...) with another provider.

Step 3: Session id (required)

Pydantic AI resolves gen_ai.conversation.id per run as: explicit conversation_id= > id on the last message of message_history > fresh UUID7. Pass it explicitly on EVERY run call, from the app's chat/thread/conversation id:

result = await agent.run(text, conversation_id=chat_id, message_history=history)

async with agent.run_stream(text, conversation_id=chat_id, message_history=history) as run:
    async for delta in run.stream_text(delta=True):
        yield delta

• Same for run_sync(, iter(, run_stream_events(, and the resume run that sends DeferredToolResults.
• The id must be stable per conversation and unique across conversations. No process-wide constants, no uuid4() per request, no module-level default.
• No id available in the app → ask the user where the conversation boundary is; if it's a single-shot script, generate one uuid per conversation (not per run) and reuse it.
• Streaming: keep the async with open until the stream is consumed (in FastAPI, inside the generator passed to StreamingResponse), or the invoke_agent span ends early.

Step 4: Content

• Content is on by default (include_content=True): gen_ai.input.messages, gen_ai.output.messages, gen_ai.system_instructions on chat spans; tool args/results on execute_tool spans. Maple's transcript needs these.
• Always set include_binary_content=False (base64 media repeats in every later chat span).
• User wants content off → include_content=False globally, or per agent: Agent(..., capabilities=[Instrumentation(settings=InstrumentationSettings(include_content=False))]) (from pydantic_ai.capabilities import Instrumentation; omit tracer_provider to use the global one). Tell them the transcript keeps roles but no text.
• Logfire scrubbing does not redact message content. For pattern redaction of prompts, recommend an OTel Collector redaction/transform processor.

Step 5: Tools, errors, sub-agents

1. Name every agent: Agent(..., name="support"). Unnamed agents become agent and share one lane.
2. Tool failures the model should see: raise ToolFailed("message") (from pydantic_ai import ToolFailed, >= 2.16). Span ERROR, message recorded as gen_ai.tool.call.result, run continues, no retry budget used.
   • ModelRetry also marks the span ERROR (one failed call per retry): use it only for real retries.
   • Replace return {"error": ...} / return "Error: ..." in tools with raise ToolFailed(...) only where the user agrees; returned errors show as successful calls.
   • Uncaught exceptions mark the tool span and the run ERROR and abort the run.
3. Delegation (a tool that calls another agent): pass the caller's id and usage to EVERY nested run:

@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

Without conversation_id=ctx.conversation_id each delegate mints a UUID7, so one trace carries several ids and Maple may file the trace under the wrong session or split the turn. 4. Sequential pipelines of top-level runs (orchestrator run, then summary run): pass the same conversation_id= to each; each run is its own trace/turn in the same session.

Step 6: Flush

• The SDK flushes at normal interpreter exit. Add an explicit flush where that doesn't happen:
  • AWS Lambda / Cloud Functions / Cloud Run jobs: trace.get_tracer_provider().force_flush() in a finally in the handler.
  • Scripts, CLIs, one-shot jobs: provider.shutdown() at the end (finally).
  • Celery/RQ/multiprocessing workers, notebooks: force_flush() after each task/cell that runs an agent.
  • Logfire: logfire.force_flush() / logfire.shutdown().

Step 7: Verify

Run one real conversation: 2+ messages with the same id, at least one tool call, one streamed message if the app streams, and a sub-agent call if the app delegates. If the app has no scriptable entry point (server, UI only), write a small driver for this run: one conversation id, 2+ turns, a tool call, flush before exit. Then check (Maple → Agent Sessions, filter by the service name; wait ~30 s):

• Exactly one session per conversation; session id = the id you passed (not a UUID7 you didn't create). A second conversation is a different session.
• Framework shows as Pydantic AI.
• One turn per run(); transcript shows user prompts, assistant replies and tool calls.
• Spans: invoke_agent <name>, chat <model>, execute_tool <tool>; each chat/execute_tool is inside its run's invoke_agent in the same trace.
• Every chat span has input and output tokens, including the streamed one.
• Tool calls have their real names, arguments and results.
• A failing tool is marked failed with its message; successful tools are not.
• Sub-agents appear as separate lanes with their name=, all under the caller's session.
• Cost shows on sessions whose model Pydantic AI can price; "unpriced" otherwise.
• No attribute contains an API key or Bearer token.

With the Maple MCP: list_agent_sessions with search=<conversation id> returns one row.

Quick local cue: Pydantic AI 2.51 prints an observability: off banner on the first run when no instrumentation is set; it disappears once instrument_all() (or logfire.instrument_pydantic_ai()) has run.

Without Maple access (or with MAPLE_TEST), both must hold; silence alone proves nothing (no spans is silent too):

• The run exits with no Failed to export / 401 lines on stderr.
• A console exporter shows the spans with gen_ai.conversation.id, identical across runs of one conversation and on delegate spans. Add it temporarily: plain path provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())); Logfire path logfire.configure(..., additional_span_processors=[SimpleSpanProcessor(ConsoleSpanExporter())]) (both from opentelemetry.sdk.trace.export). On the Logfire path its own console prints <agent> run / running tool: <x>; exported span names are still invoke_agent / execute_tool.

Known behavior (tell the user when relevant)

• Tool failure matrix: ToolFailed → span ERROR, model sees message, run continues; ModelRetry → ERROR, one failed call per retry; other exception → ERROR, run raises, failed turn; return {"error": ...} → UNSET, shows as successful call.
• Approval-gated tools (requires_approval=True) pause the run with no tool span; the execute_tool span appears only in the resumed run (the one sending DeferredToolResults). Same conversation_id= → paused and resumed runs are two turns of one session, both labeled with the original request.
• Delegation rendering: an execute_tool <x> span whose only child is invoke_agent <worker> shows as a delegation; the tool's args/result become the lane's input/output. Several delegation tools in one model reply run concurrently, so lanes overlap in time.
• include_content=False also drops exception messages (only the exception type is kept).
• Instrumentation version: 5 default (tested); 2-4 deprecated with PydanticAIDeprecationWarning (version 2 used different span names); 6 is opt-in and sends tool results with role: "tool". Fix the warning by removing version=.
• Tokens: chat spans carry gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, plus gen_ai.usage.cache_read.input_tokens / gen_ai.usage.cache_creation.input_tokens when the provider caches. invoke_agent carries run totals under gen_ai.aggregated_usage.*, which Maple does not add to the session total. Delegate tokens stay on the delegate's spans.
• Streaming: Pydantic AI requests usage on OpenAI-compatible streams (stream_options.include_usage), so streamed calls have tokens. Time-to-first-chunk is recorded under a key Maple doesn't read yet.
• Cost: Pydantic AI writes operation.cost on chat spans when it can price the model, and Maple shows it. Maple never prices tokens itself; a model Pydantic AI can't price shows as unpriced.
• Logfire with send_to_logfire=True (or a Logfire token in env) sends to both Logfire and Maple.
• Provider extras: swap [openai] for anthropic, google, openrouter, ...; the full pydantic-ai package also works.
• Duplicate spans: another instrumentor on the model client (Logfire instrument_openai(), OpenInference, OpenLLMetry) double-traces model calls; keep Pydantic AI's.
• Export 413 / huge chat spans: binary content recorded as base64; set include_binary_content=False.

Do not

• Do not keep event_mode="logs": content moves to logs, which Maple doesn't read (empty transcript).
• Do not add session.id or maple_ai.session.id: Maple reads gen_ai.conversation.id here; maple_ai.session.id re-vendors the span.