Agents code examples¶

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')
SYSTEM_PROMPT = "YOUR SYSTEM PROMPT"

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 favorite model
        mcp_servers=[tinybird],
        system_prompt=SYSTEM_PROMPT
    )

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


asyncio.run(main())

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")
SYSTEM_PROMPT = "YOUR SYSTEM PROMPT"

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 favorite model
            instructions=SYSTEM_PROMPT
        )
        await agent.aprint_response("top 5 pages with the most visits in the last 24 hours", stream=True)

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

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();
const SYSTEM_PROMPT = "YOUR SYSTEM PROMPT"

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 favorite model
    messages,
    maxSteps: 5,
    tools: {...tbTools},
    system: SYSTEM_PROMPT
  });

  console.log(result.text);
}

main();

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

SYSTEM_PROMPT = """
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
"""

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_FAVORITE_MODEL",  # e.g., "openai:gpt-4"
                mcp_servers=[server],
                instructions=SYSTEM_PROMPT
            )
            
            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())

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

SYSTEM_PROMPT = "YOUR SYSTEM PROMPT"

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. "anthropic:claude-3-5-sonnet-20241022",
        mcp_servers=[server],
        system_prompt=SYSTEM_PROMPT
    )

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