raindrop-agno 0.0.7

Raindrop observability integration for Agno AI agents

raindrop-agno

Raindrop observability integration for 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

pip install raindrop-agno agno

Quick Start

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:

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

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
    project_id="support-prod",          # Optional: route events to a specific project (slug)
    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.

Projects

Route events to a specific project by passing its slug as project_id:

rd = 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.

User Identification

Use identify() to associate metadata with a user:

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

Signal Tracking

Track custom signals for an event:

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:

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:

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:

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

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