Phase 2: Enhanced TUI/CLI with search, export/import, and custom tools for Proxmox and Ollama

This commit is contained in:
truetype committed 2026-03-20 16:08:26 +01:00
1 parent f5026965f7
commit 9f3298fa13
7 files changed
+1353 -16

No files matched your search

+50
View File
@@ -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`:
+253 -4
View File
@@ -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()
+289 -10
View File
@@ -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)
+67 -2
View File
@@ -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']:
+14
View File
@@ -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'
]
+360
View File
@@ -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
+320
View File
@@ -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