Model Context Protocol (MCP)

Introduction

The Model Context Protocol (MCP) is an open standard for connecting AI agents to external data sources and tools. Agno’s MCPTools allows you to seamlessly integrate any MCP server into your agents.

What is MCP?

MCP enables agents to:

Access tools from external MCP servers
Query resources like databases, APIs, and file systems
Use prompts defined by MCP servers
Maintain context across different services

Learn more at modelcontextprotocol.io

Installation

pip install mcp

Quick Start

Connect to an MCP filesystem server:

import asyncio
from pathlib import Path
from agno.agent import Agent
from agno.tools.mcp import MCPTools
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Initialize the MCP server
    server_params = StdioServerParameters(
        command="npx",
        args=[
            "-y",
            "@modelcontextprotocol/server-filesystem",
            str(Path.cwd())
        ],
    )
    
    # Create a client session
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize MCPTools with the session
            mcp_tools = MCPTools(session=session)
            await mcp_tools.initialize()

            # Create agent with MCP tools
            agent = Agent(tools=[mcp_tools])

            # Use the agent
            await agent.aprint_response(
                "Read and summarize the LICENSE file",
                stream=True
            )

asyncio.run(main())

Connection Methods

Method 1: Simplified Command-based Connection

The easiest way to connect using a command string:

from agno.agent import Agent
from agno.tools.mcp import MCPTools

# Connect to filesystem server
mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp"
)

agent = Agent(tools=[mcp_tools])

Method 2: Using Server Parameters

For more control over the connection:

from agno.tools.mcp import MCPTools
from mcp import StdioServerParameters

server_params = StdioServerParameters(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    env={"NODE_ENV": "production"}
)

mcp_tools = MCPTools(server_params=server_params)

Method 3: Pre-initialized Session

When you need full control over the session:

import asyncio
from agno.agent import Agent
from agno.tools.mcp import MCPTools
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            mcp_tools = MCPTools(session=session)
            await mcp_tools.initialize()
            
            agent = Agent(tools=[mcp_tools])
            await agent.aprint_response("List all files")

asyncio.run(main())

Transport Types

STDIO Transport (Default)

For local MCP servers running as processes:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    transport="stdio"
)

SSE Transport

For MCP servers accessible via Server-Sent Events:

from agno.tools.mcp import MCPTools
from agno.tools.mcp.params import SSEClientParams

mcp_tools = MCPTools(
    url="https://mcp-server.example.com/sse",
    transport="sse",
    server_params=SSEClientParams(
        url="https://mcp-server.example.com/sse",
        headers={"Authorization": "Bearer token"},
        timeout=30
    )
)

Streamable HTTP Transport

For HTTP-based MCP servers:

from agno.tools.mcp import MCPTools
from agno.tools.mcp.params import StreamableHTTPClientParams

mcp_tools = MCPTools(
    url="https://mcp-server.example.com",
    transport="streamable-http",
    server_params=StreamableHTTPClientParams(
        url="https://mcp-server.example.com",
        headers={"Authorization": "Bearer token"},
        timeout=30
    )
)

Tool Filtering

Control which MCP tools are available to your agent:

Include Specific Tools

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    include_tools=["read_file", "list_directory"]  # Only these tools
)

Exclude Specific Tools

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    exclude_tools=["write_file", "delete_file"]  # All except these
)

Add Tool Name Prefix

Avoid name collisions when using multiple MCP servers:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    tool_name_prefix="fs"  # Tools become: fs_read_file, fs_list_directory, etc.
)

Dynamic Headers

For HTTP-based transports, you can provide dynamic headers using a header provider function:

from agno.tools.mcp import MCPTools
from agno.run import RunContext

def get_auth_headers(run_context: RunContext) -> dict:
    """Generate dynamic headers based on run context."""
    user_id = run_context.session_state.get("user_id", "anonymous")
    return {
        "Authorization": f"Bearer {get_token_for_user(user_id)}",
        "X-Session-ID": run_context.session_id
    }

mcp_tools = MCPTools(
    url="https://mcp-server.example.com",
    transport="streamable-http",
    header_provider=get_auth_headers  # Called on each agent run
)

Dynamic headers are only supported with SSE and Streamable HTTP transports.

Configuration Options

command

str

Command to start the MCP server (for stdio transport)

url

str

URL endpoint for SSE or Streamable HTTP connection

env

dict

Environment variables to pass to the server

transport

str

Transport protocol: “stdio”, “sse”, or “streamable-http”

server_params

Union[StdioServerParameters, SSEClientParams, StreamableHTTPClientParams]

Detailed server connection parameters

session

ClientSession

Pre-initialized MCP client session

timeout_seconds

int

default:"10"

Read timeout in seconds for the MCP client

include_tools

list[str]

List of tool names to include (if None, includes all)

exclude_tools

list[str]

List of tool names to exclude

refresh_connection

bool

default:"False"

If True, refresh connection and tools on each agent run

tool_name_prefix

str

Prefix to add to all tool names from this server

header_provider

Callable

Function to generate dynamic HTTP headers (HTTP transports only)

Popular MCP Servers

Filesystem Server

Access local file systems:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp"
)

Available Tools:

read_file - Read file contents
write_file - Write to a file
list_directory - List directory contents
create_directory - Create a directory
move_file - Move/rename files
search_files - Search for files

GitHub Server

Interact with GitHub repositories:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-github",
    env={"GITHUB_PERSONAL_ACCESS_TOKEN": "your-token"}
)

Available Tools:

create_repository
get_repository
create_issue
create_pull_request
search_repositories
search_code

Brave Search Server

Web search capabilities:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-brave-search",
    env={"BRAVE_API_KEY": "your-api-key"}
)

PostgreSQL Server

Database access:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-postgres",
    env={"POSTGRES_CONNECTION_STRING": "postgresql://localhost/mydb"}
)

Slack Server

Slack workspace integration:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-slack",
    env={
        "SLACK_BOT_TOKEN": "your-bot-token",
        "SLACK_TEAM_ID": "your-team-id"
    }
)

Multiple MCP Servers

Use multiple MCP servers in a single agent:

import asyncio
from agno.agent import Agent
from agno.tools.mcp import MCPTools

async def main():
    # Filesystem access
    fs_tools = MCPTools(
        command="npx -y @modelcontextprotocol/server-filesystem /tmp",
        tool_name_prefix="fs"
    )
    
    # GitHub access  
    github_tools = MCPTools(
        command="npx -y @modelcontextprotocol/server-github",
        env={"GITHUB_PERSONAL_ACCESS_TOKEN": "token"},
        tool_name_prefix="github"
    )
    
    # Initialize both
    async with fs_tools as fs, github_tools as gh:
        agent = Agent(
            tools=[fs, gh],
            instructions="You can access files and GitHub."
        )
        
        await agent.aprint_response(
            "Read the README.md file and create a GitHub issue "
            "summarizing its contents"
        )

asyncio.run(main())

Connection Management

Automatic Connection

MCPTools automatically connects when the agent first uses it:

mcp_tools = MCPTools(command="npx -y @modelcontextprotocol/server-filesystem /tmp")

# Connection happens automatically on first use
agent = Agent(tools=[mcp_tools])
agent.print_response("List files")

Manual Connection Control

For more control, use the async context manager:

import asyncio
from agno.tools.mcp import MCPTools

async def main():
    mcp_tools = MCPTools(
        command="npx -y @modelcontextprotocol/server-filesystem /tmp"
    )
    
    # Manually connect
    async with mcp_tools:
        agent = Agent(tools=[mcp_tools])
        await agent.aprint_response("List files")
    
    # Connection automatically closed after context

asyncio.run(main())

Refresh Connections

Refresh the connection and tools on each agent run:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    refresh_connection=True  # Reconnect on each agent run
)

Error Handling

Handle MCP connection errors gracefully:

import asyncio
from agno.agent import Agent
from agno.tools.mcp import MCPTools

async def main():
    try:
        mcp_tools = MCPTools(
            command="npx -y @modelcontextprotocol/server-filesystem /tmp",
            timeout_seconds=30
        )
        
        async with mcp_tools:
            # Check if connection is alive
            if await mcp_tools.is_alive():
                agent = Agent(tools=[mcp_tools])
                await agent.aprint_response("List files")
            else:
                print("MCP server is not responding")
                
    except Exception as e:
        print(f"Error connecting to MCP server: {e}")

asyncio.run(main())

Best Practices

Use Tool Filtering

Only include the MCP tools your agent actually needs to reduce context size and improve performance.

Set Appropriate Timeouts

Configure timeout_seconds based on your server’s expected response time.

Add Tool Name Prefixes

When using multiple MCP servers, use tool_name_prefix to avoid naming conflicts.

Handle Connection Errors

Always wrap MCP operations in try-except blocks to handle connection failures gracefully.

Use Async Context Managers

Use async with to ensure proper cleanup of MCP connections.

Secure Credentials

Store API keys and tokens in environment variables, never hardcode them.

Advanced: Creating Custom MCP Servers

You can create your own MCP servers to expose custom tools. See the MCP documentation for details on building MCP servers.

Troubleshooting

Connection Timeout

Increase the timeout if your MCP server needs more time:

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-filesystem /tmp",
    timeout_seconds=60  # Increase from default 10s
)

Tool Not Available

Check that the tool is not being filtered:

# List all available tools
async with mcp_tools:
    print(mcp_tools.functions.keys())

Environment Variables Not Working

Ensure environment variables are properly set:

import os

mcp_tools = MCPTools(
    command="npx -y @modelcontextprotocol/server-github",
    env={
        **os.environ,  # Include existing env vars
        "GITHUB_PERSONAL_ACCESS_TOKEN": "your-token"
    }
)

Next Steps

MCP Documentation

Learn more about the Model Context Protocol

MCP Servers

Browse official MCP server implementations

Custom Tools

Create your own custom tools

Built-in Tools

Explore Agno’s built-in tools

Get Started

Core Building Blocks

Production Runtime

Memory & State

Knowledge & RAG

Advanced Features

Models

Tools & Integrations

​Introduction

​What is MCP?

​Installation

​Quick Start

​Connection Methods

​Method 1: Simplified Command-based Connection

​Method 2: Using Server Parameters

​Method 3: Pre-initialized Session

​Transport Types

​STDIO Transport (Default)

​SSE Transport

​Streamable HTTP Transport

​Tool Filtering

​Include Specific Tools

​Exclude Specific Tools

​Add Tool Name Prefix

​Dynamic Headers

​Configuration Options

​Popular MCP Servers

​Filesystem Server

​GitHub Server

​Brave Search Server

​PostgreSQL Server

​Slack Server

​Multiple MCP Servers

​Connection Management

​Automatic Connection

​Manual Connection Control

​Refresh Connections

​Error Handling

​Best Practices

​Advanced: Creating Custom MCP Servers

​Troubleshooting

​Connection Timeout

​Tool Not Available

​Environment Variables Not Working

​Next Steps

MCP Documentation

MCP Servers

Custom Tools

Built-in Tools

Build docs developers (and LLMs) love

Introduction

What is MCP?

Installation

Quick Start

Connection Methods

Method 1: Simplified Command-based Connection

Method 2: Using Server Parameters

Method 3: Pre-initialized Session

Transport Types

STDIO Transport (Default)

SSE Transport

Streamable HTTP Transport

Tool Filtering

Include Specific Tools

Exclude Specific Tools

Add Tool Name Prefix

Dynamic Headers

Configuration Options

Popular MCP Servers

Filesystem Server

GitHub Server

Brave Search Server

PostgreSQL Server

Slack Server

Multiple MCP Servers

Connection Management

Automatic Connection

Manual Connection Control

Refresh Connections

Error Handling

Best Practices

Advanced: Creating Custom MCP Servers

Troubleshooting

Connection Timeout

Tool Not Available

Environment Variables Not Working

Next Steps