File size: 8,352 Bytes
a2ec7b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 | #!/usr/bin/env python3
"""
Centralized Configuration Schema for MCPMark
=============================================
This module provides a unified configuration system with validation,
type safety, and support for multiple configuration sources.
"""
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Optional
import yaml
from dotenv import load_dotenv
from src.logger import get_logger
logger = get_logger(__name__)
# Lazy import to avoid circular dependencies
def get_service_definition(service_name: str) -> dict:
from src.services import get_service_definition as _get_service_def
return _get_service_def(service_name)
@dataclass
class ConfigValue:
"""Represents a configuration value with metadata."""
key: str
value: Any
source: str # 'env', 'file', 'default'
required: bool = True
description: str = ""
validator: Optional[callable] = None
def validate(self) -> bool:
"""Validate the configuration value."""
if self.required and self.value is None:
raise ValueError(f"Required configuration '{self.key}' is missing")
if self.validator and self.value is not None:
if not self.validator(self.value):
raise ValueError(f"Invalid value for '{self.key}': {self.value}")
return True
class ConfigSchema(ABC):
"""Abstract base class for service configuration schemas."""
def __init__(self, service_name: str):
self.service_name = service_name
self._values: Dict[str, ConfigValue] = {}
self._load_dotenv()
self._define_schema()
self._load_values()
self._validate()
@abstractmethod
def _define_schema(self) -> None:
"""Define the configuration schema for this service."""
pass
def _load_dotenv(self) -> None:
"""Load environment variables from .mcp_env file."""
load_dotenv(dotenv_path=".mcp_env", override=False)
def _add_config(
self,
key: str,
env_var: Optional[str] = None,
default: Any = None,
required: bool = True,
description: str = "",
validator: Optional[callable] = None,
transform: Optional[callable] = None,
) -> None:
"""Add a configuration value to the schema."""
# Try to get value from environment first
value = None
source = "default"
if env_var:
env_value = os.getenv(env_var)
if env_value is not None:
value = transform(env_value) if transform else env_value
source = "env"
# Use default if no environment value
if value is None and default is not None:
value = default
source = "default"
self._values[key] = ConfigValue(
key=key,
value=value,
source=source,
required=required,
description=description,
validator=validator,
)
def _load_values(self) -> None:
"""Load configuration values from file if available."""
config_file = Path(f"config/{self.service_name}.yaml")
if config_file.exists():
with open(config_file) as f:
file_config = yaml.safe_load(f)
for key, value in file_config.items():
if key in self._values and self._values[key].value is None:
self._values[key].value = value
self._values[key].source = "file"
def _validate(self) -> None:
"""Validate all configuration values."""
for config_value in self._values.values():
config_value.validate()
def get(self, key: str, default: Any = None) -> Any:
"""Get a configuration value."""
if key in self._values:
return self._values[key].value
return default
def get_all(self) -> Dict[str, Any]:
"""Get all configuration values as a dictionary."""
return {k: v.value for k, v in self._values.items()}
def get_debug_info(self) -> Dict[str, Dict[str, Any]]:
"""Get detailed configuration information for debugging."""
return {
k: {
"value": v.value,
"source": v.source,
"required": v.required,
"description": v.description,
}
for k, v in self._values.items()
}
class GenericConfigSchema(ConfigSchema):
"""Generic configuration schema that reads from service definitions."""
def __init__(self, service_name: str):
# Get service definition before calling parent init
self.service_definition = get_service_definition(service_name)
super().__init__(service_name)
def _define_schema(self) -> None:
"""Define schema from service definition."""
config_schema = self.service_definition.get("config_schema", {})
for key, config in config_schema.items():
# Handle transform strings
transform = None
transform_str = config.get("transform")
if transform_str == "bool":
transform = lambda x: x.lower() in ["true", "1", "yes"]
elif transform_str == "int":
transform = int
elif transform_str == "path":
transform = lambda x: Path(x) if x else None
elif transform_str == "list":
transform = lambda x: [t.strip() for t in x.split(",")] if x else []
# Handle validator strings
validator = None
validator_str = config.get("validator")
if validator_str == "port":
validator = lambda x: 1 <= x <= 65535
elif validator_str and validator_str.startswith("in:"):
valid_values = validator_str[3:].split(",")
validator = lambda x, values=valid_values: x in values
self._add_config(
key=key,
env_var=config.get("env_var"),
default=config.get("default"),
required=config.get("required", True),
description=config.get("description", ""),
validator=validator,
transform=transform,
)
# Configuration Registry
class ConfigRegistry:
"""Central registry for all service configurations."""
_instances: Dict[str, ConfigSchema] = {}
@classmethod
def get_config(cls, service_name: str) -> ConfigSchema:
"""Get or create configuration for a service."""
if service_name not in cls._instances:
cls._instances[service_name] = GenericConfigSchema(service_name)
return cls._instances[service_name]
@classmethod
def validate_all(cls) -> Dict[str, bool]:
"""Validate all registered configurations."""
from src.services import get_supported_mcp_services
results = {}
for service_name in get_supported_mcp_services():
try:
cls.get_config(service_name)
results[service_name] = True
except Exception as e:
logger.error(f"Configuration validation failed for {service_name}: {e}")
results[service_name] = False
return results
@classmethod
def export_template(cls, service_name: str, output_path: Path) -> None:
"""Export a configuration template for a service."""
config = cls.get_config(service_name)
template = {"service": service_name, "configuration": {}}
for key, config_value in config._values.items():
template["configuration"][key] = {
"value": config_value.value
if config_value.source == "default"
else None,
"description": config_value.description,
"required": config_value.required,
"env_var": f"${{{key.upper()}}}",
}
with open(output_path, "w") as f:
yaml.dump(template, f, default_flow_style=False, sort_keys=False)
# Utility Functions
def get_service_config(service_name: str) -> Dict[str, Any]:
"""Get service configuration as a dictionary."""
return ConfigRegistry.get_config(service_name).get_all()
|