Model Context Protocol (MCP): Complete Guide to the 2026 Standard for AI Agent Tool Integration

# Old approach: Custom integrations for each tool

# GitHub integration
class GitHubTool:
    def __init__(self, token):
        self.token = token
    
    def create_issue(self, repo, title, body):
        # Custom API call
        pass

# Slack integration  
class SlackTool:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    
    def send_message(self, channel, text):
        # Custom implementation
        pass

# File system integration
class FileSystemTool:
    def read_file(self, path):
        # Custom implementation
        pass

# Each tool requires unique code = N*M integrations

# New approach: MCP standardized protocol
# Every tool follows the same interface

# MCP Server implements standard protocol
# MCP Client connects to any server uniformly

// Initialize Request
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {},
      "resources": {}
    },
    "clientInfo": {
      "name": "claude-desktop",
      "version": "1.0.0"
    }
  }
}

// Initialize Response
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {
        "listChanged": true
      },
      "resources": {
        "subscribe": true,
        "listChanged": true
      }
    },
    "serverInfo": {
      "name": "filesystem-server",
      "version": "1.0.0"
    }
  }
}

// tools/list response
{
  "tools": [
    {
      "name": "read_file",
      "description": "Read contents of a file",
      "inputSchema": {
        "type": "object",
        "properties": {
          "path": {
            "type": "string",
            "description": "Path to the file to read"
          },
          "encoding": {
            "type": "string",
            "enum": ["utf-8", "ascii", "base64"],
            "default": "utf-8"
          },
          "limit": {
            "type": "integer",
            "description": "Maximum number of bytes to read"
          }
        },
        "required": ["path"]
      }
    },
    {
      "name": "execute_command",
      "description": "Execute a shell command",
      "inputSchema": {
        "type": "object",
        "properties": {
          "command": {
            "type": "string",
            "description": "The command to execute"
          },
          "timeout": {
            "type": "integer",
            "default": 30000,
            "description": "Timeout in milliseconds"
          },
          "workingDirectory": {
            "type": "string"
          }
        },
        "required": ["command"]
      }
    }
  ]
}

// tools/call request
{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/home/user/project/README.md"
    }
  }
}

// tools/call response
{
  "jsonrpc": "2.0",
  "id": 5,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "# My Project\n\nThis is the README..."
      }
    ]
  }
}

#!/usr/bin/env python3
"""MCP Server - Filesystem Operations"""

import json
import os
from pathlib import Path
from typing import Any

# MCP Protocol imports
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from mcp.server.notification_params import NotificationParams

app = Server("filesystem-server")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Define available tools"""
    return [
        Tool(
            name="read_file",
            description="Read contents of a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file to read"
                    },
                    "encoding": {
                        "type": "string",
                        "enum": ["utf-8", "ascii", "base64"],
                        "default": "utf-8"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="Write content to a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file to write"
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="list_directory",
            description="List files in a directory",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Directory path to list"
                    },
                    "recursive": {
                        "type": "boolean",
                        "default": False,
                        "description": "List recursively"
                    }
                },
                "required": ["path"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """Handle tool invocations"""
    
    if name == "read_file":
        return await handle_read_file(arguments)
    elif name == "write_file":
        return await handle_write_file(arguments)
    elif name == "list_directory":
        return await handle_list_directory(arguments)
    else:
        raise ValueError(f"Unknown tool: {name}")

async def handle_read_file(args: dict) -> list[TextContent]:
    """Read file implementation"""
    path = Path(args["path"])
    encoding = args.get("encoding", "utf-8")
    
    if not path.exists():
        return [TextContent(type="text", text=f"Error: File not found: {path}")]
    
    try:
        content = path.read_text(encoding=encoding)
        return [TextContent(type="text", text=content)]
    except Exception as e:
        return [TextContent(type="text", text=f"Error reading file: {str(e)}")]

async def handle_write_file(args: dict) -> list[TextContent]:
    """Write file implementation"""
    path = Path(args["path"])
    content = args["content"]
    
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
        return [TextContent(type="text", text=f"Successfully wrote to {path}")]
    except Exception as e:
        return [TextContent(type="text", text=f"Error writing file: {str(e)}")]

async def handle_list_directory(args: dict) -> list[TextContent]:
    """List directory implementation"""
    path = Path(args["path"])
    recursive = args.get("recursive", False)
    
    if not path.is_dir():
        return [TextContent(type="text", text=f"Error: Not a directory: {path}")]
    
    try:
        if recursive:
            items = [str(p) for p in path.rglob("*")]
        else:
            items = [str(p) for p in path.iterdir()]
        
        return [TextContent(type="text", text="\n".join(items))]
    except Exception as e:
        return [TextContent(type="text", text=f"Error listing directory: {str(e)}")]

async def main():
    """Run the MCP server"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

#!/usr/bin/env node
/**
 * MCP Server - Database Operations
 */

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';
import pg from 'pg';

const { Pool } = pg;

// Database connection pool
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
});

class DatabaseServer {
  constructor() {
    this.server = new Server(
      { name: 'database-server', version: '1.0.0' },
      { capabilities: { tools: {}, resources: {} } }
    );
    
    this.setupHandlers();
  }
  
  setupHandlers() {
    // List available tools
    this.server.setRequestHandler(ListToolsRequestSchema, async () => {
      return {
        tools: [
          {
            name: 'query_database',
            description: 'Execute a SQL query against the database',
            inputSchema: {
              type: 'object',
              properties: {
                sql: {
                  type: 'string',
                  description: 'SQL query to execute'
                },
                params: {
                  type: 'array',
                  description: 'Query parameters',
                  items: { type: 'string' }
                }
              },
              required: ['sql']
            }
          },
          {
            name: 'list_tables',
            description: 'List all tables in the database',
            inputSchema: {
              type: 'object',
              properties: {}
            }
          },
          {
            name: 'get_table_schema',
            description: 'Get schema for a specific table',
            inputSchema: {
              type: 'object',
              properties: {
                table: {
                  type: 'string',
                  description: 'Table name'
                }
              },
              required: ['table']
            }
          }
        ]
      };
    });
    
    // Handle tool calls
    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      const { name, arguments: args } = request.params;
      
      try {
        switch (name) {
          case 'query_database':
            return await this.queryDatabase(args.sql, args.params);
          case 'list_tables':
            return await this.listTables();
          case 'get_table_schema':
            return await this.getTableSchema(args.table);
          default:
            throw new Error(`Unknown tool: ${name}`);
        }
      } catch (error) {
        return {
          content: [
            {
              type: 'text',
              text: `Error: ${error.message}`
            }
          ]
        };
      }
    });
  }
  
  async queryDatabase(sql, params = []) {
    const result = await pool.query(sql, params);
    
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify(result.rows, null, 2)
        }
      ],
      isError: false
    };
  }
  
  async listTables() {
    const result = await pool.query(`
      SELECT table_name 
      FROM information_schema.tables 
      WHERE table_schema = 'public'
      ORDER BY table_name
    `);
    
    const tables = result.rows.map(r => r.table_name).join('\n');
    
    return {
      content: [
        {
          type: 'text',
          text: tables || 'No tables found'
        }
      ]
    };
  }
  
  async getTableSchema(table) {
    const result = await pool.query(`
      SELECT column_name, data_type, is_nullable, column_default
      FROM information_schema.columns
      WHERE table_name = $1
      ORDER BY ordinal_position
    `, [table]);
    
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify(result.rows, null, 2)
        }
      ]
    };
  }
  
  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
    console.error('Database MCP Server running on stdio');
  }
}

const server = new DatabaseServer();
server.start().catch(console.error);

/**
 * MCP Server - REST API Integration
 */

import { Server } from '@modelcontextprotocol/sdk/server';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio';
import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types';
import fetch from 'node-fetch';

interface RESTConfig {
  baseUrl: string;
  headers?: Record<string, string>;
}

class RESTAPIServer {
  private config: RESTConfig;
  private server: Server;

  constructor(config: RESTConfig) {
    this.config = config;
    this.server = new Server(
      { name: 'rest-api-server', version: '1.0.0' },
      { capabilities: { tools: {} } }
    );
    
    this.setupHandlers();
  }

  private setupHandlers() {
    this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
      tools: [
        {
          name: 'api_get',
          description: 'Make GET request to REST API',
          inputSchema: {
            type: 'object',
            properties: {
              endpoint: { type: 'string', description: 'API endpoint' },
              params: { type: 'object', description: 'Query parameters' }
            },
            required: ['endpoint']
          }
        },
        {
          name: 'api_post',
          description: 'Make POST request to REST API',
          inputSchema: {
            type: 'object',
            properties: {
              endpoint: { type: 'string', description: 'API endpoint' },
              body: { type: 'object', description: 'Request body' }
            },
            required: ['endpoint', 'body']
          }
        }
      ]
    }));

    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      const { name, arguments: args } = request.params;

      try {
        const url = new URL(args.endpoint, this.config.baseUrl);
        if (args.params) {
          Object.entries(args.params).forEach(([k, v]) => 
            url.searchParams.append(k, String(v))
          );
        }

        const options: RequestInit = {
          headers: {
            'Content-Type': 'application/json',
            ...this.config.headers
          }
        };

        let response;
        if (name === 'api_get') {
          response = await fetch(url.toString(), options);
        } else if (name === 'api_post') {
          options.method = 'POST';
          options.body = JSON.stringify(args.body);
          response = await fetch(url.toString(), options);
        } else {
          throw new Error(`Unknown tool: ${name}`);
        }

        const data = await response.json();

        return {
          content: [{ type: 'text', text: JSON.stringify(data, null, 2) }]
        };
      } catch (error: any) {
        return {
          content: [{ type: 'text', text: `Error: ${error.message}` }],
          isError: true
        };
      }
    });
  }

  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
  }
}

// Usage
const server = new RESTAPIServer({
  baseUrl: process.env.API_BASE_URL || 'https://api.example.com',
  headers: {
    'Authorization': `Bearer ${process.env.API_TOKEN}`
  }
});

server.start();

#!/usr/bin/env python3
"""MCP Client - Connect to MCP Servers"""

import asyncio
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClient:
    def __init__(self, server_command: list[str], server_env: dict = None):
        self.server_command = server_command
        self.server_env = server_env or {}
        self.session = None
    
    async def connect(self):
        """Connect to MCP server"""
        server_params = StdioServerParameters(
            command=self.server_command[0],
            args=self.server_command[1:],
            env=self.server_env
        )
        
        async with stdio_client(server_params) as (read, write):
            self.session = ClientSession(read, write)
            await self.session.initialize()
            
            # Get server capabilities
            initialize_result = await self.session.initialize()
            return initialize_result
    
    async def list_tools(self) -> list[dict]:
        """List available tools from server"""
        result = await self.session.list_tools()
        return result.tools
    
    async def call_tool(self, tool_name: str, arguments: dict) -> Any:
        """Call a specific tool"""
        result = await self.session.call_tool(tool_name, arguments)
        return result.content
    
    async def close(self):
        """Close connection"""
        if self.session:
            await self.session.close()

# Example: Connect to filesystem server
async def main():
    client = MCPClient(
        server_command=["python3", "/path/to/filesystem_server.py"],
        server_env={"HOME": "/home/user"}
    )
    
    try:
        await client.connect()
        
        # List available tools
        tools = await client.list_tools()
        print("Available tools:")
        for tool in tools:
            print(f"  - {tool.name}: {tool.description}")
        
        # Read a file
        result = await client.call_tool("read_file", {"path": "/home/user/README.md"})
        print("\nFile content:")
        print(result[0].text)
        
    finally:
        await client.close()

asyncio.run(main())

/**
 * MCP Client for Browser/Node.js
 */

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js';

class MCPClient {
  constructor(serverCommand, serverEnv = {}) {
    this.client = new Client({
      name: 'mcp-client',
      version: '1.0.0'
    }, {
      capabilities: {
        tools: {}
      }
    });
    
    this.serverCommand = serverCommand;
    this.serverEnv = serverEnv;
  }
  
  async connect() {
    const transport = new StdioClientTransport({
      command: this.serverCommand[0],
      args: this.serverCommand.slice(1),
      env: this.serverEnv
    });
    
    await this.client.connect(transport);
    
    const result = await this.client.request(
      { method: 'initialize' },
      {
        protocolVersion: '2024-11-05',
        capabilities: {},
        clientInfo: { name: 'my-client', version: '1.0.0' }
      }
    );
    
    return result;
  }
  
  async listTools() {
    const result = await this.client.request(
      { method: 'tools/list' },
      {}
    );
    
    return result.tools;
  }
  
  async callTool(name, args) {
    const result = await this.client.request(
      { method: 'tools/call' },
      { name, arguments: args }
    );
    
    return result.content;
  }
}

// Usage
async function main() {
  const client = new MCPClient(
    ['python3', '/path/to/server.py'],
    { HOME: process.env.HOME }
  );
  
  try {
    await client.connect();
    
    const tools = await client.listTools();
    console.log('Available tools:', tools.map(t => t.name));
    
    const result = await client.callTool('read_file', {
      path: '/home/user/README.md'
    });
    
    console.log('Result:', result);
  } catch (error) {
    console.error('Error:', error);
  }
}

main();

# Using npm (for Node.js servers)
npm install -g @modelcontextprotocol/server-filesystem
npm install -g @modelcontextprotocol/server-github
npm install -g @modelcontextprotocol/server-slack

# Using pip (for Python servers)
pip install mcp-server-filesystem
pip install mcp-server-github
pip install mcp-server-database

// claude_desktop_config.json
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"],
      "env": {}
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    },
    "postgres": {
      "command": "python3",
      "args": ["/path/to/postgres_server.py"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    }
  }
}

# MCP-powered development assistant

async def development_workflow():
    """Automate common development tasks"""
    
    # Connect to multiple MCP servers
    filesystem = await connect_mcp_server(["npx", "-y", "@modelcontextprotocol/server-filesystem", "./project"])
    github = await connect_mcp_server(["npx", "-y", "@modelcontextprotocol/server-github"])
    terminal = await connect_mcp_server(["python3", "./terminal_server.py"])
    
    # AI orchestrates the workflow:
    # 1. Read SPEC.md
    spec = await filesystem.call_tool("read_file", {"path": "SPEC.md"})
    
    # 2. Create issue for new feature
    issue = await github.call_tool("create_issue", {
        "owner": "myorg",
        "repo": "myproject",
        "title": "Implement user authentication",
        "body": spec
    })
    
    # 3. Create feature branch
    await terminal.call_tool("execute_command", {
        "command": f"git checkout -b feature/auth-{issue.number}"
    })

# Connect AI to database for analytics

async def analytics_workflow(question: str):
    """Answer analytics questions using database"""
    
    db = await connect_mcp_server(["python3", "./postgres_server.py"])
    
    # Get database schema
    tables = await db.call_tool("list_tables", {})
    
    # AI analyzes question and constructs query
    # "What are the top 10 customers by revenue?"
    query = """
        SELECT customer_name, SUM(order_total) as revenue
        FROM orders
        GROUP BY customer_name
        ORDER BY revenue DESC
        LIMIT 10
    """
    
    result = await db.call_tool("query_database", {"sql": query})
    return format_as_table(result)

# Connect to search, browser, and knowledge bases

async def research_workflow(topic: str):
    """Research a topic using multiple sources"""
    
    # Connect to search engine
    search = await connect_mcp_server(["npx", "-y", "@modelcontextprotocol/server-brave-search"])
    
    # Connect to browser automation
    browser = await connect_mcp_server(["npx", "-y", "@modelcontextprotocol/server-puppeteer"])
    
    # Connect to local knowledge base
    knowledge = await connect_mcp_server(["python3", "./vector_db_server.py"])
    
    # Search web
    web_results = await search.call_tool("search", {"query": topic})
    
    # Search local docs
    local_results = await knowledge.call_tool("semantic_search", {
        "query": topic,
        "limit": 5
    })
    
    # Combine and summarize
    return await synthesize_results(web_results, local_results)

# Restricted filesystem server - security best practices

class SecureFilesystemServer:
    def __init__(self, allowed_directories: list[str]):
        self.allowed_directories = [
            Path(d).resolve() for d in allowed_directories
        ]
    
    def _verify_path(self, path: str) -> Path:
        """Verify path is within allowed directories"""
        resolved = Path(path).resolve()
        
        for allowed in self.allowed_directories:
            try:
                resolved.relative_to(allowed)
                return resolved
            except ValueError:
                continue
        
        raise PermissionError(f"Path not allowed: {path}")
    
    async def read_file(self, path: str) -> str:
        """Read file with path verification"""
        verified_path = self._verify_path(path)
        
        if not verified_path.exists():
            raise FileNotFoundError(f"File not found: {path}")
        
        return verified_path.read_text()

# MCP Server with authentication

class AuthenticatedServer:
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    async def handle_request(self, request: dict):
        """Verify authentication on each request"""
        
        # Extract API key from request headers
        auth_header = request.get("headers", {}).get("Authorization", "")
        
        if not auth_header.startswith("Bearer "):
            raise UnauthorizedError("Missing authentication")
        
        token = auth_header[7:]  # Remove "Bearer "
        
        if token != self.api_key:
            raise UnauthorizedError("Invalid API key")
        
        # Process request
        return await self._process_request(request)

# Rate limiting for MCP tools

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = window_seconds
        self.calls = defaultdict(list)
    
    def check(self, client_id: str) -> bool:
        """Check if client can make a call"""
        now = time.time()
        
        # Remove old calls outside window
        self.calls[client_id] = [
            t for t in self.calls[client_id]
            if now - t < self.window
        ]
        
        if len(self.calls[client_id]) >= self.max_calls:
            return False
        
        self.calls[client_id].append(now)
        return True

# Usage in tool handler
@server.call_tool()
async def rate_limited_tool(name: str, args: dict):
    client_id = args.get("_client_id", "anonymous")
    
    if not rate_limiter.check(client_id):
        raise ToolError("Rate limit exceeded. Try again later.")
    
    # Process tool call
    return await process_tool(name, args)

# Robust error handling in MCP servers

async def safe_tool_handler(tool_name: str, args: dict):
    """Wrap tool handler with error handling"""
    try:
        # Validate inputs
        validate_arguments(args)
        
        # Execute tool
        result = await execute_tool(tool_name, args)
        
        # Return success
        return format_success_result(result)
        
    except ValidationError as e:
        return format_error_result("Invalid arguments", str(e), "validation_error")
        
    except PermissionError as e:
        return format_error_result("Access denied", str(e), "permission_denied")
        
    except FileNotFoundError as e:
        return format_error_result("File not found", str(e), "not_found")
        
    except TimeoutError:
        return format_error_result("Operation timed out", 
                                  "The operation took too long", "timeout")
        
    except Exception as e:
        # Log for debugging
        log_error(tool_name, args, e)
        return format_error_result("Internal error", 
                                  "An unexpected error occurred", "internal_error")

# Write clear, descriptive tool definitions

Tool(
    name="search_documents",
    description="""Search for documents using full-text search.
    
    Use this tool when:
    - User asks to find specific documents or files
    - User wants to search by content (not just filename)
    - User needs to find files matching keywords
    
    Returns a list of matching documents with relevance scores.
    """,
    inputSchema={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query string. Supports boolean operators (AND, OR, NOT) and phrases."
            },
            "limit": {
                "type": "integer",
                "default": 10,
                "description": "Maximum number of results to return (1-100)"
            },
            "filters": {
                "type": "object",
                "description": "Optional filters",
                "properties": {
                    "file_type": {"type": "string", "enum": ["pdf", "doc", "txt"]},
                    "date_from": {"type": "string", "format": "date"},
                    "date_to": {"type": "string", "format": "date"}
                }
            }
        },
        "required": ["query"]
    }
)

# Testing MCP server implementations

import pytest
from mcp.server.testing import create_test_client

@pytest.fixture
async def test_server():
    """Create test instance of MCP server"""
    server = MyMCPServer()
    return server

@pytest.mark.asyncio
async def test_list_tools(test_server):
    """Test tool listing"""
    async with create_test_client(test_server) as client:
        tools = await client.list_tools()
        
        assert len(tools) > 0
        assert any(t.name == "my_tool" for t in tools)

@pytest.mark.asyncio
async def test_call_tool(test_server):
    """Test tool invocation"""
    async with create_test_client(test_server) as client:
        result = await client.call_tool(
            "my_tool",
            {"arg1": "value1"}
        )
        
        assert result[0].type == "text"
        assert "expected" in result[0].text.lower()

@pytest.mark.asyncio
async def test_error_handling(test_server):
    """Test error responses"""
    async with create_test_client(test_server) as client:
        result = await client.call_tool(
            "invalid_tool",
            {}
        )
        
        assert result.isError
        assert "unknown tool" in result.content[0].text.lower()