diff --git a/README.md b/README.md index 393c3a5..6d26b37 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ Un agente configurabile basato su DeepAgents che supporta più provider LLM. - **Web UI**: Interfaccia web per gestione configurazioni - **Comandi integrati**: Comandi speciali con prefisso "/" per gestione rapida - **Monitoraggio**: Sistema di logging e monitoraggio delle performance +- **Sistema di plugin**: Estendi le funzionalità con plugin personalizzati ## Installazione @@ -206,6 +207,7 @@ Quando interagisci con l'agente, puoi usare comandi speciali preceduti da "/" pe /config use - Cambia la configurazione attiva /models - Elenca i modelli disponibili per il provider attivo /clear - Pulisce lo schermo +/plugins - Elenca i plugin disponibili /exit, /quit - Esci dall'agente ``` @@ -402,6 +404,67 @@ Gli strumenti Ollama specifici sono ancora disponibili per retrocompatibilità: - `ollama_status`: Controlla se Ollama è in esecuzione - `ollama_running_models`: Mostra i modelli attualmente in esecuzione +## Sistema di Plugin + +MaxClaw Agent include un sistema di plugin per estendere le funzionalità. + +### Plugin Inclusi + +- **Example Plugin**: Plugin di esempio che dimostra come creare plugin + +### Creare un Plugin + +Per creare un plugin, crea un file Python nella directory `plugins/`: + +```python +from plugins.base import BasePlugin +from langchain.tools import Tool + +class MyPlugin(BasePlugin): + def __init__(self): + super().__init__( + name="my_plugin", + version="1.0.0", + description="My custom plugin" + ) + + def get_tools(self): + return [ + Tool( + name="my_tool", + func=self.my_tool_func, + description="My custom tool" + ) + ] + + def get_commands(self): + return { + "/mycommand": self.my_command_func + } + + def my_tool_func(self): + return "Hello from my plugin!" + + def my_command_func(self, args=None): + return "My command executed!" +``` + +### Gestire i Plugin + +```bash +# Elenca i plugin disponibili +/plugins + +# Carica tutti i plugin +python -c "from plugins.manager import plugin_manager; plugin_manager.load_all_plugins()" + +# Usa un tool di un plugin +"Use my_tool from my_plugin" + +# Usa un comando di un plugin +/mycommand +``` + ## Test Esegui la suite di test: @@ -462,6 +525,7 @@ Per personalizzare l'agente, modifica `main.py`: - Cambia il prompt di sistema predefinito in `create_maxclaw_agent` - Aggiungi strumenti personalizzati alla chiamata `create_deep_agent` - Configura i sub-agenti o altre funzionalità di DeepAgents +- Crea plugin per estendere le funzionalità ## Monitoraggio e Logging diff --git a/plugins/__init__.py b/plugins/__init__.py new file mode 100644 index 0000000..d77db28 --- /dev/null +++ b/plugins/__init__.py @@ -0,0 +1,13 @@ +""" +MaxClaw Agent Plugins Package +Provides plugin system for extensibility +""" + +from .base import BasePlugin +from .manager import PluginManager, plugin_manager + +__all__ = [ + 'BasePlugin', + 'PluginManager', + 'plugin_manager' +] \ No newline at end of file diff --git a/plugins/base.py b/plugins/base.py new file mode 100644 index 0000000..6901c96 --- /dev/null +++ b/plugins/base.py @@ -0,0 +1,80 @@ +""" +Base plugin class for MaxClaw Agent +""" +from abc import ABC, abstractmethod +from typing import List, Dict, Any, Optional +from langchain.tools import Tool + + +class BasePlugin(ABC): + """Base class for MaxClaw plugins""" + + def __init__(self, name: str, version: str = "1.0.0", description: str = ""): + """ + Initialize plugin + + Args: + name: Plugin name + version: Plugin version + description: Plugin description + """ + self.name = name + self.version = version + self.description = description + self.enabled = True + + @abstractmethod + def get_tools(self) -> List[Tool]: + """ + Get tools provided by this plugin + + Returns: + List of tools + """ + pass + + @abstractmethod + def get_commands(self) -> Dict[str, callable]: + """ + Get commands provided by this plugin + + Returns: + Dictionary mapping command names to functions + """ + pass + + def on_load(self): + """Called when plugin is loaded""" + pass + + def on_unload(self): + """Called when plugin is unloaded""" + pass + + def get_info(self) -> Dict[str, Any]: + """ + Get plugin information + + Returns: + Dictionary with plugin info + """ + return { + "name": self.name, + "version": self.version, + "description": self.description, + "enabled": self.enabled, + "tools_count": len(self.get_tools()), + "commands_count": len(self.get_commands()) + } + + def enable(self): + """Enable the plugin""" + self.enabled = True + + def disable(self): + """Disable the plugin""" + self.enabled = False + + def is_enabled(self) -> bool: + """Check if plugin is enabled""" + return self.enabled \ No newline at end of file diff --git a/plugins/example_plugin.py b/plugins/example_plugin.py new file mode 100644 index 0000000..cb797d2 --- /dev/null +++ b/plugins/example_plugin.py @@ -0,0 +1,137 @@ +""" +Example plugin for MaxClaw Agent +Demonstrates how to create a plugin +""" +import os +import sys +from typing import List, Dict, Any +from langchain.tools import Tool + +# 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 ExamplePlugin(BasePlugin): + """Example plugin that provides simple tools and commands""" + + def __init__(self): + """Initialize example plugin""" + super().__init__( + name="example", + version="1.0.0", + description="Example plugin demonstrating plugin system capabilities" + ) + + def get_tools(self) -> List[Tool]: + """Get tools provided by this plugin""" + return [ + Tool( + name="example_hello", + func=self.hello_tool, + description="Say hello from the example plugin" + ), + Tool( + name="example_echo", + func=self.echo_tool, + description="Echo back the input message" + ), + Tool( + name="example_math", + func=self.math_tool, + description="Perform basic math operations" + ) + ] + + def get_commands(self) -> Dict[str, callable]: + """Get commands provided by this plugin""" + return { + "/example": self.example_command, + "/hello": self.hello_command, + "/math": self.math_command + } + + def hello_tool(self) -> str: + """Hello tool implementation""" + return "👋 Hello from the Example Plugin! This is a tool demonstration." + + def echo_tool(self, message: str) -> str: + """Echo tool implementation""" + return f"🔊 Echo: {message}" + + def math_tool(self, operation: str, a: float, b: float) -> str: + """Math tool implementation""" + try: + if operation == "add": + result = a + b + return f"➕ {a} + {b} = {result}" + elif operation == "subtract": + result = a - b + return f"➖ {a} - {b} = {result}" + elif operation == "multiply": + result = a * b + return f"✖️ {a} × {b} = {result}" + elif operation == "divide": + if b == 0: + return "❌ Error: Division by zero" + result = a / b + return f"➗ {a} ÷ {b} = {result}" + else: + return f"❌ Unknown operation: {operation}" + except Exception as e: + return f"❌ Error: {str(e)}" + + def example_command(self, args: List[str] = None) -> str: + """Example command implementation""" + if args: + return f"📋 Example command executed with args: {', '.join(args)}" + else: + return "📋 Example command executed without arguments" + + def hello_command(self, args: List[str] = None) -> str: + """Hello command implementation""" + if args: + name = " ".join(args) + return f"👋 Hello, {name}! This is a command from the Example Plugin." + else: + return "👋 Hello! This is a command from the Example Plugin." + + def math_command(self, args: List[str] = None) -> str: + """Math command implementation""" + if not args or len(args) < 3: + return "❌ Usage: /math \nOperations: add, subtract, multiply, divide" + + try: + operation = args[0].lower() + a = float(args[1]) + b = float(args[2]) + + if operation == "add": + result = a + b + return f"➕ {a} + {b} = {result}" + elif operation == "subtract": + result = a - b + return f"➖ {a} - {b} = {result}" + elif operation == "multiply": + result = a * b + return f"✖️ {a} × {b} = {result}" + elif operation == "divide": + if b == 0: + return "❌ Error: Division by zero" + result = a / b + return f"➗ {a} ÷ {b} = {result}" + else: + return f"❌ Unknown operation: {operation}" + except ValueError: + return "❌ Error: Invalid numbers" + except Exception as e: + return f"❌ Error: {str(e)}" + + def on_load(self): + """Called when plugin is loaded""" + print(f"📦 Example plugin loaded: {self.name} v{self.version}") + + def on_unload(self): + """Called when plugin is unloaded""" + print(f"🗑️ Example plugin unloaded: {self.name}") \ No newline at end of file diff --git a/plugins/manager.py b/plugins/manager.py new file mode 100644 index 0000000..3e9bda6 --- /dev/null +++ b/plugins/manager.py @@ -0,0 +1,302 @@ +""" +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() \ No newline at end of file