MCP Server Hero

"""Base plugin system.""" import asyncio from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional, TYPE_CHECKING from dataclasses import dataclass from ..utils.logging import get_logger if TYPE_CHECKING: from ..core.server import MCPServerHero logger = get_logger(__name__) @dataclass class PluginInfo: """Plugin information metadata.""" name: str version: str description: str author: str dependencies: List[str] enabled: bool = True class BasePlugin(ABC): """Base class for all plugins.""" def __init__(self, name: str, version: str = "1.0.0") -> None: """Initialize plugin. Args: name: Plugin name version: Plugin version """ self.name = name self.version = version self.enabled = True self.server: Optional["MCPServerHero"] = None self._logger = get_logger(f"{__name__}.{name}") @abstractmethod async def initialize(self, server: "MCPServerHero") -> None: """Initialize the plugin with server instance. Args: server: MCP server instance """ pass @abstractmethod async def cleanup(self) -> None: """Cleanup plugin resources.""" pass @abstractmethod def get_info(self) -> PluginInfo: """Get plugin information. Returns: Plugin information """ pass async def enable(self) -> None: """Enable the plugin.""" if not self.enabled: self.enabled = True await self.on_enable() self._logger.info(f"Plugin {self.name} enabled") async def disable(self) -> None: """Disable the plugin.""" if self.enabled: self.enabled = False await self.on_disable() self._logger.info(f"Plugin {self.name} disabled") async def on_enable(self) -> None: """Called when plugin is enabled.""" pass async def on_disable(self) -> None: """Called when plugin is disabled.""" pass def is_compatible(self, server_version: str) -> bool: """Check if plugin is compatible with server version. Args: server_version: Server version Returns: True if compatible """ return True # Override in specific plugins class PluginManager: """Manager for handling plugins.""" def __init__(self) -> None: """Initialize plugin manager.""" self.plugins: Dict[str, BasePlugin] = {} self.plugin_order: List[str] = [] self._logger = get_logger(__name__) self._server: Optional["MCPServerHero"] = None async def set_server(self, server: "MCPServerHero") -> None: """Set the server instance. Args: server: MCP server instance """ self._server = server async def register_plugin(self, plugin: BasePlugin) -> None: """Register a plugin. Args: plugin: Plugin instance to register """ if plugin.name in self.plugins: self._logger.warning(f"Plugin {plugin.name} is already registered") return # Check dependencies info = plugin.get_info() for dep in info.dependencies: if dep not in self.plugins: raise RuntimeError(f"Plugin {plugin.name} requires dependency: {dep}") self.plugins[plugin.name] = plugin self.plugin_order.append(plugin.name) # Initialize if server is available if self._server: await plugin.initialize(self._server) plugin.server = self._server self._logger.info(f"Registered plugin: {plugin.name} v{plugin.version}") async def unregister_plugin(self, plugin_name: str) -> None: """Unregister a plugin. Args: plugin_name: Name of plugin to unregister """ if plugin_name not in self.plugins: self._logger.warning(f"Plugin {plugin_name} is not registered") return # Check if other plugins depend on this one for name, plugin in self.plugins.items(): if name != plugin_name: info = plugin.get_info() if plugin_name in info.dependencies: raise RuntimeError(f"Cannot unregister {plugin_name}: {name} depends on it") plugin = self.plugins[plugin_name] await plugin.cleanup() del self.plugins[plugin_name] self.plugin_order.remove(plugin_name) self._logger.info(f"Unregistered plugin: {plugin_name}") async def enable_plugin(self, plugin_name: str) -> None: """Enable a plugin. Args: plugin_name: Name of plugin to enable """ if plugin_name not in self.plugins: raise ValueError(f"Plugin {plugin_name} is not registered") await self.plugins[plugin_name].enable() async def disable_plugin(self, plugin_name: str) -> None: """Disable a plugin. Args: plugin_name: Name of plugin to disable """ if plugin_name not in self.plugins: raise ValueError(f"Plugin {plugin_name} is not registered") await self.plugins[plugin_name].disable() async def initialize_all(self) -> None: """Initialize all registered plugins.""" if not self._server: raise RuntimeError("Server instance not set") for plugin_name in self.plugin_order: plugin = self.plugins[plugin_name] if plugin.enabled: try: await plugin.initialize(self._server) plugin.server = self._server self._logger.info(f"Initialized plugin: {plugin_name}") except Exception as e: self._logger.error(f"Failed to initialize plugin {plugin_name}: {e}") plugin.enabled = False async def cleanup_all(self) -> None: """Cleanup all plugins.""" # Cleanup in reverse order for plugin_name in reversed(self.plugin_order): plugin = self.plugins[plugin_name] try: await plugin.cleanup() self._logger.info(f"Cleaned up plugin: {plugin_name}") except Exception as e: self._logger.error(f"Error cleaning up plugin {plugin_name}: {e}") def get_plugin(self, plugin_name: str) -> Optional[BasePlugin]: """Get a plugin by name. Args: plugin_name: Name of plugin to get Returns: Plugin instance or None """ return self.plugins.get(plugin_name) def list_plugins(self) -> List[PluginInfo]: """List all registered plugins. Returns: List of plugin information """ return [plugin.get_info() for plugin in self.plugins.values()] def get_enabled_plugins(self) -> List[str]: """Get list of enabled plugin names. Returns: List of enabled plugin names """ return [name for name, plugin in self.plugins.items() if plugin.enabled] def get_plugin_stats(self) -> Dict[str, Any]: """Get plugin statistics. Returns: Plugin statistics """ total = len(self.plugins) enabled = len(self.get_enabled_plugins()) return { "total_plugins": total, "enabled_plugins": enabled, "disabled_plugins": total - enabled, "plugin_names": list(self.plugins.keys()), }