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.

Getting Started

SDK

Integrations

Platform

Organization

Cookbooks

Workshop

Query API

MCP

Security

Support

Pydantic AI (Beta)

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

​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

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