From 9f3298fa13fead56360f589fed63daf5d67739f9 Mon Sep 17 00:00:00 2001 From: truetype Date: Fri, 20 Mar 2026 16:08:26 +0100 Subject: [PATCH] Phase 2: Enhanced TUI/CLI with search, export/import, and custom tools for Proxmox and Ollama --- README.md | 50 ++++++ config_cli.py | 257 ++++++++++++++++++++++++++++- config_tui.py | 299 ++++++++++++++++++++++++++++++++-- main.py | 69 +++++++- tools/__init__.py | 14 ++ tools/ollama_tools.py | 360 +++++++++++++++++++++++++++++++++++++++++ tools/proxmox_tools.py | 320 ++++++++++++++++++++++++++++++++++++ 7 files changed, 1353 insertions(+), 16 deletions(-) create mode 100644 tools/__init__.py create mode 100644 tools/ollama_tools.py create mode 100644 tools/proxmox_tools.py diff --git a/README.md b/README.md index 985a64f..e5be1ea 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ Un agente configurabile basato su DeepAgents che supporta più provider LLM. - **Configurazione semplice**: Imposta le variabili d'ambiente per scegliere il tuo provider e modello - **Configurazione persistente**: Gestisci e conserva le configurazioni tramite CLI o TUI - **Estensibile**: Aggiungi strumenti personalizzati, modifica i prompt e configura i sub-agenti secondo le tue esigenze +- **Strumenti personalizzati**: Include strumenti per gestire Proxmox e Ollama ## Installazione @@ -65,6 +66,9 @@ Utilizza la CLI o la TUI per gestire le configurazioni che persistono tra le ses # Elenca le configurazioni python config_cli.py list +# Cerca configurazioni +python config_cli.py search "ollama" + # Aggiungi una nuova configurazione interattivamente python config_cli.py add @@ -82,6 +86,12 @@ python config_cli.py active # Testa la connessione a una configurazione python config_cli.py test "mio-ollama" + +# Esporta configurazioni in JSON +python config_cli.py export --format json --output backup + +# Importa configurazioni da JSON +python config_cli.py import backup.json ``` #### Utilizzo della TUI @@ -92,6 +102,12 @@ python config_tui.py Naviga usando le frecce direzionali, premi Invio per selezionare e 'q' per uscire. +La TUI ora include: +- Ricerca e filtro delle configurazioni +- Esportazione/importazione in JSON/YAML +- Test di connessione diretto +- Navigazione migliorata + ## Utilizzo Esegui l'agente: @@ -101,6 +117,40 @@ python main.py Puoi quindi interagire con l'agente nel terminale. Digita 'exit' per uscire. +## Strumenti Personalizzati + +MaxClaw Agent include strumenti personalizzati per gestire il tuo ambiente: + +### Strumenti Proxmox +Configura le variabili d'ambiente per abilitare gli strumenti Proxmox: +```bash +export PROXMOX_HOST="192.168.0.100" +export PROXMOX_USER="root" +export PROXMOX_PASSWORD="tua-password" # oppure +export PROXMOX_TOKEN="tuo-api-token" +``` + +Strumenti disponibili: +- `proxmox_list_vms`: Elenca tutte le VM nel cluster +- `proxmox_vm_status`: Ottieni lo stato di una VM specifica +- `proxmox_start_vm`: Avvia una VM +- `proxmox_stop_vm`: Ferma una VM +- `proxmox_restart_vm`: Riavvia una VM +- `proxmox_create_snapshot`: Crea uno snapshot +- `proxmox_list_snapshots`: Elenca gli snapshot di una VM + +### Strumenti Ollama +Gli strumenti Ollama sono sempre disponibili e utilizzano l'URL configurato in `OLLAMA_BASE_URL`. + +Strumenti disponibili: +- `ollama_list_models`: Elenca tutti i modelli disponibili +- `ollama_model_info`: Ottieni informazioni dettagliate su un modello +- `ollama_pull_model`: Scarica un nuovo modello +- `ollama_delete_model`: Elimina un modello +- `ollama_generate`: Genera una completion +- `ollama_status`: Controlla se Ollama è in esecuzione +- `ollama_running_models`: Mostra i modelli attualmente in esecuzione + ## Personalizzazione Per personalizzare l'agente, modifica `main.py`: diff --git a/config_cli.py b/config_cli.py index 5063247..ad61272 100644 --- a/config_cli.py +++ b/config_cli.py @@ -1,22 +1,27 @@ """ Command-line interface for managing MaxClaw Agent configurations +Enhanced with export/import, search, and better formatting """ import sys import argparse -from typing import Optional +import json +import yaml +from pathlib import Path +from typing import Optional, Dict, Any from config_manager import config_manager -def print_configurations(): +def print_configurations(configs=None, title="MaxClaw Agent Configurations"): """Print all configurations in a formatted way""" - configs = config_manager.get_configurations() + if configs is None: + configs = config_manager.get_configurations() active = config_manager.get_active_configuration() if not configs: print("No configurations found.") return - print("\nMaxClaw Agent Configurations:") + print(f"\n{title}:") print("=" * 50) for name, config in configs.items(): @@ -41,6 +46,41 @@ def print_configurations(): print() +def filter_configurations(configs: Dict[str, Any], search_term: str) -> Dict[str, Any]: + """Filter configurations based on search term""" + if not search_term: + return configs + + search_term = search_term.lower() + filtered = {} + + for name, config in configs.items(): + # Search in name + if search_term in name.lower(): + filtered[name] = config + continue + + # Search in provider + provider = config.get("provider", "").lower() + if search_term in provider: + filtered[name] = config + continue + + # Search in model name + model_name = config.get("model_name", "").lower() + if search_term in model_name: + filtered[name] = config + continue + + # Search in base URL + base_url = config.get("base_url", "").lower() + if search_term in base_url: + filtered[name] = config + continue + + return filtered + + def add_configuration_interactive(): """Interactively add a new configuration""" print("\nAdd New Configuration") @@ -247,6 +287,122 @@ def test_configuration(): print(f"❌ {message}") +def search_configurations(): + """Search/filter configurations""" + configs = config_manager.get_configurations() + if not configs: + print("No configurations to search.") + return + + search_term = input("Search term (name, provider, model, URL): ").strip() + if not search_term: + print("Search cancelled.") + return + + filtered = filter_configurations(configs, search_term) + + if not filtered: + print(f"No configurations found matching '{search_term}'.") + else: + print_configurations(filtered, f"Search Results for '{search_term}'") + + +def export_configurations(): + """Export configurations to JSON or YAML""" + configs = config_manager.get_configurations() + if not configs: + print("No configurations to export.") + return + + print("\nExport Configurations") + print("=" * 30) + + # Select format + print("Select export format:") + print("1. JSON") + print("2. YAML") + + format_choice = input("Enter choice (1-2): ").strip() + if format_choice not in ["1", "2"]: + print("Invalid choice.") + return + + # Get filename + filename = input("Filename (without extension): ").strip() + if not filename: + print("Cancelled.") + return + + # Export configurations + try: + if format_choice == "1": # JSON + export_file = Path(f"{filename}.json") + with open(export_file, 'w') as f: + json.dump(configs, f, indent=2) + else: # YAML + export_file = Path(f"{filename}.yaml") + with open(export_file, 'w') as f: + yaml.dump(configs, f, default_flow_style=False) + + print(f"✅ Configurations exported to {export_file}") + except Exception as e: + print(f"❌ Error exporting configurations: {e}") + + +def import_configurations(): + """Import configurations from JSON or YAML""" + print("\nImport Configurations") + print("=" * 30) + + # Get filename + filename = input("Filename (with extension .json or .yaml): ").strip() + if not filename: + print("Cancelled.") + return + + # Check if file exists + file_path = Path(filename) + if not file_path.exists(): + print(f"❌ Error: File '{filename}' not found.") + return + + # Import configurations + try: + with open(file_path, 'r') as f: + if filename.endswith('.json'): + imported_configs = json.load(f) + elif filename.endswith(('.yaml', '.yml')): + imported_configs = yaml.safe_load(f) + else: + print("❌ Error: Unsupported file format. Use .json or .yaml") + return + + # Validate imported configurations + if not isinstance(imported_configs, dict): + print("❌ Error: Invalid configuration format.") + return + + # Ask for confirmation + confirm = input(f"Import {len(imported_configs)} configurations? (y/N): ").strip().lower() + if confirm != 'y': + print("Cancelled.") + return + + # Import configurations + imported_count = 0 + for name, config in imported_configs.items(): + try: + config_manager.add_configuration(name, config) + imported_count += 1 + print(f"✅ Imported configuration: {name}") + except Exception as e: + print(f"⚠️ Warning: Could not import '{name}': {e}") + + print(f"\n✅ Successfully imported {imported_count} configurations.") + except Exception as e: + print(f"❌ Error importing configurations: {e}") + + def main(): """Main CLI entry point""" parser = argparse.ArgumentParser(description="MaxClaw Agent Configuration Manager") @@ -255,6 +411,10 @@ def main(): # List command subparsers.add_parser('list', help='List all configurations') + # Search command + search_parser = subparsers.add_parser('search', help='Search configurations') + search_parser.add_argument('term', nargs='?', help='Search term') + # Add command add_parser = subparsers.add_parser('add', help='Add a new configuration') add_parser.add_argument('--non-interactive', action='store_true', @@ -281,11 +441,33 @@ def main(): test_parser = subparsers.add_parser('test', help='Test connection to a configuration') test_parser.add_argument('name', nargs='?', help='Configuration name to test') + # Export command + export_parser = subparsers.add_parser('export', help='Export configurations to JSON/YAML') + export_parser.add_argument('--format', choices=['json', 'yaml'], help='Export format') + export_parser.add_argument('--output', help='Output filename (without extension)') + + # Import command + import_parser = subparsers.add_parser('import', help='Import configurations from JSON/YAML') + import_parser.add_argument('file', nargs='?', help='File to import') + args = parser.parse_args() if args.command == 'list': print_configurations() + elif args.command == 'search': + if args.term: + # Non-interactive search + configs = config_manager.get_configurations() + filtered = filter_configurations(configs, args.term) + if not filtered: + print(f"No configurations found matching '{args.term}'.") + else: + print_configurations(filtered, f"Search Results for '{args.term}'") + else: + # Interactive search + search_configurations() + elif args.command == 'add': if args.non_interactive: # Non-interactive mode @@ -368,6 +550,73 @@ def main(): # Interactive test test_configuration() + elif args.command == 'export': + if args.format and args.output: + # Non-interactive export + configs = config_manager.get_configurations() + if not configs: + print("No configurations to export.") + return 1 + + try: + if args.format == 'json': + export_file = Path(f"{args.output}.json") + with open(export_file, 'w') as f: + json.dump(configs, f, indent=2) + else: # yaml + export_file = Path(f"{args.output}.yaml") + with open(export_file, 'w') as f: + yaml.dump(configs, f, default_flow_style=False) + + print(f"✅ Configurations exported to {export_file}") + except Exception as e: + print(f"❌ Error exporting configurations: {e}") + return 1 + else: + # Interactive export + export_configurations() + + elif args.command == 'import': + if args.file: + # Non-interactive import + file_path = Path(args.file) + if not file_path.exists(): + print(f"❌ Error: File '{args.file}' not found.") + return 1 + + try: + with open(file_path, 'r') as f: + if args.file.endswith('.json'): + imported_configs = json.load(f) + elif args.file.endswith(('.yaml', '.yml')): + imported_configs = yaml.safe_load(f) + else: + print("❌ Error: Unsupported file format. Use .json or .yaml") + return 1 + + # Validate imported configurations + if not isinstance(imported_configs, dict): + print("❌ Error: Invalid configuration format.") + return 1 + + # Import configurations + imported_count = 0 + for name, config in imported_configs.items(): + try: + config_manager.add_configuration(name, config) + imported_count += 1 + print(f"✅ Imported configuration: {name}") + except Exception as e: + print(f"⚠️ Warning: Could not import '{name}': {e}") + + print(f"\n✅ Successfully imported {imported_count} configurations.") + except Exception as e: + print(f"❌ Error importing configurations: {e}") + return 1 + else: + # Interactive import + import_configurations() + else: parser.print_help() diff --git a/config_tui.py b/config_tui.py index 23a6e48..965d43d 100644 --- a/config_tui.py +++ b/config_tui.py @@ -1,19 +1,22 @@ """ Text User Interface for MaxClaw Agent configuration management +Enhanced with search/filter, export/import, and connection testing """ import curses import sys +import json +import yaml +from pathlib import Path from typing import Optional, List, Dict, Any from config_manager import config_manager -def draw_menu(stdscr, selected_row_idx, options): +def draw_menu(stdscr, selected_row_idx, options, title="MaxClaw Agent Configuration TUI"): """Draw a menu with options""" stdscr.clear() h, w = stdscr.getmaxyx() # Title - title = "MaxClaw Agent Configuration TUI" stdscr.attron(curses.A_BOLD) stdscr.addstr(0, (w - len(title)) // 2, title) stdscr.attroff(curses.A_BOLD) @@ -87,20 +90,61 @@ def show_message(stdscr, message, wait_for_key=True): stdscr.getch() +def filter_configurations(configs: Dict[str, Any], search_term: str) -> Dict[str, Any]: + """Filter configurations based on search term""" + if not search_term: + return configs + + search_term = search_term.lower() + filtered = {} + + for name, config in configs.items(): + # Search in name + if search_term in name.lower(): + filtered[name] = config + continue + + # Search in provider + provider = config.get("provider", "").lower() + if search_term in provider: + filtered[name] = config + continue + + # Search in model name + model_name = config.get("model_name", "").lower() + if search_term in model_name: + filtered[name] = config + continue + + # Search in base URL + base_url = config.get("base_url", "").lower() + if search_term in base_url: + filtered[name] = config + continue + + return filtered + + def main_tui(stdscr): """Main TUI function""" # Initialize colors if available if curses.has_colors(): curses.start_color() curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_WHITE) + curses.init_pair(2, curses.COLOR_GREEN, curses.COLOR_BLACK) + curses.init_pair(3, curses.COLOR_RED, curses.COLOR_BLACK) # Main menu options main_options = [ "View Configurations", + "Search Configurations", "Add Configuration", "Remove Configuration", "Set Active Configuration", "View Active Configuration", + "Test Configuration", + "Export Configurations", + "Import Configurations", "Exit" ] @@ -119,30 +163,38 @@ def main_tui(stdscr): # Enter key pressed if current_selection == 0: # View Configurations view_configurations_tui(stdscr) - elif current_selection == 1: # Add Configuration + elif current_selection == 1: # Search Configurations + search_configurations_tui(stdscr) + elif current_selection == 2: # Add Configuration add_configuration_tui(stdscr) - elif current_selection == 2: # Remove Configuration + elif current_selection == 3: # Remove Configuration remove_configuration_tui(stdscr) - elif current_selection == 3: # Set Active Configuration + elif current_selection == 4: # Set Active Configuration set_active_configuration_tui(stdscr) - elif current_selection == 4: # View Active Configuration + elif current_selection == 5: # View Active Configuration view_active_configuration_tui(stdscr) - elif current_selection == 5: # Exit + elif current_selection == 6: # Test Configuration + test_configuration_tui(stdscr) + elif current_selection == 7: # Export Configurations + export_configurations_tui(stdscr) + elif current_selection == 8: # Import Configurations + import_configurations_tui(stdscr) + elif current_selection == 9: # Exit break elif key == ord('q') or key == ord('Q'): break -def view_configurations_tui(stdscr): +def view_configurations_tui(stdscr, configs=None, title="Configurations"): """View all configurations in TUI""" - configs = config_manager.get_configurations() + if configs is None: + configs = config_manager.get_configurations() active = config_manager.get_active_configuration() stdscr.clear() h, w = stdscr.getmaxyx() # Title - title = "Configurations" stdscr.attron(curses.A_BOLD) stdscr.addstr(0, (w - len(title)) // 2, title) stdscr.attroff(curses.A_BOLD) @@ -174,6 +226,35 @@ def view_configurations_tui(stdscr): stdscr.getch() +def search_configurations_tui(stdscr): + """Search/filter configurations""" + stdscr.clear() + h, w = stdscr.getmaxyx() + + # Title + title = "Search Configurations" + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + # Get search term + search_term = get_string_input(stdscr, "Search term (name, provider, model, URL): ", 2, 0, 50) + + if not search_term: + show_message(stdscr, "Search cancelled.") + return + + # Get all configurations and filter + all_configs = config_manager.get_configurations() + filtered_configs = filter_configurations(all_configs, search_term) + + # Show filtered results + if not filtered_configs: + show_message(stdscr, f"No configurations found matching '{search_term}'.") + else: + view_configurations_tui(stdscr, filtered_configs, f"Search Results for '{search_term}'") + + def add_configuration_tui(stdscr): """Add a new configuration via TUI""" stdscr.clear() @@ -480,6 +561,204 @@ def view_active_configuration_tui(stdscr): stdscr.getch() +def test_configuration_tui(stdscr): + """Test connection to a configuration""" + configs = config_manager.get_configurations() + if not configs: + show_message(stdscr, "No configurations to test.") + return + + stdscr.clear() + h, w = stdscr.getmaxyx() + + # Title + title = "Test Configuration Connection" + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + # List configurations + config_names = list(configs.keys()) + selection = 0 + + while True: + stdscr.clear() + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + stdscr.addstr(2, 0, "Select configuration to test:") + + for idx, name in enumerate(config_names): + y = 3 + idx + if y >= h - 4: # Prevent overflow + break + + if idx == selection: + stdscr.attron(curses.A_REVERSE) + stdscr.addstr(y, 2, f"> {name}") + stdscr.attroff(curses.A_REVERSE) + else: + stdscr.addstr(y, 2, f" {name}") + + stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to test, 'c' to cancel") + stdscr.refresh() + + key = stdscr.getch() + if key == curses.KEY_UP and selection > 0: + selection -= 1 + elif key == curses.KEY_DOWN and selection < len(config_names) - 1: + selection += 1 + elif key == curses.KEY_ENTER or key in [10, 13]: + # Test configuration + name_to_test = config_names[selection] + show_message(stdscr, f"Testing connection to '{name_to_test}'...", False) + + success, message = config_manager.test_configuration_connection(name_to_test) + + if success: + show_message(stdscr, f"✅ {message}") + else: + show_message(stdscr, f"❌ {message}") + return + elif key == ord('c') or key == ord('C'): + show_message(stdscr, "Cancelled.") + return + + +def export_configurations_tui(stdscr): + """Export configurations to JSON or YAML""" + configs = config_manager.get_configurations() + if not configs: + show_message(stdscr, "No configurations to export.") + return + + stdscr.clear() + h, w = stdscr.getmaxyx() + + # Title + title = "Export Configurations" + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + # Select format + formats = ["JSON", "YAML"] + format_selection = 0 + + while True: + stdscr.clear() + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + stdscr.addstr(2, 0, "Select export format:") + + for idx, fmt in enumerate(formats): + y = 3 + idx + if idx == format_selection: + stdscr.attron(curses.A_REVERSE) + stdscr.addstr(y, 2, f"> {fmt}") + stdscr.attroff(curses.A_REVERSE) + else: + stdscr.addstr(y, 2, f" {fmt}") + + stdscr.addstr(h-2, 0, "Use arrow keys to select, Enter to confirm, 'c' to cancel") + stdscr.refresh() + + key = stdscr.getch() + if key == curses.KEY_UP and format_selection > 0: + format_selection -= 1 + elif key == curses.KEY_DOWN and format_selection < len(formats) - 1: + format_selection += 1 + elif key == curses.KEY_ENTER or key in [10, 13]: + break + elif key == ord('c') or key == ord('C'): + show_message(stdscr, "Cancelled.") + return + + # Get filename + filename = get_string_input(stdscr, "Filename (without extension): ", 5, 0, 30) + if not filename: + show_message(stdscr, "Cancelled.") + return + + # Export configurations + try: + if format_selection == 0: # JSON + export_file = Path(f"{filename}.json") + with open(export_file, 'w') as f: + json.dump(configs, f, indent=2) + else: # YAML + export_file = Path(f"{filename}.yaml") + with open(export_file, 'w') as f: + yaml.dump(configs, f, default_flow_style=False) + + show_message(stdscr, f"Configurations exported to {export_file}") + except Exception as e: + show_message(stdscr, f"Error exporting configurations: {e}") + + +def import_configurations_tui(stdscr): + """Import configurations from JSON or YAML""" + stdscr.clear() + h, w = stdscr.getmaxyx() + + # Title + title = "Import Configurations" + stdscr.attron(curses.A_BOLD) + stdscr.addstr(0, (w - len(title)) // 2, title) + stdscr.attroff(curses.A_BOLD) + + # Get filename + filename = get_string_input(stdscr, "Filename (with extension .json or .yaml): ", 2, 0, 50) + if not filename: + show_message(stdscr, "Cancelled.") + return + + # Check if file exists + file_path = Path(filename) + if not file_path.exists(): + show_message(stdscr, f"Error: File '{filename}' not found.") + return + + # Import configurations + try: + with open(file_path, 'r') as f: + if filename.endswith('.json'): + imported_configs = json.load(f) + elif filename.endswith(('.yaml', '.yml')): + imported_configs = yaml.safe_load(f) + else: + show_message(stdscr, "Error: Unsupported file format. Use .json or .yaml") + return + + # Validate imported configurations + if not isinstance(imported_configs, dict): + show_message(stdscr, "Error: Invalid configuration format.") + return + + # Ask for confirmation + confirm_prompt = f"Import {len(imported_configs)} configurations? (y/n): " + confirm = get_string_input(stdscr, confirm_prompt, 4, 0, 1) + if confirm.lower() != 'y': + show_message(stdscr, "Cancelled.") + return + + # Import configurations + imported_count = 0 + for name, config in imported_configs.items(): + try: + config_manager.add_configuration(name, config) + imported_count += 1 + except Exception as e: + show_message(stdscr, f"Warning: Could not import '{name}': {e}") + + show_message(stdscr, f"Successfully imported {imported_count} configurations.") + except Exception as e: + show_message(stdscr, f"Error importing configurations: {e}") + + def main(): """Main entry point for the TUI""" curses.wrapper(main_tui) diff --git a/main.py b/main.py index d4a28ec..b63883c 100644 --- a/main.py +++ b/main.py @@ -2,13 +2,23 @@ """ MaxClaw Agent - A configurable agent harness using DeepAgents Supports multiple LLM providers via LangChain's init_chat_model +Enhanced with custom tools for Proxmox and Ollama management """ import os import sys -from typing import Optional +from typing import Optional, List from deepagents import create_deep_agent from langchain.chat_models import init_chat_model from langchain_core.messages import HumanMessage, SystemMessage +from langchain.tools import Tool + +# Import custom tools +try: + from tools import get_proxmox_tools, get_ollama_tools + TOOLS_AVAILABLE = True +except ImportError: + TOOLS_AVAILABLE = False + print("Warning: Custom tools not available. Install required dependencies.") def load_model_from_config(): @@ -53,6 +63,49 @@ def load_model_from_config(): return init_chat_model(f"openai:{model_name}") +def get_custom_tools() -> List[Tool]: + """ + Get custom tools for the MaxClaw agent + + Returns: + List of custom tools + """ + custom_tools = [] + + if not TOOLS_AVAILABLE: + return custom_tools + + # Add Proxmox tools if configured + proxmox_host = os.getenv("PROXMOX_HOST") + if proxmox_host: + proxmox_user = os.getenv("PROXMOX_USER", "root") + proxmox_password = os.getenv("PROXMOX_PASSWORD") + proxmox_token = os.getenv("PROXMOX_TOKEN") + + try: + proxmox_tools = get_proxmox_tools( + host=proxmox_host, + user=proxmox_user, + password=proxmox_password, + api_token=proxmox_token + ) + custom_tools.extend(proxmox_tools) + print(f"✅ Added {len(proxmox_tools)} Proxmox tools") + except Exception as e: + print(f"⚠️ Could not load Proxmox tools: {e}") + + # Add Ollama tools if configured + ollama_url = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") + try: + ollama_tools = get_ollama_tools(base_url=ollama_url) + custom_tools.extend(ollama_tools) + print(f"✅ Added {len(ollama_tools)} Ollama tools") + except Exception as e: + print(f"⚠️ Could not load Ollama tools: {e}") + + return custom_tools + + def create_maxclaw_agent(system_prompt: Optional[str] = None): """ Create and return a DeepAgent instance with configurable model. @@ -60,13 +113,23 @@ def create_maxclaw_agent(system_prompt: Optional[str] = None): if system_prompt is None: system_prompt = """You are MaxClaw, a helpful AI agent that can plan, use tools, manage context, and delegate to subagents. You are designed to be configurable - to work with various LLM providers.""" + to work with various LLM providers. + + You have access to custom tools for managing Proxmox virtual machines and Ollama models. + Use these tools when asked about system administration, virtualization, or AI model management.""" model = load_model_from_config() + + # Get custom tools + custom_tools = get_custom_tools() + + # Create the agent with custom tools agent = create_deep_agent( model=model, + tools=custom_tools, # Add custom tools to the agent system_prompt=system_prompt ) + return agent @@ -82,6 +145,8 @@ def main(): # Simple test interaction print("\n💬 You can now interact with the agent. Type 'exit' to quit.") + print("💡 Try asking about Proxmox VMs or Ollama models!") + while True: user_input = input("\n👤 You: ").strip() if user_input.lower() in ['exit', 'quit']: diff --git a/tools/__init__.py b/tools/__init__.py new file mode 100644 index 0000000..ca684e9 --- /dev/null +++ b/tools/__init__.py @@ -0,0 +1,14 @@ +""" +MaxClaw Agent Tools Package +Provides custom tools for Proxmox and Ollama management +""" + +from .proxmox_tools import get_proxmox_tools, ProxmoxTools +from .ollama_tools import get_ollama_tools, OllamaTools + +__all__ = [ + 'get_proxmox_tools', + 'ProxmoxTools', + 'get_ollama_tools', + 'OllamaTools' +] \ No newline at end of file diff --git a/tools/ollama_tools.py b/tools/ollama_tools.py new file mode 100644 index 0000000..0172d1b --- /dev/null +++ b/tools/ollama_tools.py @@ -0,0 +1,360 @@ +""" +Ollama management tools for MaxClaw Agent +Provides tools for managing Ollama models and instances +""" +import requests +import json +from typing import Dict, List, Optional, Any +from langchain.tools import Tool + + +class OllamaTools: + """Tools for managing Ollama instances and models""" + + def __init__(self, base_url: str = "http://localhost:11434"): + """ + Initialize Ollama tools + + Args: + base_url: Base URL of the Ollama instance + """ + self.base_url = base_url.rstrip('/') + self.api_url = f"{self.base_url}/api" + + def _make_request(self, endpoint: str, method: str = "GET", data: Optional[Dict] = None) -> Dict[str, Any]: + """ + Make a request to the Ollama API + + Args: + endpoint: API endpoint + method: HTTP method (GET, POST, etc.) + data: Optional data for POST requests + + Returns: + Dictionary with response data + """ + try: + url = f"{self.api_url}/{endpoint}" + + if method.upper() == "GET": + response = requests.get(url, timeout=10) + elif method.upper() == "POST": + response = requests.post(url, json=data, timeout=30) + else: + return {"success": False, "error": f"Unsupported method: {method}"} + + if response.status_code == 200: + try: + return {"success": True, "data": response.json()} + except json.JSONDecodeError: + return {"success": True, "data": response.text} + else: + return {"success": False, "error": f"HTTP {response.status_code}: {response.text}"} + + except requests.exceptions.Timeout: + return {"success": False, "error": "Request timed out"} + except requests.exceptions.ConnectionError: + return {"success": False, "error": f"Connection failed to {self.base_url}"} + except Exception as e: + return {"success": False, "error": str(e)} + + def list_models(self) -> str: + """ + List all available models on the Ollama instance + + Returns: + Formatted string with model information + """ + result = self._make_request("tags") + + if not result["success"]: + return f"Error listing models: {result['error']}" + + models = result["data"].get("models", []) + if not models: + return "No models found on the Ollama instance." + + output = "Available Ollama Models:\n" + output += "=" * 50 + "\n" + + for model in models: + name = model.get("name", "Unknown") + size = model.get("size", 0) + modified = model.get("modified_at", "") + + # Convert size to human readable format + if size > 1024 * 1024 * 1024: + size_str = f"{size / (1024 * 1024 * 1024):.1f} GB" + elif size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.1f} MB" + else: + size_str = f"{size / 1024:.1f} KB" + + # Format modification date + if modified: + from datetime import datetime + try: + dt = datetime.fromisoformat(modified.replace('Z', '+00:00')) + modified_str = dt.strftime("%Y-%m-%d %H:%M:%S") + except: + modified_str = modified + else: + modified_str = "N/A" + + output += f"Name: {name}\n" + output += f"Size: {size_str}\n" + output += f"Modified: {modified_str}\n" + output += "-" * 30 + "\n" + + return output + + def get_model_info(self, model_name: str) -> str: + """ + Get detailed information about a specific model + + Args: + model_name: Name of the model + + Returns: + Formatted string with model details + """ + result = self._make_request("show", method="POST", data={"name": model_name}) + + if not result["success"]: + return f"Error getting model info: {result['error']}" + + model_info = result["data"] + + output = f"Model Information: {model_name}\n" + output += "=" * 50 + "\n" + + # Basic information + if "license" in model_info: + output += f"License: {model_info['license']}\n" + + if "modelfile" in model_info: + output += f"Modelfile: {model_info['modelfile'][:100]}...\n" + + if "parameters" in model_info: + output += f"Parameters: {model_info['parameters']}\n" + + if "template" in model_info: + output += f"Template: {model_info['template'][:100]}...\n" + + # System information + if "system" in model_info: + output += f"System: {model_info['system'][:100]}...\n" + + # Details + if "details" in model_info: + details = model_info["details"] + output += "\nModel Details:\n" + output += "-" * 30 + "\n" + + if "parent_model" in details: + output += f"Parent Model: {details['parent_model']}\n" + + if "format" in details: + output += f"Format: {details['format']}\n" + + if "family" in details: + output += f"Family: {details['family']}\n" + + if "families" in details: + output += f"Families: {', '.join(details['families'])}\n" + + if "parameter_size" in details: + output += f"Parameter Size: {details['parameter_size']}\n" + + if "quantization_level" in details: + output += f"Quantization Level: {details['quantization_level']}\n" + + return output + + def pull_model(self, model_name: str) -> str: + """ + Pull a model from the Ollama library + + Args: + model_name: Name of the model to pull + + Returns: + Result message + """ + result = self._make_request("pull", method="POST", data={"name": model_name}) + + if result["success"]: + return f"✅ Model '{model_name}' pulled successfully." + else: + return f"❌ Failed to pull model '{model_name}': {result['error']}" + + def delete_model(self, model_name: str) -> str: + """ + Delete a model from the Ollama instance + + Args: + model_name: Name of the model to delete + + Returns: + Result message + """ + result = self._make_request("delete", method="POST", data={"name": model_name}) + + if result["success"]: + return f"✅ Model '{model_name}' deleted successfully." + else: + return f"❌ Failed to delete model '{model_name}': {result['error']}" + + def generate_completion(self, model: str, prompt: str, system: Optional[str] = None) -> str: + """ + Generate a completion using a model + + Args: + model: Model name + prompt: Prompt text + system: Optional system prompt + + Returns: + Generated text + """ + data = { + "model": model, + "prompt": prompt, + "stream": False + } + + if system: + data["system"] = system + + result = self._make_request("generate", method="POST", data=data) + + if not result["success"]: + return f"Error generating completion: {result['error']}" + + response = result["data"] + return response.get("response", "No response generated") + + def check_ollama_status(self) -> str: + """ + Check if the Ollama instance is running and responsive + + Returns: + Status message + """ + try: + response = requests.get(f"{self.base_url}/", timeout=5) + if response.status_code == 200: + return f"✅ Ollama instance at {self.base_url} is running and responsive." + else: + return f"⚠️ Ollama instance at {self.base_url} returned status {response.status_code}." + except requests.exceptions.ConnectionError: + return f"❌ Cannot connect to Ollama instance at {self.base_url}." + except requests.exceptions.Timeout: + return f"❌ Connection to Ollama instance at {self.base_url} timed out." + except Exception as e: + return f"❌ Error checking Ollama status: {str(e)}" + + def get_running_models(self) -> str: + """ + Get information about currently running models + + Returns: + Formatted string with running model information + """ + result = self._make_request("ps") + + if not result["success"]: + return f"Error getting running models: {result['error']}" + + running = result["data"].get("models", []) + if not running: + return "No models currently running." + + output = "Currently Running Models:\n" + output += "=" * 50 + "\n" + + for model in running: + name = model.get("name", "Unknown") + size = model.get("size", 0) + digest = model.get("digest", "") + expires_at = model.get("expires_at", "") + + # Convert size to human readable format + if size > 1024 * 1024 * 1024: + size_str = f"{size / (1024 * 1024 * 1024):.1f} GB" + elif size > 1024 * 1024: + size_str = f"{size / (1024 * 1024):.1f} MB" + else: + size_str = f"{size / 1024:.1f} KB" + + # Format expiration time + if expires_at: + from datetime import datetime + try: + dt = datetime.fromisoformat(expires_at.replace('Z', '+00:00')) + expires_str = dt.strftime("%Y-%m-%d %H:%M:%S") + except: + expires_str = expires_at + else: + expires_str = "N/A" + + output += f"Name: {name}\n" + output += f"Size: {size_str}\n" + output += f"Digest: {digest[:20]}...\n" + output += f"Expires: {expires_str}\n" + output += "-" * 30 + "\n" + + return output + + +def get_ollama_tools(base_url: str = "http://localhost:11434") -> List[Tool]: + """ + Get a list of Ollama tools for the MaxClaw agent + + Args: + base_url: Base URL of the Ollama instance + + Returns: + List of LangChain tools for Ollama management + """ + ollama = OllamaTools(base_url) + + tools = [ + Tool( + name="ollama_list_models", + func=lambda: ollama.list_models(), + description="List all available models on the Ollama instance" + ), + Tool( + name="ollama_model_info", + func=lambda model_name: ollama.get_model_info(model_name), + description="Get detailed information about a specific model" + ), + Tool( + name="ollama_pull_model", + func=lambda model_name: ollama.pull_model(model_name), + description="Pull a model from the Ollama library" + ), + Tool( + name="ollama_delete_model", + func=lambda model_name: ollama.delete_model(model_name), + description="Delete a model from the Ollama instance" + ), + Tool( + name="ollama_generate", + func=lambda model, prompt, system="": ollama.generate_completion(model, prompt, system if system else None), + description="Generate a completion using a model with optional system prompt" + ), + Tool( + name="ollama_status", + func=lambda: ollama.check_ollama_status(), + description="Check if the Ollama instance is running and responsive" + ), + Tool( + name="ollama_running_models", + func=lambda: ollama.get_running_models(), + description="Get information about currently running models" + ) + ] + + return tools \ No newline at end of file diff --git a/tools/proxmox_tools.py b/tools/proxmox_tools.py new file mode 100644 index 0000000..a8a7acb --- /dev/null +++ b/tools/proxmox_tools.py @@ -0,0 +1,320 @@ +""" +Proxmox management tools for MaxClaw Agent +Provides tools for managing Proxmox virtual machines and containers +""" +import subprocess +import json +from typing import Dict, List, Optional, Any +from langchain.tools import Tool + + +class ProxmoxTools: + """Tools for managing Proxmox virtual environment""" + + def __init__(self, host: str = "192.168.0.100", user: str = "root", + password: Optional[str] = None, api_token: Optional[str] = None): + """ + Initialize Proxmox tools + + Args: + host: Proxmox host address + user: Username for authentication + password: Password for authentication (optional if using API token) + api_token: API token for authentication (optional if using password) + """ + self.host = host + self.user = user + self.password = password + self.api_token = api_token + self.base_url = f"https://{host}:8006/api2/json" + + def _run_pvesh_command(self, command: str) -> Dict[str, Any]: + """ + Run a pvesh command and return the result + + Args: + command: The pvesh command to run + + Returns: + Dictionary with command result + """ + try: + # Build the full command + full_command = f"pvesh {command}" + + # Add authentication if needed + if self.api_token: + full_command = f"pvesh {command} --apitoken {self.api_token}" + elif self.password: + # Note: In production, use proper authentication methods + full_command = f"pvesh {command} --username {self.user} --password {self.password}" + + # Run the command + result = subprocess.run( + full_command, + shell=True, + capture_output=True, + text=True, + timeout=30 + ) + + if result.returncode == 0: + try: + return {"success": True, "data": json.loads(result.stdout)} + except json.JSONDecodeError: + return {"success": True, "data": result.stdout} + else: + return {"success": False, "error": result.stderr} + + except subprocess.TimeoutExpired: + return {"success": False, "error": "Command timed out"} + except Exception as e: + return {"success": False, "error": str(e)} + + def list_vms(self) -> str: + """ + List all virtual machines in the Proxmox cluster + + Returns: + Formatted string with VM information + """ + result = self._run_pvesh_command("get /cluster/resources --type vm") + + if not result["success"]: + return f"Error listing VMs: {result['error']}" + + vms = result["data"] + if not vms: + return "No virtual machines found." + + output = "Virtual Machines:\n" + output += "=" * 50 + "\n" + + for vm in vms: + vmid = vm.get("vmid", "N/A") + name = vm.get("name", "Unknown") + status = vm.get("status", "Unknown") + node = vm.get("node", "Unknown") + cpu = vm.get("cpu", 0) * 100 + mem = vm.get("mem", 0) / (1024 * 1024) # Convert to MB + maxmem = vm.get("maxmem", 0) / (1024 * 1024 * 1024) # Convert to GB + + output += f"VMID: {vmid}\n" + output += f"Name: {name}\n" + output += f"Status: {status}\n" + output += f"Node: {node}\n" + output += f"CPU Usage: {cpu:.1f}%\n" + output += f"Memory: {mem:.1f} MB / {maxmem:.1f} GB\n" + output += "-" * 30 + "\n" + + return output + + def get_vm_status(self, vmid: int) -> str: + """ + Get the status of a specific virtual machine + + Args: + vmid: Virtual machine ID + + Returns: + Formatted string with VM status + """ + result = self._run_pvesh_command(f"get /nodes/proxmox/qemu/{vmid}/status/current") + + if not result["success"]: + return f"Error getting VM status: {result['error']}" + + vm = result["data"] + status = vm.get("status", "Unknown") + cpu = vm.get("cpu", 0) * 100 + mem = vm.get("mem", 0) / (1024 * 1024) # Convert to MB + maxmem = vm.get("maxmem", 0) / (1024 * 1024 * 1024) # Convert to GB + uptime = vm.get("uptime", 0) + + # Convert uptime to human readable format + days = uptime // (24 * 3600) + hours = (uptime % (24 * 3600)) // 3600 + minutes = (uptime % 3600) // 60 + + output = f"VM {vmid} Status:\n" + output += "=" * 30 + "\n" + output += f"Status: {status}\n" + output += f"CPU Usage: {cpu:.1f}%\n" + output += f"Memory: {mem:.1f} MB / {maxmem:.1f} GB\n" + output += f"Uptime: {days} days, {hours} hours, {minutes} minutes\n" + + return output + + def start_vm(self, vmid: int) -> str: + """ + Start a virtual machine + + Args: + vmid: Virtual machine ID + + Returns: + Result message + """ + result = self._run_pvesh_command(f"create /nodes/proxmox/qemu/{vmid}/status/start") + + if result["success"]: + return f"✅ VM {vmid} started successfully." + else: + return f"❌ Failed to start VM {vmid}: {result['error']}" + + def stop_vm(self, vmid: int) -> str: + """ + Stop a virtual machine + + Args: + vmid: Virtual machine ID + + Returns: + Result message + """ + result = self._run_pvesh_command(f"create /nodes/proxmox/qemu/{vmid}/status/stop") + + if result["success"]: + return f"✅ VM {vmid} stopped successfully." + else: + return f"❌ Failed to stop VM {vmid}: {result['error']}" + + def restart_vm(self, vmid: int) -> str: + """ + Restart a virtual machine + + Args: + vmid: Virtual machine ID + + Returns: + Result message + """ + result = self._run_pvesh_command(f"create /nodes/proxmox/qemu/{vmid}/status/reboot") + + if result["success"]: + return f"✅ VM {vmid} restarted successfully." + else: + return f"❌ Failed to restart VM {vmid}: {result['error']}" + + def create_snapshot(self, vmid: int, snapname: str, description: str = "") -> str: + """ + Create a snapshot of a virtual machine + + Args: + vmid: Virtual machine ID + snapname: Snapshot name + description: Optional description + + Returns: + Result message + """ + command = f"create /nodes/proxmox/qemu/{vmid}/snapshot" + command += f" --snapname {snapname}" + if description: + command += f" --description \"{description}\"" + + result = self._run_pvesh_command(command) + + if result["success"]: + return f"✅ Snapshot '{snapname}' created for VM {vmid}." + else: + return f"❌ Failed to create snapshot for VM {vmid}: {result['error']}" + + def list_snapshots(self, vmid: int) -> str: + """ + List all snapshots for a virtual machine + + Args: + vmid: Virtual machine ID + + Returns: + Formatted string with snapshot information + """ + result = self._run_pvesh_command(f"get /nodes/proxmox/qemu/{vmid}/snapshot") + + if not result["success"]: + return f"Error listing snapshots: {result['error']}" + + snapshots = result["data"] + if not snapshots: + return f"No snapshots found for VM {vmid}." + + output = f"Snapshots for VM {vmid}:\n" + output += "=" * 40 + "\n" + + for snap in snapshots: + name = snap.get("name", "Unknown") + description = snap.get("description", "") + snaptime = snap.get("snaptime", 0) + + # Convert timestamp to readable format + if snaptime: + from datetime import datetime + dt = datetime.fromtimestamp(snaptime) + time_str = dt.strftime("%Y-%m-%d %H:%M:%S") + else: + time_str = "N/A" + + output += f"Name: {name}\n" + output += f"Description: {description}\n" + output += f"Created: {time_str}\n" + output += "-" * 30 + "\n" + + return output + + +def get_proxmox_tools(host: str = "192.168.0.100", user: str = "root", + password: Optional[str] = None, api_token: Optional[str] = None) -> List[Tool]: + """ + Get a list of Proxmox tools for the MaxClaw agent + + Args: + host: Proxmox host address + user: Username for authentication + password: Password for authentication + api_token: API token for authentication + + Returns: + List of LangChain tools for Proxmox management + """ + proxmox = ProxmoxTools(host, user, password, api_token) + + tools = [ + Tool( + name="proxmox_list_vms", + func=lambda: proxmox.list_vms(), + description="List all virtual machines in the Proxmox cluster with their status and resource usage" + ), + Tool( + name="proxmox_vm_status", + func=lambda vmid: proxmox.get_vm_status(int(vmid)), + description="Get detailed status of a specific virtual machine by VMID" + ), + Tool( + name="proxmox_start_vm", + func=lambda vmid: proxmox.start_vm(int(vmid)), + description="Start a virtual machine by VMID" + ), + Tool( + name="proxmox_stop_vm", + func=lambda vmid: proxmox.stop_vm(int(vmid)), + description="Stop a virtual machine by VMID" + ), + Tool( + name="proxmox_restart_vm", + func=lambda vmid: proxmox.restart_vm(int(vmid)), + description="Restart a virtual machine by VMID" + ), + Tool( + name="proxmox_create_snapshot", + func=lambda vmid, snapname, description="": proxmox.create_snapshot(int(vmid), snapname, description), + description="Create a snapshot of a virtual machine with optional description" + ), + Tool( + name="proxmox_list_snapshots", + func=lambda vmid: proxmox.list_snapshots(int(vmid)), + description="List all snapshots for a specific virtual machine" + ) + ] + + return tools \ No newline at end of file