Model Context Protocol (MCP): Complete Guide [2025]

# Example MCP Communication Flow

# 1. Client discovers available tools
GET /mcp/tools
Response: [
  {
    "name": "search_database",
    "description": "Search customer database",
    "parameters": {
      "query": "string",
      "limit": "number"
    }
  }
]

# 2. AI decides to use the tool
User: "Find customers in California"
AI: I'll use the search_database tool

# 3. Client calls the tool
POST /mcp/tools/search_database
{
  "query": "state:California",
  "limit": 10
}

# 4. Server executes and returns results
Response: {
  "results": [
    {"name": "John Doe", "email": "john@example.com", "state": "CA"},
    {"name": "Jane Smith", "email": "jane@example.com", "state": "CA"}
  ]
}

# 5. AI uses results to respond
AI: "I found 2 customers in California: John Doe and Jane Smith..."

# Secure MCP Server Implementation
from mcp.server import Server
from mcp.types import Tool, TextContent
import re
import logging
from functools import wraps

server = Server("secure-server")
logger = logging.getLogger(__name__)

# Authentication decorator
def require_auth(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # Check authentication token
        auth_token = kwargs.get("_auth_token")
        if not verify_token(auth_token):
            return [TextContent(
                type="text",
                text="Authentication failed"
            )]
        return await func(*args, **kwargs)
    return wrapper

# Input validation
def validate_email(email: str) -> bool:
    """Validate email format"""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def sanitize_sql_input(value: str) -> str:
    """Remove potentially dangerous SQL characters"""
    # Only allow alphanumeric, spaces, and basic punctuation
    return re.sub(r'[^a-zA-Z0-9\s@._-]', '', value)

# Secure tool implementation
@server.tool("create_user")
@require_auth
async def create_user(
    name: str,
    email: str,
    _auth_token: str = None
) -> list[TextContent]:
    """Create a new user (requires authentication)"""
    
    # Input validation
    if not name or len(name) > 100:
        return [TextContent(
            type="text",
            text="Invalid name: must be 1-100 characters"
        )]
    
    if not validate_email(email):
        return [TextContent(
            type="text",
            text="Invalid email format"
        )]
    
    # Sanitize inputs
    name = sanitize_sql_input(name)
    email = sanitize_sql_input(email)
    
    try:
        # Use parameterized query (prevents SQL injection)
        conn = get_db_connection()
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            (name, email)  # Parameters, not string concatenation
        )
        conn.commit()
        user_id = cursor.lastrowid
        
        # Audit log
        logger.info(f"User created: id={user_id}, name={name}, email={email}")
        
        return [TextContent(
            type="text",
            text=f"User created successfully with ID {user_id}"
        )]
        
    except Exception as e:
        # Log error but don't expose details to client
        logger.error(f"Error creating user: {e}")
        return [TextContent(
            type="text",
            text="An error occurred while creating the user"
        )]
    finally:
        conn.close()

# Rate limiting
from collections import defaultdict
from datetime import datetime, timedelta

rate_limits = defaultdict(list)
MAX_REQUESTS_PER_MINUTE = 10

def check_rate_limit(client_id: str) -> bool:
    """Check if client has exceeded rate limit"""
    now = datetime.now()
    minute_ago = now - timedelta(minutes=1)
    
    # Remove old requests
    rate_limits[client_id] = [
        req_time for req_time in rate_limits[client_id]
        if req_time > minute_ago
    ]
    
    # Check limit
    if len(rate_limits[client_id]) >= MAX_REQUESTS_PER_MINUTE:
        return False
    
    # Add current request
    rate_limits[client_id].append(now)
    return True