Pydantic AI (Beta)

pip install raindrop-pydantic-ai pydantic-ai

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

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
)

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

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

from raindrop import Raindrop

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

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

import asyncio

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

asyncio.run(main())

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

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

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

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