from mcp.server import Server from mcp.types import Tool, TextContent import asyncpg import logging from datetime import datetime class PostgresMCPServer: """Production-ready MCP server for PostgreSQL""" def __init__(self, config): self.config = config self.pool = None self.server = Server("postgres-server") self.query_log = [] self._register_tools() async def start(self): """Initialize connection pool with retry logic""" try: self.pool = await asyncpg.create_pool( self.config.database_url, min_size=self.config.min_connections, max_size=self.config.max_connections, timeout=self.config.timeout, command_timeout=self.config.command_timeout ) logging.info("✅ Database pool initialized") except Exception as e: logging.error(f"❌ Failed to connect: {e}") raise def _register_tools(self): """Register available database operations""" @self.server.tool() async def query_database(sql: str, params: list = None): """Execute read-only SQL query""" # Security: Only allow SELECT queries if not sql.strip().upper().startswith('SELECT'): return TextContent( text="❌ Error: Only SELECT queries allowed", mime_type="text/plain" ) try: async with self.pool.acquire() as conn: # Execute with timeout rows = await conn.fetch(sql, *(params or [])) # Audit logging self.query_log.append({ 'timestamp': datetime.utcnow().isoformat(), 'query': sql, 'rows_returned': len(rows) }) return TextContent( text=json.dumps([dict(r) for r in rows], indent=2), mime_type="application/json" ) except Exception as e: logging.error(f"Query failed: {e}") return TextContent( text=f"❌ Error: {str(e)}", mime_type="text/plain" ) # Usage config = DatabaseConfig( database_url="postgresql://user:pass@localhost/db", min_connections=5, max_connections=20, timeout=30, command_timeout=10 ) server = PostgresMCPServer(config) await server.start()

class SafeDatabaseMCP: """Secure database access with sandboxing""" def __init__(self): self.allowed_tables = ['users', 'products', 'orders'] self.read_only = True self.max_rows = 1000 async def query(self, table, conditions): # Validate table access if table not in self.allowed_tables: raise PermissionError(f"❌ Access to {table} not allowed") # Enforce row limits query = f"SELECT * FROM {table} WHERE {conditions} LIMIT {self.max_rows}" # Execute safely return await self.execute_read_only(query)

class APIGatewayMCP: """Multi-API gateway with intelligent caching""" def __init__(self): self.rate_limiter = RateLimiter(calls=100, period=3600) self.cache = TTLCache(maxsize=1000, ttl=300) async def call_api(self, endpoint, params): # Check cache first cache_key = f"{endpoint}:{hash(str(params))}" if cache_key in self.cache: return self.cache[cache_key] # Apply rate limiting await self.rate_limiter.acquire() # Make API call result = await self.http_client.get(endpoint, params=params) # Cache result self.cache[cache_key] = result return result

class FileSystemMCP: """Secure file operations with path validation""" def __init__(self, allowed_paths): self.allowed_paths = [Path(p).resolve() for p in allowed_paths] self.max_file_size = 10_000_000 # 10MB async def read_file(self, path): # Validate path is within allowed directories target = Path(path).resolve() if not any(target.is_relative_to(allowed) for allowed in self.allowed_paths): raise PermissionError(f"❌ Access denied: {path}") # Check file size if target.stat().st_size > self.max_file_size: raise ValueError(f"❌ File too large: {path}") # Read safely async with aiofiles.open(target, 'r') as f: return await f.read()

# production_config.py from dataclasses import dataclass from typing import Dict, List @dataclass class MCPProductionConfig: """Production-grade MCP configuration""" # Connection Settings min_connections: int = 5 max_connections: int = 20 connection_timeout: int = 30 command_timeout: int = 10 # Security Settings require_authentication: bool = True allowed_origins: List[str] = None api_key_header: str = "X-API-Key" # Rate Limiting rate_limit_calls: int = 100 rate_limit_period: int = 3600 # 1 hour # Caching cache_enabled: bool = True cache_ttl: int = 300 # 5 minutes cache_max_size: int = 1000 # Monitoring enable_metrics: bool = True enable_tracing: bool = True log_level: str = "INFO" # Resource Limits max_query_time: int = 30 # seconds max_result_size: int = 10_000_000 # 10MB max_concurrent_requests: int = 100 # Usage config = MCPProductionConfig( min_connections=10, max_connections=50, rate_limit_calls=1000, cache_ttl=600 )

# Connection pooling pool = await create_pool( min_size=5, max_size=20, max_inactive_connection_lifetime=300 ) # Response caching @cache(ttl=300) async def expensive_operation(): """Cache results for 5 minutes""" pass # Batch operations async def batch_query(queries: list): """Execute multiple queries efficiently""" async with pool.acquire() as conn: return await conn.fetch_many(queries) # Parallel execution async def parallel_tools(tool_calls: list): """Execute tools concurrently""" return await asyncio.gather(*[ execute_tool(call) for call in tool_calls ])

# Problem: Bot needs access to tickets and knowledge base servers = { "support_db": "mcp-server-postgres", "knowledge": "mcp-server-elasticsearch", "slack": "mcp-server-slack", "email": "mcp-server-sendgrid" } # Result: Unified support experience response = await bot.handle_inquiry( "What's the status of ticket #12345?" )

# Problem: Analyst needs SQL + Python + visualization servers = { "warehouse": "mcp-server-snowflake", "jupyter": "mcp-server-jupyter", "charts": "mcp-server-plotly", "reports": "mcp-server-s3" } # Result: End-to-end analysis automation analysis = await analyst.analyze( "Show me Q4 sales trends by region with forecast" )

# Problem: Manage infrastructure through chat servers = { "k8s": "mcp-server-kubernetes", "metrics": "mcp-server-prometheus", "logs": "mcp-server-elasticsearch", "incidents": "mcp-server-pagerduty" } # Result: ChatOps for infrastructure await devops.execute( "Scale production to 10 replicas and monitor for 5 minutes" )

def calculate_mcp_roi( num_integrations: int, num_llms: int, avg_integration_cost: float = 10000, avg_integration_time_hours: int = 80 ): """Calculate ROI of using MCP vs custom integrations""" # Without MCP: M × N integrations without_mcp_cost = num_integrations * num_llms * avg_integration_cost without_mcp_time = num_integrations * num_llms * avg_integration_time_hours # With MCP: M + N integrations with_mcp_cost = (num_integrations + num_llms) * avg_integration_cost * 0.3 with_mcp_time = (num_integrations + num_llms) * avg_integration_time_hours * 0.2 savings = without_mcp_cost - with_mcp_cost time_saved = without_mcp_time - with_mcp_time return { "cost_savings": f"${savings:,.0f}", "time_saved_hours": f"{time_saved:,.0f}", "roi_percentage": f"{(savings / with_mcp_cost * 100):.0f}%" } # Example: 10 integrations, 3 LLMs roi = calculate_mcp_roi(10, 3) # Result: $210,000 saved, 1,760 hours saved, 350% ROI