""" Plugin manager for LangCode Agent """ import os import sys import importlib import inspect from typing import Dict, List, Optional, Any, Type from pathlib import Path # Add parent directory to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from plugins.base import BasePlugin class PluginManager: """Manages plugins for LangCode Agent""" def __init__(self, plugin_dir: Optional[str] = None): """ Initialize plugin manager Args: plugin_dir: Directory containing plugins. If None, uses ./plugins """ if plugin_dir is None: self.plugin_dir = Path(__file__).parent else: self.plugin_dir = Path(plugin_dir) self.plugins: Dict[str, BasePlugin] = {} self.loaded_modules = {} def discover_plugins(self) -> List[str]: """ Discover available plugins Returns: List of plugin module names """ plugin_files = [] # Look for Python files in plugin directory for file_path in self.plugin_dir.glob("*.py"): if file_path.name != "__init__.py" and file_path.name != "base.py" and file_path.name != "manager.py": module_name = file_path.stem plugin_files.append(module_name) return plugin_files def load_plugin(self, module_name: str) -> bool: """ Load a plugin from module name Args: module_name: Name of the module to load Returns: True if successful, False otherwise """ try: # Import the module module_path = f"plugins.{module_name}" module = importlib.import_module(module_path) # Find plugin classes in the module for name, obj in inspect.getmembers(module): if (inspect.isclass(obj) and issubclass(obj, BasePlugin) and obj != BasePlugin): # Instantiate the plugin plugin = obj() # Check if plugin with same name already exists if plugin.name in self.plugins: print(f"⚠️ Plugin '{plugin.name}' already loaded") return False # Load the plugin plugin.on_load() self.plugins[plugin.name] = plugin self.loaded_modules[plugin.name] = module print(f"✅ Loaded plugin: {plugin.name} v{plugin.version}") return True print(f"⚠️ No plugin class found in module: {module_name}") return False except Exception as e: print(f"❌ Error loading plugin {module_name}: {e}") return False def load_all_plugins(self): """Load all discovered plugins""" plugin_modules = self.discover_plugins() for module_name in plugin_modules: self.load_plugin(module_name) def unload_plugin(self, plugin_name: str) -> bool: """ Unload a plugin Args: plugin_name: Name of the plugin to unload Returns: True if successful, False otherwise """ if plugin_name not in self.plugins: print(f"⚠️ Plugin '{plugin_name}' not found") return False try: # Unload the plugin plugin = self.plugins[plugin_name] plugin.on_unload() # Remove from dictionaries del self.plugins[plugin_name] if plugin_name in self.loaded_modules: del self.loaded_modules[plugin_name] print(f"✅ Unloaded plugin: {plugin_name}") return True except Exception as e: print(f"❌ Error unloading plugin {plugin_name}: {e}") return False def get_plugin(self, plugin_name: str) -> Optional[BasePlugin]: """ Get a plugin by name Args: plugin_name: Name of the plugin Returns: Plugin instance or None if not found """ return self.plugins.get(plugin_name) def get_all_plugins(self) -> List[BasePlugin]: """ Get all loaded plugins Returns: List of plugin instances """ return list(self.plugins.values()) def get_enabled_plugins(self) -> List[BasePlugin]: """ Get all enabled plugins Returns: List of enabled plugin instances """ return [plugin for plugin in self.plugins.values() if plugin.is_enabled()] def get_all_tools(self) -> List: """ Get all tools from enabled plugins Returns: List of tools """ all_tools = [] for plugin in self.get_enabled_plugins(): try: tools = plugin.get_tools() all_tools.extend(tools) except Exception as e: print(f"⚠️ Error getting tools from plugin {plugin.name}: {e}") return all_tools def get_all_commands(self) -> Dict[str, callable]: """ Get all commands from enabled plugins Returns: Dictionary mapping command names to functions """ all_commands = {} for plugin in self.get_enabled_plugins(): try: commands = plugin.get_commands() all_commands.update(commands) except Exception as e: print(f"⚠️ Error getting commands from plugin {plugin.name}: {e}") return all_commands def enable_plugin(self, plugin_name: str) -> bool: """ Enable a plugin Args: plugin_name: Name of the plugin to enable Returns: True if successful, False otherwise """ plugin = self.get_plugin(plugin_name) if not plugin: print(f"⚠️ Plugin '{plugin_name}' not found") return False plugin.enable() print(f"✅ Enabled plugin: {plugin_name}") return True def disable_plugin(self, plugin_name: str) -> bool: """ Disable a plugin Args: plugin_name: Name of the plugin to disable Returns: True if successful, False otherwise """ plugin = self.get_plugin(plugin_name) if not plugin: print(f"⚠️ Plugin '{plugin_name}' not found") return False plugin.disable() print(f"✅ Disabled plugin: {plugin_name}") return True def get_plugin_info(self, plugin_name: str) -> Optional[Dict[str, Any]]: """ Get information about a plugin Args: plugin_name: Name of the plugin Returns: Dictionary with plugin info or None if not found """ plugin = self.get_plugin(plugin_name) if not plugin: return None return plugin.get_info() def get_all_plugins_info(self) -> List[Dict[str, Any]]: """ Get information about all plugins Returns: List of dictionaries with plugin info """ return [plugin.get_info() for plugin in self.plugins.values()] def reload_plugin(self, plugin_name: str) -> bool: """ Reload a plugin Args: plugin_name: Name of the plugin to reload Returns: True if successful, False otherwise """ # Find the module name for this plugin module_name = None for name, module in self.loaded_modules.items(): if name == plugin_name: # Get the module name from the module object module_name = module.__name__.split('.')[-1] break if not module_name: print(f"⚠️ Could not find module for plugin '{plugin_name}'") return False # Unload the plugin if not self.unload_plugin(plugin_name): return False # Reload the module try: module_path = f"plugins.{module_name}" if module_path in sys.modules: importlib.reload(sys.modules[module_path]) except Exception as e: print(f"⚠️ Error reloading module: {e}") # Load the plugin again return self.load_plugin(module_name) # Global plugin manager instance plugin_manager = PluginManager()