MCP Server Hero

"""Enterprise MCP server example with all advanced features.""" import sys from pathlib import Path # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent.parent)) import asyncio import mcp.types as types from mcp.server.stdio import stdio_server from mcp_server_hero.core.server import MCPServerHero from mcp_server_hero.middleware.rate_limit import RateLimitMiddleware from mcp_server_hero.auth.base import BaseAuthProvider, AuthContext, Permission from mcp_server_hero.plugins.base import BasePlugin, PluginInfo from mcp_server_hero.cache.memory import MemoryCache from mcp_server_hero.metrics.base import MetricsCollector class AdminPlugin(BasePlugin): """Administrative plugin for server management.""" def __init__(self) -> None: super().__init__("admin", "1.0.0") async def initialize(self, server: MCPServerHero) -> None: """Initialize the admin plugin.""" self.server = server # Add admin tools server.add_tool( name="server_stats", tool_func=self.get_server_stats, description="Get comprehensive server statistics", schema={"type": "object", "properties": {}}, ) server.add_tool( name="health_check", tool_func=self.health_check, description="Perform server health check", schema={"type": "object", "properties": {}}, ) server.add_tool( name="cache_stats", tool_func=self.get_cache_stats, description="Get cache statistics", schema={"type": "object", "properties": {}}, ) self._logger.info("Admin plugin initialized") async def cleanup(self) -> None: """Cleanup admin plugin.""" self._logger.info("Admin plugin cleanup") def get_info(self) -> PluginInfo: """Get plugin information.""" return PluginInfo( name=self.name, version=self.version, description="Administrative tools for server management", author="MCP Server Hero", dependencies=[], enabled=self.enabled, ) async def get_server_stats(self) -> str: """Get server statistics.""" if not self.server: return "Server not initialized" stats = self.server.get_server_stats() return f"Server Statistics:\n{stats}" async def health_check(self) -> str: """Perform health check.""" if not self.server: return "Server not initialized" health = await self.server.health_check() return f"Health Check:\n{health}" async def get_cache_stats(self) -> str: """Get cache statistics.""" if not self.server: return "Server not initialized" cache_stats = self.server.cache_manager.get_all_stats() return f"Cache Statistics:\n{cache_stats}" class SimpleAuthProvider(BaseAuthProvider): """Simple authentication provider for demonstration.""" def __init__(self) -> None: super().__init__("simple") self.valid_tokens = { "admin-token": AuthContext( user_id="admin", authenticated=True, permissions={Permission.ADMIN}, metadata={"role": "administrator"} ), "user-token": AuthContext( user_id="user", authenticated=True, permissions={Permission.READ_TOOLS, Permission.CALL_TOOLS, Permission.READ_RESOURCES}, metadata={"role": "user"}, ), } async def authenticate(self, credentials: dict[str, any]) -> AuthContext: """Authenticate using token.""" token = credentials.get("token", "") if token in self.valid_tokens: return self.valid_tokens[token] return AuthContext(authenticated=False) async def authorize(self, context: AuthContext, resource: str, action: str) -> bool: """Authorize access to resource.""" if not context.authenticated: return False # Simple authorization logic if context.has_permission(Permission.ADMIN): return True if action == "read" and context.has_permission(Permission.READ_TOOLS): return True if action == "execute" and context.has_permission(Permission.CALL_TOOLS): return True return False async def advanced_calculation(operation: str, a: float, b: float) -> float: """Advanced calculation with various operations.""" operations = { "add": lambda x, y: x + y, "subtract": lambda x, y: x - y, "multiply": lambda x, y: x * y, "divide": lambda x, y: x / y if y != 0 else 0, "power": lambda x, y: x**y, "sqrt": lambda x, y: x**0.5, # y is ignored "log": lambda x, y: __import__("math").log(x) if x > 0 else 0, } if operation not in operations: raise ValueError(f"Unknown operation: {operation}") return operations[operation](a, b) async def get_system_info() -> str: """Get system information.""" import platform import psutil return f"""System Information: Platform: {platform.platform()} Python: {platform.python_version()} CPU: {psutil.cpu_count()} cores, {psutil.cpu_percent()}% usage Memory: {psutil.virtual_memory().percent}% usage """ async def create_analysis_prompt(data_type: str, analysis_type: str = "basic") -> str: """Create data analysis prompt.""" analysis_templates = { "basic": "Please provide a basic analysis of the following {data_type} data:", "detailed": "Please provide a comprehensive, detailed analysis of the following {data_type} data:", "statistical": "Please provide a statistical analysis of the following {data_type} data:", "predictive": "Please provide a predictive analysis based on the following {data_type} data:", } template = analysis_templates.get(analysis_type, analysis_templates["basic"]) return template.format(data_type=data_type) async def main() -> None: """Run the enterprise MCP server example.""" # Create server with advanced configuration server = MCPServerHero(name="Enterprise MCP Server", debug=True) # Add custom middleware server.add_middleware( RateLimitMiddleware(tool_limit=50, resource_limit=100, prompt_limit=25, refill_rate=1.0, window_size=60) ) # Setup authentication auth_provider = SimpleAuthProvider() server.enable_auth_provider(auth_provider) # Add custom cache server.cache_manager.add_cache( MemoryCache("enterprise", max_size=500, default_ttl=1800) # 30 min TTL ) # Register advanced tools server.add_tool( name="calculate", tool_func=advanced_calculation, description="Perform advanced mathematical calculations", schema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["add", "subtract", "multiply", "divide", "power", "sqrt", "log"], "description": "Mathematical operation to perform", }, "a": {"type": "number", "description": "First operand"}, "b": {"type": "number", "description": "Second operand"}, }, "required": ["operation", "a", "b"], }, ) # Register resource server.add_resource( uri="system://info", resource_func=get_system_info, name="system_info", description="Current system information and statistics", ) # Register prompt server.add_prompt( name="analyze", prompt_func=create_analysis_prompt, description="Create data analysis prompts", arguments=[ {"name": "data_type", "description": "Type of data to analyze", "required": True}, {"name": "analysis_type", "description": "Type of analysis to perform", "required": False}, ], ) # Load admin plugin admin_plugin = AdminPlugin() await server.plugin_manager.register_plugin(admin_plugin) # Initialize server await server.initialize() print("🚀 Enterprise MCP Server started with advanced features:") print(" ✅ Middleware: Validation, Logging, Timing, Rate Limiting") print(" ✅ Authentication: Simple token-based auth") print(" ✅ Caching: Memory caches with TTL") print(" ✅ Metrics: Performance monitoring") print(" ✅ Plugins: Admin plugin loaded") print(" ✅ Advanced Tools: Mathematical calculations") print(" ✅ System Resources: Real-time system info") print(" ✅ Analysis Prompts: Data analysis templates") try: # Run the server async with stdio_server() as streams: await server.run(streams[0], streams[1], server.create_initialization_options()) finally: await server.shutdown() if __name__ == "__main__": asyncio.run(main())