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

# Agno (Beta)

> Automatic tracing for Agno agents, teams, and workflows with Raindrop

## Installation

```bash theme={null}
pip install raindrop-agno agno
```

## Quick Start

```python theme={null}
from raindrop_agno import RaindropAgno
from agno.agent import Agent
from agno.models.openai import OpenAIChat

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

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

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

raindrop.shutdown()
```

## What Gets Traced

The Agno integration automatically captures:

* **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, and duration
* **Model calls** — LLM invocations as nested child spans under the agent run
* **Team delegation** — member agent calls appear as nested spans under the team
* **Errors** — captured (with error message) 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 theme={null}
raindrop = 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
    project_id="support-prod",          # Optional: route events to a specific project (slug)
    tracing_enabled=True,               # Enables the Agno OpenInference instrumentor (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 with a hierarchical span tree:

```
Agent run
├── Model call (gpt-4o-mini) — 1.2s
├── Tool: get_stock_price — 0.1ms
├── Model call (gpt-4o-mini) — 0.8s
└── Tool: calculate — 0.1ms
```

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

## Projects

Route events to a specific [project](/platform/projects) by passing its slug as `project_id`:

```python theme={null}
raindrop = RaindropAgno(
    api_key="your-write-key",
    project_id="support-prod",
)
```

`project_id` sets the `X-Raindrop-Project-Id` header on every event. Omit it (or pass `"default"`) to use your org's default **Production** project, which is the existing behavior. The same option is accepted by the `create_raindrop_agno(...)` factory. Invalid slugs are ignored with a warning and no header is sent.

### Multiple projects in one process

Available in `raindrop-ai>=0.0.56`. When one service runs several Agno agents
that should report to **different** projects, create one `RaindropAgno` wrapper
per project. Each wrapper owns its own `raindrop.Raindrop` client, so the two
route independently — there is no shared module-level state:

```python theme={null}
from raindrop_agno import RaindropAgno

# One long-lived wrapper per project, created at startup and reused.
rd_support = RaindropAgno(api_key="rk_...", project_id="support-prod")
rd_billing = RaindropAgno(api_key="rk_...", project_id="billing-prod")

support_agent = rd_support.wrap(support_agent)   # -> support-prod
billing_agent = rd_billing.wrap(billing_agent)   # -> billing-prod

rd_support.flush()
rd_billing.flush()
```

Each wrapper owns its configuration and delivery pipeline, so agents handled by
different wrappers route independently. To share a single client across wrappers
(or with the module-level API), construct a `raindrop.Raindrop` yourself and pass
it via `client=`:

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

client = Raindrop(api_key="rk_...", project_id="support-prod")
raindrop = RaindropAgno(client=client)
```

## Wrapping Agents

Use `raindrop.wrap()` to instrument an agent. The wrapped agent behaves identically
to the original — `run()` and `arun()` return the same `RunOutput` objects:

```python theme={null}
from agno.agent import Agent
from agno.models.openai import OpenAIChat

agent = Agent(
    name="Stock Price Agent",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[get_stock_price],
    instructions="Answer questions in the style of a stock analyst.",
)
wrapped = raindrop.wrap(agent)

result = wrapped.run("What is the current price of AAPL?")
print(result.content)
```

## Tool Call Tracking

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

```python theme={null}
def get_stock_price(symbol: str) -> str:
    """Get the current stock price."""
    return "189.50"

agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[get_stock_price],
)
wrapped = raindrop.wrap(agent)

result = wrapped.run("What is the price of AAPL?")
# Tool calls appear as nested spans with:
#   - tool name, input arguments, output result
#   - errors (if any), duration
```

The number of tool calls is also included in event properties as
`agno.tool_calls_count`.

## Teams

For Agno Teams, wrap each member agent. The OpenInference instrumentor
automatically traces team coordination, member delegation, and nested agent calls:

```python theme={null}
from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat

researcher = Agent(name="Researcher", model=OpenAIChat(id="gpt-4o-mini"),
    tools=[search_tool], role="Research analyst")
writer = Agent(name="Writer", model=OpenAIChat(id="gpt-4o-mini"),
    role="Content writer")

raindrop.wrap(researcher)
raindrop.wrap(writer)

team = Team(name="ResearchTeam", mode="coordinate",
    model=OpenAIChat(id="gpt-4o-mini"),
    members=[researcher, writer])

result = team.run("Write a report on AI trends")
```

## Wrapping Workflows

The same `wrap()` function works with Agno Workflows:

```python theme={null}
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow import Workflow

class ResearchWorkflow(Workflow):
    researcher: Agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        instructions="Research the given topic thoroughly.",
    )
    writer: Agent = Agent(
        model=OpenAIChat(id="gpt-4o"),
        instructions="Write a summary based on the research.",
    )

workflow = ResearchWorkflow()
raindrop.wrap(workflow.researcher)
raindrop.wrap(workflow.writer)
wrapped = raindrop.wrap(workflow)

result = wrapped.run("Explain quantum computing")
print(result.content)
```

## Async Usage

The wrapper supports both sync and async agent runs:

```python theme={null}
import asyncio

async def main():
    result = await wrapped.arun("What is quantum computing?")
    print(result.content)
    raindrop.shutdown()

asyncio.run(main())
```

## Captured Properties

Each event includes the following properties when available:

| Property                     | Description                                          |
| ---------------------------- | ---------------------------------------------------- |
| `ai.usage.prompt_tokens`     | Input token count                                    |
| `ai.usage.completion_tokens` | Output token count                                   |
| `ai.usage.cached_tokens`     | Cached (cache-read) token count (provider-dependent) |
| `ai.model`                   | Model name                                           |
| `agno.finish_reason`         | Model finish/stop reason (e.g. `"stop"`, `"length"`) |
| `agno.run_id`                | Agno run identifier                                  |
| `agno.session_id`            | Agno session identifier                              |
| `agno.agent_id`              | Agno agent identifier                                |
| `agno.agent_name`            | Agent name                                           |
| `agno.workflow_id`           | Workflow identifier (if applicable)                  |
| `agno.tool_calls_count`      | Number of tool calls in the run                      |

## Flushing and Shutdown

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

```python theme={null}
raindrop.shutdown()  # flush + release resources
```

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

<Tip>
  The `RaindropAgno` class also provides `identify()` and `track_signal()` methods:

  ```python theme={null}
  raindrop.identify("user-123", traits={"plan": "pro", "company": "Acme"})
  raindrop.track_signal(
      event_id="evt-abc",
      name="thumbs_up",
      signal_type="feedback",
      sentiment="POSITIVE",
      comment="Great response!",
  )
  ```
</Tip>