Metadata-Version: 2.4
Name: raindrop-agno
Version: 0.0.3
Summary: Raindrop observability integration for Agno AI agents
Project-URL: Homepage, https://raindrop.ai
Project-URL: Repository, https://github.com/invisible-tools/dawn/tree/main/packages/agno-python
Project-URL: Documentation, https://docs.raindrop.ai
Author-email: Raindrop AI <sdk@raindrop.ai>
License-Expression: MIT
License-File: LICENSE
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: agno>=2.5.0
Requires-Dist: openinference-instrumentation-agno>=0.1.28
Requires-Dist: raindrop-ai>=0.0.42
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == 'dev'
Description-Content-Type: text/markdown

# raindrop-agno

Raindrop observability integration for [Agno](https://github.com/agno-agi/agno) — a Python agent framework for building AI applications.

Wraps Agno `Agent`, `Team`, and `Workflow` objects to automatically capture runs and ship telemetry to Raindrop. When tracing is enabled, the Agno OpenInference instrumentor provides properly nested spans (agent → model → tool calls).

## Installation

```bash
pip install raindrop-agno agno
```

## Quick Start

```python
from raindrop_agno import RaindropAgno
from agno.agent import Agent
from agno.models.openai import OpenAIChat

rd = RaindropAgno(
    api_key="your-write-key",
    user_id="user-123",
    tracing_enabled=True,
)

agent = Agent(model=OpenAIChat(id="gpt-4o"))
wrapped = rd.wrap(agent)

result = wrapped.run("What is the capital of France?")
print(result.content)

rd.shutdown()
```

### Factory Function (Legacy)

The `create_raindrop_agno()` factory function is still available and returns a `RaindropAgno` instance. Dict-style access (`rd["wrap"]`, `rd["flush"]`, `rd["shutdown"]`) is supported for backward compatibility:

```python
from raindrop_agno import create_raindrop_agno

rd = create_raindrop_agno(api_key="rk_...", user_id="user-123")
wrapped = rd["wrap"](agent)
rd["shutdown"]()
```

## What Gets Traced

- **Agent runs** — input prompt, output text, model name
- **Token usage** — input_tokens, output_tokens, and cached_tokens (cache_read_tokens) from the Agno RunOutput metrics
- **Finish reason** — extracted from `model_provider_data` or last assistant message's `provider_data` when available
- **Tool calls** — nested spans with name, arguments, result, errors, duration (requires `tracing_enabled=True`)
- **Model calls** — LLM invocations as nested child spans (requires `tracing_enabled=True`)
- **Team delegation** — member agent calls appear as nested spans under the team run
- **Errors** — captured with error type/message metadata and re-raised to the caller
- **Async support** — both `run()` (sync) and `arun()` (async) are instrumented
- **Agno identity** — run_id, session_id, agent_name forwarded as properties

## Configuration

```python
rd = RaindropAgno(
    api_key="your-write-key",           # Your Raindrop API key (optional — omit to disable telemetry)
    user_id="user-123",                 # Optional: associate events with a user
    convo_id="convo-456",               # Optional: conversation/thread ID
    tracing_enabled=True,               # Enables nested trace spans (default: True)
    bypass_otel_for_tools=True,         # Bypass OTEL for tool spans (default: True)
    debug=True,                         # Optional: enable DEBUG-level logging
)
```

When `tracing_enabled=True`, the integration enables the Agno OpenInference instrumentor (`Instruments.AGNO`), which automatically creates properly nested OTEL spans for agent runs, model calls, and tool executions. This gives full trace visibility in the Raindrop dashboard.

When `debug=True`, the `raindrop_agno` logger is set to `DEBUG` level, which outputs detailed information about telemetry extraction and any issues encountered.

## User Identification

Use `identify()` to associate metadata with a user:

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

## Signal Tracking

Track custom signals for an event:

```python
rd.track_signal(event_id="evt-abc", name="thumbs_up")
```

## Tool Call Tracking

When your agent uses tools and tracing is enabled, each tool execution appears as a nested span in the trace view with input arguments, output, and duration:

```python
def get_stock_price(symbol: str) -> str:
    return "189.50"

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[get_stock_price],
)
wrapped = rd.wrap(agent)
result = wrapped.run("What is the price of AAPL?")
```

Tool call count is also captured in event properties as `agno.tool_calls_count`.

## Wrapping Agents and Workflows

The `wrap()` method works with Agno Agents and Workflows. For Teams, wrap each member agent individually:

```python
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(model=OpenAIChat(id="gpt-4o"))
wrapped_agent = rd.wrap(agent)
```

## Flushing and Shutdown

Always call `shutdown()` before your process exits to ensure all telemetry is shipped:

```python
rd.shutdown()  # flush + release resources
```

## Payload size bounds

Structured run inputs and structured `RunOutput` content (Pydantic models,
dicts, lists) are serialized with a hard 1,000,000-character budget and a
`...[truncated by raindrop]` marker. The bound is enforced *during*
serialization (cost proportional to the cap, not the payload), so a multi-MB
structured payload can't stall your event loop. Plain-string messages and
output text are capped by the Raindrop SDK's own per-field limit
(`max_text_field_chars`, raindrop-ai >= 0.0.51).

## Known Limitations

- **Streaming**: `run(stream=True)` does not produce events, but trace spans are still captured when `tracing_enabled=True`.
- **Multi-step agent runs**: The event captures the final result. Individual LLM and tool calls appear as nested trace spans when `tracing_enabled=True`.

## Testing

```bash
cd packages/agno-python
pip install -e .
pip install pytest
pytest
```
