config_manager.py•10.9 kB
"""
Configuration system for the Sectional MCP Panel.
"""
import json
import jsonschema
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field
# Configuration schema definition
CONFIG_SCHEMA = {
"type": "object",
"properties": {
"configSchemaVersion": {"type": "string"},
"panelName": {"type": "string"},
"globalDefaults": {
"type": "object",
"properties": {
"settings": {
"type": "object",
"properties": {
"environmentVars": {"type": "object"},
"resourceLimits": {
"type": "object",
"properties": {
"cpuRequest": {"type": ["number", "null"]},
"cpuLimit": {"type": ["number", "null"]},
"memoryRequestMB": {"type": ["integer", "null"]},
"memoryLimitMB": {"type": ["integer", "null"]}
}
},
"runtimeOptions": {
"type": "object",
"properties": {
"restartPolicy": {"type": "string", "enum": ["no", "on_failure", "always"]},
"maxRestarts": {"type": "integer"},
"terminationGracePeriodSeconds": {"type": "integer"}
}
},
"securityContext": {
"type": "object",
"properties": {
"runAsUser": {"type": ["integer", "null"]},
"runAsGroup": {"type": ["integer", "null"]},
"readOnlyRootFilesystem": {"type": "boolean"}
}
}
}
}
},
"required": ["settings"]
},
"sections": {
"type": "array",
"items": {
"type": "object",
"properties": {
"sectionName": {"type": "string"},
"description": {"type": "string"},
"settings": {
"type": "object",
"properties": {
"environmentVars": {"type": "object"},
"resourceLimits": {
"type": "object",
"properties": {
"cpuRequest": {"type": ["number", "null"]},
"cpuLimit": {"type": ["number", "null"]},
"memoryRequestMB": {"type": ["integer", "null"]},
"memoryLimitMB": {"type": ["integer", "null"]}
}
},
"runtimeOptions": {
"type": "object",
"properties": {
"restartPolicy": {"type": "string", "enum": ["no", "on_failure", "always"]},
"maxRestarts": {"type": "integer"},
"terminationGracePeriodSeconds": {"type": "integer"}
}
},
"securityContext": {
"type": "object",
"properties": {
"runAsUser": {"type": ["integer", "null"]},
"runAsGroup": {"type": ["integer", "null"]},
"readOnlyRootFilesystem": {"type": "boolean"}
}
}
}
},
"servers": {
"type": "array",
"items": {
"type": "object",
"properties": {
"serverName": {"type": "string"},
"description": {"type": "string"},
"runtimeDefinition": {
"type": "object",
"properties": {
"type": {"type": "string"},
"command": {"type": "string"},
"args": {"type": "array", "items": {"type": "string"}},
"workingDirectory": {"type": "string"},
"ports": {
"type": "array",
"items": {
"type": "object",
"properties": {
"containerPort": {"type": "integer"},
"protocol": {"type": "string", "enum": ["TCP", "UDP"]}
},
"required": ["containerPort"]
}
}
},
"required": ["type"]
},
"settings": {
"type": "object",
"properties": {
"environmentVars": {"type": "object"},
"resourceLimits": {
"type": "object",
"properties": {
"cpuRequest": {"type": ["number", "null"]},
"cpuLimit": {"type": ["number", "null"]},
"memoryRequestMB": {"type": ["integer", "null"]},
"memoryLimitMB": {"type": ["integer", "null"]}
}
},
"runtimeOptions": {
"type": "object",
"properties": {
"restartPolicy": {"type": "string", "enum": ["no", "on_failure", "always"]},
"maxRestarts": {"type": "integer"},
"terminationGracePeriodSeconds": {"type": "integer"}
}
},
"securityContext": {
"type": "object",
"properties": {
"runAsUser": {"type": ["integer", "null"]},
"runAsGroup": {"type": ["integer", "null"]},
"readOnlyRootFilesystem": {"type": "boolean"}
}
}
}
}
},
"required": ["serverName", "runtimeDefinition"]
}
}
},
"required": ["sectionName"]
}
}
},
"required": ["configSchemaVersion", "panelName"]
}
class ConfigManager:
"""Configuration manager for the Sectional MCP Panel."""
def __init__(self):
self.schema = CONFIG_SCHEMA
def validate_config(self, config: Dict[str, Any]) -> bool:
"""Validate a configuration against the schema."""
try:
jsonschema.validate(instance=config, schema=self.schema)
return True
except jsonschema.exceptions.ValidationError:
return False
def resolve_server_settings(
self,
global_settings: Dict[str, Any],
section_settings: Dict[str, Any],
server_settings: Dict[str, Any]
) -> Dict[str, Any]:
"""
Resolve effective settings for a server by applying the hierarchy:
Global -> Section -> Server
"""
# Start with global settings
effective_settings = self._deep_copy(global_settings)
# Apply section settings
if section_settings:
effective_settings = self._deep_merge(effective_settings, section_settings)
# Apply server settings
if server_settings:
effective_settings = self._deep_merge(effective_settings, server_settings)
return effective_settings
def _deep_copy(self, obj: Dict[str, Any]) -> Dict[str, Any]:
"""Create a deep copy of a dictionary."""
return json.loads(json.dumps(obj))
def _deep_merge(self, base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
"""
Deep merge two dictionaries, with values from override taking precedence.
Lists are replaced, not merged.
"""
result = self._deep_copy(base)
for key, value in override.items():
# If both values are dictionaries, merge them recursively
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._deep_merge(result[key], value)
else:
# Otherwise, override the value
result[key] = self._deep_copy(value)
return result
def load_config_from_file(self, file_path: str) -> Dict[str, Any]:
"""Load configuration from a JSON file."""
with open(file_path, 'r') as f:
config = json.load(f)
if not self.validate_config(config):
raise ValueError("Invalid configuration format")
return config
def save_config_to_file(self, config: Dict[str, Any], file_path: str) -> None:
"""Save configuration to a JSON file."""
if not self.validate_config(config):
raise ValueError("Invalid configuration format")
with open(file_path, 'w') as f:
json.dump(config, f, indent=2)