MCP Server Hero

"""Plugin registry for managing plugin metadata and state.""" import json from pathlib import Path from typing import Any, Dict, List, Optional from dataclasses import asdict from .base import PluginInfo from ..utils.logging import get_logger logger = get_logger(__name__) class PluginRegistry: """Registry for plugin metadata and configuration.""" def __init__(self, registry_file: Optional[str | Path] = None) -> None: """Initialize plugin registry. Args: registry_file: Path to registry file (optional) """ self.registry_file = Path(registry_file) if registry_file else Path("plugins.json") self.plugins_info: Dict[str, Dict[str, Any]] = {} self.plugin_configs: Dict[str, Dict[str, Any]] = {} self._logger = get_logger(__name__) # Load existing registry self._load_registry() def _load_registry(self) -> None: """Load plugin registry from file.""" if self.registry_file.exists(): try: with open(self.registry_file, "r") as f: data = json.load(f) self.plugins_info = data.get("plugins", {}) self.plugin_configs = data.get("configs", {}) self._logger.info(f"Loaded plugin registry from {self.registry_file}") except Exception as e: self._logger.error(f"Failed to load plugin registry: {e}") self.plugins_info = {} self.plugin_configs = {} def _save_registry(self) -> None: """Save plugin registry to file.""" try: data = { "plugins": self.plugins_info, "configs": self.plugin_configs, } # Ensure directory exists self.registry_file.parent.mkdir(parents=True, exist_ok=True) with open(self.registry_file, "w") as f: json.dump(data, f, indent=2) self._logger.debug(f"Saved plugin registry to {self.registry_file}") except Exception as e: self._logger.error(f"Failed to save plugin registry: {e}") def register_plugin_info(self, info: PluginInfo) -> None: """Register plugin information. Args: info: Plugin information to register """ self.plugins_info[info.name] = asdict(info) self._save_registry() self._logger.debug(f"Registered plugin info: {info.name}") def unregister_plugin_info(self, plugin_name: str) -> None: """Unregister plugin information. Args: plugin_name: Name of plugin to unregister """ if plugin_name in self.plugins_info: del self.plugins_info[plugin_name] # Also remove config if plugin_name in self.plugin_configs: del self.plugin_configs[plugin_name] self._save_registry() self._logger.debug(f"Unregistered plugin info: {plugin_name}") def get_plugin_info(self, plugin_name: str) -> Optional[PluginInfo]: """Get plugin information by name. Args: plugin_name: Name of plugin Returns: Plugin information or None """ if plugin_name in self.plugins_info: data = self.plugins_info[plugin_name] return PluginInfo(**data) return None def list_registered_plugins(self) -> List[PluginInfo]: """List all registered plugins. Returns: List of plugin information """ return [PluginInfo(**data) for data in self.plugins_info.values()] def is_plugin_enabled(self, plugin_name: str) -> bool: """Check if plugin is enabled. Args: plugin_name: Name of plugin Returns: True if enabled, False otherwise """ info = self.get_plugin_info(plugin_name) return info.enabled if info else False def set_plugin_enabled(self, plugin_name: str, enabled: bool) -> None: """Set plugin enabled state. Args: plugin_name: Name of plugin enabled: Whether plugin should be enabled """ if plugin_name in self.plugins_info: self.plugins_info[plugin_name]["enabled"] = enabled self._save_registry() self._logger.info(f"Plugin {plugin_name} {'enabled' if enabled else 'disabled'}") def set_plugin_config(self, plugin_name: str, config: Dict[str, Any]) -> None: """Set plugin configuration. Args: plugin_name: Name of plugin config: Plugin configuration """ self.plugin_configs[plugin_name] = config self._save_registry() self._logger.debug(f"Updated config for plugin: {plugin_name}") def get_plugin_config(self, plugin_name: str) -> Dict[str, Any]: """Get plugin configuration. Args: plugin_name: Name of plugin Returns: Plugin configuration (empty dict if none) """ return self.plugin_configs.get(plugin_name, {}) def update_plugin_config(self, plugin_name: str, updates: Dict[str, Any]) -> None: """Update plugin configuration. Args: plugin_name: Name of plugin updates: Configuration updates """ if plugin_name not in self.plugin_configs: self.plugin_configs[plugin_name] = {} self.plugin_configs[plugin_name].update(updates) self._save_registry() self._logger.debug(f"Updated config for plugin: {plugin_name}") def get_plugins_by_dependency(self, dependency: str) -> List[str]: """Get plugins that depend on a specific plugin. Args: dependency: Plugin name that others depend on Returns: List of plugin names that depend on the specified plugin """ dependents = [] for plugin_name, info_data in self.plugins_info.items(): dependencies = info_data.get("dependencies", []) if dependency in dependencies: dependents.append(plugin_name) return dependents def validate_dependencies(self) -> Dict[str, List[str]]: """Validate plugin dependencies. Returns: Dictionary mapping plugin names to list of missing dependencies """ missing_deps = {} registered_plugins = set(self.plugins_info.keys()) for plugin_name, info_data in self.plugins_info.items(): dependencies = info_data.get("dependencies", []) missing = [dep for dep in dependencies if dep not in registered_plugins] if missing: missing_deps[plugin_name] = missing return missing_deps def get_load_order(self) -> List[str]: """Get recommended plugin load order based on dependencies. Returns: List of plugin names in load order """ # Simple topological sort remaining = set(self.plugins_info.keys()) ordered = [] while remaining: # Find plugins with no unmet dependencies ready = [] for plugin_name in remaining: info_data = self.plugins_info[plugin_name] dependencies = set(info_data.get("dependencies", [])) # Check if all dependencies are already ordered if dependencies.issubset(set(ordered)): ready.append(plugin_name) if not ready: # Circular dependency or missing dependency # Add remaining plugins anyway (will fail at load time) ready = list(remaining) self._logger.warning("Circular or missing dependencies detected") # Add ready plugins to order for plugin_name in sorted(ready): # Sort for deterministic order ordered.append(plugin_name) remaining.remove(plugin_name) return ordered def export_registry(self, export_path: str | Path) -> None: """Export plugin registry to a different file. Args: export_path: Path to export file """ export_path = Path(export_path) try: data = { "plugins": self.plugins_info, "configs": self.plugin_configs, "metadata": { "exported_from": str(self.registry_file), "export_timestamp": str(Path().resolve()), }, } export_path.parent.mkdir(parents=True, exist_ok=True) with open(export_path, "w") as f: json.dump(data, f, indent=2) self._logger.info(f"Exported plugin registry to {export_path}") except Exception as e: self._logger.error(f"Failed to export plugin registry: {e}") raise def import_registry(self, import_path: str | Path, merge: bool = True) -> None: """Import plugin registry from a file. Args: import_path: Path to import file merge: Whether to merge with existing registry (True) or replace (False) """ import_path = Path(import_path) if not import_path.exists(): raise FileNotFoundError(f"Registry file not found: {import_path}") try: with open(import_path, "r") as f: data = json.load(f) imported_plugins = data.get("plugins", {}) imported_configs = data.get("configs", {}) if merge: self.plugins_info.update(imported_plugins) self.plugin_configs.update(imported_configs) else: self.plugins_info = imported_plugins self.plugin_configs = imported_configs self._save_registry() self._logger.info(f"Imported plugin registry from {import_path}") except Exception as e: self._logger.error(f"Failed to import plugin registry: {e}") raise