Chapter 24: MCP Server Development Guide¶

from mcp.server.fastmcp import FastMCP

# Create an MCP server instance
mcp = FastMCP("My Custom Server")

# Define a tool
@mcp.tool()
def calculate_sum(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

# Define a resource
@mcp.resource("config://weather")
def get_weather_config() -> str:
    """Get weather configuration."""
    return '{"api_key": "secret", "units": "metric"}'

# Define a prompt
@mcp.prompt()
def weather_report(location: str) -> str:
    """Generate a weather report prompt."""
    return f"Create a detailed weather report for {location}."

if __name__ == "__main__":
    mcp.run()

# my_server.py
from mcp.server.fastmcp import FastMCP
from typing import Any, List, Optional
import asyncio
import json

class MyCustomMCPServer:
    def __init__(self, name: str, version: str = "1.0.0"):
        self.mcp = FastMCP(name)
        self.setup_tools()
        self.setup_resources()
        self.setup_prompts()

    def setup_tools(self):
        """Register all tool functions."""

        @self.mcp.tool()
        def list_files(directory: str, pattern: Optional[str] = None) -> List[str]:
            """
            List files in a directory with optional pattern matching.

            Args:
                directory: Path to the directory to list
                pattern: Optional glob pattern for filtering files
            """
            import glob
            import os

            if pattern:
                search_path = os.path.join(directory, pattern)
                return glob.glob(search_path)

            return os.listdir(directory)

        @self.mcp.tool()
        def get_file_info(file_path: str) -> dict:
            """
            Get detailed information about a file.

            Args:
                file_path: Path to the file to analyze
            """
            import os
            import hashlib

            if not os.path.exists(file_path):
                raise FileNotFoundError(f"File not found: {file_path}")

            stat = os.stat(file_path)

            # Calculate file hash
            with open(file_path, 'rb') as f:
                file_hash = hashlib.md5(f.read()).hexdigest()

            return {
                "path": file_path,
                "size": stat.st_size,
                "modified": stat.st_mtime,
                "hash": file_hash,
                "readable": os.access(file_path, os.R_OK),
                "writable": os.access(file_path, os.W_OK)
            }

    def setup_resources(self):
        """Register all resource endpoints."""

        @self.mcp.resource("file://{path}")
        def get_file_content(path: str) -> str:
            """
            Get the content of a file as a resource.

            Args:
                path: Path to the file to read
            """
            with open(path, 'r') as f:
                return f.read()

    def setup_prompts(self):
        """Register all prompt templates."""

        @self.mcp.prompt()
        def analyze_code(file_path: str, focus_areas: Optional[List[str]] = None) -> str:
            """
            Generate a prompt for code analysis.

            Args:
                file_path: Path to the code file to analyze
                focus_areas: Optional list of specific areas to focus on
            """
            focus_text = ""
            if focus_areas:
                focus_text = f"Focus on: {', '.join(focus_areas)}"

            return f"""
            Analyze the code in {file_path}. {focus_text}

            Please provide:
            1. Code quality assessment
            2. Potential bugs or issues
            3. Performance suggestions
            4. Documentation recommendations
            """

    def run(self):
        """Start the MCP server."""
        self.mcp.run()

# Usage example
if __name__ == "__main__":
    server = MyCustomMCPServer("file-analyzer", "1.0.0")
    server.run()

from mcp.server.fastmcp import FastMCP
from mcp.server.models import ToolCallResult, ToolCallError
import logging

mcp = FastMCP("advanced-server")

@mcp.tool()
def risky_operation(param: str) -> str:
    """Tool with custom error handling."""
    try:
        # Potentially failing operation
        result = perform_operation(param)
        return result
    except ValueError as e:
        # Return structured error for better client handling
        raise ToolCallError(
            code=400,
            message=f"Invalid parameter: {e}",
            data={"parameter": param, "valid_values": ["opt1", "opt2"]}
        )
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        raise ToolCallError(
            code=500,
            message="Internal server error",
            data={"error_type": type(e).__name__}
        )

def perform_operation(param: str) -> str:
    """Simulated operation that might fail."""
    if param not in ["opt1", "opt2"]:
        raise ValueError(f"Invalid parameter: {param}")
    return f"Processed {param}"

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("streaming-server")

@mcp.tool(streaming=True)
async def stream_large_file(file_path: str) -> str:
    """
    Stream file content in chunks for large files.

    Args:
        file_path: Path to the file to stream
    """
    chunk_size = 1024  # 1KB chunks

    try:
        with open(file_path, 'r') as f:
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                yield chunk  # Stream this chunk
                await asyncio.sleep(0.01)  # Prevent overwhelming the client
    except Exception as e:
        yield f"Error reading file: {e}"

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("subscription-server")

@mcp.resource("log://application", subscribe=True)
async def get_application_logs() -> str:
    """Get application logs with subscription support."""
    return read_latest_logs()

@mcp.resource("config://settings")
def get_settings() -> str:
    """Get current configuration."""
    return json.dumps(load_current_settings())

# Notification system for subscription updates
@mcp.notification()
async def notify_config_changes():
    """Notify clients of configuration changes."""
    await mcp.send_notification(
        "notifications/resource/updated",
        {
            "uri": "config://settings",
            "change": "configuration_modified"
        }
    )

# test_my_server.py
import pytest
import asyncio
from unittest.mock import patch, mock_open
from my_server import MyCustomMCPServer

class TestMyCustomMCPServer:
    @pytest.fixture
    def server(self):
        return MyCustomMCPServer("test-server")

    @pytest.mark.asyncio
    async def test_list_files_success(self, server):
        """Test successful file listing."""
        with patch('os.listdir', return_value=['file1.txt', 'file2.py']):
            result = await server.list_files("/test/directory")
            assert result == ['file1.txt', 'file2.py']

    @pytest.mark.asyncio
    async def test_list_files_with_pattern(self, server):
        """Test file listing with pattern matching."""
        with patch('glob.glob', return_value=['/test/file1.txt']):
            result = await server.list_files("/test", "*.txt")
            assert result == ['/test/file1.txt']

    @pytest.mark.asyncio
    async def test_get_file_info_success(self, server):
        """Test successful file info retrieval."""
        mock_stat = {
            'st_size': 1024,
            'st_mtime': 1234567890
        }

        with patch('os.path.exists', return_value=True), \
             patch('os.stat', return_value=type('MockStat', (), mock_stat)()), \
             patch('builtins.open', mock_open(read_data=b'test content')), \
             patch('os.access', return_value=True):

            result = await server.get_file_info("/test/file.txt")

            assert result['path'] == "/test/file.txt"
            assert result['size'] == 1024
            assert result['hash']  # MD5 hash should be calculated

    @pytest.mark.asyncio
    async def test_get_file_info_not_found(self, server):
        """Test file info with non-existent file."""
        with patch('os.path.exists', return_value=False):
            with pytest.raises(FileNotFoundError):
                await server.get_file_info("/nonexistent/file.txt")

# test_integration.py
import asyncio
import json
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client

async def test_server_integration():
    """Test server with actual MCP client."""
    # Start server process
    server_process = await asyncio.create_subprocess_exec(
        'python', '-m', 'my_server',
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    try:
        # Create MCP client connection
        async with stdio_client(server_process.stdin, server_process.stdout) as (read, write):
            async with ClientSession(read, write) as session:
                # Initialize session
                await session.initialize()

                # Test tool availability
                tools = await session.list_tools()
                assert 'list_files' in [tool.name for tool in tools.tools]

                # Test tool execution
                result = await session.call_tool('list_files', {'directory': '/tmp'})
                assert not result.isError
                assert isinstance(result.content[0].text, str)

    finally:
        server_process.terminate()
        await server_process.wait()

if __name__ == "__main__":
    asyncio.run(test_server_integration())

# src/my_mcp_server/cli.py
import argparse
import asyncio
import sys
from .server import MyCustomMCPServer

async def main():
    parser = argparse.ArgumentParser(description="My Custom MCP Server")
    parser.add_argument("--host", default="localhost", help="Host to bind to")
    parser.add_argument("--port", type=int, default=8080, help="Port to bind to")
    parser.add_argument("--config", help="Path to configuration file")

    args = parser.parse_args()

    server = MyCustomMCPServer("my-server")

    if args.host == "localhost" and args.port == 8080:
        # Default to stdio transport
        server.run()
    else:
        # Use HTTP transport for remote access
        await server.run_http(args.host, args.port)

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

# Security-focused server implementation
from mcp.server.fastmcp import FastMCP
import os
import re
from pathlib import Path

mcp = FastMCP("secure-server")

# Define safe patterns
SAFE_PATH_PATTERN = re.compile(r'^[\w\-./]+$')
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB

@mcp.tool()
def safe_read_file(file_path: str) -> str:
    """
    Safely read file contents with extensive validation.

    Args:
        file_path: Path to the file to read
    """
    # Validate input
    if not file_path:
        raise ValueError("File path cannot be empty")

    if not SAFE_PATH_PATTERN.match(file_path):
        raise ValueError("Invalid characters in file path")

    # Resolve path and prevent directory traversal
    resolved_path = Path(file_path).resolve()

    # Add your security logic here
    if not resolved_path.exists():
        raise FileNotFoundError("File not found")

    if resolved_path.stat().st_size > MAX_FILE_SIZE:
        raise ValueError("File too large")

    try:
        with open(resolved_path, 'r', encoding='utf-8') as f:
            return f.read()
    except UnicodeDecodeError:
        raise ValueError("File contains invalid characters")

from mcp.server.fastmcp import FastMCP
import time
from collections import defaultdict
import threading

mcp = FastMCP("rate-limited-server")

# Simple in-memory rate limiter
class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
        self.lock = threading.Lock()

    def is_allowed(self, client_id: str) -> bool:
        with self.lock:
            now = time.time()
            client_requests = self.requests[client_id]

            # Remove old requests
            self.requests[client_id] = [
                req_time for req_time in client_requests
                if now - req_time < self.time_window
            ]

            if len(self.requests[client_id]) >= self.max_requests:
                return False

            self.requests[client_id].append(now)
            return True

rate_limiter = RateLimiter(max_requests=10, time_window=60)

@mcp.tool()
def expensive_operation(input_data: str, client_id: str = "default") -> str:
    """
    Expensive operation with rate limiting.

    Args:
        input_data: Data to process
        client_id: Client identifier for rate limiting
    """
    if not rate_limiter.is_allowed(client_id):
        raise Exception("Rate limit exceeded. Please try again later.")

    # Perform expensive operation
    time.sleep(1)  # Simulate work
    return f"Processed: {input_data}"

from functools import lru_cache
import time

class CacheManager:
    def __init__(self, max_size: int = 1000, ttl: int = 300):
        self.max_size = max_size
        self.ttl = ttl
        self.cache = {}
        self.timestamps = {}

    def get(self, key: str) -> Optional[Any]:
        if key not in self.cache:
            return None

        # Check TTL
        if time.time() - self.timestamps[key] > self.ttl:
            del self.cache[key]
            del self.timestamps[key]
            return None

        return self.cache[key]

    def set(self, key: str, value: Any) -> None:
        # Implement LRU eviction if needed
        if len(self.cache) >= self.max_size:
            oldest_key = min(self.timestamps.keys(), 
                           key=self.timestamps.get)
            del self.cache[oldest_key]
            del self.timestamps[oldest_key]

        self.cache[key] = value
        self.timestamps[key] = time.time()

# Usage in server
cache = CacheManager()

@mcp.tool()
def cached_operation(data: str) -> str:
    """Operation with caching to improve performance."""
    cache_key = hashlib.md5(data.encode()).hexdigest()

    # Try cache first
    cached_result = cache.get(cache_key)
    if cached_result:
        return cached_result

    # Perform expensive operation
    result = expensive_computation(data)

    # Cache the result
    cache.set(cache_key, result)
    return result

import asyncio
from concurrent.futures import ThreadPoolExecutor

@mcp.tool()
async def parallel_file_operations(file_paths: List[str]) -> List[dict]:
    """
    Process multiple files in parallel for better performance.

    Args:
        file_paths: List of files to process
    """

    # Create thread pool for I/O operations
    with ThreadPoolExecutor(max_workers=4) as executor:
        # Submit all tasks
        loop = asyncio.get_event_loop()
        tasks = [
            loop.run_in_executor(executor, process_single_file, path)
            for path in file_paths
        ]

        # Wait for all to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Filter exceptions and return successful results
        successful_results = [
            result for result in results 
            if not isinstance(result, Exception)
        ]

        return successful_results

def process_single_file(file_path: str) -> dict:
    """Process a single file (runs in thread pool)."""
    # Simulate file processing
    time.sleep(0.1)  # I/O bound operation
    return {"path": file_path, "size": os.path.getsize(file_path)}

import logging
from mcp.server.fastmcp import FastMCP

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('mcp-server.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

mcp = FastMCP("debug-server")

@mcp.tool()
def debug_operation(param: str) -> str:
    """Operation with comprehensive debugging information."""
    logger.info(f"Starting debug_operation with param: {param}")

    try:
        result = perform_computation(param)
        logger.debug(f"Computation result: {result}")
        return result
    except Exception as e:
        logger.exception("Operation failed")
        raise

@mcp.tool()
async def debug_async_operation(data: dict) -> str:
    """Async operation with debug context tracking."""
    logger.info(f"Async operation started with data keys: {list(data.keys())}")

    # Add async context debugging
    current_task = asyncio.current_task()
    logger.debug(f"Running in task: {current_task.get_name() if current_task else 'unknown'}")

    # Process data
    result = await async_data_processing(data)
    logger.info("Async operation completed successfully")
    return result