Model Context Protocol (MCP)

# For Gemini (Google ADK)
def get_weather(city: str) -> dict:
    """Gets the weather for a city."""
    return {"temp": 25, "condition": "sunny"}
 
# For OpenAI
weather_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Gets the weather for a city",
        "parameters": {...}
    }
}
 
# For Anthropic Claude
weather_tool = {
    "name": "get_weather",
    "description": "Gets the weather for a city",
    "input_schema": {...}  # Slightly different schema format!
}

pip install mcp

# notes_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import json
from pathlib import Path
from datetime import datetime
 
# Persistent storage
NOTES_FILE = Path("notes.json")
 
def load_notes() -> dict:
    """Load notes from disk."""
    if NOTES_FILE.exists():
        return json.loads(NOTES_FILE.read_text())
    return {}
 
def save_notes(notes: dict):
    """Save notes to disk."""
    NOTES_FILE.write_text(json.dumps(notes, indent=2))
 
# Create the MCP server instance
server = Server("notes-server")
 
@server.list_tools()
async def list_tools() -> list[Tool]:
    """Declare the tools this server provides."""
    return [
        Tool(
            name="create_note",
            description="Creates a new note with a title and content. Use this when the user wants to save information for later.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "A short, descriptive title for the note"
                    },
                    "content": {
                        "type": "string",
                        "description": "The full content/body of the note"
                    }
                },
                "required": ["title", "content"]
            }
        ),
        Tool(
            name="list_notes",
            description="Lists all saved notes by title. Use this to see what notes exist.",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="get_note",
            description="Retrieves the full content of a specific note by its title.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "The exact title of the note to retrieve"
                    }
                },
                "required": ["title"]
            }
        ),
        Tool(
            name="delete_note",
            description="Permanently deletes a note by title.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "The title of the note to delete"
                    }
                },
                "required": ["title"]
            }
        )
    ]
 
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle incoming tool calls."""
    notes = load_notes()
    
    if name == "create_note":
        title = arguments["title"]
        content = arguments["content"]
        notes[title] = {
            "content": content,
            "created_at": datetime.now().isoformat()
        }
        save_notes(notes)
        return [TextContent(
            type="text",
            text=f"✅ Note '{title}' created successfully."
        )]
    
    elif name == "list_notes":
        if not notes:
            return [TextContent(type="text", text="No notes found. Create one first!")]
        
        note_list = "\n".join(f"• {title}" for title in sorted(notes.keys()))
        return [TextContent(
            type="text",
            text=f"📝 Your notes ({len(notes)} total):\n{note_list}"
        )]
    
    elif name == "get_note":
        title = arguments["title"]
        if title in notes:
            note = notes[title]
            return [TextContent(
                type="text",
                text=f"# {title}\n\n{note['content']}\n\n---\n_Created: {note.get('created_at', 'unknown')}_"
            )]
        return [TextContent(
            type="text",
            text=f"❌ Note '{title}' not found. Use list_notes to see available notes."
        )]
    
    elif name == "delete_note":
        title = arguments["title"]
        if title in notes:
            del notes[title]
            save_notes(notes)
            return [TextContent(type="text", text=f"🗑️ Note '{title}' deleted.")]
        return [TextContent(type="text", text=f"❌ Note '{title}' not found.")]
    
    return [TextContent(type="text", text=f"Unknown tool: {name}")]
 
async def main():
    """Run the MCP server over stdio."""
    async with stdio_server() as (read, write):
        await server.run(read, write)
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

{
  "mcpServers": {
    "notes": {
      "command": "python",
      "args": ["/absolute/path/to/notes_server.py"]
    }
  }
}

# agent.py
import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters
 
# Create an agent that uses our Notes MCP server
root_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="notes_assistant",
    description="An assistant that can manage your notes.",
    instruction="""You are a helpful assistant that manages the user's notes.
    Use the available tools to create, list, and retrieve notes.
    When creating notes, use clear, descriptive titles.
    When the user asks about their notes, list them first to see what's available.""",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="python",
                    args=["/path/to/notes_server.py"]
                )
            )
            # Optional: only expose specific tools
            # tool_filter=["create_note", "list_notes", "get_note"]
        )
    ]
)

my_agent/
    __init__.py
    agent.py
    .env

# my_agent/__init__.py
from . import agent

cd parent_of_my_agent
adk web

import asyncio
import os
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters
 
load_dotenv()
 
async def main():
    # Create the MCP toolset
    notes_toolset = McpToolset(
        connection_params=StdioConnectionParams(
            server_params=StdioServerParameters(
                command="python",
                args=["notes_server.py"]
            )
        )
    )
    
    # Create the agent
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="notes_assistant",
        instruction="You help users manage their notes.",
        tools=[notes_toolset]
    )
    
    # Set up session management
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="notes_app",
        user_id="user_1"
    )
    
    # Create runner
    runner = Runner(
        agent=agent,
        app_name="notes_app",
        session_service=session_service
    )
    
    # Interact with the agent
    queries = [
        "Create a note titled 'MCP Tutorial' with content about what I learned today",
        "List all my notes",
        "What's in the MCP Tutorial note?"
    ]
    
    for query in queries:
        print(f"\n👤 User: {query}")
        content = types.Content(
            role="user",
            parts=[types.Part(text=query)]
        )
        
        # Use async for with run_async
        async for event in runner.run_async(
            user_id="user_1",
            session_id=session.id,
            new_message=content
        ):
            if event.content and event.content.parts:
                final_text = event.content.parts[0].text
                if final_text:
                    print(f"🤖 Agent: {final_text}")
    
    # Clean up
    await notes_toolset.close()
 
if __name__ == "__main__":
    asyncio.run(main())

from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import SseConnectionParams
 
# For HTTP+SSE remote servers
remote_toolset = McpToolset(
    connection_params=SseConnectionParams(
        url="https://your-mcp-server.com/sse",
        headers={"Authorization": "Bearer your-token"}
    )
)

import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters
 
# Create an agent with multiple MCP servers
super_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="super_assistant",
    instruction="You can manage notes, files, and GitHub repos.",
    tools=[
        # Notes server
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="python",
                    args=["notes_server.py"]
                )
            )
        ),
        # Filesystem server (community)
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-filesystem", "/allowed/path"]
                )
            ),
            tool_filter=["read_file", "list_directory"]  # Security: limit tools
        ),
        # GitHub server (community)
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-github"],
                    env={"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]}
                )
            )
        )
    ]
)

McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/safe/directory"]
        )
    ),
    # Only allow read operations, not write/delete
    tool_filter=["read_file", "read_multiple_files", "list_directory", "search_files"]
)

# Just use native ADK tools - simpler, faster, less overhead
def get_weather(city: str) -> dict:
    """Gets the weather for a city."""
    return fetch_weather_api(city)
 
agent = LlmAgent(
    model="gemini-2.0-flash",
    tools=[get_weather]  # Direct function, no MCP needed
)

# This WON'T work well with MCP - needs agent context
@server.call_tool()
async def tool_that_needs_context(name: str, args: dict):
    # ❌ Can't access: agent.session, agent.state, agent.memory
    # ❌ Can't access: user authentication, conversation history
    pass

# Don't build an MCP server for this
# Just call the API directly
import requests
 
def quick_lookup():
    response = requests.get("https://api.example.com/data")
    return response.json()

# Install the filesystem MCP server
npx -y @modelcontextprotocol/server-filesystem /path/to/allowed/directory

McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()]
        )
    )
)

import os
import asyncio
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.google_search_tool import google_search
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters
 
load_dotenv()
 
# Research assistant that can search and save findings
research_agent = LlmAgent(
    model="gemini-2.0-flash",
    name="research_assistant",
    description="A research assistant that can search the web and save notes.",
    instruction="""You are a research assistant. Your workflow:
    1. When the user asks about a topic, search for information
    2. Summarize your findings clearly
    3. Offer to save important findings as notes for later reference
    
    Be thorough in research but concise in summaries.""",
    tools=[
        google_search,  # Built-in web search
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="python",
                    args=["notes_server.py"]
                )
            )
        )
    ]
)
 
async def run_research_session():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="research_app",
        user_id="researcher"
    )
    
    runner = Runner(
        agent=research_agent,
        app_name="research_app",
        session_service=session_service
    )
    
    # Research session
    queries = [
        "Research the current state of fusion energy technology",
        "Save the key points as a note titled 'Fusion Energy 2026'",
        "List my research notes"
    ]
    
    for query in queries:
        print(f"\n👤 {query}")
        content = types.Content(role="user", parts=[types.Part(text=query)])
        
        async for event in runner.run_async(
            user_id="researcher",
            session_id=session.id,
            new_message=content
        ):
            if event.content and event.content.parts:
                text = event.content.parts[0].text
                if text:
                    print(f"🔬 {text}")
 
if __name__ == "__main__":
    asyncio.run(run_research_session())