627 lines
22 KiB
Python
627 lines
22 KiB
Python
"""
|
|
Command-line interface for managing MaxClaw Agent configurations
|
|
Enhanced with export/import, search, and better formatting
|
|
"""
|
|
import sys
|
|
import argparse
|
|
import json
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any
|
|
from config_manager import config_manager
|
|
|
|
|
|
def print_configurations(configs=None, title="MaxClaw Agent Configurations"):
|
|
"""Print all configurations in a formatted way"""
|
|
if configs is None:
|
|
configs = config_manager.get_configurations()
|
|
active = config_manager.get_active_configuration()
|
|
|
|
if not configs:
|
|
print("No configurations found.")
|
|
return
|
|
|
|
print(f"\n{title}:")
|
|
print("=" * 50)
|
|
|
|
for name, config in configs.items():
|
|
marker = " * " if name == active else " "
|
|
provider = config.get("provider", "unknown")
|
|
model = config.get("model_name", "unknown")
|
|
print(f"{marker}{name}:")
|
|
print(f" Provider: {provider}")
|
|
print(f" Model: {model}")
|
|
|
|
# Show additional info based on provider
|
|
if provider == "openai":
|
|
print(f" API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
elif provider == "ollama":
|
|
base_url = config.get("base_url", "http://localhost:11434")
|
|
print(f" Base URL: {base_url}")
|
|
elif provider == "openrouter":
|
|
print(f" API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
base_url = config.get("base_url", "https://openrouter.ai/api/v1")
|
|
print(f" Base URL: {base_url}")
|
|
|
|
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")
|
|
print("=" * 30)
|
|
|
|
name = input("Configuration name: ").strip()
|
|
if not name:
|
|
print("Error: Configuration name cannot be empty")
|
|
return
|
|
|
|
if config_manager.configuration_exists(name):
|
|
overwrite = input(f"Configuration '{name}' already exists. Overwrite? (y/N): ").strip().lower()
|
|
if overwrite != 'y':
|
|
print("Cancelled.")
|
|
return
|
|
|
|
print("\nSelect provider:")
|
|
print("1. OpenAI")
|
|
print("2. Ollama")
|
|
print("3. OpenRouter")
|
|
|
|
provider_choice = input("Enter choice (1-3): ").strip()
|
|
provider_map = {
|
|
"1": "openai",
|
|
"2": "ollama",
|
|
"3": "openrouter"
|
|
}
|
|
|
|
if provider_choice not in provider_map:
|
|
print("Invalid choice.")
|
|
return
|
|
|
|
provider = provider_map[provider_choice]
|
|
config = {"provider": provider}
|
|
|
|
# Get model name
|
|
model_name = input(f"Model name for {provider} (e.g., gpt-4o, gemma2:9b): ").strip()
|
|
if not model_name:
|
|
print("Error: Model name cannot be empty")
|
|
return
|
|
config["model_name"] = model_name
|
|
|
|
# Get provider-specific settings
|
|
if provider == "openai":
|
|
api_key = input("OpenAI API key: ").strip()
|
|
if not api_key:
|
|
print("Warning: API key is required for OpenAI")
|
|
config["api_key"] = api_key
|
|
|
|
elif provider == "ollama":
|
|
base_url = input(f"Ollama base URL (default: http://localhost:11434): ").strip()
|
|
if not base_url:
|
|
base_url = "http://localhost:11434"
|
|
config["base_url"] = base_url
|
|
|
|
elif provider == "openrouter":
|
|
api_key = input("OpenRouter API key: ").strip()
|
|
if not api_key:
|
|
print("Warning: API key is required for OpenRouter")
|
|
config["api_key"] = api_key
|
|
|
|
base_url = input(f"OpenRouter base URL (default: https://openrouter.ai/api/v1): ").strip()
|
|
if not base_url:
|
|
base_url = "https://openrouter.ai/api/v1"
|
|
config["base_url"] = base_url
|
|
|
|
# Ask if this should be the active configuration
|
|
set_active = input("Set as active configuration? (y/N): ").strip().lower()
|
|
|
|
# Save the configuration
|
|
try:
|
|
config_manager.add_configuration(name, config)
|
|
print(f"\nConfiguration '{name}' saved successfully.")
|
|
|
|
if set_active == 'y':
|
|
config_manager.set_active_configuration(name)
|
|
print(f"Configuration '{name}' set as active.")
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
|
|
|
|
def remove_configuration_interactive():
|
|
"""Interactively remove a configuration"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
print("No configurations to remove.")
|
|
return
|
|
|
|
print("\nRemove Configuration")
|
|
print("=" * 30)
|
|
print_configurations()
|
|
|
|
name = input("Enter configuration name to remove: ").strip()
|
|
if not name:
|
|
print("Error: Configuration name cannot be empty")
|
|
return
|
|
|
|
if not config_manager.configuration_exists(name):
|
|
print(f"Error: Configuration '{name}' not found.")
|
|
return
|
|
|
|
active = config_manager.get_active_configuration()
|
|
if name == active:
|
|
confirm = input(f"'{name}' is the active configuration. Remove it anyway? (y/N): ").strip().lower()
|
|
if confirm != 'y':
|
|
print("Cancelled.")
|
|
return
|
|
|
|
confirm = input(f"Are you sure you want to remove configuration '{name}'? (y/N): ").strip().lower()
|
|
if confirm == 'y':
|
|
if config_manager.remove_configuration(name):
|
|
print(f"Configuration '{name}' removed successfully.")
|
|
if name == active:
|
|
print("Active configuration has been unset.")
|
|
else:
|
|
print(f"Error: Failed to remove configuration '{name}'.")
|
|
else:
|
|
print("Cancelled.")
|
|
|
|
|
|
def set_active_configuration_interactive():
|
|
"""Interactively set the active configuration"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
print("No configurations available.")
|
|
return
|
|
|
|
print("\nSet Active Configuration")
|
|
print("=" * 30)
|
|
print_configurations()
|
|
|
|
name = input("Enter configuration name to set as active: ").strip()
|
|
if not name:
|
|
print("Error: Configuration name cannot be empty")
|
|
return
|
|
|
|
if not config_manager.configuration_exists(name):
|
|
print(f"Error: Configuration '{name}' not found.")
|
|
return
|
|
|
|
if config_manager.set_active_configuration(name):
|
|
print(f"Configuration '{name}' set as active.")
|
|
else:
|
|
print(f"Error: Failed to set configuration '{name}' as active.")
|
|
|
|
|
|
def show_active_configuration():
|
|
"""Show the currently active configuration"""
|
|
active_name = config_manager.get_active_configuration()
|
|
if not active_name:
|
|
print("No active configuration set.")
|
|
return
|
|
|
|
config = config_manager.get_configuration(active_name)
|
|
if not config:
|
|
print(f"Error: Active configuration '{active_name}' not found.")
|
|
return
|
|
|
|
print("\nActive Configuration")
|
|
print("=" * 30)
|
|
print(f"Name: {active_name}")
|
|
print(f"Provider: {config.get('provider', 'unknown')}")
|
|
print(f"Model: {config.get('model_name', 'unknown')}")
|
|
|
|
# Show additional info based on provider
|
|
provider = config.get('provider')
|
|
if provider == "openai":
|
|
print(f"API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
elif provider == "ollama":
|
|
base_url = config.get('base_url', 'http://localhost:11434')
|
|
print(f"Base URL: {base_url}")
|
|
elif provider == "openrouter":
|
|
print(f"API Key: {'Set' if config.get('api_key') else 'Not set'}")
|
|
base_url = config.get('base_url', 'https://openrouter.ai/api/v1')
|
|
print(f"Base URL: {base_url}")
|
|
|
|
|
|
def test_configuration():
|
|
"""Test connection to a configuration"""
|
|
configs = config_manager.get_configurations()
|
|
if not configs:
|
|
print("No configurations to test.")
|
|
return
|
|
|
|
print("\nTest Configuration Connection")
|
|
print("=" * 30)
|
|
print_configurations()
|
|
|
|
name = input("Enter configuration name to test: ").strip()
|
|
if not name:
|
|
print("Error: Configuration name cannot be empty")
|
|
return
|
|
|
|
if not config_manager.configuration_exists(name):
|
|
print(f"Error: Configuration '{name}' not found.")
|
|
return
|
|
|
|
print(f"\nTesting connection to '{name}'...")
|
|
success, message = config_manager.test_configuration_connection(name)
|
|
|
|
if success:
|
|
print(f"✅ {message}")
|
|
else:
|
|
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")
|
|
subparsers = parser.add_subparsers(dest='command', help='Available commands')
|
|
|
|
# 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',
|
|
help='Non-interactive mode (requires all parameters)')
|
|
add_parser.add_argument('--name', help='Configuration name')
|
|
add_parser.add_argument('--provider', choices=['openai', 'ollama', 'openrouter'], help='Provider')
|
|
add_parser.add_argument('--model', help='Model name')
|
|
add_parser.add_argument('--api-key', help='API key (for OpenAI/OpenRouter)')
|
|
add_parser.add_argument('--base-url', help='Base URL (for Ollama/OpenRouter)')
|
|
add_parser.add_argument('--set-active', action='store_true', help='Set as active configuration')
|
|
|
|
# Remove command
|
|
remove_parser = subparsers.add_parser('remove', help='Remove a configuration')
|
|
remove_parser.add_argument('name', nargs='?', help='Configuration name to remove')
|
|
|
|
# Set active command
|
|
active_parser = subparsers.add_parser('set-active', help='Set active configuration')
|
|
active_parser.add_argument('name', nargs='?', help='Configuration name to set as active')
|
|
|
|
# Show active command
|
|
subparsers.add_parser('active', help='Show active configuration')
|
|
|
|
# Test command
|
|
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
|
|
if not all([args.name, args.provider, args.model]):
|
|
print("Error: --name, --provider, and --model are required for non-interactive mode")
|
|
return 1
|
|
|
|
config = {
|
|
"provider": args.provider,
|
|
"model_name": args.model
|
|
}
|
|
|
|
if args.provider in ['openai', 'openrouter']:
|
|
if not args.api_key:
|
|
print("Error: --api-key is required for OpenAI/OpenRouter providers")
|
|
return 1
|
|
config["api_key"] = args.api_key
|
|
|
|
if args.provider in ['ollama', 'openrouter']:
|
|
config["base_url"] = args.base_url or (
|
|
"http://localhost:11434" if args.provider == "ollama"
|
|
else "https://openrouter.ai/api/v1"
|
|
)
|
|
|
|
try:
|
|
config_manager.add_configuration(args.name, config)
|
|
print(f"Configuration '{args.name}' added successfully.")
|
|
|
|
if args.set_active:
|
|
config_manager.set_active_configuration(args.name)
|
|
print(f"Configuration '{args.name}' set as active.")
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
return 1
|
|
else:
|
|
# Interactive mode
|
|
add_configuration_interactive()
|
|
|
|
elif args.command == 'remove':
|
|
if args.name:
|
|
# Non-interactive removal
|
|
if config_manager.remove_configuration(args.name):
|
|
print(f"Configuration '{args.name}' removed successfully.")
|
|
active = config_manager.get_active_configuration()
|
|
if args.name == active:
|
|
print("Active configuration has been unset.")
|
|
else:
|
|
print(f"Error: Configuration '{args.name}' not found.")
|
|
return 1
|
|
else:
|
|
# Interactive removal
|
|
remove_configuration_interactive()
|
|
|
|
elif args.command == 'set-active':
|
|
if args.name:
|
|
# Non-interactive setting
|
|
if config_manager.set_active_configuration(args.name):
|
|
print(f"Configuration '{args.name}' set as active.")
|
|
else:
|
|
print(f"Error: Configuration '{args.name}' not found.")
|
|
return 1
|
|
else:
|
|
# Interactive setting
|
|
set_active_configuration_interactive()
|
|
|
|
elif args.command == 'active':
|
|
show_active_configuration()
|
|
|
|
elif args.command == 'test':
|
|
if args.name:
|
|
# Non-interactive test
|
|
print(f"Testing connection to '{args.name}'...")
|
|
success, message = config_manager.test_configuration_connection(args.name)
|
|
if success:
|
|
print(f"✅ {message}")
|
|
else:
|
|
print(f"❌ {message}")
|
|
return 1
|
|
else:
|
|
# 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()
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |