--- name: mcp-server description: Generic MCP (Model Context Protocol) server development patterns. Provides reusable architecture and best practices for building MCP servers that expose any domain-specific operations as tools for AI agents. Framework-agnostic implementation supporting async operations, error handling, and enterprise-grade features. license: MIT --- # Generic MCP Server Development This skill provides comprehensive patterns and reusable code for building MCP (Model Context Protocol) servers that can expose any domain operations as tools for AI agents. Follows 2025 best practices for performance, security, and maintainability. ## When to Use This Skill Use this skill when you need to: - Build an MCP server for any domain (not just todos) - Expose database operations as MCP tools - Create AI-agent accessible APIs - Implement async MCP tool handlers - Add proper error handling and validation - Support rate limiting and caching - Build enterprise-grade MCP servers - Integrate with multiple storage backends ## 1. Generic MCP Server Architecture ```python # mcp_server/core.py #!/usr/bin/env python3 """ Generic MCP Server Base Architecture Provides reusable patterns for any MCP server implementation """ import asyncio import json import logging from abc import ABC, abstractmethod from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Sequence, Union, Callable from contextlib import asynccontextmanager from dataclasses import dataclass, field from enum import Enum import redis.asyncio as redis from mcp.server import Server, NotificationOptions, stdio from mcp.server.models import InitializationOptions from mcp.server.stdio import stdio_server from mcp.types import ( Resource, Tool, TextContent, ImageContent, EmbeddedResource, LoggingLevel, CallToolRequest, EmptyResult, ListResourcesRequest, ListToolsRequest, ReadResourceRequest, GetPromptRequest, ListPromptsRequest ) from pydantic import BaseModel, Field, validator import aiofiles import yaml from pathlib import Path # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("mcp_server") class ServerConfig(BaseModel): """MCP Server configuration""" name: str version: str = "1.0.0" description: str debug: bool = False redis_url: Optional[str] = None rate_limit_requests: int = 100 rate_limit_window: int = 60 cache_ttl: int = 300 max_retries: int = 3 timeout: int = 30 class Config: extra = "allow" @dataclass class RequestContext: """Request context for tool calls""" user_id: str session_id: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) timestamp: datetime = field(default_factory=datetime.utcnow) class RateLimiter: """Redis-based rate limiter for MCP operations""" def __init__(self, redis_url: str, requests: int, window: int): self.redis_url = redis_url self.requests = requests self.window = window self._redis = None async def _get_redis(self): if not self._redis: self._redis = await redis.from_url(self.redis_url) return self._redis async def is_allowed(self, key: str) -> bool: """Check if request is allowed""" r = await self._get_redis() current = await r.incr(f"rate_limit:{key}") if current == 1: await r.expire(f"rate_limit:{key}", self.window) return current <= self.requests async def get_remaining(self, key: str) -> int: """Get remaining requests""" r = await self._get_redis() current = await r.get(f"rate_limit:{key}") return max(0, self.requests - int(current or 0)) class CacheManager: """Redis-based caching for MCP responses""" def __init__(self, redis_url: str, ttl: int = 300): self.redis_url = redis_url self.ttl = ttl self._redis = None async def _get_redis(self): if not self._redis: self._redis = await redis.from_url(self.redis_url) return self._redis def _make_key(self, tool_name: str, args: Dict[str, Any]) -> str: """Generate cache key from tool name and arguments""" import hashlib args_str = json.dumps(args, sort_keys=True) return f"cache:{tool_name}:{hashlib.md5(args_str.encode()).hexdigest()}" async def get(self, tool_name: str, args: Dict[str, Any]) -> Optional[Any]: """Get cached result""" r = await self._get_redis() key = self._make_key(tool_name, args) result = await r.get(key) return json.loads(result) if result else None async def set(self, tool_name: str, args: Dict[str, Any], value: Any): """Cache result""" r = await self._get_redis() key = self._make_key(tool_name, args) await r.setex(key, self.ttl, json.dumps(value)) class BaseMCPServer: """Base MCP Server with common functionality""" def __init__(self, config: ServerConfig): self.config = config self.server = Server(config.name) self.tools: Dict[str, Callable] = {} self.rate_limiter: Optional[RateLimiter] = None self.cache: Optional[CacheManager] = None # Setup optional components if config.redis_url: self.rate_limiter = RateLimiter( config.redis_url, config.rate_limit_requests, config.rate_limit_window ) self.cache = CacheManager( config.redis_url, config.cache_ttl ) # Register handlers self._register_handlers() logger.info(f"MCP Server '{config.name}' initialized") def _register_handlers(self): """Register MCP handlers""" @self.server.list_tools() async def handle_list_tools() -> List[Tool]: """Return list of available tools""" return await self.list_tools() @self.server.call_tool() async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: """Handle tool call with rate limiting and caching""" return await self.call_tool(name, arguments) def register_tool(self, name: str, handler: Callable, schema: Dict[str, Any]): """Register a new tool""" self.tools[name] = { "handler": handler, "schema": schema } logger.info(f"Registered tool: {name}") async def list_tools(self) -> List[Tool]: """List all available tools""" tools = [] for name, tool_info in self.tools.items(): tools.append(Tool( name=name, description=tool_info["schema"].get("description", ""), inputSchema=tool_info["schema"].get("inputSchema", {}) )) return tools async def call_tool(self, name: str, arguments: Dict[str, Any]) -> List[TextContent]: """Execute a tool call with full middleware pipeline""" start_time = datetime.utcnow() try: # Extract context from arguments context = self._extract_context(arguments) # Rate limiting check if self.rate_limiter: rate_key = f"{context.user_id}:{name}" if not await self.rate_limiter.is_allowed(rate_key): return [TextContent( type="text", text=json.dumps({ "status": "error", "error": "Rate limit exceeded", "remaining": await self.rate_limiter.get_remaining(rate_key) }) )] # Check cache if self.cache and self._is_cacheable(name): cached_result = await self.cache.get(name, arguments) if cached_result: logger.info(f"Cache hit for tool: {name}") return [TextContent( type="text", text=json.dumps(cached_result) )] # Validate tool exists if name not in self.tools: raise ValueError(f"Unknown tool: {name}") # Validate arguments schema = self.tools[name]["schema"] self._validate_arguments(arguments, schema) # Execute tool handler = self.tools[name]["handler"] result = await self._execute_tool(handler, arguments, context) # Cache result if applicable if self.cache and self._is_cacheable(name) and result.get("status") != "error": await self.cache.set(name, arguments, result) # Log execution duration = (datetime.utcnow() - start_time).total_seconds() logger.info(f"Tool {name} executed in {duration:.2f}s for user {context.user_id}") return [TextContent( type="text", text=json.dumps(result, default=str) )] except Exception as e: logger.error(f"Error executing tool {name}: {str(e)}", exc_info=True) duration = (datetime.utcnow() - start_time).total_seconds() logger.error(f"Tool {name} failed after {duration:.2f}s") return [TextContent( type="text", text=json.dumps({ "status": "error", "error": str(e), "tool": name, "timestamp": datetime.utcnow().isoformat() }) )] def _extract_context(self, arguments: Dict[str, Any]) -> RequestContext: """Extract request context from arguments""" user_id = arguments.pop("_user_id", "anonymous") session_id = arguments.pop("_session_id", None) metadata = arguments.pop("_metadata", {}) return RequestContext( user_id=user_id, session_id=session_id, metadata=metadata ) def _validate_arguments(self, arguments: Dict[str, Any], schema: Dict[str, Any]): """Validate tool arguments against schema""" # Basic validation - can be extended with pydantic input_schema = schema.get("inputSchema", {}) required = input_schema.get("required", []) properties = input_schema.get("properties", {}) # Check required fields for field in required: if field not in arguments: raise ValueError(f"Missing required field: {field}") # Validate field types for field, value in arguments.items(): if field in properties: field_schema = properties[field] expected_type = field_schema.get("type") if expected_type == "string" and not isinstance(value, str): raise ValueError(f"Field {field} must be a string") elif expected_type == "integer" and not isinstance(value, int): raise ValueError(f"Field {field} must be an integer") elif expected_type == "array" and not isinstance(value, list): raise ValueError(f"Field {field} must be an array") # Check enum values if "enum" in field_schema and value not in field_schema["enum"]: raise ValueError(f"Field {field} must be one of {field_schema['enum']}") def _is_cacheable(self, tool_name: str) -> bool: """Determine if tool result should be cached""" # Non-mutating operations are cacheable non_mutating = ["get", "list", "search", "find", "read"] return any(op in tool_name.lower() for op in non_mutating) async def _execute_tool(self, handler: Callable, arguments: Dict[str, Any], context: RequestContext) -> Dict[str, Any]: """Execute tool handler with error handling""" try: # Pass context to handler if it accepts it import inspect sig = inspect.signature(handler) if 'context' in sig.parameters: result = await handler(arguments, context=context) else: result = await handler(arguments) return result except Exception as e: logger.error(f"Tool handler failed: {str(e)}") return { "status": "error", "error": str(e), "timestamp": datetime.utcnow().isoformat() } async def run(self): """Start the MCP server""" logger.info(f"Starting MCP server: {self.config.name}") async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, InitializationOptions( server_name=self.config.name, server_version=self.config.version, capabilities=self.server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ) ) ) def tool( name: Optional[str] = None, description: str = "", input_schema: Optional[Dict[str, Any]] = None ): """Decorator for registering MCP tools""" def decorator(func): tool_name = name or func.__name__ schema = { "description": description or func.__doc__ or "", "inputSchema": input_schema or {} } # Store schema on function for later registration func._mcp_tool_schema = schema func._mcp_tool_name = tool_name return func return decorator ``` ## 2. Database Integration Patterns ```python # mcp_server/database.py """ Generic Database Integration for MCP Servers Supports multiple ORMs and connection patterns """ import asyncio from abc import ABC, abstractmethod from contextlib import asynccontextmanager from typing import Any, Dict, List, Optional, TypeVar, Generic, Union from datetime import datetime import json from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, Text, Boolean, select, update, delete, insert from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import sessionmaker, declarative_base from sqlalchemy.pool import NullPool import asyncpg import motor.motor_asyncio from redis.asyncio import Redis # Type variables T = TypeVar('T') class DatabaseBackend(ABC): """Abstract base for database backends""" @abstractmethod async def connect(self): """Establish connection""" pass @abstractmethod async def disconnect(self): """Close connection""" pass @abstractmethod async def execute_query(self, query: str, params: Dict[str, Any] = None) -> List[Dict[str, Any]]: """Execute a query and return results""" pass @abstractmethod async def execute_command(self, command: str, params: Dict[str, Any] = None) -> Any: """Execute a command (INSERT, UPDATE, DELETE)""" pass class PostgresBackend(DatabaseBackend): """PostgreSQL backend using asyncpg""" def __init__(self, connection_string: str): self.connection_string = connection_string self.pool: Optional[asyncpg.Pool] = None async def connect(self): self.pool = await asyncpg.create_pool( self.connection_string, min_size=5, max_size=20, command_timeout=60 ) async def disconnect(self): if self.pool: await self.pool.close() async def execute_query(self, query: str, params: Dict[str, Any] = None) -> List[Dict[str, Any]]: async with self.pool.acquire() as conn: rows = await conn.fetch(query, *params.values() if params else []) return [dict(row) for row in rows] async def execute_command(self, command: str, params: Dict[str, Any] = None) -> Any: async with self.pool.acquire() as conn: return await conn.execute(command, *params.values() if params else []) class SQLAlchemyBackend(DatabaseBackend): """SQLAlchemy backend for multiple databases""" def __init__(self, database_url: str, async_mode: bool = True): self.database_url = database_url self.async_mode = async_mode self.engine = None self.session_factory = None async def connect(self): if self.async_mode: self.engine = create_async_engine( self.database_url, pool_pre_ping=True, pool_recycle=300, echo=False ) self.session_factory = async_sessionmaker( self.engine, class_=AsyncSession, expire_on_commit=False ) else: self.engine = create_engine( self.database_url, pool_pre_ping=True, pool_recycle=300, echo=False ) self.session_factory = sessionmaker( bind=self.engine, expire_on_commit=False ) async def disconnect(self): if self.engine: await self.engine.dispose() @asynccontextmanager async def get_session(self): """Get database session""" async with self.session_factory() as session: try: yield session if self.async_mode: await session.commit() else: session.commit() except Exception: if self.async_mode: await session.rollback() else: session.rollback() raise finally: if self.async_mode: await session.close() else: session.close() async def execute_query(self, query: Any, params: Dict[str, Any] = None) -> List[Dict[str, Any]]: """Execute SQLAlchemy query""" async with self.get_session() as session: if isinstance(query, str): # Raw SQL query result = await session.execute(query, params or {}) rows = result.fetchall() return [dict(row._mapping) for row in rows] else: # SQLAlchemy ORM query result = await session.execute(query) rows = result.fetchall() return [dict(row._mapping) for row in rows] async def execute_command(self, command: Any, params: Dict[str, Any] = None) -> Any: """Execute SQLAlchemy command""" async with self.get_session() as session: if isinstance(command, str): # Raw SQL command result = await session.execute(command, params or {}) await session.commit() return result else: # SQLAlchemy ORM command await session.execute(command, params or {}) await session.commit() return None class MongoBackend(DatabaseBackend): """MongoDB backend using motor""" def __init__(self, connection_string: str, database_name: str): self.connection_string = connection_string self.database_name = database_name self.client = None self.db = None async def connect(self): self.client = motor.motor_asyncio.AsyncIOMotorClient(self.connection_string) self.db = self.client[self.database_name] async def disconnect(self): if self.client: self.client