AnyDocs MCP Server

#!/usr/bin/env python3 """ Configuration Manager Provides unified configuration management with support for environment variables, YAML files, and database storage. """ import os import yaml from pathlib import Path from typing import Any, Dict, List, Optional, Union, Type, TypeVar from dataclasses import dataclass, field, asdict from contextlib import contextmanager from ..utils import get_logger, get_env_bool, get_env_int, get_env_float, get_env_list from ..database.models import Configuration from ..database.manager import DatabaseManager T = TypeVar('T') logger = get_logger(__name__) @dataclass class ServerConfig: """Server configuration.""" host: str = '127.0.0.1' port: int = 8000 debug: bool = False reload: bool = False workers: int = 1 max_connections: int = 100 timeout: int = 30 @classmethod def from_env(cls) -> 'ServerConfig': """Create from environment variables.""" return cls( host=os.getenv('SERVER_HOST', '127.0.0.1'), port=get_env_int('SERVER_PORT', 8000), debug=get_env_bool('DEBUG', False), reload=get_env_bool('RELOAD', False), workers=get_env_int('WORKERS', 1), max_connections=get_env_int('MAX_CONNECTIONS', 100), timeout=get_env_int('TIMEOUT', 30), ) @dataclass class DatabaseConfig: """Database configuration.""" type: str = 'sqlite' url: str = 'sqlite:///anydocs.db' pool_size: int = 5 max_overflow: int = 10 pool_timeout: int = 30 pool_recycle: int = 3600 echo: bool = False @classmethod def from_env(cls) -> 'DatabaseConfig': """Create from environment variables.""" return cls( type=os.getenv('DATABASE_TYPE', 'sqlite'), url=os.getenv('DATABASE_URL', 'sqlite:///anydocs.db'), pool_size=get_env_int('DATABASE_POOL_SIZE', 5), max_overflow=get_env_int('DATABASE_MAX_OVERFLOW', 10), pool_timeout=get_env_int('DATABASE_POOL_TIMEOUT', 30), pool_recycle=get_env_int('DATABASE_POOL_RECYCLE', 3600), echo=get_env_bool('DATABASE_ECHO', False), ) @dataclass class AuthConfig: """Authentication configuration.""" methods: List[str] = field(default_factory=lambda: ['api_key']) jwt_secret: str = '' jwt_algorithm: str = 'HS256' jwt_expiration: int = 3600 api_key_prefix: str = 'ak' oauth2_providers: Dict[str, Dict[str, str]] = field(default_factory=dict) @classmethod def from_env(cls) -> 'AuthConfig': """Create from environment variables.""" return cls( methods=get_env_list('AUTH_METHODS', default=['api_key']), jwt_secret=os.getenv('JWT_SECRET', ''), jwt_algorithm=os.getenv('JWT_ALGORITHM', 'HS256'), jwt_expiration=get_env_int('JWT_EXPIRATION', 3600), api_key_prefix=os.getenv('API_KEY_PREFIX', 'ak'), oauth2_providers={}, # Load from config file ) @dataclass class CacheConfig: """Cache configuration.""" enabled: bool = True type: str = 'memory' ttl: int = 3600 max_size: int = 1000 redis_url: str = '' @classmethod def from_env(cls) -> 'CacheConfig': """Create from environment variables.""" return cls( enabled=get_env_bool('CACHE_ENABLED', True), type=os.getenv('CACHE_TYPE', 'memory'), ttl=get_env_int('CACHE_TTL', 3600), max_size=get_env_int('CACHE_MAX_SIZE', 1000), redis_url=os.getenv('REDIS_URL', ''), ) @dataclass class LoggingConfig: """Logging configuration.""" level: str = 'INFO' format: str = '%(asctime)s [%(levelname)8s] %(name)s: %(message)s' file: str = '' max_size: int = 10 * 1024 * 1024 # 10MB backup_count: int = 5 json_logs: bool = False @classmethod def from_env(cls) -> 'LoggingConfig': """Create from environment variables.""" return cls( level=os.getenv('LOG_LEVEL', 'INFO'), format=os.getenv('LOG_FORMAT', '%(asctime)s [%(levelname)8s] %(name)s: %(message)s'), file=os.getenv('LOG_FILE', ''), max_size=get_env_int('LOG_MAX_SIZE', 10 * 1024 * 1024), backup_count=get_env_int('LOG_BACKUP_COUNT', 5), json_logs=get_env_bool('JSON_LOGS', False), ) @dataclass class ContentConfig: """Content processing configuration.""" max_file_size: int = 50 * 1024 * 1024 # 50MB supported_formats: List[str] = field(default_factory=lambda: ['markdown', 'html', 'text']) image_max_size: int = 5 * 1024 * 1024 # 5MB image_formats: List[str] = field(default_factory=lambda: ['jpg', 'jpeg', 'png', 'gif', 'webp']) enable_syntax_highlighting: bool = True @classmethod def from_env(cls) -> 'ContentConfig': """Create from environment variables.""" return cls( max_file_size=get_env_int('CONTENT_MAX_FILE_SIZE', 50 * 1024 * 1024), supported_formats=get_env_list('CONTENT_SUPPORTED_FORMATS', default=['markdown', 'html', 'text']), image_max_size=get_env_int('CONTENT_IMAGE_MAX_SIZE', 5 * 1024 * 1024), image_formats=get_env_list('CONTENT_IMAGE_FORMATS', default=['jpg', 'jpeg', 'png', 'gif', 'webp']), enable_syntax_highlighting=get_env_bool('CONTENT_ENABLE_SYNTAX_HIGHLIGHTING', True), ) @dataclass class MCPConfig: """MCP server configuration.""" name: str = 'AnyDocs MCP Server' version: str = '1.0.0' description: str = 'Transform documentation into MCP-compatible server' max_resources: int = 1000 max_tools: int = 50 enable_discovery: bool = True @classmethod def from_env(cls) -> 'MCPConfig': """Create from environment variables.""" return cls( name=os.getenv('MCP_NAME', 'AnyDocs MCP Server'), version=os.getenv('MCP_VERSION', '1.0.0'), description=os.getenv('MCP_DESCRIPTION', 'Transform documentation into MCP-compatible server'), max_resources=get_env_int('MCP_MAX_RESOURCES', 1000), max_tools=get_env_int('MCP_MAX_TOOLS', 50), enable_discovery=get_env_bool('MCP_ENABLE_DISCOVERY', True), ) @dataclass class AppConfig: """Main application configuration.""" server: ServerConfig = field(default_factory=ServerConfig) database: DatabaseConfig = field(default_factory=DatabaseConfig) auth: AuthConfig = field(default_factory=AuthConfig) cache: CacheConfig = field(default_factory=CacheConfig) logging: LoggingConfig = field(default_factory=LoggingConfig) content: ContentConfig = field(default_factory=ContentConfig) mcp: MCPConfig = field(default_factory=MCPConfig) @classmethod def from_env(cls) -> 'AppConfig': """Create from environment variables.""" return cls( server=ServerConfig.from_env(), database=DatabaseConfig.from_env(), auth=AuthConfig.from_env(), cache=CacheConfig.from_env(), logging=LoggingConfig.from_env(), content=ContentConfig.from_env(), mcp=MCPConfig.from_env(), ) def to_dict(self) -> Dict[str, Any]: """Convert to dictionary.""" return asdict(self) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'AppConfig': """Create from dictionary.""" return cls( server=ServerConfig(**data.get('server', {})), database=DatabaseConfig(**data.get('database', {})), auth=AuthConfig(**data.get('auth', {})), cache=CacheConfig(**data.get('cache', {})), logging=LoggingConfig(**data.get('logging', {})), content=ContentConfig(**data.get('content', {})), mcp=MCPConfig(**data.get('mcp', {})), ) class ConfigManager: """Configuration manager with multiple sources.""" def __init__( self, config_file: Optional[str] = None, db_manager: Optional[DatabaseManager] = None, auto_reload: bool = False ): """ Initialize configuration manager. Args: config_file: Path to YAML configuration file db_manager: Database manager for persistent config auto_reload: Enable automatic config reloading """ self.config_file = config_file self.db_manager = db_manager self.auto_reload = auto_reload self._config: Optional[AppConfig] = None self._file_mtime: Optional[float] = None logger.info("Configuration manager initialized", config_file=config_file) def load_config(self, force_reload: bool = False) -> AppConfig: """Load configuration from all sources. Args: force_reload: Force reload even if cached Returns: Application configuration """ if self._config is not None and not force_reload and not self._should_reload(): return self._config logger.info("Loading configuration") # Start with environment variables config = AppConfig.from_env() # Override with file configuration if self.config_file: file_config = self._load_file_config() if file_config: config = self._merge_configs(config, file_config) # Override with database configuration if self.db_manager: db_config = self._load_db_config() if db_config: config = self._merge_configs(config, db_config) self._config = config self._update_file_mtime() logger.info("Configuration loaded successfully") return config def save_config(self, config: AppConfig, target: str = 'file') -> bool: """Save configuration to target. Args: config: Configuration to save target: Target ('file' or 'database') Returns: True if successful """ try: if target == 'file' and self.config_file: return self._save_file_config(config) elif target == 'database' and self.db_manager: return self._save_db_config(config) else: logger.error("Invalid save target or missing manager", target=target) return False except Exception as e: logger.error("Failed to save configuration", target=target, error=str(e)) return False def get_config(self) -> AppConfig: """Get current configuration. Returns: Current application configuration """ return self.load_config() def update_config(self, updates: Dict[str, Any], save_to: Optional[str] = None) -> bool: """Update configuration with partial updates. Args: updates: Configuration updates save_to: Target to save to ('file', 'database', or None) Returns: True if successful """ try: current_config = self.get_config() config_dict = current_config.to_dict() # Apply updates self._deep_update(config_dict, updates) # Create new config new_config = AppConfig.from_dict(config_dict) # Save if requested if save_to: if not self.save_config(new_config, save_to): return False # Update cached config self._config = new_config logger.info("Configuration updated", updates=updates) return True except Exception as e: logger.error("Failed to update configuration", error=str(e)) return False def get_setting(self, key: str, default: Any = None) -> Any: """Get specific setting value. Args: key: Setting key (dot notation supported) default: Default value if not found Returns: Setting value """ config = self.get_config() config_dict = config.to_dict() # Navigate through nested keys keys = key.split('.') value = config_dict for k in keys: if isinstance(value, dict) and k in value: value = value[k] else: return default return value def set_setting(self, key: str, value: Any, save_to: Optional[str] = None) -> bool: """Set specific setting value. Args: key: Setting key (dot notation supported) value: Setting value save_to: Target to save to Returns: True if successful """ keys = key.split('.') updates = {} # Build nested update dictionary current = updates for k in keys[:-1]: current[k] = {} current = current[k] current[keys[-1]] = value return self.update_config(updates, save_to) def _load_file_config(self) -> Optional[AppConfig]: """Load configuration from YAML file.""" if not self.config_file or not Path(self.config_file).exists(): return None try: with open(self.config_file, 'r', encoding='utf-8') as f: data = yaml.safe_load(f) if not data: return None return AppConfig.from_dict(data) except Exception as e: logger.error("Failed to load file configuration", file=self.config_file, error=str(e)) return None def _save_file_config(self, config: AppConfig) -> bool: """Save configuration to YAML file.""" if not self.config_file: return False try: # Ensure directory exists Path(self.config_file).parent.mkdir(parents=True, exist_ok=True) with open(self.config_file, 'w', encoding='utf-8') as f: yaml.dump(config.to_dict(), f, default_flow_style=False, indent=2) self._update_file_mtime() logger.info("Configuration saved to file", file=self.config_file) return True except Exception as e: logger.error("Failed to save file configuration", file=self.config_file, error=str(e)) return False def _load_db_config(self) -> Optional[AppConfig]: """Load configuration from database.""" if not self.db_manager: return None try: with self.db_manager.get_session() as session: configs = session.query(Configuration).all() if not configs: return None # Build config dictionary config_dict = {} for config in configs: keys = config.key.split('.') current = config_dict for key in keys[:-1]: if key not in current: current[key] = {} current = current[key] # Parse value based on type if config.value_type == 'int': current[keys[-1]] = int(config.value) elif config.value_type == 'float': current[keys[-1]] = float(config.value) elif config.value_type == 'bool': current[keys[-1]] = config.value.lower() == 'true' elif config.value_type == 'list': current[keys[-1]] = config.value.split(',') else: current[keys[-1]] = config.value return AppConfig.from_dict(config_dict) except Exception as e: logger.error("Failed to load database configuration", error=str(e)) return None def _save_db_config(self, config: AppConfig) -> bool: """Save configuration to database.""" if not self.db_manager: return False try: with self.db_manager.get_session() as session: # Clear existing configuration session.query(Configuration).delete() # Flatten config dictionary flat_config = self._flatten_dict(config.to_dict()) # Save each setting for key, value in flat_config.items(): value_type = 'str' if isinstance(value, int): value_type = 'int' elif isinstance(value, float): value_type = 'float' elif isinstance(value, bool): value_type = 'bool' elif isinstance(value, list): value_type = 'list' value = ','.join(str(v) for v in value) config_obj = Configuration( key=key, value=str(value), value_type=value_type, description=f"Configuration setting: {key}" ) session.add(config_obj) session.commit() logger.info("Configuration saved to database") return True except Exception as e: logger.error("Failed to save database configuration", error=str(e)) return False def _merge_configs(self, base: AppConfig, override: AppConfig) -> AppConfig: """Merge two configurations.""" base_dict = base.to_dict() override_dict = override.to_dict() self._deep_update(base_dict, override_dict) return AppConfig.from_dict(base_dict) def _deep_update(self, base: Dict[str, Any], updates: Dict[str, Any]) -> None: """Deep update dictionary.""" for key, value in updates.items(): if key in base and isinstance(base[key], dict) and isinstance(value, dict): self._deep_update(base[key], value) else: base[key] = value def _flatten_dict(self, d: Dict[str, Any], parent_key: str = '', sep: str = '.') -> Dict[str, Any]: """Flatten nested dictionary.""" items = [] for k, v in d.items(): new_key = f"{parent_key}{sep}{k}" if parent_key else k if isinstance(v, dict): items.extend(self._flatten_dict(v, new_key, sep=sep).items()) else: items.append((new_key, v)) return dict(items) def _should_reload(self) -> bool: """Check if configuration should be reloaded.""" if not self.auto_reload or not self.config_file: return False try: current_mtime = Path(self.config_file).stat().st_mtime return self._file_mtime is None or current_mtime > self._file_mtime except Exception: return False def _update_file_mtime(self) -> None: """Update file modification time.""" if self.config_file: try: self._file_mtime = Path(self.config_file).stat().st_mtime except Exception: self._file_mtime = None @contextmanager def config_context(self, **overrides): """Context manager for temporary configuration overrides.""" original_config = self._config try: if overrides: current_config = self.get_config() config_dict = current_config.to_dict() self._deep_update(config_dict, overrides) self._config = AppConfig.from_dict(config_dict) yield self._config finally: self._config = original_config # Global configuration manager instance _config_manager: Optional[ConfigManager] = None def init_config_manager( config_file: Optional[str] = None, db_manager: Optional[DatabaseManager] = None, auto_reload: bool = False ) -> ConfigManager: """Initialize global configuration manager. Args: config_file: Path to configuration file db_manager: Database manager instance auto_reload: Enable automatic reloading Returns: ConfigManager instance """ global _config_manager _config_manager = ConfigManager(config_file, db_manager, auto_reload) return _config_manager def get_config_manager() -> Optional[ConfigManager]: """Get global configuration manager. Returns: ConfigManager instance or None """ return _config_manager def get_config() -> AppConfig: """Get current application configuration. Returns: Application configuration """ if _config_manager: return _config_manager.get_config() else: # Fallback to environment-only config return AppConfig.from_env() def get_setting(key: str, default: Any = None) -> Any: """Get configuration setting. Args: key: Setting key default: Default value Returns: Setting value """ if _config_manager: return _config_manager.get_setting(key, default) else: # Fallback to environment variables env_key = key.upper().replace('.', '_') return os.getenv(env_key, default)