raindrop-pydantic-ai

Raindrop integration for Pydantic AI. Automatically captures Agent run() and run_sync() calls including input, output, model name, token usage, and finish reason.

Installation

pip install raindrop-pydantic-ai pydantic-ai

Quick Start

from raindrop_pydantic_ai import RaindropPydanticAI
from pydantic_ai import Agent

raindrop = RaindropPydanticAI(
    api_key="your-write-key",
    user_id="user-123",
)

agent = Agent("openai:gpt-4o", system_prompt="Be helpful")
raindrop.wrap(agent)

result = agent.run_sync("What is the capital of France?")
print(result.output)

raindrop.flush()

Async Usage

import asyncio
from raindrop_pydantic_ai import RaindropPydanticAI
from pydantic_ai import Agent

raindrop = RaindropPydanticAI(api_key="rk_...", user_id="user-123")
agent = Agent("openai:gpt-4o")
raindrop.wrap(agent)

async def main():
    result = await agent.run("What is the capital of France?")
    print(result.output)
    raindrop.flush()

asyncio.run(main())

Factory Function (Legacy)

The create_raindrop_pydantic_ai() factory function is still available for backwards compatibility:

from raindrop_pydantic_ai import create_raindrop_pydantic_ai

raindrop = create_raindrop_pydantic_ai(api_key="rk_...", user_id="user-123")

Projects

Route events to a specific project by passing its slug as project_id:

raindrop = RaindropPydanticAI(
    api_key="your-write-key",
    project_id="support-prod",
)

project_id sets the X-Raindrop-Project-Id header on every event. Omit it (or pass "default") to use your org's default Production project, which is the existing behavior. The same option is accepted by the create_raindrop_pydantic_ai(...) factory. Invalid slugs are ignored with a warning and no header is sent.

What Gets Captured

• Agent runs: input prompt, output text (including structured output), model name
• Token usage: input_tokens and output_tokens from the result
• Finish reason: pydantic_ai.finish_reason captured from the last model response (e.g. "stop", "length", "tool_call")
• Errors: error type and message captured in event properties, then re-raised
• Async support: both run() (async) and run_sync() (sync) are instrumented
• Double-wrap guard: calling wrap() twice on the same agent is a safe no-op

API Reference

RaindropPydanticAI(api_key, user_id=None, convo_id=None, project_id=None, tracing_enabled=True, bypass_otel_for_tools=True, disable_auto_instrument=True, debug=False)

Create a new Raindrop wrapper instance.

Library auto-instrumentation is opt-in

As of 0.0.4, disable_auto_instrument defaults to True: the integration no longer lets Traceloop monkey-patch every LLM client library it recognizes in your process (OpenAI, Anthropic, botocore, google-genai, etc.). The wrapper captures input/output, token usage, model name, and finish_reason directly from run() / run_sync() results, so no library patching is needed for full dashboards.

Model spans need no library instrumentation: the wrapper records its own (see Model spans). If you specifically want the instrumentor's spans instead and have verified compatibility in your environment, opt back in with disable_auto_instrument=False; the wrapper then adds no span for calls that instrumentation already records.

Debug Mode

raindrop = RaindropPydanticAI(
    api_key="rk_...",
    debug=True,  # enables DEBUG-level logging
)

Per-call properties and identity

run / run_sync has no channel for your own event data, so wrap the call in event_context. Its properties merge into every event captured inside the block, and user_id / convo_id replace the constructor values for those events:

from raindrop_pydantic_ai import event_context

with event_context(user_id="user-123", convo_id="conv-9", properties={"plan": "pro"}):
    result = agent.run_sync("What is the capital of France?")

Updating the last event

last_event_id is the event_id of the most recent event this wrapper captured. Use it to attach a signal, or reopen the event with resume_interaction() (defaults to last_event_id) and update it through the core SDK's Interaction:

interaction = raindrop.resume_interaction()
if interaction is not None:
    interaction.set_properties({"reviewed": True})
    interaction.finish()

Key clashes. If one of your properties has the same key as one the wrapper derives itself, your value wins, whichever update carries it (a debug-level log line names the key). The wrapper's own keys are ai.* (for example ai.usage.prompt_tokens), error/error.* and pydantic_ai.*; avoid them unless you mean to replace the wrapper's value.

Property-only updates need raindrop-ai 0.0.77 or newer; older core versions drop an update that carries no input or output.

last_event_id is per wrapper instance; under concurrent calls it is whichever event was captured last, so pass event_id= when you need to correlate concurrent calls.

identify(user_id, traits=None)

Identify a user with optional traits:

raindrop.identify("user-123", traits={"name": "Alice", "plan": "pro", "age": 30})

track_signal(event_id, name, signal_type="default", ...)

Track a signal event (feedback, edits, etc.):

raindrop.track_signal(
    event_id="evt-abc",
    name="thumbs_up",
    signal_type="feedback",
    sentiment="POSITIVE",
    comment="Great answer!",
)

flush() / shutdown()

Flush pending events before your process exits:

raindrop.flush()     # flush pending data
raindrop.shutdown()  # flush + release resources

Methods

Tool catalog capture (ai.prompt.tools)

Every model span records the tools Pydantic AI handed the model on that request as ai.prompt.tools: an OTel string array, one JSON document per tool with name, description and the full JSON Schema as inputSchema ({"type": "function", ...}), or {"type": "provider-defined", "name": "web_search"} for native tools such as WebSearchTool. The capture is exact, per request: Pydantic AI passes every Model.request() / Model.request_stream() a ModelRequestParameters whose function_tools, output_tools and native_tools are precisely what goes into the provider request. wrap(agent) wraps those two methods on the agent's model (and on a Model instance passed to run(model=...)) and binds that list through the core's prompt_tools_scope mechanism for the duration of the request, so the model span started inside it (the provider instrumentor's or the wrapper's own) carries the attribute. A request with no tools records [].

With disable_auto_instrument=False, model spans come from the OpenLLMetry instrumentor of the underlying provider client (the OpenAI instrumentor for OpenAIChatModel). OpenLLMetry 0.62 writes the same list to gen_ai.tool.definitions, not llm.request.functions.*; raindrop-ai 0.0.70 / 0.0.71 do not rebuild ai.prompt.tools from that attribute and 0.0.72 does, so the wrapper-side scope is what puts the attribute on the span on every core version, and it takes precedence over the rebuilt list when both are present. With the default disable_auto_instrument=True, the wrapper's own model span (see below) carries it. Pydantic AI's own instrument=True spans are not model spans in this sense and do not receive the attribute.

Model resolution: an agent created with a "provider:model" string has that string resolved to a Model once, on the first wrapped run, so agent.model becomes the resolved instance; a string passed per run (run(model="openai:...")) is resolved the same way and the wrapped instance is handed to run() in its place. A Model instance shared by several agents is wrapped once and still reports each agent's own tools, because the list is read from the request, not from the agent. A tools= override is scoped to the run it belongs to: a wrapped agent run from inside another wrapped agent's tool still reports its own list (or its own override), not the outer agent's. An explicit raindrop.prompt_tools(...) scope, whether opened around the run or from inside one of its tools, still wins over the inferred list.

Override:

rd = RaindropPydanticAI(api_key="...", user_id="u1", disable_auto_instrument=False,
                        tools=[{"type": "function", "name": "get_weather", "inputSchema": {...}}])
agent = rd.wrap(Agent(model))               # constructor default applies to every run
agent = rd.wrap(Agent(model), tools=[...])  # per-wrap value replaces the default

tools=[] at either level records an empty catalog. An enclosing raindrop.prompt_tools(...) / begin(..., tools=...) scope is also treated as an override and is never shadowed by the request's list.

Semantics shared with every Raindrop SDK:

• Attribute absent: the wrapper could not see the tool list (or there is no model span, see below).
• []: the model had no tools.
• Override replaces, never merges. A tools= you pass wins over anything inferred; tools=[] records an empty catalog.
• Content gate. Tool definitions are prompt content. With TRACELOOP_TRACE_CONTENT=false nothing is recorded, overrides included.
• Malformed entries are skipped with a debug log; nothing here raises into your application.

Accepted shapes for tools=: canonical {"type": "function", "name", "description", "inputSchema"}, OpenAI {"type": "function", "function": {...}}, Anthropic {"name", "description", "input_schema"}, and {"type": "provider-defined", "name", ...}. The core normalises them; the JSON Schema is kept exactly as given.

Needs raindrop-ai>=0.0.70; on an older core the wrapper logs at debug and records nothing.

Application Git metadata

RaindropPydanticAI(...) and create_raindrop_pydantic_ai(...) accept the keyword-only app_git option. It defaults to True: explicit Raindrop Git environment or deployment context is applied immediately, and the base SDK may perform one bounded background local-Git lookup from the process working directory. Event capture, flush, and shutdown never wait for that lookup. Pass False to disable enrichment, or pass an AppGitOptions mapping with commit_sha, commit_dirty, branch, source_directory, detect_branch, and/or auto_detect. Automatic branch discovery remains opt-in through detect_branch=True (or RAINDROP_GIT_DETECT_BRANCH=true).

For an ordinary in-process application, the process working directory is treated as the application-under-test checkout. A remote, coding, workflow, or observer process must not rely on its own checkout: pass app_git=False, provide explicit revision values, or set source_directory to the actual application checkout. Canonical per-operation properties remain authoritative. When supplying client=, configure app_git while constructing that Raindrop client; the supplied client is authoritative and the wrapper's app_git argument does not reconfigure it.

Release order is deliberate: first publish the base SDK feature, then publish the wrapper feature release with its minimum dependency coordinated to that base release. The existing raindrop-ai lower bound remains compatible, but application Git metadata is unavailable on an older core and must not be claimed complete until the base is upgraded. Until coordination assigns a released version, the wrapper checks for an explicit base app_git parameter and omits the option when unsupported. Explicit non-default configuration is debug-logged and omitted. Unsupported app_git is determined by signature inspection before construction, not by retrying initialization after a TypeError; Git configuration adds no initialization attempts and does not change any existing framework-specific initialization fallback.

User id

Every event needs a real user id: set user_id= on the wrapper or per call with event_context(user_id=...). Without one (missing, blank, or a placeholder such as "unknown" or "anonymous"), the event is not sent and the wrapper logs one warning; it never raises. Updating an event you resumed with resume_interaction(event_id=...) needs no user id, since ingest keeps the one it stored.

Model spans

With tracing on (the default), the wrapper records each model request of a run as its own span on the Raindrop client's tracing, with that request's usage and the model that answered it. A model set with Agent.override(model=...) is observed the same way. (A model the wrapper cannot wrap gets one span for the whole run, with the run's total usage, and its output tools are read from the agent's output schema, or, when that cannot be read, recognised by Pydantic AI's default names final_result / final_result_<Type> unless they are registered function tools.) Each span carries:

• the model: gen_ai.provider.name (canonical name, such as openai or amazon-bedrock), gen_ai.request.model and gen_ai.response.model;
• the token usage the provider reports: gen_ai.usage.input_tokens, output_tokens, cache_read_input_tokens, cache_write_input_tokens and reasoning_tokens;
• the event it belongs to: ai.telemetry.metadata.raindrop.eventId and the traceloop.association.properties.* keys.

The run's total token usage also stays on the event as ai.usage.* properties. When an OpenTelemetry instrumentation that sees the call already records it (the provider's own instrumentation, for example with auto_instrument=True on the client), the wrapper adds no span of its own, so a call is never counted twice. An instrumentation for another provider or library does not suppress the wrapper's span.

Tool calls

Each tool call in a run becomes a tool span on the run's event, with its arguments, result, duration and any error. The calls are read from the run's new messages: a tool's ModelRetry is recorded as an error, and output tools (those Pydantic AI declares as output tools for the request) are not tool calls. When a run raises, the tool calls that completed before the failure are still recorded on its error event. Spans need tracing enabled (the default); input and output follow the core SDK's content setting (TRACELOOP_TRACE_CONTENT).

Standard options and flush results

When the wrapper builds its own client, it accepts the core SDK's standard options by the same names and passes them on:

raindrop = RaindropPydanticAI(
    api_key="rk_...",
    user_id="user-123",
    redact_pii=True,        # local PII redaction of event input/output
    flush_timeout=5,        # seconds; deadline for flush() and shutdown()
    debug=True,             # also raises the core SDK's log verbosity
    max_queue_size=5_000,   # any other option `Raindrop(...)` declares
)

result = raindrop.flush()   # the core's FlushResult; safe to ignore

• An option the installed raindrop-ai does not support is ignored with a debug log, never an error. flush_timeout and the FlushResult return value need raindrop-ai 0.0.77 or later; on older releases flush() and shutdown() return None.
• With client= (a Raindrop you built), configure these options on that client; the wrapper ignores them (debug log).
• flush() and shutdown() never raise. The result is truthy when everything was delivered, and reports pending, dropped and timed_out.

Testing

cd packages/pydantic-ai-python
pip install -e ".[dev]"
python -m pytest tests/ -v   # unit tests (no external services)

End-to-end behavior is verified by the cross-SDK conformance harness. This package ships a thin conformance driver at conformance/driver.py that maps the shared scenario corpus onto the wrapper's public API; known gaps are tracked as ticket-linked entries in conformance/failures.txt. The fault lane runs on every PR touching packages/*-python/** (.github/workflows/conformance-wrappers-python.yml) against a local capture server; the prod lane verifies delivery by reading back through the public Query API. The harness is pinned by commit SHA (HARNESS_REF). See the harness docs: HOW-IT-WORKS · AGENTS · README.

Full Documentation

See Raindrop Pydantic AI Integration Docs for full documentation.

License

MIT