Raindrop Python SDK

The official Python SDK for Raindrop AI — track AI events, collect user signals, and instrument LLM applications with OpenTelemetry-based tracing.

Installation

pip install raindrop-ai

Requires Python 3.10+

Quick Start

import raindrop.analytics as raindrop

raindrop.init(api_key="your-api-key", tracing_enabled=True)

# Track an AI event
raindrop.track_ai(
    user_id="user-123",
    event="chat-completion",
    model="gpt-4",
    input="What is the weather?",
    output="It's sunny and 72°F.",
    convo_id="conv-456",
)

Track a plain, non-AI event through the same local queue and batched events/track transport:

raindrop.track(
    user_id="user-123",
    event="cache-hit",
    properties={"region": "us-east-1", "tags": ["hot", "shared"]},
    event_id="event-456",       # optional; generated when omitted
    timestamp="2026-07-13T18:00:00Z",
    attachments=[
        raindrop.Attachment(type="text", value="cached response", role="output")
    ],
)

Plain events have no conversation association — there is no convo_id parameter. To group events into a conversation, use track_ai / begin (partials), which carry convo_id on the AI-data branch.

Instance-based clients (multiple projects in one process)

raindrop.Raindrop is the instance-shaped counterpart of the module API — the same shape as the JS (new Raindrop({...})), Go, Rust, and Java SDKs. Each instance owns its configuration and event pipeline, so one service can route different agents to different projects concurrently:

from raindrop import Raindrop

rd_support = Raindrop(api_key=KEY, project_id="support-agent", tracing_enabled=True)
rd_billing = Raindrop(api_key=KEY, project_id="billing-agent", tracing_enabled=True)

# Everything hangs off the instance — same method names as the module API.
rd_support.track(user_id="u1", event="session-started", properties={"plan": "pro"})
interaction = rd_billing.begin(user_id="u1", event="billing-chat", input="...")
interaction.track_tool(name="invoice_lookup", input=..., output=...)
interaction.finish(output="...")

rd_support.track_ai(user_id="u1", event="support-chat", input="q", output="a")
rd_support.track_signal(event_id=..., name="thumbs_up")

Create instances once at startup and reuse them (see example/fastapi_multi_project.py for a two-agent FastAPI app). Manual events (track_ai, begin/finish, signals, identify) ship on each instance's own connections with its own headers — fully isolated per instance. The module-level API keeps working unchanged; it is simply the default, process-wide instance of the same machinery.

• rd.track_ai_partial(event) streams an incremental track_ai patch (a PartialTrackAIEvent) into that instance's own buffers — the per-instance counterpart used by integration wrappers that emit partial events. Its read-only rd.write_key property exposes the key backing the instance (e.g. to compare whether two clients are equivalent) without reaching into internals.

Tracing with multiple instances

OpenTelemetry tracing is a process singleton (one tracer provider, one exporter, global auto-instrumentation). The first instance constructed with tracing_enabled=True initializes it; later tracing instances share it. Span routing is per-span: begin() binds the current request/task to its client's project until the matching finish() (which restores the previous binding — abandoned interactions release theirs when garbage collected), and every span started while bound carries a raindrop.project_id attribute that the ingest boundary routes on. For LLM/library calls made outside an interaction, scope them explicitly:

with rd_billing.as_current():
    openai_client.chat.completions.create(...)   # spans route to billing-agent

Two limitations, by design:

• One tracing key/org per process. Instances created with a different api_key than the pipeline owner's get a constructor warning, and from that point the export guard only ships spans positively attributed to the owner: the foreign client's spans and any unattributed spans (emitted outside begin()/as_current()) are dropped at export — nothing can ride the wrong org's credential. Single-key processes are unaffected (spans stay byte-identical). Manual events always route correctly. Run a second org's tracing in its own process.
• interaction.track_tool() with bypass_otel_for_tools=True skips the shared pipeline entirely and ships on the instance's own connection — those spans route per instance with no caveats.

Payload size limits

As of 0.0.52, text fields (ai input/output, tool span I/O, LLM span content) are capped at 1,000,000 characters per field by default and truncated with a ...[truncated by raindrop] marker. Fields up to 1M chars — i.e. everything the ingest API accepts today — round-trip unchanged. The cap is enforced before (or during) serialization, so oversized payloads cost the cap — not the payload — on your calling thread, and a single capped ASCII field still fits under the 1 MB event-level ingest limit. Tune it via:

raindrop.init(api_key="...", max_text_field_chars=250_000)   # process-wide default

# Instance clients cap themselves only (never other clients or the default):
rd = Raindrop(api_key="...", max_text_field_chars=100_000)

A stricter OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT env var is still honored for span content. All outbound HTTP carries finite timeouts, and the atexit shutdown flush runs under a 10s deadline so a dead network can never wedge your process exit.

Projects

If your org has a single project, you don't need to do anything — events go to the default Production project automatically. When you have more than one project, route events to a specific one by passing its slug as project_id to init(). When set, every outbound request attaches an X-Raindrop-Project-Id: <slug> header so events land under the named project instead of the org default:

raindrop.init(api_key="your-api-key", project_id="my-project")

The header is attached to every channel — manual events (track_ai, identify, track_signal, partial events, direct tool/trace POSTs), the local Workshop mirror, and auto-instrumented OpenTelemetry spans exported via Traceloop — so all telemetry routes to the same project.

Slugs must match ^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$ (lowercase alphanumerics and dashes, not starting or ending with a dash). Invalid values are logged as a warning and ignored — no exception is raised and no header is sent. Omitting project_id (or passing "default") is fully backward compatible: events fall back to your org's default Production project.

Interactions

Use begin() and finish() for multi-step AI workflows:

interaction = raindrop.begin(
    user_id="user-123",
    event="agent-run",
    input="Search for weather data",
    convo_id="conv-456",
)

# Update incrementally
interaction.set_property("region", "us-east")
interaction.add_attachments([
    raindrop.Attachment(type="code", value="print('hello')", language="python")
])

# Complete the interaction
interaction.finish(output="Found weather data for NYC")

Resuming Interactions

Access the current interaction from nested functions:

@raindrop.tool("sentiment_analyzer")
def analyze_sentiment(text: str):
    interaction = raindrop.resume_interaction()
    interaction.set_property("sentiment", "positive")
    return {"sentiment": "positive"}

Decorators

Instrument functions with automatic span creation:

@raindrop.interaction("my_workflow")
def run_workflow():
    ...

@raindrop.task("process_data")
def process():
    ...

@raindrop.tool("search")
def search(query: str):
    ...

Spans

Context Managers

with raindrop.task_span("process_data"):
    result = do_processing()

with raindrop.tool_span("web_search"):
    results = search(query)

Manual Spans

For async or distributed operations where you need explicit control:

span = raindrop.start_span(kind="tool", name="async_search")
span.record_input({"query": "weather"})

# ... later, when the result arrives
span.record_output({"result": "sunny"})
span.end()

Retroactive Tool Logging

Log tool calls after they complete, without wrapping them in spans:

interaction = raindrop.begin(user_id="user-123", event="agent-run")

interaction.track_tool(
    name="web_search",
    input={"query": "weather in NYC"},
    output={"results": ["Sunny, 72°F"]},
    duration_ms=150,
)

interaction.track_tool(
    name="database_query",
    input={"query": "SELECT * FROM users"},
    duration_ms=50,
    error=ConnectionError("Connection timeout"),
)

interaction.finish(output="Done")

Signals

Track user feedback on AI outputs:

# Basic signal
raindrop.track_signal(event_id="evt-123", name="thumbs_up")

# Feedback with comment
raindrop.track_signal(
    event_id="evt-123",
    name="user_feedback",
    signal_type="feedback",
    comment="This answer was helpful",
    sentiment="POSITIVE",
)

# Edit signal
raindrop.track_signal(
    event_id="evt-123",
    name="user_edit",
    signal_type="edit",
    after="The corrected response text",
)

User Identification

raindrop.identify("user-123", traits={"plan": "pro", "company": "Acme"})

PII Redaction

Enable automatic redaction of emails, phone numbers, credit cards, SSNs, and other PII from AI inputs and outputs:

raindrop.set_redact_pii(True)

Auto-Instrumentation

By default, Raindrop auto-instruments detected LLM libraries (OpenAI, Anthropic, Bedrock, etc.) via Traceloop. To disable:

raindrop.init(api_key="your-key", tracing_enabled=True, auto_instrument=False)

Or selectively control which libraries are instrumented:

from raindrop.analytics import Instruments

raindrop.init(
    api_key="your-key",
    tracing_enabled=True,
    instruments={Instruments.OPENAI},
)

Note: When auto-instrumentation is enabled, the SDK automatically suppresses noisy warnings from instrumentors for providers you don't use (e.g. "Error initializing MistralAI instrumentor") and from OTel attribute type validation (e.g. provider SDKs using sentinel types like Omit). Enable set_debug_logs(True) to see these messages for troubleshooting.

Buffering and Performance

All event-tracking calls (track, track_ai, identify, track_signal, and Interaction.set_input / set_properties / add_attachments / finish) are non-blocking from the caller's perspective. They append to an in-memory buffer that a background daemon thread drains every second by POSTing to the Raindrop API. The HTTP request never runs on the calling thread, so it is safe to call these from a request hot path.

shutdown() is registered via atexit and drains any still-pending events before the process exits. Call flush() explicitly if you need to force a drain at a known point. flush() is a Class-2 operation and may block on network delivery. Plain track() batches use the bounded delivery policy on the caller thread (up to three attempts with backoff), while other event batches retain a single attempt during explicit flush. Do not call flush() from a latency-sensitive hot path. All delivery paths treat non-429 4xx responses as permanent rejections and do not retry them.

# Tune the in-memory buffer size (default 10_000 events)
import raindrop.analytics as raindrop
raindrop.max_queue_size = 500

Configuration

Environment Variables

Development

# Install dependencies
pip install poetry
poetry install

# Run tests
poetry run pytest

# Run with coverage
poetry run pytest --cov=raindrop

# Run specific test file
poetry run pytest tests/test_analytics.py -v

Cross-SDK conformance

Beyond pytest, behavioral parity across Raindrop SDKs is verified by the cross-SDK conformance harness, which replays a shared scenario corpus against every SDK. It runs a fault lane (local capture server, on every PR) and a prod lane (real ingest verified via the public Query API readback). This SDK's thin driver lives in conformance/driver.py (public API only) and known gaps are ratcheted with ticket links in conformance/failures.txt, against a commit-SHA-pinned HARNESS_REF. See AGENTS.md for how to run it and how to add a new integration.

License

MIT