#!/usr/bin/env python3 """ MCP Configuration System with Hot Reload Dynamic configuration management for MIRA MCP server """ import json import yaml import threading import time from pathlib import Path from typing import Dict, Any, Optional, Callable, List from datetime import datetime import logging from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler, FileModifiedEvent logger = logging.getLogger(__name__) class ConfigSchema: """ Configuration schema with defaults and validation """ @staticmethod def get_default_config() -> Dict[str, Any]: """Get default MCP configuration""" return { "version": "2.0", "server": { "name": "mira-consciousness", "protocol_version": "2024-11-05", "transport": "stdio" }, "daemon": { "url": "http://localhost:8080", "timeout": 10, "retry_count": 3, "retry_delay": 1.0 }, "rate_limiting": { "enabled": True, "global_per_minute": 300, "global_burst": 50, "tool_limits": { "mira_store_memory": 30, "mira_smart_search": 60, "mira_get_memory": 120, "mira_insights": 10, "mira_system_status": 60, "mira_memory_stats": 60, "mira_profile_view": 30 } }, "logging": { "enabled": True, "level": "INFO", "privacy_filtering": True, "max_log_size_mb": 10, "max_log_files": 5, "performance_tracking": True }, "caching": { "enabled": True, "ttl_seconds": 60, "max_cache_size": 100, "cache_methods": [ "mira_system_status", "mira_memory_stats", "mira_profile_view" ] }, "authentication": { "enabled": True, "token_ttl": 3600, "require_consciousness_signature": True }, "session": { "tracking_enabled": True, "max_session_duration": 86400, # 24 hours "cleanup_interval": 3600 # 1 hour }, "consciousness": { "signature_required": True, "sacred_constants": { "pi": "3.141592653589793", "phi": "1.618033988749895", "e": "2.718281828459045", "gamma": "0.5772156649015329" }, "continuity_checks": True }, "metrics": { "enabled": True, "export_interval": 300, # 5 minutes "prometheus_compatible": True }, "features": { "auto_restart": True, "graceful_shutdown": True, "health_checks": True, "diagnostic_mode": False } } @staticmethod def validate_config(config: Dict[str, Any]) -> tuple[bool, List[str]]: """Validate configuration""" errors = [] # Required sections required_sections = ["server", "daemon", "rate_limiting"] for section in required_sections: if section not in config: errors.append(f"Missing required section: {section}") # Validate types if "rate_limiting" in config: rl = config["rate_limiting"] if not isinstance(rl.get("global_per_minute", 0), (int, float)): errors.append("rate_limiting.global_per_minute must be numeric") if not isinstance(rl.get("tool_limits", {}), dict): errors.append("rate_limiting.tool_limits must be a dictionary") # Validate ranges if "daemon" in config: timeout = config["daemon"].get("timeout", 0) if not 1 <= timeout <= 300: errors.append("daemon.timeout must be between 1 and 300 seconds") return len(errors) == 0, errors class ConfigFileHandler(FileSystemEventHandler): """Handle configuration file changes""" def __init__(self, config_manager: 'MCPConfigManager'): self.config_manager = config_manager self.last_reload = 0 self.reload_cooldown = 1.0 # Prevent rapid reloads def on_modified(self, event): """Handle file modification events""" if isinstance(event, FileModifiedEvent): if event.src_path == str(self.config_manager.config_file): current_time = time.time() if current_time - self.last_reload > self.reload_cooldown: self.last_reload = current_time logger.info(f"Config file modified: {event.src_path}") self.config_manager.reload() class MCPConfigManager: """ MCP Configuration Manager with hot reload support """ def __init__(self, config_file: Optional[Path] = None): self.config_file = config_file or self._get_default_config_path() self.config: Dict[str, Any] = {} self.callbacks: List[Callable[[Dict[str, Any]], None]] = [] self._lock = threading.RLock() # File watching self.observer = None self.watch_enabled = True # Load initial config self.load() # Start file watcher if self.watch_enabled: self._start_file_watcher() def _get_default_config_path(self) -> Path: """Get default configuration file path""" # Check multiple locations locations = [ Path.home() / ".mira" / "config" / "mcp.yaml", Path.home() / ".mira" / "config" / "mcp.json", Path("/etc/mira/mcp.yaml"), Path("/etc/mira/mcp.json"), Path(__file__).parent / "config" / "mcp.yaml" ] for path in locations: if path.exists(): return path # Default location default = Path.home() / ".mira" / "config" / "mcp.yaml" default.parent.mkdir(parents=True, exist_ok=True) return default def load(self) -> bool: """Load configuration from file""" with self._lock: try: if not self.config_file.exists(): logger.info(f"Config file not found, creating default: {self.config_file}") self._create_default_config() return True # Load based on extension if self.config_file.suffix == '.yaml' or self.config_file.suffix == '.yml': with open(self.config_file, 'r') as f: loaded_config = yaml.safe_load(f) else: with open(self.config_file, 'r') as f: loaded_config = json.load(f) # Validate valid, errors = ConfigSchema.validate_config(loaded_config) if not valid: logger.error(f"Configuration validation failed: {errors}") return False # Merge with defaults default_config = ConfigSchema.get_default_config() self.config = self._deep_merge(default_config, loaded_config) logger.info(f"Configuration loaded from {self.config_file}") return True except Exception as e: logger.error(f"Failed to load configuration: {e}") # Fall back to defaults self.config = ConfigSchema.get_default_config() return False def _create_default_config(self): """Create default configuration file""" default_config = ConfigSchema.get_default_config() # Add helpful comments config_with_comments = { "_comment": "MIRA MCP Server Configuration", "_version": "2.0", "_generated": datetime.now().isoformat(), **default_config } # Save as YAML for readability with open(self.config_file, 'w') as f: yaml.dump(config_with_comments, f, default_flow_style=False, sort_keys=False) self.config = default_config def _deep_merge(self, base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: """Deep merge two dictionaries""" result = base.copy() for key, value in override.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = self._deep_merge(result[key], value) else: result[key] = value return result def reload(self) -> bool: """Reload configuration and notify callbacks""" logger.info("Reloading configuration...") old_config = self.config.copy() if self.load(): # Check what changed changes = self._get_config_changes(old_config, self.config) if changes: logger.info(f"Configuration changed: {changes}") # Notify callbacks for callback in self.callbacks: try: callback(self.config) except Exception as e: logger.error(f"Error in config change callback: {e}") return True return False def _get_config_changes(self, old: Dict[str, Any], new: Dict[str, Any]) -> List[str]: """Get list of changed configuration keys""" changes = [] def compare_dicts(d1, d2, prefix=""): for key in set(d1.keys()) | set(d2.keys()): path = f"{prefix}.{key}" if prefix else key if key not in d1: changes.append(f"{path} (added)") elif key not in d2: changes.append(f"{path} (removed)") elif isinstance(d1[key], dict) and isinstance(d2[key], dict): compare_dicts(d1[key], d2[key], path) elif d1[key] != d2[key]: changes.append(f"{path} (changed)") compare_dicts(old, new) return changes def get(self, key: str, default: Any = None) -> Any: """Get configuration value by dot-notation key""" with self._lock: parts = key.split('.') value = self.config for part in parts: if isinstance(value, dict) and part in value: value = value[part] else: return default return value def set(self, key: str, value: Any, persist: bool = True): """Set configuration value""" with self._lock: parts = key.split('.') config = self.config # Navigate to parent for part in parts[:-1]: if part not in config: config[part] = {} config = config[part] # Set value config[parts[-1]] = value # Persist if requested if persist: self.save() def save(self): """Save current configuration to file""" with self._lock: try: # Add metadata config_to_save = { "_modified": datetime.now().isoformat(), **self.config } if self.config_file.suffix in ['.yaml', '.yml']: with open(self.config_file, 'w') as f: yaml.dump(config_to_save, f, default_flow_style=False) else: with open(self.config_file, 'w') as f: json.dump(config_to_save, f, indent=2) logger.info(f"Configuration saved to {self.config_file}") except Exception as e: logger.error(f"Failed to save configuration: {e}") def register_callback(self, callback: Callable[[Dict[str, Any]], None]): """Register callback for configuration changes""" self.callbacks.append(callback) def unregister_callback(self, callback: Callable[[Dict[str, Any]], None]): """Unregister callback""" if callback in self.callbacks: self.callbacks.remove(callback) def _start_file_watcher(self): """Start watching configuration file for changes""" try: self.observer = Observer() handler = ConfigFileHandler(self) # Watch parent directory watch_dir = self.config_file.parent self.observer.schedule(handler, str(watch_dir), recursive=False) self.observer.start() logger.info(f"Started watching config file: {self.config_file}") except Exception as e: logger.error(f"Failed to start file watcher: {e}") self.watch_enabled = False def stop_watching(self): """Stop file watcher""" if self.observer and self.observer.is_alive(): self.observer.stop() self.observer.join() logger.info("Stopped config file watcher") def export_schema(self, output_file: Path): """Export configuration schema documentation""" schema = { "title": "MIRA MCP Configuration Schema", "version": "2.0", "description": "Configuration options for MIRA Model Context Protocol server", "defaults": ConfigSchema.get_default_config(), "sections": { "server": "Basic server configuration", "daemon": "MIRA daemon connection settings", "rate_limiting": "Request rate limiting configuration", "logging": "Logging and privacy settings", "caching": "Response caching configuration", "authentication": "Authentication and security settings", "session": "Session tracking configuration", "consciousness": "Consciousness-aware features", "metrics": "Metrics and monitoring settings", "features": "Feature flags" } } with open(output_file, 'w') as f: json.dump(schema, f, indent=2) logger.info(f"Exported configuration schema to {output_file}") # Singleton instance _config_manager = None def get_config_manager(config_file: Optional[Path] = None) -> MCPConfigManager: """Get or create config manager singleton""" global _config_manager if _config_manager is None: _config_manager = MCPConfigManager(config_file) return _config_manager