302 lines
8.9 KiB
Python
302 lines
8.9 KiB
Python
"""
|
|
Plugin manager for MaxClaw 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 MaxClaw 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() |