maple-agent-tracing-pydantic-ai

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

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.
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.
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().
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

Name every agent: Agent(..., name="support"). Unnamed agents become agent and share one lane.
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.
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.