​
Installation

To use the Raindrop SDK, start by installing with pip: 
pip install raindrop-ai

​
Configuration

First, import the SDK and initialize it with your write key. You’ll see your write key when you log into app.raindrop.ai 
import os
import raindrop.analytics as raindrop

# Recommended: load from env var
raindrop.init(os.getenv("RAINDROP_WRITE_KEY") or "YOUR_WRITE_KEY")

# Optional: enable tracing integration for task/tool decorators
# raindrop.init("YOUR_WRITE_KEY", tracing_enabled=True)

​
Tracking AI Interactions

To track AI interactions, you can use the track_ai function. It takes the following parameters:
  • user_id (str): The unique identifier of the user.
  • event (str): The name of the AI event you want to track.
  • event_id (Optional[str]): A unique identifier for this specific event. If not provided, a UUID will be generated.
  • model (Optional[str]): The name of the AI model used.
  • input (Optional[str]): The input provided to the AI model. (Either this or output is required if AI data is logged)
  • output (Optional[str]): The output generated by the AI. (Either this or input is required if AI data is logged)
  • convo_id (Optional[str]): The conversation ID associated with the interaction. Helpful in AI apps with multiple conversations per user (e.g. ChatGPT).
  • properties (Optional[Dict[str, Any]]): Additional properties associated with the AI event.
  • timestamp (Optional[str]): An ISO 8601 formatted timestamp for the event. If not provided, the SDK generates a UTC timestamp.
  • attachments (Optional[List[Attachment]]): A list of attachments associated with the event. See the Attachments section below.
Example usage: 
raindrop.track_ai(
    user_id="user123",
    event="user_message",
    model="gpt_4",
    input="What is the weather like today?",  # this or output is required
    output="The weather is sunny and warm.",  # this or input is required
    convo_id="conv789",  # optional
    properties={
        "tool_call": "reasoning_engine",
        "system_prompt": "you are a helpful...",
        "experiment": "experiment_a",
    },
    attachments=[
        {
            "type": "text",
            "name": "Additional Info",
            "value": "A very long document",
            "role": "input",
        },
        {
            "type": "image",
            "value": "https://example.com/image.png",
            "role": "output",
        },
        {
            "type": "iframe",
            "name": "Generated UI",
            "value": "https://newui.generated.com",
            "role": "output",
        },
    ],
)

​
Attachments

Attachments allow you to include context from the user (e.g. an attached image), or stuff that the model outputted (whether that is an image, document, code, or even an entire web page). Each attachment is an object with the following properties:
  • type (string): The type of attachment. Can be “code”, “text”, “image”, or “iframe”.
  • name (optional string): A name for the attachment.
  • value (string): The content or URL of the attachment.
  • role (string): Either “input” or “output”, indicating whether the attachment is part of the user input or AI output.
  • language (optional string): For code attachments, specifies the programming language.
Example of different attachment types: 
attachments = [
    {
        "type": "code",
        "name": "Example Code",
        "value": "console.log('Hello, World!');",
        "role": "input",
        "language": "javascript",
    },
    {
        "type": "text",
        "name": "Additional Info",
        "value": "Some extra text",
        "role": "input",
    },
    {"type": "image", "value": "https://example.com/image.png", "role": "output"},
    {"type": "iframe", "value": "https://example.com/embed", "role": "output"},
]
Each event has a limit of 1 MB. Properties will be truncated for larger events. Contact us if you have custom requirements.

​
Identifying Users

To associate traits with users, you can use the identify function. It takes the following parameters:
  • user_id (str): The unique identifier of the user.
  • traits (Dict[str, Union[str, int, bool, float]]): The traits associated with the user.
Example usage: 
raindrop.identify(
    user_id="user123",
    traits={
        "name": "John Doe",
        "email": "john@example.com",
        "age": 30,
        "plan": "paid" #we recommend 'free', 'paid', 'trial'
    }
)

​
Partial Event Tracking (Interactions)

For multi-turn conversations or when event data arrives incrementally, you can use the begin() and Interaction object to send partial updates.

​
begin()

Starts or resumes an interaction and returns an Interaction helper object.
  • user_id (str): The user’s identifier.
  • event (str): The name of the event.
  • event_id (Optional[str]): A unique ID for the event. If not provided, one is generated.
  • properties (Optional[Dict[str, Any]]): Initial properties for the event.
  • input (Optional[str]): Initial input for the AI.
  • attachments (Optional[List[Attachment]]): Initial attachments.
  • convo_id (Optional[str]): Conversation ID.
interaction = raindrop.begin(
    user_id="user456",
    event="chatbot_session_started",
    input="Hello chatbot!"
)
# interaction.id contains the event_id

​
resume_interaction()

If you already have an event_id for an ongoing interaction, you can get an Interaction object: 
interaction = raindrop.resume_interaction(event_id="existing_event_id")

​
Interaction Object Methods

The Interaction object has the following methods to update the event:
  • interaction.set_input(text: str): Updates the AI input.
  • interaction.add_attachments(attachments: List[Attachment]): Adds more attachments.
  • interaction.set_properties(props: Dict[str, Any]): Merges new properties with existing ones.
  • interaction.set_property(key: str, value: str): Convenience for setting a single property.
  • interaction.finish(output: Optional[str] = None, **extra): Marks the interaction as complete.
    • output: The final AI output.
    • **extra: Any other top-level TrackAIEvent fields to update (e.g., properties, attachments).
The SDK automatically sends updates to the backend after a short period of inactivity or when finish() is called. Example usage: 
# Start an interaction
interaction = raindrop.begin(user_id="user789", event="code_generation", input="Write a python function for fibonacci")

# ... later, user adds more context
interaction.add_attachments([{"type": "text", "value": "It should be recursive", "role": "input"}])

# ... AI generates output
interaction.finish(output="def fib(n): if n <= 1: return n else: return fib(n-1) + fib(n-2)")

​
Tracking Signals

Signals are used to attach user feedback (such as thumbs down or thumbs up) or other labels to existing events. Use the track_signal function:
  • event_id (str): The ID of the event to attach the signal to.
  • name (str): Name of the signal (e.g., “thumbs_up”, “copied_code”).
  • signal_type (Literal[“default”, “feedback”, “edit”]): Type of signal. Defaults to "default".
    • For "feedback" signals, a "comment" string must be included in the properties.
    • For "edit" signals, an "after" string (representing the content after edit) must be included in the properties.
  • timestamp (Optional[str]): ISO 8601 formatted timestamp. Defaults to current UTC time.
  • properties (Optional[Dict[str, Any]]): Additional properties for the signal.
  • attachment_id (Optional[str]): ID of a specific attachment within the original event to associate this signal with.
  • comment (Optional[str]): Convenience parameter for feedback signals. If provided, it’s added to properties as {"comment": "your comment"}.
  • after (Optional[str]): Convenience parameter for edit signals. If provided, it’s added to properties as {"after": "new content"}.
Example usage: 
# Example: Tracking a thumbs-up signal
raindrop.track_signal(
    event_id="evt_abc123", # ID of the event being signaled
    name="thumbs_up",
    signal_type="default",
    sentiment="POSITIVE"
)

# Example: Tracking a thumbs-down signal
raindrop.track_signal(
    event_id="evt_abc123", # ID of the event being signaled
    name="thumbs_down",
    signal_type="default",
    sentiment="NEGATIVE"
)

# Example: Tracking feedback
raindrop.track_signal(
    event_id="evt_abc123",
    name="user_feedback",
    signal_type="feedback",
    comment="The AI's response was very helpful!"
)


### **Timestamp**

For functions like `track_ai` and `track_signal`, you can optionally provide a `timestamp` parameter (an ISO 8601 formatted string) if you need to specify a custom time for the event. If not provided, the SDK generates a UTC timestamp at the moment of the call.

### **Flushing Events**

The Raindrop SDK uses a buffering mechanism to efficiently send events in batches. The events are automatically flushed when the buffer reaches a certain size or after a specified timeout.

You can manually flush the events by calling the `flush` function. Make sure this happens before the process exits or you will lose events:

```python
raindrop.flush()

​
Shutting Down

To ensure all events are processed before your application exits, call the shutdown function: 
raindrop.shutdown()
This will also flush any pending partial events from interactions.

​
Error Handling

The SDK will retry a request up to 3 times. Failed requests will be logged, regardless of if debug_logs is true.

​
Configuration

The SDK has several configurable parameters:
  • max_queue_size: Maximum number of events to store in the buffer (default: 10_000)
  • upload_size: Number of events to send in a single API request (default: 10)
  • upload_interval: Time interval in seconds between automatic flushes (default: 1.0) You can modify these parameters if needed:
raindrop.max_queue_size = 20_000
raindrop.upload_size = 200
raindrop.upload_interval = 2.0

​
Debugging

If you want to enable debug logs to see the events being added to the buffer, you can use the set_debug_logs function: 
raindrop.set_debug_logs(True)
That’s it! You should be ready to go. Please let us know if you have any questions.

​
Tracing (Beta)

Tracing is currently in beta. We’d love your feedback as we continue to improve the experience. Email: founders@raindrop.ai
  1. Enable tracing by passing tracing_enabled=True to raindrop.init(...).
  2. Decorate your entry-point function with @raindrop.interaction.
  3. Decorate tool functions with @raindrop.tool.
The example below traces OpenAI tool calls. It enables tracing, decorates a tool, starts an interaction with begin(...), and finishes it later via resume_interaction(). 
import json
import os
from openai import OpenAI
import raindrop.analytics as raindrop

@raindrop.tool("get_current_weather")
def get_current_weather(location: str, unit: str = "celsius"):
    """Mock weather tool."""
    return {"location": location, "temperature": 22, "unit": unit}

def send_to_user(text: str) -> None:
    # Resume the current interaction from the tracing context and finish elsewhere
    raindrop.resume_interaction().finish(output=text)
    print(f"Sending to user: {text}")

@raindrop.interaction("weather_interaction")
def main() -> None:
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

    # Create an interaction for observability (begin → finish)
    interaction = raindrop.begin(
        user_id="user-001",
        event="weather_query",
        input="What's the weather in Boston, MA today?",
        convo_id="convo-weather-001",
    )

    messages = [
        {"role": "system", "content": "You are helpful. Use tools when needed."},
        {"role": "user", "content": "What's the weather in Boston, MA today?"},
    ]

    # Let the model request tool invocations if needed
    first = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=[{
            "type": "function",
            "function": {
                "name": "get_current_weather",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string"},
                        "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                    },
                    "required": ["location"],
                },
            },
        }],
        tool_choice="auto",
        temperature=0.2,
    )

    choice = first.choices[0]
    tool_calls = getattr(choice.message, "tool_calls", None)

    if tool_calls:
        messages.append({
            "role": "assistant",
            "content": getattr(choice.message, "content", None),
            "tool_calls": [{
                "id": tc.id,
                "type": "function",
                "function": {"name": tc.function.name, "arguments": tc.function.arguments},
            } for tc in tool_calls],
        })

        for tc in tool_calls:
            args = json.loads(tc.function.arguments or "{}")
            result = (
                get_current_weather(**args)
                if tc.function.name == "get_current_weather"
                else {"error": "unknown tool"}
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "name": tc.function.name,
                "content": json.dumps(result),
            })

        # Final model response after tools
        second = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, temperature=0.2
        )
        final_text = second.choices[0].message.content or ""
    else:
        final_text = choice.message.content or ""

    print("Assistant:\n", final_text)
    send_to_user(final_text)
    raindrop.flush()
    raindrop.shutdown()

if __name__ == "__main__":
    raindrop.init(os.getenv("RAINDROP_WRITE_KEY"), tracing_enabled=True)
    main()

​
Using the @raindrop.interaction() decorator (Beta)

You can wrap your flow with the @raindrop.interaction("name") decorator to ensure a tracing context exists, which allows resume_interaction() to find the current Interaction without passing an event_id: 
import os
import raindrop.analytics as raindrop

raindrop.init(os.getenv("RAINDROP_WRITE_KEY"), tracing_enabled=True)

def send_to_user():
    interaction = raindrop.resume_interaction()
    interaction.finish(output="It's sunny!")

@raindrop.interaction("weather_flow")
def run_weather_flow():
    interaction = raindrop.begin(user_id="user-001", event="weather_query", input="What's the weather?")
    send_to_user()
    

run_weather_flow()
  • resume_interaction() resolves the current Interaction by reading the active tracing context (current span). It will only find an existing Interaction when called under the same traced execution that called begin(...) (for example, inside a function decorated with @raindrop.task, @raindrop.tool, or @raindrop.interaction).
  • If your code runs outside that tracing context (separate thread/process, lost OpenTelemetry context, background job, etc.), pass the event ID explicitly: resume_interaction(event_id="...").
  • If neither a matching trace context nor an event_id is available, a new Interaction instance will be created.