How to build an analytics agent with Agno and Tinybird: Step-by-step

cd datasources && touch events.datasource

SCHEMA >
    `timestamp` DateTime,
    `session_id` String,
    `user_id` String,
    `event_type` String,
    `page_url` String,
    `referrer` String,
    `duration_ms` Int32,
    `country` String

ENGINE "MergeTree"
ENGINE_PARTITION_KEY "toYYYYMM(timestamp)"
ENGINE_SORTING_KEY "timestamp, session_id"

tb deploy

tb mock events --rows 100000 --prompt "Generate timestamps within the last 30 days. All pages should have a tinybird.co domain. Use a mix of referrers including Google, X/Twitter, LinkedIn, HackerNews, ChatGPT, and Perplexity. Use pageview, click, or form_submit as event types."

tb datasource append events fixtures/events.ndjson

DESCRIPTION >
    'Returns the top N pages by page_view count within a specific date range'
NODE top_pages_node
SQL >
    SELECT 
        page_url,
        COUNT() as visits
    FROM events
    WHERE timestamp >= {{DateTime(date_from, '2024-01-01')}}
      AND timestamp < {{DateTime(date_to, '2024-01-02')}}
      AND event_type = 'pageview'
    GROUP BY page_url
    ORDER BY visits DESC
    LIMIT {{Int32(N, 10)}}

tb create --prompt "Create an API endpoint to return the top pages by clicks. Include date_from and date_to parameters, as well as an N parameter which limits the number of results."

tb create --prompt "Create a materialized view to calculate session metrics based on session_id. The resulting data source should store session_id, referrer, utm parameters for the session, session duration, total pageviews, and arrays containing all pages viewed and all actions taken"

tb deploy

python -m venv venv
source venv/bin/activate
pip install -U agno python-dotenv

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.models.google import Gemini
from agno.tools.mcp import MCPTools
from dotenv import load_dotenv
import os
import asyncio

load_dotenv()

async def create_analytics_agent():
    # Choose your model (Gemini for larger context, Claude for reasoning)
    model = Gemini(
        id="gemini-2.5-flash",
        vertexai=True,
        project_id=os.getenv("GOOGLE_CLOUD_PROJECT"),
        location=os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
    )
    
    # Alternative: Use Claude for strong reasoning
    # model = Claude(id="claude-4-sonnet-20250514")
    
    agent = Agent(
        model=model,
        role="Analytics Data Analyst",
        description="""You are an expert data analyst with access to web analytics data. 
        You can explore data, identify trends, and provide actionable insights.""",
        instructions=[
            "Always provide specific metrics and timeframes in your analysis",
            "When exploring data, explain what the numbers mean for the business",
            "Include relevant context about data patterns and anomalies"
        ],
        markdown=True
    )
    
    return agent

async def main():
    agent = await create_analytics_agent()
    
    async:
        # Test the agent
        await agent.aprint_response(
            "Hi, I'm Cameron. How are you?",
            stream=True
        )

if __name__ == "__main__":
    asyncio.run(main())

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools
from agno.tools.slack import SlackTools
from agno.tools.resend import ResendTools
import os
import json
from datetime import datetime

async def create_advanced_analytics_agent():
    # Tinybird MCP configuration
    tinybird_api_key = os.getenv("TINYBIRD_TOKEN")
    server_url = f"https://mcp.tinybird.co?token={tinybird_api_key}"
    
    mcp_tools = MCPTools(
        transport="streamable-http",
        url=server_url,
        timeout_seconds=300
    )

    # Choose your model (Gemini for larger context, Claude for reasoning)
    model = Gemini(
        id="gemini-2.5-flash",
        vertexai=True,
        project_id=os.getenv("GOOGLE_CLOUD_PROJECT"),
        location=os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
    )
    
    # Alternative: Use Claude for strong reasoning
    # model = Claude(id="claude-4-sonnet-20250514")
    
    agent = Agent(
        model=model,
        tools=[mcp_tools],
        role="Senior Analytics Agent",
        name="DataExplorer",
        description="""You are a senior data analyst with deep expertise in web analytics, 
        user behavior analysis, and business intelligence. You have access to real-time 
        analytics data through Tinybird and can explore data to answer complex questions.""",
        instructions=[
            "Use the explore_data tool for complex analytical questions",
            "Always include timeframes and specific metrics in your analysis. Default to the last 24 hours unless a time frame is provided.", 
            "Provide business context and actionable recommendations.",
            "When detecting anomalies, investigate potential root causes.",
            "Format responses clearly with key findings highlighted."
        ],
        markdown=True,
        show_tool_calls=True,
        debug_mode=True
    )
    
    return agent, mcp_tools

# Usage example
async def main():
    agent, mcp_tools = await create_advanced_analytics_agent()
    
    async with mcp_tools:
        await agent.aprint_response(
            """Investigate any unusual traffic patterns in the last 24 hours. 
            Look for spikes in page views, unusual referrer patterns, or 
            geographic anomalies. Provide a summary with key metrics.""",
            stream=True
        )

# The agent can call your endpoints directly
await agent.aprint_response(
    "Use the top_pages endpoint to show me the most popular pages today"
)

# Or use natural language exploration
await agent.aprint_response(
    "Explore user session data to find patterns in user engagement by country"
)

SYSTEM_PROMPT = """
You are an expert data analyst for Tinybird metrics. You have MCP tools to get schemas, endpoints and data. The explore_data tool is an agent capable of exploring data autonomously.
"""

MISSION_PROMPT = """
You are an expert data analyst for web analytics. You have access to real-time data 
through Tinybird MCP tools including page views, user sessions, and behavioral data.

<role_definition>
Your role is to:
- Analyze web analytics data to provide actionable insights
- Investigate performance anomalies and traffic patterns  
- Monitor key business metrics and identify trends
- Provide clear, data-driven recommendations
</role_definition>

<data_access_rules>
- Use explore_data for complex analytical questions requiring data exploration
- Use specific endpoint tools when you know the exact data needed
- Always include timeframes in your analysis (default to last 24 hours if not specified)
- Provide both raw metrics and business context
</data_access_rules>

<response_formatting>
- Structure responses with clear sections: Summary, Key Findings, Metrics, Recommendations
- Use tables for comparative data
- Highlight significant changes or anomalies
- Include data confidence levels when relevant
</response_formatting>

Current date and time: {current_date}
"""

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools
from agno.tools.slack import SlackTools
from agno.tools.resend import ResendTools
import os
import json
from datetime import datetime
from prompts import SYSTEM_PROMPT

async def create_advanced_analytics_agent():
    # Tinybird MCP configuration
    tinybird_api_key = os.getenv("TINYBIRD_TOKEN")
    server_url = f"https://mcp.tinybird.co?token={tinybird_api_key}"
    
    mcp_tools = MCPTools(
        transport="streamable-http",
        url=server_url,
        timeout_seconds=300
    )

    # Choose your model (Gemini for larger context, Claude for reasoning)
    model = Gemini(
        id="gemini-2.5-flash",
        vertexai=True,
        project_id=os.getenv("GOOGLE_CLOUD_PROJECT"),
        location=os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
    )

    mission="""
    Your job is to analyze data in the Tinybird workspace to identify performance issues and make recommendations for how to fix them. Your goal is to effectively answer the user request.

    <exploration_instructions>
    - If list_service_datasources returns organization data sources, you must append "use organization service data sources" in the explore_data tool call, otherwise answer with an error message
    - You MUST include a time filter in every call to the explore_data tool if not provided by the user in the prompt
    - You MUST do one call to the explore_data tool per data source requested by the user
    - Do not ask follow up questions, do a best effort to answer the user request, if you make any assumptions, report them in the response.
    </exploration_instructions>

    <output_instructions>
    - Format output using Markdown
    - Use clearly structured headers, text formatting, and lists
    </output_instructions>
    """
    
    agent = Agent(
        model=model,
        tools=[mcp_tools],
        role="Senior Analytics Agent",
        name="PerformanceInvestigator",
        description=dedent(SYSTEM_PROMPT),
        instructions=mission
        markdown=True,
        show_tool_calls=True,
        debug_mode=True
    )
    
    return agent, mcp_tools

# Usage example
async def investigate_traffic_spike():
    agent, mcp_tools = await create_advanced_analytics_agent()
    
    async with mcp_tools:
        await agent.aprint_response(
            """Investigate any unusual traffic patterns in the last 24 hours. 
            Look for spikes in page views, unusual referrer patterns, or 
            geographic anomalies. Provide a summary with key metrics.""",
            stream=True
        )

Show me the top 5 pages by pageviews in the last 30 days

Which of those pages has the highest form submission conversion rate?

Which referrers get me the most traffic to the top converting page?

import asyncio
from agno.agent import Agent

async def cli_analytics_agent():
    agent, mcp_tools = await create_analytics_agent()
    
    print("🤖 Analytics Agent CLI")
    print("Ask questions about your data or type 'exit' to quit")
    print("-" * 50)
    
    async with mcp_tools:
        while True:
            try:
                user_input = input("\n💬 Your question: ").strip()
                if user_input.lower() == "exit":
                    print("👋 Goodbye!")
                    break
                
                await agent.aprint_response(
                    user_input,
                    stream=True,
                    user_id="analyst"
                )
            except KeyboardInterrupt:
                print("\n👋 Goodbye!")
                break
            except Exception as e:
                print(f"❌ Error: {e}")

# Run with: python analytics_agent.py
if __name__ == "__main__":
    asyncio.run(cli_analytics_agent())

from agno.tools.resend import ResendTools

async def email_analytics_report():
    agent = Agent(
        model=Claude(id="claude-4-sonnet-20250514"),
        tools=[
            MCPTools(transport="streamable-http", url=server_url),
            ResendTools(from_email="analytics@company.com")
        ],
        instructions=[
            "Generate comprehensive HTML email reports",
            "Include executive summary and key metrics",
            "Use tables and visual formatting"
        ]
    )
    
    async with agent.tools[0]:  # MCP tools
        response = await agent.arun(
            """Generate a weekly analytics report with:
            1. Traffic overview and trends
            2. Top performing content
            3. User engagement metrics
            4. Key insights and recommendations
            
            Send this as an HTML email to team@company.com with subject 
            'Weekly Analytics Report - [Current Date]'""",
            user_id="system"
        )
        
        return response

# Schedule with cron or GitHub Actions
# 0 9 * * 1 cd /path/to/agent && python -c "import asyncio; asyncio.run(email_analytics_report())"

from agno.tools.slack import SlackTools

async def slack_analytics_bot():
    agent = Agent(
        model=Claude(id="claude-4-sonnet-20250514"),
        tools=[
            MCPTools(transport="streamable-http", url=server_url),
            SlackTools(token=os.getenv("SLACK_BOT_TOKEN"))
        ],
        instructions=[
            "Respond in Slack-friendly format with clear sections",
            "Use code blocks for data tables",
            "Include relevant metrics in thread responses",
            "Tag relevant team members for critical alerts"
        ]
    )
    
    # Alert example
    await agent.arun(
        """Check for any traffic anomalies in the last hour.
        If significant issues are found, send an alert to #analytics-alerts 
        channel with details and suggested actions."""
    )
    
    # Interactive query handling
    await agent.arun(
        """A user in #general asked: 'How is our mobile traffic trending this week?'
        Analyze mobile traffic patterns and respond in the thread with insights."""
    )

# Webhook handler for Slack events
async def handle_slack_message(event):
    if "analytics" in event.get("text", "").lower():
        await slack_analytics_bot()

from agno.agent import Agent

async def multi_agent_analysis():
    # Data Explorer Agent
    explorer = Agent(
        model=model,
        tools=[mcp_tools],
        role="Data Explorer",
        instructions=["Focus on data discovery and pattern identification"]
    )
    
    # Insights Analyst Agent  
    analyst = Agent(
        model=model,
        tools=[mcp_tools],
        role="Insights Analyst",
        instructions=["Focus on business interpretation and recommendations"]
    )
    
    # Report Writer Agent
    writer = Agent(
        model=model,
        tools=[slack_tools, email_tools],
        role="Report Writer", 
        instructions=["Create clear, actionable reports for stakeholders"]
    )
    
    async with mcp_tools:
        # Step 1: Explore data
        exploration = await explorer.arun(
            "Identify interesting patterns in user behavior this week"
        )
        
        # Step 2: Analyze insights
        analysis = await analyst.arun(
            f"Based on this data exploration: {exploration.content}, "
            "what are the key business insights and recommendations?"
        )
        
        # Step 3: Create and distribute report
        await writer.arun(
            f"Create a summary report based on this analysis: {analysis.content}. "
            "Send to #analytics channel and email to stakeholders@company.com"
        )

# Schedule for comprehensive weekly analysis

from agno.memory.v2.db.postgres import PostgresMemoryDb
from agno.memory.v2.memory import Memory
from agno.storage.postgres import PostgresStorage

async def create_persistent_agent():
    # Configure memory
    memory = Memory(
        model=Claude(id="claude-4-sonnet-20250514"),
        db=PostgresMemoryDb(
            table_name="analytics_memories",
            db_url=os.getenv("DATABASE_URL")
        )
    )
    
    # Configure storage for session history
    storage = PostgresStorage(
        table_name="analytics_sessions",
        db_url=os.getenv("DATABASE_URL")
    )
    
    agent = Agent(
        model=model,
        tools=[mcp_tools],
        memory=memory,
        storage=storage,
        enable_agentic_memory=True,
        add_history_to_messages=True,
        num_history_runs=10,
        instructions=[
            "Remember insights from previous analyses",
            "Build context over time about data patterns",
            "Reference historical findings when relevant"
        ]
    )
    
    return agent

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def resilient_query(agent, query, user_id="system"):
    try:
        async with agent.tools[0]:  # MCP tools context
            response = await agent.arun(query, user_id=user_id)
            logger.info(f"Query successful: {query[:50]}...")
            return response
    except Exception as e:
        logger.error(f"Query failed: {query[:50]}... - Error: {e}")
        raise

async def safe_analytics_run(query):
    try:
        agent, mcp_tools = await create_analytics_agent()
        return await resilient_query(agent, query)
    except Exception as e:
        logger.error(f"Analytics run failed completely: {e}")
        return {"error": str(e), "success": False}

tb --cloud deploy

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "analytics_agent.py"]

# docker-compose.yml
version: '3.8'
services:
  analytics-agent:
    build: .
    environment:
      - TINYBIRD_TOKEN=${TINYBIRD_TOKEN}
      - SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN}
      - DATABASE_URL=${DATABASE_URL}
    restart: unless-stopped
    
  postgres:
    image: postgres:15
    environment:
      - POSTGRES_DB=analytics_agent
      - POSTGRES_USER=agent
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data: