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

What Gets Traced

The Pydantic AI integration automatically captures:

• Agent runs — input prompt, output text (including structured Pydantic model output), model name
• Token usage — input_tokens and output_tokens from the agent 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 to the caller
• 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

Configuration

raindrop = RaindropPydanticAI(
    api_key="your-write-key",           # Raindrop API key (None = disabled mode)
    user_id="user-123",                 # Optional: associate events with a user
    convo_id="convo-456",               # Optional: conversation/thread ID
    tracing_enabled=True,               # Optional: enable/disable tracing
    bypass_otel_for_tools=True,         # Optional: bypass OTEL for tool calls
    debug=True,                         # Optional: enable DEBUG-level logging
)

Structured Output

The integration handles Pydantic AI’s structured output types — the output is serialized to JSON for telemetry:

from pydantic import BaseModel

class CityInfo(BaseModel):
    name: str
    country: str
    population: int

agent = Agent("openai:gpt-4o", output_type=CityInfo)
raindrop.wrap(agent)

result = agent.run_sync("Tell me about Paris")
print(result.output)  # CityInfo(name='Paris', country='France', population=2161000)

raindrop.flush()

Async Usage

The wrapper supports both sync and async agent runs:

import asyncio

async def main():
    result = await agent.run("What is quantum computing?")
    print(result.output)
    raindrop.flush()

asyncio.run(main())

Identifying Users

Use identify() to associate a user with traits:

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

Traits values must be str, int, bool, or float.

Tracking Signals

Use track_signal() to record feedback, edits, or custom signals:

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

# Edit signal
raindrop.track_signal(
    event_id="evt-abc",
    name="user_edit",
    signal_type="edit",
    after="The corrected answer is ...",
)

Flushing and Shutdown

Always call flush() before your process exits to ensure all telemetry is shipped:

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

Factory Function (Legacy)

The create_raindrop_pydantic_ai() factory is 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",
    tracing_enabled=True,
    bypass_otel_for_tools=True,
)

Known Limitations

• run_stream() is not instrumented — only run() and run_sync() are captured. Streaming runs produce no telemetry.
• Multi-step agent runs: In agents with multiple LLM calls (e.g., tool use loops), only the final result’s data is captured. Intermediate LLM calls are not tracked individually.