"""
Configuration management for the MCP system
"""
import os
import json
import logging
from dataclasses import dataclass
from typing import Optional, Dict, Any
from sqlalchemy import create_engine, Engine
from shared.models import DatabaseConnection, EmbeddingConfig
from shared.exceptions import ConfigurationError
logger = logging.getLogger(__name__)
@dataclass
class LLMConfig:
"""Configuration for LLM integration"""
provider: str # "openai", "anthropic", "local", etc.
model: str
api_key: Optional[str] = None
base_url: Optional[str] = None
temperature: float = 0.7
max_tokens: int = 4000
@dataclass
class AppConfig:
"""Main application configuration"""
database: DatabaseConnection
llm: LLMConfig
embeddings: EmbeddingConfig
log_level: str = "INFO"
cache_ttl: int = 3600 # Cache TTL in seconds
@property
def engine(self) -> Engine:
"""Get SQLAlchemy engine"""
if not hasattr(self, '_engine'):
self._engine = create_engine(
self.database.get_connection_string(),
pool_size=5,
max_overflow=10,
pool_pre_ping=True
)
return self._engine
def load_config(config_path: Optional[str] = None) -> AppConfig:
"""
Load configuration from file or environment variables
Args:
config_path: Optional path to configuration file
Returns:
AppConfig object
Raises:
ConfigurationError: If configuration is invalid or missing
"""
try:
if config_path and os.path.exists(config_path):
return _load_from_file(config_path)
else:
return _load_from_environment()
except Exception as e:
logger.error(f"Failed to load configuration: {e}")
raise ConfigurationError(f"Configuration loading failed: {e}")
def _load_from_file(config_path: str) -> AppConfig:
"""Load configuration from JSON file"""
try:
with open(config_path, 'r') as f:
config_data = json.load(f)
return AppConfig(
database=DatabaseConnection(
host=config_data['database']['host'],
port=config_data['database']['port'],
database=config_data['database']['database'],
username=config_data['database']['username'],
password=config_data['database']['password'],
schema=config_data['database'].get('schema')
),
llm=LLMConfig(
provider=config_data['llm']['provider'],
model=config_data['llm']['model'],
api_key=config_data['llm'].get('api_key'),
base_url=config_data['llm'].get('base_url'),
temperature=config_data['llm'].get('temperature', 0.7),
max_tokens=config_data['llm'].get('max_tokens', 4000)
),
embeddings=EmbeddingConfig(
model=config_data['embeddings']['model'],
dimension=config_data['embeddings']['dimension'],
provider=config_data['embeddings']['provider'],
api_key=config_data['embeddings'].get('api_key')
),
log_level=config_data.get('log_level', 'INFO'),
cache_ttl=config_data.get('cache_ttl', 3600)
)
except KeyError as e:
raise ConfigurationError(f"Missing required configuration key: {e}")
except json.JSONDecodeError as e:
raise ConfigurationError(f"Invalid JSON in config file: {e}")
def _load_from_environment() -> AppConfig:
"""Load configuration from environment variables"""
try:
# Database configuration
db_config = DatabaseConnection(
host=_get_env_var('DB_HOST', 'localhost'),
port=int(_get_env_var('DB_PORT', '5432')),
database=_get_env_var('DB_NAME', required=True),
username=_get_env_var('DB_USER', required=True),
password=_get_env_var('DB_PASSWORD', required=True),
schema=_get_env_var('DB_SCHEMA')
)
# LLM configuration
llm_config = LLMConfig(
provider=_get_env_var('LLM_PROVIDER', 'openai'),
model=_get_env_var('LLM_MODEL', 'gpt-3.5-turbo'),
api_key=_get_env_var('LLM_API_KEY'),
base_url=_get_env_var('LLM_BASE_URL'),
temperature=float(_get_env_var('LLM_TEMPERATURE', '0.7')),
max_tokens=int(_get_env_var('LLM_MAX_TOKENS', '4000'))
)
# Embeddings configuration
embeddings_config = EmbeddingConfig(
model=_get_env_var('EMBEDDINGS_MODEL', 'text-embedding-ada-002'),
dimension=int(_get_env_var('EMBEDDINGS_DIMENSION', '1536')),
provider=_get_env_var('EMBEDDINGS_PROVIDER', 'openai'),
api_key=_get_env_var('EMBEDDINGS_API_KEY')
)
return AppConfig(
database=db_config,
llm=llm_config,
embeddings=embeddings_config,
log_level=_get_env_var('LOG_LEVEL', 'INFO'),
cache_ttl=int(_get_env_var('CACHE_TTL', '3600'))
)
except ValueError as e:
raise ConfigurationError(f"Invalid environment variable value: {e}")
def _get_env_var(name: str, default: Optional[str] = None, required: bool = False) -> str:
"""Get environment variable with optional default and required check"""
value = os.getenv(name, default)
if required and value is None:
raise ConfigurationError(f"Required environment variable {name} is not set")
return value
def create_sample_config(output_path: str = "config.json"):
"""Create a sample configuration file"""
sample_config = {
"database": {
"host": "localhost",
"port": 5432,
"database": "your_database",
"username": "your_username",
"password": "your_password",
"schema": "public"
},
"llm": {
"provider": "openai",
"model": "gpt-3.5-turbo",
"api_key": "your_openai_api_key",
"temperature": 0.7,
"max_tokens": 4000
},
"embeddings": {
"model": "text-embedding-ada-002",
"dimension": 1536,
"provider": "openai",
"api_key": "your_openai_api_key"
},
"log_level": "INFO",
"cache_ttl": 3600
}
with open(output_path, 'w') as f:
json.dump(sample_config, f, indent=2)
logger.info(f"Sample configuration created at: {output_path}")
