raindrop-pydantic-ai 0.0.4

Raindrop integration for Pydantic AI — automatic observability for Agent run() and run_sync() calls

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")

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, tracing_enabled=True, bypass_otel_for_tools=True, disable_auto_instrument=True, debug=False)

Create a new Raindrop wrapper instance.

Parameter	Type	Default	Description
api_key	str | None	None	Raindrop API key. When None or empty, telemetry is disabled
user_id	str | None	None	Associate all events with a user
convo_id	str | None	None	Group events into a conversation
tracing_enabled	bool	True	Enable/disable tracing in raindrop.init()
bypass_otel_for_tools	bool	True	Bypass OpenTelemetry for tool calls
disable_auto_instrument	bool	True	Library auto-instrumentation is opt-in (see below)
debug	bool	False	Enable DEBUG-level logging for the package

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.

If you specifically want LLM-call-level spans from library instrumentation and have verified compatibility in your environment, opt back in with disable_auto_instrument=False.

Debug Mode

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

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

Method	Description
wrap(agent)	Instrument a Pydantic AI Agent
flush()	Flush pending events to Raindrop
shutdown()	Flush and shut down the client
identify(user_id, traits=None)	Identify a user with optional traits
track_signal(event_id, name, signal_type="default", ...)	Track a custom signal event

Testing

cd packages/pydantic-ai-python
pip install -e ".[dev]"
python -m pytest tests/ -v

Full Documentation

See Raindrop Pydantic AI Integration Docs for full documentation.

License

MIT