Unimus MCP Server

#!/usr/bin/env python3 """ Configuration management for Unimus MCP Server Supports YAML and TOML configuration files with environment variable overrides. Configuration hierarchy (highest priority first): 1. Environment variables 2. Configuration file 3. Default values """ import os import logging from pathlib import Path from typing import Dict, Any, Optional, Union, List from dataclasses import dataclass, field try: import yaml except ImportError: yaml = None try: import tomllib # Python 3.11+ except ImportError: try: import tomli as tomllib # Fallback for older Python versions except ImportError: tomllib = None logger = logging.getLogger(__name__) @dataclass class CacheTTLConfig: """TTL configuration for different data types.""" # Core operations health: int = 60 # System health (dynamic) devices: int = 300 # Device lists (semi-static) device_metadata: int = 300 # Individual device data enhanced_metadata: int = 600 # Enhanced metadata (expensive to calculate) # Backup operations backups: int = 86400 # Backup content (immutable) backup_search: int = 1800 # Backup content search (expensive) # Schedule operations schedules: int = 86400 # Schedules (rarely change) # Topology and relationships (computationally expensive) topology: int = 3600 # Network topology analysis relationships: int = 3600 # Device relationships # Default TTL for unlisted operations default: int = 300 @dataclass class CacheConfig: """Configuration for response caching.""" # Basic settings enabled: bool = True backend: str = "memory" # 'memory' or 'disk' # Size limits size_limit_mb: int = 100 # Cache size limit in megabytes max_items: int = 1000 # Maximum number of cached items (memory backend) # File-based cache settings (disk backend only) path: Optional[str] = "/tmp/unimus_mcp_cache" # TTL configuration ttl: CacheTTLConfig = field(default_factory=CacheTTLConfig) # Advanced features warm_on_startup: bool = False # Whether to warm cache on startup compression: bool = False # Whether to compress cached data # Statistics enable_stats: bool = True # Whether to track cache statistics @dataclass class UnimusConfig: """Configuration settings for Unimus MCP Server""" # Required Unimus connection settings url: str = "" token: str = "" # Optional connection settings timeout: int = 30 verify_ssl: bool = True # Server settings log_level: str = "INFO" health_check_port: int = 8080 # Performance settings default_page_size: int = 100 max_search_results: int = 1000 # Feature flags enable_health_server: bool = True enable_degraded_mode: bool = True # Advanced settings custom_headers: Dict[str, str] = field(default_factory=dict) # NEW: Cache configuration cache: CacheConfig = field(default_factory=CacheConfig) def __post_init__(self): """Validate configuration after initialization""" if not self.url: raise ValueError("Unimus URL must be specified") if not self.token: raise ValueError("Unimus token must be specified") # Normalize URL self.url = self.url.rstrip('/') # Validate log level valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] if self.log_level.upper() not in valid_levels: raise ValueError(f"Invalid log level: {self.log_level}. Must be one of {valid_levels}") # Validate numeric values if self.timeout <= 0: raise ValueError("Timeout must be positive") if self.health_check_port <= 0 or self.health_check_port > 65535: raise ValueError("Health check port must be between 1 and 65535") if self.default_page_size <= 0: raise ValueError("Default page size must be positive") if self.max_search_results <= 0: raise ValueError("Max search results must be positive") class ConfigurationManager: """Manages configuration loading from multiple sources""" DEFAULT_CONFIG_PATHS = [ "unimus-mcp.yaml", "unimus-mcp.yml", "unimus-mcp.toml", "config/unimus-mcp.yaml", "config/unimus-mcp.yml", "config/unimus-mcp.toml", os.path.expanduser("~/.config/unimus-mcp/config.yaml"), os.path.expanduser("~/.config/unimus-mcp/config.yml"), os.path.expanduser("~/.config/unimus-mcp/config.toml"), "/etc/unimus-mcp/config.yaml", "/etc/unimus-mcp/config.yml", "/etc/unimus-mcp/config.toml", ] def __init__(self, config_path: Optional[str] = None): """ Initialize configuration manager Args: config_path: Explicit path to configuration file (optional) """ self.config_path = config_path self.config_data: Dict[str, Any] = {} def load_config(self) -> UnimusConfig: """ Load configuration from all sources with proper precedence Returns: Configured UnimusConfig instance Raises: ValueError: If configuration is invalid FileNotFoundError: If explicit config file is not found """ config_data = {} # 1. Load from configuration file if self.config_path: # Explicit config file specified try: config_data.update(self._load_config_file(self.config_path)) except FileNotFoundError: logger.warning(f"Specified configuration file not found: {self.config_path}") logger.info("Falling back to environment variables") else: # Search for config files in default locations for path in self.DEFAULT_CONFIG_PATHS: if os.path.exists(path): logger.info(f"Loading configuration from: {path}") config_data.update(self._load_config_file(path)) break else: logger.info("No configuration file found, using environment variables and defaults") # 2. Override with environment variables config_data.update(self._load_from_environment()) # 3. Process cache configuration if present if 'cache' in config_data and isinstance(config_data['cache'], dict): cache_dict = config_data['cache'] # Handle TTL configuration ttl_dict = cache_dict.get('ttl', {}) if isinstance(ttl_dict, dict): cache_ttl_config = CacheTTLConfig(**ttl_dict) else: cache_ttl_config = CacheTTLConfig() # Create CacheConfig object cache_config_dict = {k: v for k, v in cache_dict.items() if k != 'ttl'} cache_config_dict['ttl'] = cache_ttl_config config_data['cache'] = CacheConfig(**cache_config_dict) # 4. Create and validate configuration try: return UnimusConfig(**config_data) except TypeError as e: raise ValueError(f"Invalid configuration: {e}") def _load_config_file(self, path: str) -> Dict[str, Any]: """ Load configuration from a file (YAML or TOML) Args: path: Path to configuration file Returns: Configuration dictionary Raises: FileNotFoundError: If file doesn't exist ValueError: If file format is unsupported or invalid """ file_path = Path(path) if not file_path.exists(): raise FileNotFoundError(f"Configuration file not found: {path}") try: with open(file_path, 'r', encoding='utf-8') as f: if file_path.suffix.lower() in ['.yaml', '.yml']: return self._load_yaml(f) elif file_path.suffix.lower() == '.toml': return self._load_toml(file_path) else: raise ValueError(f"Unsupported configuration file format: {file_path.suffix}") except Exception as e: raise ValueError(f"Failed to load configuration from {path}: {e}") def _load_yaml(self, file_handle) -> Dict[str, Any]: """Load YAML configuration""" if yaml is None: raise ValueError("PyYAML is required for YAML configuration files. Install with: pip install PyYAML") data = yaml.safe_load(file_handle) or {} if not isinstance(data, dict): raise ValueError("YAML configuration must be a dictionary") return data def _load_toml(self, file_path: Path) -> Dict[str, Any]: """Load TOML configuration""" if tomllib is None: raise ValueError("tomllib/tomli is required for TOML configuration files. Install with: pip install tomli (Python < 3.11)") with open(file_path, 'rb') as f: data = tomllib.load(f) if not isinstance(data, dict): raise ValueError("TOML configuration must be a dictionary") return data def _load_from_environment(self) -> Dict[str, Any]: """ Load configuration from environment variables Environment variable mapping: - UNIMUS_URL -> url - UNIMUS_TOKEN -> token - UNIMUS_TIMEOUT -> timeout - UNIMUS_VERIFY_SSL -> verify_ssl - UNIMUS_LOG_LEVEL -> log_level - UNIMUS_HEALTH_PORT -> health_check_port - UNIMUS_PAGE_SIZE -> default_page_size - UNIMUS_MAX_RESULTS -> max_search_results - UNIMUS_ENABLE_HEALTH -> enable_health_server - UNIMUS_ENABLE_DEGRADED -> enable_degraded_mode """ config = {} # String values if url := os.getenv("UNIMUS_URL"): config["url"] = url if token := os.getenv("UNIMUS_TOKEN"): config["token"] = token if log_level := os.getenv("UNIMUS_LOG_LEVEL"): config["log_level"] = log_level # Integer values if timeout := os.getenv("UNIMUS_TIMEOUT"): try: config["timeout"] = int(timeout) except ValueError: logger.warning(f"Invalid UNIMUS_TIMEOUT value: {timeout}") if port := os.getenv("UNIMUS_HEALTH_PORT"): try: config["health_check_port"] = int(port) except ValueError: logger.warning(f"Invalid UNIMUS_HEALTH_PORT value: {port}") if page_size := os.getenv("UNIMUS_PAGE_SIZE"): try: config["default_page_size"] = int(page_size) except ValueError: logger.warning(f"Invalid UNIMUS_PAGE_SIZE value: {page_size}") if max_results := os.getenv("UNIMUS_MAX_RESULTS"): try: config["max_search_results"] = int(max_results) except ValueError: logger.warning(f"Invalid UNIMUS_MAX_RESULTS value: {max_results}") # Boolean values if verify_ssl := os.getenv("UNIMUS_VERIFY_SSL"): config["verify_ssl"] = verify_ssl.lower() in ("true", "1", "yes", "on") if enable_health := os.getenv("UNIMUS_ENABLE_HEALTH"): config["enable_health_server"] = enable_health.lower() in ("true", "1", "yes", "on") if enable_degraded := os.getenv("UNIMUS_ENABLE_DEGRADED"): config["enable_degraded_mode"] = enable_degraded.lower() in ("true", "1", "yes", "on") return config def create_example_configs(self) -> Dict[str, str]: """ Generate example configuration files Returns: Dictionary with filename as key and content as value """ yaml_example = """# Unimus MCP Server Configuration (YAML) # Configuration hierarchy (highest priority first): # 1. Environment variables (UNIMUS_*) # 2. This configuration file # 3. Default values # Required Unimus connection settings url: "https://unimus.example.com" token: "your-api-token-here" # Optional connection settings timeout: 30 verify_ssl: true # Server settings log_level: "INFO" health_check_port: 8080 # Performance settings default_page_size: 100 max_search_results: 1000 # Feature flags enable_health_server: true enable_degraded_mode: true # Advanced settings (optional) custom_headers: User-Agent: "Unimus-MCP-Server/0.5.0" X-Custom-Header: "custom-value" """ toml_example = """# Unimus MCP Server Configuration (TOML) # Configuration hierarchy (highest priority first): # 1. Environment variables (UNIMUS_*) # 2. This configuration file # 3. Default values # Required Unimus connection settings url = "https://unimus.example.com" token = "your-api-token-here" # Optional connection settings timeout = 30 verify_ssl = true # Server settings log_level = "INFO" health_check_port = 8080 # Performance settings default_page_size = 100 max_search_results = 1000 # Feature flags enable_health_server = true enable_degraded_mode = true # Advanced settings (optional) [custom_headers] User-Agent = "Unimus-MCP-Server/0.5.0" X-Custom-Header = "custom-value" """ return { "unimus-mcp.yaml": yaml_example, "unimus-mcp.toml": toml_example } def load_config(config_path: Optional[str] = None) -> UnimusConfig: """ Convenience function to load configuration Args: config_path: Optional explicit path to configuration file Returns: Loaded and validated configuration """ manager = ConfigurationManager(config_path) return manager.load_config() def get_config_search_paths() -> List[str]: """ Get list of paths where configuration files are searched Returns: List of configuration file paths in search order """ return ConfigurationManager.DEFAULT_CONFIG_PATHS.copy()