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