mcp-server

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

# 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)