> ## Documentation Index
> Fetch the complete documentation index at: https://raindrop.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Pydantic AI (Beta)

> Automatic tracing for Pydantic AI agents with Raindrop

## Installation

```bash theme={null}
pip install raindrop-pydantic-ai pydantic-ai
```

## Quick Start

```python theme={null}
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

```python theme={null}
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](/platform/projects) (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](/platform/projects) by passing its slug as `project_id`:

```python theme={null}
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:

```python theme={null}
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=`:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
# 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:

```python theme={null}
raindrop.flush()     # flush pending data
raindrop.shutdown()  # flush + release resources
```

## Factory Function (Legacy)

The `create_raindrop_pydantic_ai()` factory is available for backwards compatibility:

```python theme={null}
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.