Pydantic AI (Beta)

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
    project_id="support-prod",          # Optional: route events to a specific project (slug)
    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
)

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
`project_id`	`str \| None`	`None`	Route events to a specific project (slug); omit for the default Production project
`tracing_enabled`	`bool`	`True`	Enable/disable tracing in `raindrop.init()`
`bypass_otel_for_tools`	`bool`	`True`	Bypass OpenTelemetry for tool calls
`debug`	`bool`	`False`	Enable DEBUG-level logging for the package

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.

Multiple projects in one process

Available in raindrop-ai>=0.0.56. When one service runs several agents that should report to different projects, create one RaindropPydanticAI wrapper per project. Each wrapper owns its own raindrop.Raindrop client, so the two route independently — there is no shared module-level state:

from raindrop_pydantic_ai import RaindropPydanticAI

# One long-lived wrapper per project, created at startup and reused.
rd_support = RaindropPydanticAI(api_key="your-write-key", project_id="support-prod")
rd_billing = RaindropPydanticAI(api_key="your-write-key", project_id="billing-prod")

rd_support.wrap(support_agent)   # -> support-prod
rd_billing.wrap(billing_agent)   # -> billing-prod

rd_support.flush()
rd_billing.flush()

Each wrapper owns its configuration and delivery pipeline, so agents handled by different wrappers route independently. To share a single client across wrappers (or with the module-level API), construct a raindrop.Raindrop yourself and pass it via client=:

from raindrop import Raindrop

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

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

Parameter	Type	Default	Description
`event_id`	`str`	required	The event ID to attach the signal to
`name`	`str`	required	Signal name
`signal_type`	`"default" \| "feedback" \| "edit"`	`"default"`	Type of signal
`timestamp`	`str \| None`	`None`	ISO-8601 timestamp
`properties`	`dict \| None`	`None`	Extra properties
`attachment_id`	`str \| None`	`None`	Attachment identifier
`comment`	`str \| None`	`None`	Comment text
`after`	`str \| None`	`None`	”After” value for edit signals
`sentiment`	`"POSITIVE" \| "NEGATIVE" \| None`	`None`	Sentiment

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.

​Installation

​Quick Start

​What Gets Traced

​Configuration

​Projects

​Multiple projects in one process

​Structured Output

​Async Usage

​Identifying Users

​Tracking Signals

​Flushing and Shutdown

​Factory Function (Legacy)

​Known Limitations