Agents code examples

Here are some of examples of simple agents using Tinybird's MCP Server:

  • They are based on the web analytics project but you can adapt it to your own project by using your TINYBIRD_TOKEN.
  • Model and libraries set up (such as API keys and other environmenta variables) is omitted

Building an agent? Want to know which LLM generates best SQL queries? Explore the results in the LLM Benchmark.

Basic Query Execution with Pydantic AI

import os
from dotenv import load_dotenv
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
import asyncio

load_dotenv()

tinybird_token = os.getenv('TINYBIRD_TOKEN')

async def main():
    tinybird = MCPServerStreamableHTTP(
        f"https://mcp.tinybird.co?token={tinybird_token}",
        timeout=120,
    )

    agent = Agent(
        model="anthropic:claude-4-opus-20250514",  # use your favourite model
        mcp_servers=[tinybird],
    )

    async with agent.run_mcp_servers():
        result = await agent.run("top 5 pages with more visits in the last 24 hours")
        print(result.output)


asyncio.run(main())

Basic Query Execution with Agno

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

import asyncio
import os

tinybird_api_key = os.getenv("TINYBIRD_TOKEN")

async def main():
    async with MCPTools(
        transport="streamable-http",
        url=f"https://mcp.tinybird.co?token={tinybird_api_key}",
        timeout_seconds=120) as mcp_tools:
        agent = Agent(model=Claude(id="claude-4-opus-20250514"), tools=[mcp_tools])  # use your favourite model
        await agent.aprint_response("top 5 pages with more visits in the last 24 hours", stream=True)

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

Basic Query Execution with Vercel AI SDK

import { anthropic } from "@ai-sdk/anthropic";
import {
  generateText,
  experimental_createMCPClient as createMCPClient,
  type Message,
} from "ai";
import {
  StreamableHTTPClientTransport,
} from "@modelcontextprotocol/sdk/client/streamableHttp";
import * as dotenv from 'dotenv';

dotenv.config();

async function main() {
  const messages: Message[] = [{
    id: "1",
    role: "user",
    content: "top 5 pages with more visits in the last 24 hours"
  }];

  const url = new URL(
    `https://mcp.tinybird.co?token=${process.env.TINYBIRD_TOKEN}`
  );

  const mcpClient = await createMCPClient({
    transport: new StreamableHTTPClientTransport(url, {
      sessionId: "session_123",
    }),
  });

  const tbTools = await mcpClient.tools();

  const result = await generateText({
    model: anthropic("claude-3-7-sonnet-20250219"),  // use your favourite model
    messages,
    maxSteps: 5,
    tools: {...tbTools},
    system: "use the explore_data tool"
  });

  console.log(result.text);
}

main();

Advanced Analytics with OpenAI Agents SDK

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def analyze_user_behavior():
    try:
        server = MCPServerStreamableHttp(
            name="tinybird",
            params={
                "url": "https://mcp.tinybird.co?token=TINYBIRD_TOKEN",
            },
        )

        async with server:
            agent = Agent(
                name="user_behavior_analyst", 
                model="YOUR_FAVOURITE_MODEL", # e.g. "gpt4",
                mcp_servers=[server],
                instructions="""
                You are a data analyst. When analyzing user behavior:
                1. First list available endpoints to understand what data is available
                2. Use appropriate endpoints or execute_query for analysis
                3. Provide insights with specific numbers and trends
                4. Suggest actionable recommendations
                """
            )
            
            result = await Runner.run(
                agent,
                input="""
                Analyze our user engagement patterns:
                1. What are the current weekly active user trends?
                2. Which features are most popular?
                3. Are there any concerning drops in engagement?
                """
            )
            print("Engagement Analysis:", result.final_output)
            
    except Exception as e:
        print(f"Analysis failed: {e}")

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

Real-time Dashboard Assistant

import asyncio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

async def dashboard_assistant():
    server = MCPServerStdio(
        command="npx",
        args=["-y", "mcp-remote", "https://mcp.tinybird.co?token=TINYBIRD_TOKEN"],
    )

    agent = Agent(
        name="dashboard_assistant",
        model=MODEL, # e.g. "claude-3-5-sonnet-20241022",
        mcp_servers=[server]
    )

    async with agent.run_mcp_servers():
        while True:
            try:
                user_question = input("\nAsk about your data (or 'quit' to exit): ")
                if user_question.lower() == 'quit':
                    break
                
                result = await agent.run(user_question)
                print(f"Assistant: {result.output}")
                
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"Error processing question: {e}")

if __name__ == "__main__":
    asyncio.run(dashboard_assistant())
Updated