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.

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

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