Llama 4 Maverick MCP Server

""" Configuration Management Module Author: Yobie Benjamin Version: 0.9 Date: August 1, 2025 This module manages all configuration for the MCP server, including: - Environment variable loading - Default values - Configuration validation - Runtime configuration access Configuration is loaded from (in order of precedence): 1. Environment variables 2. .env file 3. Default values in code """ import os from pathlib import Path from typing import Any, Dict, Optional from dataclasses import dataclass, field from dotenv import load_dotenv from pydantic import BaseModel, Field, validator # Load environment variables from .env file load_dotenv() @dataclass class Config: """ Main configuration class for the MCP server. This class centralizes all configuration options and provides validated access to settings. Users can override any setting via environment variables or .env file. Configuration Categories: - Ollama/Llama settings - MCP server settings - Security options - Performance tuning - Feature flags - Model parameters """ # Ollama/Llama Configuration # These control the connection to Ollama and model selection llama_api_url: str = field( default_factory=lambda: os.getenv("LLAMA_API_URL", "http://localhost:11434") ) llama_model_name: str = field( default_factory=lambda: os.getenv("LLAMA_MODEL_NAME", "llama3:latest") ) llama_api_key: str = field( default_factory=lambda: os.getenv("LLAMA_API_KEY", "") ) # MCP Server Configuration # Server-level settings (mostly for future HTTP mode) server_host: str = field( default_factory=lambda: os.getenv("MCP_SERVER_HOST", "localhost") ) server_port: int = field( default_factory=lambda: int(os.getenv("MCP_SERVER_PORT", "3000")) ) log_level: str = field( default_factory=lambda: os.getenv("MCP_LOG_LEVEL", "INFO") ) # Security Configuration # Control access and authentication enable_auth: bool = field( default_factory=lambda: os.getenv("MCP_ENABLE_AUTH", "false").lower() == "true" ) api_key: str = field( default_factory=lambda: os.getenv("MCP_API_KEY", "") ) allowed_hosts: str = field( default_factory=lambda: os.getenv("MCP_ALLOWED_HOSTS", "*") ) # Performance Configuration # Tune server performance and resource usage max_context_length: int = field( default_factory=lambda: int(os.getenv("MAX_CONTEXT_LENGTH", "128000")) ) max_concurrent_requests: int = field( default_factory=lambda: int(os.getenv("MAX_CONCURRENT_REQUESTS", "10")) ) request_timeout: int = field( default_factory=lambda: int(os.getenv("REQUEST_TIMEOUT_MS", "30000")) ) cache_ttl: int = field( default_factory=lambda: int(os.getenv("CACHE_TTL", "3600")) ) cache_size: int = field( default_factory=lambda: int(os.getenv("CACHE_MAX_SIZE", "1000")) ) # Feature Flags # Enable/disable specific functionality enable_streaming: bool = field( default_factory=lambda: os.getenv("ENABLE_STREAMING", "true").lower() != "false" ) enable_function_calling: bool = field( default_factory=lambda: os.getenv("ENABLE_FUNCTION_CALLING", "true").lower() != "false" ) enable_vision: bool = field( default_factory=lambda: os.getenv("ENABLE_VISION", "false").lower() == "true" ) enable_code_execution: bool = field( default_factory=lambda: os.getenv("ENABLE_CODE_EXECUTION", "false").lower() == "true" ) enable_web_search: bool = field( default_factory=lambda: os.getenv("ENABLE_WEB_SEARCH", "true").lower() != "false" ) # Model Generation Parameters # Control how the model generates text temperature: float = field( default_factory=lambda: float(os.getenv("TEMPERATURE", "0.7")) ) top_p: float = field( default_factory=lambda: float(os.getenv("TOP_P", "0.9")) ) top_k: int = field( default_factory=lambda: int(os.getenv("TOP_K", "40")) ) repeat_penalty: float = field( default_factory=lambda: float(os.getenv("REPEAT_PENALTY", "1.1")) ) seed: Optional[int] = field( default_factory=lambda: int(os.getenv("SEED")) if os.getenv("SEED") else None ) # File System Configuration # Control file access boundaries filesystem_base_path: Path = field( default_factory=lambda: Path(os.getenv("FILE_SYSTEM_BASE_PATH", os.getcwd())) ) allow_file_writes: bool = field( default_factory=lambda: os.getenv("ALLOW_FILE_WRITES", "true").lower() != "false" ) # Database Configuration (optional) # For tools that need database access database_url: str = field( default_factory=lambda: os.getenv("DATABASE_URL", "") ) database_pool_size: int = field( default_factory=lambda: int(os.getenv("DATABASE_POOL_SIZE", "10")) ) # Debug Configuration # Development and debugging options debug_mode: bool = field( default_factory=lambda: os.getenv("DEBUG", "false").lower() == "true" ) verbose_logging: bool = field( default_factory=lambda: os.getenv("VERBOSE_LOGGING", "false").lower() == "true" ) def __post_init__(self): """ Validate configuration after initialization. This method ensures: - Values are within valid ranges - Required settings are present - Incompatible options are flagged """ # Validate temperature range if not 0.0 <= self.temperature <= 2.0: print(f"Warning: Temperature {self.temperature} out of range [0, 2], using 0.7") self.temperature = 0.7 # Validate top_p range if not 0.0 <= self.top_p <= 1.0: print(f"Warning: Top-p {self.top_p} out of range [0, 1], using 0.9") self.top_p = 0.9 # Validate paths if not self.filesystem_base_path.exists(): print(f"Warning: Base path {self.filesystem_base_path} does not exist, using current directory") self.filesystem_base_path = Path.cwd() # Security warnings if self.enable_code_execution: print("⚠️ Warning: Code execution is enabled. This can be a security risk.") if self.allow_file_writes and not self.filesystem_base_path.is_absolute(): print("⚠️ Warning: File writes enabled with relative base path. Consider using absolute path.") def to_dict(self) -> Dict[str, Any]: """ Convert configuration to dictionary. Returns: Dictionary of all configuration values with sensitive data redacted """ config_dict = {} for key, value in self.__dict__.items(): # Redact sensitive values if any(sensitive in key.lower() for sensitive in ["key", "secret", "password", "token"]): config_dict[key] = "***REDACTED***" if value else None elif isinstance(value, Path): config_dict[key] = str(value) else: config_dict[key] = value return config_dict def get_model_params(self) -> Dict[str, Any]: """ Get model generation parameters. Returns: Dictionary of model parameters for Ollama API """ params = { "temperature": self.temperature, "top_p": self.top_p, "top_k": self.top_k, "repeat_penalty": self.repeat_penalty, } if self.seed is not None: params["seed"] = self.seed return params @property def is_production(self) -> bool: """Check if running in production mode.""" return os.getenv("ENV", "development").lower() == "production" @property def is_development(self) -> bool: """Check if running in development mode.""" return not self.is_production def validate_ollama_connection(self) -> bool: """ Validate that Ollama is accessible. Returns: True if Ollama is reachable, False otherwise """ import httpx try: response = httpx.get(f"{self.llama_api_url}/api/tags", timeout=5.0) return response.status_code == 200 except: return False def __repr__(self) -> str: """String representation of configuration.""" return f"Config(model={self.llama_model_name}, api_url={self.llama_api_url}, debug={self.debug_mode})" class ConfigValidator(BaseModel): """ Pydantic model for configuration validation. This provides additional validation using Pydantic's features for more complex configuration scenarios. """ temperature: float = Field(ge=0.0, le=2.0, default=0.7) top_p: float = Field(ge=0.0, le=1.0, default=0.9) top_k: int = Field(ge=1, le=100, default=40) repeat_penalty: float = Field(ge=0.1, le=2.0, default=1.1) max_context_length: int = Field(ge=1024, le=256000, default=128000) max_concurrent_requests: int = Field(ge=1, le=100, default=10) @validator("temperature") def validate_temperature(cls, v): """Ensure temperature is within reasonable bounds.""" if v < 0.0 or v > 2.0: raise ValueError("Temperature must be between 0.0 and 2.0") return v # Singleton configuration instance _config_instance: Optional[Config] = None def get_config() -> Config: """ Get the global configuration instance. Returns: Config: Global configuration object """ global _config_instance if _config_instance is None: _config_instance = Config() return _config_instance def reset_config(): """Reset the global configuration instance.""" global _config_instance _config_instance = None