secure-mcp-proxy

"""Configuration loader for MCP proxy. This module provides functionality to load named server configurations from JSON files. """ import json import logging from pathlib import Path from mcp.client.stdio import StdioServerParameters logger = logging.getLogger(__name__) def load_server_configs( config_file_path: str, base_env: dict[str, str], ) -> dict[str, tuple[StdioServerParameters, bool]]: """Load server configurations from JSON file. Returns: Dict mapping server name to (params, auth_required) tuples. """ configs: dict[str, tuple[StdioServerParameters, bool]] = {} logger.info("Loading server configurations from: %s", config_file_path) try: with Path(config_file_path).open() as f: config_data = json.load(f) except FileNotFoundError: logger.exception("Configuration file not found: %s", config_file_path) raise except json.JSONDecodeError: logger.exception("Error decoding JSON from configuration file: %s", config_file_path) raise except Exception as e: logger.exception("Unexpected error reading configuration file %s", config_file_path) error_message = f"Could not read configuration file: {e}" raise ValueError(error_message) from e if not isinstance(config_data, dict) or "mcpServers" not in config_data: msg = f"Invalid config file format in {config_file_path}. Missing 'mcpServers' key." logger.error(msg) raise ValueError(msg) for name, server_config in config_data.get("mcpServers", {}).items(): if not isinstance(server_config, dict): logger.warning("Skipping invalid server config for '%s'. Entry is not a dictionary.", name) continue if not server_config.get("enabled", True): logger.info("Named server '%s' from config is not enabled. Skipping.", name) continue command = server_config.get("command") command_args = server_config.get("args", []) env = server_config.get("env", {}) auth_required = server_config.get("auth", False) if not command: logger.warning("Named server '%s' from config is missing 'command'. Skipping.", name) continue if not isinstance(command_args, list): logger.warning("Named server '%s' from config has invalid 'args' (must be a list). Skipping.", name) continue new_env = base_env.copy() new_env.update(env) params = StdioServerParameters( command=command, args=command_args, env=new_env, cwd=None, ) configs[name] = (params, auth_required) auth_status = "with auth" if auth_required else "without auth" logger.info("Configured server '%s' (%s): %s %s", name, auth_status, command, " ".join(command_args)) return configs