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",                 # Required: events without a user id are not sent
    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")

Per-call properties and identity

An agent's run() / arun() has no channel for your own event data, so wrap the call in event_context. Its properties merge into every event captured inside the block, and user_id / convo_id replace the wrapper's own values for those events:

from raindrop_agno import event_context

with event_context(user_id="user-123", convo_id="conv-9", properties={"plan": "pro"}):
    rd.wrap(agent).run("What is the capital of France?")

Updating the last event

last_event_id is the event_id of the most recent event this wrapper captured. Use it to attach a signal, or reopen the event with resume_interaction() (defaults to last_event_id) and update it through the core SDK's Interaction:

interaction = rd.resume_interaction()
if interaction is not None:
    interaction.set_properties({"reviewed": True})
    interaction.finish()

Key clashes. If one of your properties has the same key as one the wrapper derives itself, your value wins, whichever update carries it (a debug-level log line names the key). The wrapper's own keys are ai.* (for example ai.usage.prompt_tokens), error/error.* and agno.*; avoid them unless you mean to replace the wrapper's value.

Property-only updates need raindrop-ai 0.0.77 or newer; older core versions drop an update that carries no input or output.

last_event_id is per wrapper instance; under concurrent calls it is whichever event was captured last, so pass event_id= when you need to correlate concurrent calls.

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.

These spans come from Agno's OpenTelemetry instrumentation, which the wrapper enables with tracing. When it isn't active (for example with a client= you built without the Agno instrument), the wrapper records each tool execution from the run's tools itself, with arguments, result, timing and errors, so tool calls are never missing. Input and output follow the core SDK's content setting (TRACELOOP_TRACE_CONTENT).

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)

Tool catalog capture (ai.prompt.tools)

Every model span of a wrapped run() / arun() records the agent's tool catalog as ai.prompt.tools: the entries of agent.tools, toolkits expanded to their functions, plain callables described the way Agno does (Function.from_callable), provider builtins passed as dicts (e.g. {"type": "web_search_preview"}) kept verbatim, and — for a wrapped Team — every member's tools as well. This is the catalog the agent can hand the model, not the exact per-request list: Agno resolves the model's tools per run and adds its own (knowledge search, memory/session-state, learning, skills and team-delegation tools, per-run client tools), and the OpenInference agno instrumentor emits no tool definitions, so an inferred catalog is marked with ai.prompt.tools.source = "agno.agent.tools" (an explicit tools override carries no source). Each tool is one JSON document in an OTLP string array: {"type":"function","name","description","inputSchema"} for function tools (the JSON Schema is recorded as given — properties, enums, required and per-field descriptions intact) or {"type":"provider-defined","id","name","args"} for provider-owned tools, which get no invented schema. A target without a visible tool list records no attribute; an explicit empty list records [].

Pass tools to record a different catalog instead (it replaces the inferred list; tools=[] records an empty catalog). The constructor value is the default for every wrapped target; wrap(agent, tools=[...]) replaces it for that target:

rd = RaindropAgno(api_key="your-write-key", tools=[{"type": "function", "name": "get_weather", "inputSchema": {...}}])
wrapped = rd.wrap(agent)                    # constructor default
other = rd.wrap(other_agent, tools=[])      # this agent: "no tools"

Tool definitions are prompt content: they follow the same switch as prompts, so TRACELOOP_TRACE_CONTENT=false never records ai.prompt.tools or ai.prompt.tools.source. Recording the catalog on Agno's model spans needs raindrop-ai>=0.0.71: that release adds ai.prompt.tools.source and is the first core that recognises the OpenInference llm.model_name spans Agno emits as model calls. On raindrop-ai==0.0.70 the list is bound but no span is stamped, so ai.prompt.tools stays absent. Streaming runs (run(stream=True)) whose model calls happen after run() returns are not covered.

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

User id

Every event needs a real user id: set user_id= on the wrapper or per call with event_context(user_id=...). Without one (missing, blank, or a placeholder such as "unknown" or "anonymous"), the event is not sent and the wrapper logs one warning; it never raises. Updating an event you resumed with resume_interaction(event_id=...) needs no user id, since ingest keeps the one it stored.

Model spans

With tracing on (the default), the wrapper records each agent run (one span for the run's model calls, with the run's total usage) as a span on the Raindrop client's tracing. The span carries:

• the model: gen_ai.provider.name (canonical name, such as openai or amazon-bedrock), gen_ai.request.model and gen_ai.response.model;
• the token usage the provider reports: gen_ai.usage.input_tokens, output_tokens, cache_read_input_tokens, cache_write_input_tokens and reasoning_tokens;
• the event it belongs to: ai.telemetry.metadata.raindrop.eventId and the traceloop.association.properties.* keys.

Token usage also stays on the event as ai.usage.* properties. When an OpenTelemetry instrumentation that sees the call already records it (the provider's own instrumentation, for example with auto_instrument=True on the client; for Agno runs, OpenInference's Agno instrumentation, which the default client enables), the wrapper adds no span of its own, so a call is never counted twice. An instrumentation for another provider or library does not suppress the wrapper's span.

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.

Application Git metadata

RaindropAgno(...) and create_raindrop_agno(...) accept the keyword-only app_git option. It defaults to True: explicit Raindrop Git environment or deployment context is applied immediately, and the base SDK may perform one bounded background local-Git lookup from the process working directory. Event capture, flush, and shutdown never wait for that lookup. Pass False to disable enrichment, or pass an AppGitOptions mapping with commit_sha, commit_dirty, branch, source_directory, detect_branch, and/or auto_detect. Automatic branch discovery remains opt-in through detect_branch=True (or RAINDROP_GIT_DETECT_BRANCH=true).

For an ordinary in-process application, the process working directory is treated as the application-under-test checkout. A remote, coding, workflow, or observer process must not rely on its own checkout: pass app_git=False, provide explicit revision values, or set source_directory to the actual application checkout. Canonical per-operation properties remain authoritative. When supplying client=, configure app_git while constructing that Raindrop client; the supplied client is authoritative and the wrapper's app_git argument does not reconfigure it.

Release order is deliberate: first publish the base SDK feature, then publish the wrapper feature release with its minimum dependency coordinated to that base release. The existing raindrop-ai lower bound remains compatible, but application Git metadata is unavailable on an older core and must not be claimed complete until the base is upgraded. Until coordination assigns a released version, the wrapper checks for an explicit base app_git parameter and omits the option when unsupported. Explicit non-default configuration is debug-logged and omitted. Unsupported app_git is determined by signature inspection before construction, not by retrying initialization after a TypeError; Git configuration adds no initialization attempts and does not change any existing framework-specific initialization fallback.

Standard options and flush results

When the wrapper builds its own client, it accepts the core SDK's standard options by the same names and passes them on:

raindrop = RaindropAgno(
    api_key="rk_...",
    user_id="user-123",
    redact_pii=True,        # local PII redaction of event input/output
    flush_timeout=5,        # seconds; deadline for flush() and shutdown()
    debug=True,             # also raises the core SDK's log verbosity
    max_queue_size=5_000,   # any other option `Raindrop(...)` declares
)

result = raindrop.flush()   # the core's FlushResult; safe to ignore

• An option the installed raindrop-ai does not support is ignored with a debug log, never an error. flush_timeout and the FlushResult return value need raindrop-ai 0.0.77 or later; on older releases flush() and shutdown() return None.
• With client= (a Raindrop you built), configure these options on that client; the wrapper ignores them (debug log).
• flush() and shutdown() never raise. The result is truthy when everything was delivered, and reports pending, dropped and timed_out.

Testing

cd packages/agno-python
pip install -e .
pip install pytest
pytest   # unit tests (no external services)

End-to-end behavior is verified by the cross-SDK conformance harness. This package ships a thin conformance driver at conformance/driver.py that maps the shared scenario corpus onto the wrapper's public API; known gaps are tracked as ticket-linked entries in conformance/failures.txt. The fault lane runs on every PR touching packages/*-python/** (.github/workflows/conformance-wrappers-python.yml) against a local capture server; the prod lane verifies delivery by reading back through the public Query API. The harness is pinned by commit SHA (HARNESS_REF). See the harness docs: HOW-IT-WORKS · AGENTS · README.