Agno (Beta)

pip install raindrop-agno agno

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

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
)

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

raindrop = RaindropAgno(
    api_key="your-write-key",
    project_id="support-prod",
)

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

from raindrop import Raindrop

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

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)

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

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

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)

import asyncio

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

asyncio.run(main())

raindrop.shutdown()  # flush + release resources