feat(cli): phase 1 - local cli with docker integration

mcontainer/config.py

@@ -0,0 +1,199 @@
+import yaml
+from pathlib import Path
+from typing import Dict, Optional
+
+from .models import Config, Driver
+
+DEFAULT_CONFIG_DIR = Path.home() / ".config" / "mc"
+DEFAULT_CONFIG_FILE = DEFAULT_CONFIG_DIR / "config.yaml"
+DEFAULT_DRIVERS_DIR = Path.home() / ".config" / "mc" / "drivers"
+PROJECT_ROOT = Path(__file__).parent.parent
+BUILTIN_DRIVERS_DIR = PROJECT_ROOT / "drivers"
+
+# Default built-in driver configurations
+DEFAULT_DRIVERS = {
+    "goose": Driver(
+        name="goose",
+        description="Goose with MCP servers",
+        version="1.0.0",
+        maintainer="team@monadical.com",
+        image="monadical/mc-goose:latest",
+        ports=[8000, 22],
+    ),
+    "aider": Driver(
+        name="aider",
+        description="Aider coding assistant",
+        version="1.0.0",
+        maintainer="team@monadical.com",
+        image="monadical/mc-aider:latest",
+        ports=[22],
+    ),
+    "claude-code": Driver(
+        name="claude-code",
+        description="Claude Code environment",
+        version="1.0.0",
+        maintainer="team@monadical.com",
+        image="monadical/mc-claude-code:latest",
+        ports=[22],
+    ),
+}
+
+
+class ConfigManager:
+    def __init__(self, config_path: Optional[Path] = None):
+        self.config_path = config_path or DEFAULT_CONFIG_FILE
+        self.config_dir = self.config_path.parent
+        self.drivers_dir = DEFAULT_DRIVERS_DIR
+        self.config = self._load_or_create_config()
+
+    def _load_or_create_config(self) -> Config:
+        """Load existing config or create a new one with defaults"""
+        if self.config_path.exists():
+            try:
+                with open(self.config_path, "r") as f:
+                    config_data = yaml.safe_load(f) or {}
+                
+                # Create a new config from scratch, then update with data from file
+                config = Config(
+                    docker=config_data.get('docker', {}),
+                    defaults=config_data.get('defaults', {})
+                )
+                
+                # Add drivers
+                if 'drivers' in config_data:
+                    for driver_name, driver_data in config_data['drivers'].items():
+                        config.drivers[driver_name] = Driver.model_validate(driver_data)
+                
+                # Add sessions (stored as simple dictionaries)
+                if 'sessions' in config_data:
+                    config.sessions = config_data['sessions']
+                
+                return config
+            except Exception as e:
+                print(f"Error loading config: {e}")
+                return self._create_default_config()
+        else:
+            return self._create_default_config()
+
+    def _create_default_config(self) -> Config:
+        """Create a default configuration"""
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+        self.drivers_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Load built-in drivers from directories
+        builtin_drivers = self.load_builtin_drivers()
+        
+        # Merge with default drivers, with directory drivers taking precedence
+        drivers = {**DEFAULT_DRIVERS, **builtin_drivers}
+
+        config = Config(
+            docker={
+                "socket": "/var/run/docker.sock",
+                "network": "mc-network",
+            },
+            drivers=drivers,
+            defaults={
+                "driver": "goose",
+            },
+        )
+
+        self.save_config(config)
+        return config
+
+    def save_config(self, config: Optional[Config] = None) -> None:
+        """Save the current config to disk"""
+        if config:
+            self.config = config
+
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+        
+        # Use model_dump with mode="json" for proper serialization of enums
+        config_dict = self.config.model_dump(mode="json")
+        
+        # Write to file
+        with open(self.config_path, "w") as f:
+            yaml.dump(config_dict, f)
+
+    def get_driver(self, name: str) -> Optional[Driver]:
+        """Get a driver by name"""
+        return self.config.drivers.get(name)
+
+    def list_drivers(self) -> Dict[str, Driver]:
+        """List all available drivers"""
+        return self.config.drivers
+
+    def add_session(self, session_id: str, session_data: dict) -> None:
+        """Add a session to the config"""
+        # Store session data as a dictionary in the config
+        self.config.sessions[session_id] = session_data
+        self.save_config()
+
+    def remove_session(self, session_id: str) -> None:
+        """Remove a session from the config"""
+        if session_id in self.config.sessions:
+            del self.config.sessions[session_id]
+            self.save_config()
+
+    def list_sessions(self) -> Dict:
+        """List all sessions in the config"""
+        return self.config.sessions
+        
+    def load_driver_from_dir(self, driver_dir: Path) -> Optional[Driver]:
+        """Load a driver configuration from a directory"""
+        yaml_path = driver_dir / "mai-driver.yaml"  # Keep this name for backward compatibility
+        
+        if not yaml_path.exists():
+            return None
+            
+        try:
+            with open(yaml_path, "r") as f:
+                driver_data = yaml.safe_load(f)
+                
+            # Extract required fields
+            if not all(k in driver_data for k in ["name", "description", "version", "maintainer"]):
+                print(f"Driver config {yaml_path} missing required fields")
+                return None
+                
+            # Create driver object
+            driver = Driver(
+                name=driver_data["name"],
+                description=driver_data["description"],
+                version=driver_data["version"],
+                maintainer=driver_data["maintainer"],
+                image=f"monadical/mc-{driver_data['name']}:latest",
+                ports=driver_data.get("ports", []),
+            )
+            
+            return driver
+        except Exception as e:
+            print(f"Error loading driver from {yaml_path}: {e}")
+            return None
+            
+    def load_builtin_drivers(self) -> Dict[str, Driver]:
+        """Load all built-in drivers from the drivers directory"""
+        drivers = {}
+        
+        if not BUILTIN_DRIVERS_DIR.exists():
+            return drivers
+            
+        for driver_dir in BUILTIN_DRIVERS_DIR.iterdir():
+            if driver_dir.is_dir():
+                driver = self.load_driver_from_dir(driver_dir)
+                if driver:
+                    drivers[driver.name] = driver
+                    
+        return drivers
+        
+    def get_driver_path(self, driver_name: str) -> Optional[Path]:
+        """Get the directory path for a driver"""
+        # Check built-in drivers first
+        builtin_path = BUILTIN_DRIVERS_DIR / driver_name
+        if builtin_path.exists() and builtin_path.is_dir():
+            return builtin_path
+            
+        # Then check user drivers
+        user_path = self.drivers_dir / driver_name
+        if user_path.exists() and user_path.is_dir():
+            return user_path
+            
+        return None