KDB MCP Service

"""Configuration management for KDB MCP service.""" import os import yaml import json from typing import Any, Dict from pathlib import Path from dotenv import load_dotenv # Load environment variables load_dotenv() def load_config(config_path: str) -> Dict[str, Any]: """ Load configuration from YAML or JSON file. Args: config_path: Path to configuration file Returns: Configuration dictionary """ config_file = Path(config_path) # Check if config file exists if not config_file.exists(): # Try to find config in common locations common_paths = [ Path("config/kdb_config.yaml"), Path("config/kdb_config.json"), Path("kdb_config.yaml"), Path("kdb_config.json"), Path.home() / ".kdb-mcp" / "config.yaml", Path.home() / ".kdb-mcp" / "config.json" ] for path in common_paths: if path.exists(): config_file = path break else: # Return default configuration if no config file found return get_default_config() # Load configuration based on file extension if config_file.suffix in ['.yaml', '.yml']: with open(config_file, 'r') as f: config = yaml.safe_load(f) elif config_file.suffix == '.json': with open(config_file, 'r') as f: config = json.load(f) else: raise ValueError(f"Unsupported config file format: {config_file.suffix}") # Process environment variable substitutions config = process_env_vars(config) # Validate configuration validate_config(config) return config def process_env_vars(config: Any) -> Any: """ Recursively process environment variable substitutions in config. Environment variables can be referenced using ${VAR_NAME} or ${VAR_NAME:default} Args: config: Configuration dictionary or value Returns: Processed configuration """ if isinstance(config, dict): return {k: process_env_vars(v) for k, v in config.items()} elif isinstance(config, list): return [process_env_vars(item) for item in config] elif isinstance(config, str): # Check for environment variable pattern if config.startswith('${') and config.endswith('}'): var_content = config[2:-1] if ':' in var_content: var_name, default = var_content.split(':', 1) return os.getenv(var_name, default) else: return os.getenv(var_content, config) return config else: return config def validate_config(config: Dict[str, Any]): """ Validate configuration structure. Args: config: Configuration dictionary Raises: ValueError: If configuration is invalid """ if 'databases' not in config: raise ValueError("Configuration must include 'databases' section") databases = config['databases'] if not isinstance(databases, dict): raise ValueError("'databases' must be a dictionary") for db_name, db_config in databases.items(): # Validate required fields if 'host' not in db_config: raise ValueError(f"Database '{db_name}' missing required field 'host'") if 'port' not in db_config: raise ValueError(f"Database '{db_name}' missing required field 'port'") # Validate port is a number try: port = int(db_config['port']) if port < 1 or port > 65535: raise ValueError(f"Database '{db_name}' port must be between 1 and 65535") except (TypeError, ValueError): raise ValueError(f"Database '{db_name}' port must be a valid integer") def get_default_config() -> Dict[str, Any]: """ Get default configuration. Returns: Default configuration dictionary """ return { "databases": { "default": { "host": os.getenv("KDB_HOST", "localhost"), "port": int(os.getenv("KDB_PORT", "5000")), "username": os.getenv("KDB_USERNAME", ""), "password": os.getenv("KDB_PASSWORD", ""), "pool_size": int(os.getenv("KDB_POOL_SIZE", "5")), "description": "Default KDB+ database" } }, "logging": { "level": os.getenv("LOG_LEVEL", "INFO"), "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s" }, "server": { "name": "kdb-mcp-server", "version": "0.1.0" } } def save_config(config: Dict[str, Any], config_path: str): """ Save configuration to file. Args: config: Configuration dictionary config_path: Path to save configuration """ config_file = Path(config_path) config_file.parent.mkdir(parents=True, exist_ok=True) if config_file.suffix in ['.yaml', '.yml']: with open(config_file, 'w') as f: yaml.dump(config, f, default_flow_style=False, sort_keys=False) elif config_file.suffix == '.json': with open(config_file, 'w') as f: json.dump(config, f, indent=2) else: raise ValueError(f"Unsupported config file format: {config_file.suffix}") def merge_configs(*configs: Dict[str, Any]) -> Dict[str, Any]: """ Merge multiple configuration dictionaries. Later configs override earlier ones. Args: *configs: Configuration dictionaries to merge Returns: Merged configuration """ result = {} for config in configs: result = deep_merge(result, config) return result def deep_merge(dict1: Dict[str, Any], dict2: Dict[str, Any]) -> Dict[str, Any]: """ Deep merge two dictionaries. Args: dict1: Base dictionary dict2: Dictionary to merge in (overrides dict1) Returns: Merged dictionary """ result = dict1.copy() for key, value in dict2.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): result[key] = deep_merge(result[key], value) else: result[key] = value return result