> ## 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.

> The Raindrop SDK allows you to track user events and AI interactions in your app. This documentation provides a brief overview of how to use the Python SDK.

# Python

### **Installation**

Install with your package manager of choice:

<CodeGroup>
  ```sh pip theme={null}
  pip install raindrop-ai
  ```

  ```sh uv theme={null}
  uv add raindrop-ai
  ```

  ```sh poetry theme={null}
  poetry add raindrop-ai
  ```
</CodeGroup>

### **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](https://app.raindrop.ai).

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

# Optional: bypass OTEL exporters for interaction.track_tool() spans only
# raindrop.init("YOUR_WRITE_KEY", tracing_enabled=True, bypass_otel_for_tools=True)
```

#### Auto-instrumentation

When `tracing_enabled=True`, the SDK automatically instruments detected LLM
client libraries (OpenAI, Anthropic, Bedrock, etc.) so their calls appear as
spans in your traces.

You can use the `Instruments` enum to control which libraries are instrumented:

```python theme={null}
from raindrop.analytics import Instruments

# Only auto-instrument Anthropic
raindrop.init(
    os.getenv("RAINDROP_WRITE_KEY"),
    tracing_enabled=True,
    instruments={Instruments.ANTHROPIC},
)

# Auto-instrument everything except OpenAI
raindrop.init(
    os.getenv("RAINDROP_WRITE_KEY"),
    tracing_enabled=True,
    block_instruments={Instruments.OPENAI},
)

# Disable all auto-instrumentation (manual tracing still works)
raindrop.init(
    os.getenv("RAINDROP_WRITE_KEY"),
    tracing_enabled=True,
    auto_instrument=False,
)
```

<Tip>
  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.
</Tip>

### **Projects**

Pass `project_id` to `raindrop.init(...)` to scope events to a specific [project](/docs/platform/projects). This sets the `X-Raindrop-Project-Id` header on each request.

```python theme={null}
raindrop.init("YOUR_WRITE_KEY", project_id="support-prod")
```

Omitting `project_id` (or passing `"default"`) sends to the default **Production** project, which is the existing behavior. See [Projects](/docs/platform/projects) for isolation, archival, and the full behavior table.

#### Multiple projects in one process

Available in `raindrop-ai>=0.0.56`. When one service runs several agents that
should report to **different** projects — e.g. a FastAPI app serving a support
agent and a billing agent — create one `Raindrop` client per project instead
of using the module-level `init()`. This is the same instance-based shape as
the [TypeScript](/docs/sdk/typescript), [Go](/docs/sdk/go), [Rust](/docs/sdk/rust), and
[Java](/docs/sdk/java) SDKs:

```python theme={null}
from raindrop import Raindrop

# One long-lived client per project, created at startup and reused.
rd_support = Raindrop(api_key=KEY, project_id="support-prod", tracing_enabled=True)
rd_billing = Raindrop(api_key=KEY, project_id="billing-prod", tracing_enabled=True)

# Same method names as the module API — the instance is the routing decision.
interaction = rd_billing.begin(user_id="u1", event="billing-chat", input="...")
interaction.track_tool(name="invoice_lookup", input={...}, output={...})
interaction.finish(output="...")            # -> billing-prod

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

Each client owns its configuration and delivery pipeline, so concurrent
requests handled by different agents route independently — there is no shared
mutable project state to race on. The module-level API keeps working
unchanged; it is simply the default, process-wide client.

Per-instance configuration and lifecycle: `rd.flush()` / `rd.shutdown()`
drain that client only (an `atexit` hook drains every client at process
exit under one shared deadline); `Raindrop(max_text_field_chars=...)` caps
that client's payload fields without affecting other clients (module-level
`init(max_text_field_chars=...)` still sets the process-wide default,
1,000,000 characters per field); and `bypass_otel_for_tools=True` ships
`interaction.track_tool()` spans on the client's own connection, so they
route per instance with no shared-pipeline caveats.

**Tracing across multiple projects.** OpenTelemetry auto-instrumentation is
shared per process: the first client constructed with `tracing_enabled=True`
initializes it, and later clients share the pipeline. `begin()` scopes spans
produced in the current request/task to its client's project until the
matching `finish()` (an interaction that is never finished releases its
scope once garbage-collected). For LLM calls made *outside* an interaction,
scope them explicitly with `as_current()`:

```python theme={null}
with rd_billing.as_current():
    openai_client.chat.completions.create(...)  # spans -> billing-prod
```

<Warning>
  One tracing API key per process: all auto-instrumented spans export under the
  first tracing-enabled client's key. Clients created with a different
  `api_key` (a different workspace) get a warning, and from then on only spans
  scoped to a client via `begin()`/`as_current()` are exported — the other
  workspace's spans and any unscoped spans are dropped at export rather than
  delivered to the wrong workspace. Run a second workspace's tracing in its
  own process. Single-key processes (the normal case) are unaffected. Manual
  events (`track_ai`, `begin`/`finish`, signals, `identify`) always ship with
  each client's own key and are unaffected.
</Warning>

### **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:**

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

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

<Warning>
  Each event has a limit of 1 MB. Properties will be truncated for larger events. [Contact us](mailto:founders@raindrop.ai) if you have custom requirements.
</Warning>

### **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:**

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

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

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

```python theme={null}
# 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.
* `sentiment` (Optional\[Literal\["POSITIVE", "NEGATIVE"]]): Optional sentiment indicating whether the signal is positive or negative.
* `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:**

```python theme={null}
# 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 theme={null}
raindrop.flush()
```

### Shutting Down

To ensure all events are processed before your application exits, call the shutdown function:

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

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

```python theme={null}
raindrop.set_debug_logs(True)
```

That's it! You should be ready to go. Please let us know if you have any questions.

### **Tracing**

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

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

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`:

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

<Note>
  <strong>Resume caveats:</strong>

  * `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.
</Note>

#### Using the `tool_span` and `task_span` context managers

When decorators aren't feasible (e.g., dynamic tool selection, complex control flow), you can use context managers for fine-grained tracing control.

**Requirements:**

1. Initialize with `tracing_enabled=True`
2. Decorate your entry-point function with `@raindrop.interaction` to establish a tracing context
3. Use `with raindrop.tool_span(...)` or `with raindrop.task_span(...)` for traced blocks

Context managers support both synchronous and asynchronous code. Spans automatically inherit the current trace context and no-op when tracing is disabled.

**Available methods on the span object:**

* `record_input(data)`: Record input data for the span
* `record_output(data)`: Record output data for the span
* `set_properties(props)`: Set custom properties on the span

**Example:**

```python theme={null}
import asyncio
import os
import raindrop.analytics as raindrop
import time

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

@raindrop.task("main_task")
async def main():
    interaction = raindrop.begin(
        user_id="user789",
        event="multi_tool_query",
        input="Find and summarize docs for the search API"
    )

    # Synchronous tool call: web_search
    with raindrop.tool_span("web_search", version=1) as tool:
        tool.record_input({"query": "search API documentation"})
        time.sleep(0.1)  # Simulate API call
        search_results = ["https://docs.example.com/api", "https://blog.example.com/tutorial"]
        tool.set_properties({"results_count": len(search_results)})
        tool.record_output({"urls": search_results})

    # Async tool call: document_reranker
    async with raindrop.tool_span("document_reranker") as tool:
        tool.record_input({
            "documents": search_results, 
            "query": "search API usage"
        })
        await asyncio.sleep(0.05)  # Simulate async reranking
        best_doc = search_results[0]
        tool.record_output({"top_document": best_doc})

    # Nested task span for summarization
    with raindrop.task_span("summarization") as task:
        task.record_input({"document": best_doc})
        summary = "The search API allows filtering by..."
        task.record_output({"summary": summary})

    interaction.finish(output=summary)
    raindrop.shutdown()

if __name__ == "__main__":
    asyncio.run(main())
```

#### Using `interaction.start_span()` for Manual Spans

When the span lifecycle doesn't fit within a `with` block (e.g., the span starts in one function and ends in another), use `interaction.start_span()` to create a `ManualSpan` with an explicit `.end()` call.

```python theme={null}
span = interaction.start_span(kind="tool", name="my_tool")
```

**Parameters:**

* `kind` (Literal\["task", "tool"]): The type of span.
* `name` (str): Name of the span.
* `version` (Optional\[int]): Version number for the span.

**`ManualSpan` methods:**

* `record_input(data)`: Record input data for the span.
* `record_output(data)`: Record output data for the span.
* `set_properties(props)`: Set custom properties on the span.
* `end(error=None)`: End the span. Pass an exception to mark it as failed.

**`ManualSpan` properties:**

* `event_id`: The interaction's event\_id.

The span automatically inherits association properties (`event_id`, `user_id`, `event`, `convo_id`) from the interaction.

**Example:**

```python theme={null}
import os
import raindrop.analytics as raindrop

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

@raindrop.interaction("my_workflow")
def main():
    interaction = raindrop.begin(user_id="user123", event="process_data", input="...")

    # Start a span
    span = interaction.start_span(kind="tool", name="external_api")
    span.record_input({"query": "example"})

    try:
        result = call_external_api()
        span.record_output(result)
        span.end()
    except Exception as e:
        span.end(error=e)

    interaction.finish(output="Done")
    raindrop.shutdown()
```

#### Retroactive Tool Logging with `interaction.track_tool()`

Use `track_tool` to log a tool call after it has completed:

If your OTEL exporter pipeline is unavailable or flaky, set
`bypass_otel_for_tools=True` in `raindrop.init(...)` to ship only
`interaction.track_tool()` spans directly to `POST /v1/traces`.

```python theme={null}
interaction = raindrop.begin(
    user_id="user123",
    event="agent_run",
    input="Search for weather data",
)

# Log a completed tool call
interaction.track_tool(
    name="web_search",
    input={"query": "weather in NYC"},
    output={"results": ["Sunny, 72°F", "Clear skies"]},
    duration_ms=150,
    properties={"engine": "google"},
)

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

interaction.finish(output="Weather search complete")
```

| Parameter     | Type                       | Description                                             |
| ------------- | -------------------------- | ------------------------------------------------------- |
| `name`        | `str`                      | Name of the tool                                        |
| `input`       | `Any`                      | Input passed to the tool                                |
| `output`      | `Any`                      | Output returned by the tool                             |
| `duration_ms` | `float \| int`             | Duration in milliseconds                                |
| `start_time`  | `datetime \| int \| float` | When the tool started (defaults to `now - duration_ms`) |
| `error`       | `BaseException \| str`     | Error if the tool failed                                |
| `properties`  | `Dict[str, Any]`           | Additional metadata                                     |
| `version`     | `int`                      | Optional version number for the tool                    |

## Detached sub-agents (Beta)

A detached sub-agent runs asynchronously in another process, worker, or machine.
Unlike `resume_interaction()`, which joins the parent's event, it reports as its
**own event**. The parent mints the child event and sends a hand-off carrier; the
worker resumes that event and reports its result.

<Info>
  **Beta.** The detached sub-agent API surface may change in a minor release.
  Requires `raindrop-ai>=0.0.65`, the first published version containing these
  symbols.
</Info>

Python does not have a separate `withSubagentRun()` helper. The idiomatic form
is the `SubagentRun` context manager shown below.

<Steps>
  <Step title="Dispatch from the parent">
    Each process creates its own `Raindrop` instance. The parent allocates the
    child event before sending the job:

    ```python theme={null}
    # Parent process
    import os
    import requests
    from raindrop import Raindrop

    rd = Raindrop(
        api_key=os.environ["RAINDROP_WRITE_KEY"],
        tracing_enabled=True,
    )

    interaction = rd.begin(
        user_id="user_123",
        event="parent_turn",
        convo_id="conversation_456",
        input="Find the relevant documents",
    )

    dispatch = interaction.subagent(
        name="researcher",
        input={"task": "Find the relevant documents"},
    )
    ```
  </Step>

  <Step title="Send the hand-off">
    `subagent()` records telemetry and returns a dispatch object; it does not
    transport the job. Send the job and the carrier separately:

    ```python theme={null}
    job = {
        "task": "Find the relevant documents",
    }

    requests.post(worker_url, json=job, headers=dispatch.headers)

    interaction.finish(
        output=f"Dispatched {dispatch.name} as {dispatch.child_event_id}",
    )
    rd.flush()
    rd.shutdown()
    ```

    Send **all** of `dispatch.headers`, not only `traceparent`. The carrier
    includes `traceparent`, `baggage`, and `x-raindrop-handoff`; cherry-picking
    only `traceparent` silently unlinks the child. `dispatch.langsmith_headers`
    is also available when the worker uses LangSmith-style instrumentation.

    The child event ID is allocated before dispatch, so the link is resolvable
    while the job is queued. The dispatch span is named `launch_subagent` by
    default; pass `tool_name` to choose another name.
  </Step>

  <Step title="Resume in the worker">
    The carrier travels with the job. Use the context manager as the worker's
    entry point:

    ```python theme={null}
    # Worker process
    import os
    from raindrop import Raindrop

    rd = Raindrop(
        api_key=os.environ["RAINDROP_WRITE_KEY"],
        tracing_enabled=True,
    )

    try:
        with rd.resume_subagent(headers=request.headers) as run:
            answer = do_the_work()
            run.finish(output=answer)
    finally:
        rd.flush()
        rd.shutdown()
    ```

    `run.interaction` is the child's ordinary interaction object. Every span
    emitted while the run is open carries the reverse reference to the parent
    dispatch. If the worker exits the block without reporting an outcome, the
    context manager closes the run with an abort outcome instead of leaving the
    parent stuck in `queued`. If the body raises, it reports an abort and
    re-raises the exception. A cancellation exception is reported as cancelled
    rather than failed.
  </Step>

  <Step title="Report the outcome">
    Use the method that matches the worker's result:

    ```python theme={null}
    run.finish(output=answer)                 # successful result
    run.fail(error)                           # errored child span and abort output
    run.cancel("user aborted the request")   # cancelled terminal event
    ```

    `finish()` reports the child's output. `fail()` marks the child's telemetry
    as errored and synthesizes output from the reason. `cancel()` writes
    `raindrop.handoff.terminal="cancelled"` on the child event without marking
    the run as failed. A run that reports no outcome is what the context manager
    rescues on exit.
  </Step>
</Steps>

<Warning>
  Only accept carriers inside your own trust boundary. A carrier from an
  untrusted caller can attribute the child to another tenant's event.
</Warning>

The carrier is the same across Raindrop SDKs, so a hand-off can cross languages:
a Python parent can dispatch to a TypeScript worker, and the reverse also works.
The worker resumes the parent-minted child event either way.