Dynamic Per-User Tool Generation MCP Server

""" Configuration Loader Loads and parses field mapping configurations from JSON files. Provides type-safe access to field mappings with validation. """ import json import logging from pathlib import Path from typing import Dict, List, Optional from dataclasses import dataclass logger = logging.getLogger(__name__) @dataclass class FieldMapping: """ Represents a single field mapping configuration. Attributes: column_name: The field name in the incoming request (e.g., "impact", "assignee") primitive_type: The data type (e.g., "long", "string") reference_to: The entity type to search in Elasticsearch (e.g., "impact", "user") db_key: The target field name in the transformed request (e.g., "impactId", "technicianId") required: Whether this field is required filters: Optional contextual filters to apply during resolution """ column_name: str primitive_type: str reference_to: str db_key: str required: bool = False filters: Optional[Dict[str, any]] = None @classmethod def from_dict(cls, data: dict) -> "FieldMapping": """Create FieldMapping from dictionary.""" return cls( column_name=data["columnName"], primitive_type=data["primitiveType"], reference_to=data["referenceTo"], db_key=data["dbKey"], required=data.get("required", False), filters=data.get("filters") ) def to_dict(self) -> dict: """Convert FieldMapping to dictionary.""" result = { "columnName": self.column_name, "primitiveType": self.primitive_type, "referenceTo": self.reference_to, "dbKey": self.db_key, "required": self.required } if self.filters: result["filters"] = self.filters return result class FieldMappingLoader: """ Loads and manages field mapping configurations. Supports multiple modules (request, problem, change, etc.) with lazy loading and caching for performance. """ def __init__(self, config_path: Optional[Path] = None): """ Initialize the configuration loader. Args: config_path: Path to the field_mappings.json file. If None, uses default path relative to this file. """ if config_path is None: # Default to field_mappings.json in the same directory config_path = Path(__file__).parent / "field_mappings.json" self.config_path = config_path self._config_cache: Optional[Dict[str, List[FieldMapping]]] = None logger.info(f"FieldMappingLoader initialized with config: {self.config_path}") def _load_config(self) -> Dict[str, List[FieldMapping]]: """ Load configuration from JSON file. Returns: Dictionary mapping module names to lists of FieldMapping objects Raises: FileNotFoundError: If config file doesn't exist json.JSONDecodeError: If config file is invalid JSON ValueError: If config structure is invalid """ if not self.config_path.exists(): raise FileNotFoundError(f"Configuration file not found: {self.config_path}") logger.debug(f"Loading configuration from {self.config_path}") with open(self.config_path, 'r') as f: raw_config = json.load(f) # Parse and validate configuration parsed_config = {} for module_name, field_list in raw_config.items(): if not isinstance(field_list, list): raise ValueError(f"Invalid config for module '{module_name}': expected list, got {type(field_list)}") parsed_config[module_name] = [ FieldMapping.from_dict(field_data) for field_data in field_list ] logger.info(f"Loaded configuration for {len(parsed_config)} modules: {list(parsed_config.keys())}") return parsed_config def get_config(self, force_reload: bool = False) -> Dict[str, List[FieldMapping]]: """ Get the full configuration with caching. Args: force_reload: If True, reload from file even if cached Returns: Dictionary mapping module names to field mappings """ if self._config_cache is None or force_reload: self._config_cache = self._load_config() return self._config_cache def get_module_mappings(self, module: str) -> List[FieldMapping]: """ Get field mappings for a specific module. Args: module: Module name (e.g., "request", "problem", "change") Returns: List of FieldMapping objects for the module Raises: ValueError: If module is not found in configuration """ config = self.get_config() if module not in config: available_modules = list(config.keys()) raise ValueError( f"Module '{module}' not found in configuration. " f"Available modules: {available_modules}" ) return config[module] def get_field_mapping(self, module: str, column_name: str) -> Optional[FieldMapping]: """ Get mapping for a specific field in a module. Args: module: Module name (e.g., "request") column_name: Field name (e.g., "impact", "assignee") Returns: FieldMapping object if found, None otherwise """ mappings = self.get_module_mappings(module) for mapping in mappings: if mapping.column_name == column_name: return mapping return None def get_mapping_by_column(self, module: str) -> Dict[str, FieldMapping]: """ Get field mappings indexed by column name for fast lookup. Args: module: Module name (e.g., "request") Returns: Dictionary mapping column names to FieldMapping objects """ mappings = self.get_module_mappings(module) return {mapping.column_name: mapping for mapping in mappings} def get_available_modules(self) -> List[str]: """ Get list of available module names. Returns: List of module names (e.g., ["request", "problem", "change"]) """ config = self.get_config() return list(config.keys()) def validate_module(self, module: str) -> bool: """ Check if a module exists in the configuration. Args: module: Module name to validate Returns: True if module exists, False otherwise """ return module in self.get_config() def reload(self): """Force reload configuration from file.""" logger.info("Reloading configuration from file") self._config_cache = None self.get_config()