"""
Configuration management for OpenSCAD MCP Server.
This module provides centralized configuration with environment variable
support, type safety, and sensible defaults.
"""
import asyncio
import logging
import logging.handlers
import os
from importlib.metadata import version as _pkg_version
from pathlib import Path
from typing import Optional
import yaml
from dotenv import load_dotenv
from pydantic import BaseModel, Field, field_validator
from ..types import TransportType
class RenderingConfig(BaseModel):
"""Rendering-specific configuration."""
max_concurrent: int = Field(
5,
ge=1,
le=20,
description="Maximum concurrent rendering operations",
)
timeout_seconds: int = Field(
300,
ge=30,
le=3600,
description="Render timeout in seconds",
)
max_image_width: int = Field(
4096,
ge=100,
le=8192,
description="Maximum image width",
)
max_image_height: int = Field(
4096,
ge=100,
le=8192,
description="Maximum image height",
)
default_color_scheme: str = Field(
"Cornfield",
description="Default OpenSCAD color scheme",
)
class CacheConfig(BaseModel):
"""Cache configuration."""
enabled: bool = Field(True, description="Enable caching")
directory: Path = Field(
Path.home() / ".cache" / "openscad-mcp",
description="Cache directory",
)
max_size_mb: int = Field(
500,
ge=100,
le=10000,
description="Maximum cache size in MB",
)
ttl_hours: int = Field(
24,
ge=1,
le=168,
description="Cache time-to-live in hours",
)
@field_validator("directory")
@classmethod
def validate_directory(cls, v: Path) -> Path:
"""Validate that the directory path is absolute."""
return v
def ensure_cache_directory(self) -> None:
"""Create the cache directory if it does not exist.
Call this explicitly when you need the directory to be present,
rather than relying on validation-time side effects.
"""
self.directory.mkdir(parents=True, exist_ok=True)
class SecurityConfig(BaseModel):
"""Security configuration."""
rate_limit: int = Field(
60,
ge=0,
le=1000,
description="Max requests per minute (0 to disable)",
)
max_file_size_mb: int = Field(
10,
ge=1,
le=100,
description="Maximum SCAD file size in MB",
)
allowed_paths: Optional[list[str]] = Field(
None,
description="Allowed paths for file access",
)
class LoggingConfig(BaseModel):
"""Logging configuration."""
level: str = Field(
"INFO",
description="Logging level",
pattern="^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$",
)
file: Optional[Path] = Field(
None,
description="Log file path",
)
max_size_mb: int = Field(
100,
ge=10,
le=1000,
description="Maximum log file size in MB",
)
rotate_count: int = Field(
5,
ge=1,
le=10,
description="Number of rotated log files to keep",
)
class ServerConfig(BaseModel):
"""Server configuration."""
name: str = Field(
"OpenSCAD MCP Server",
description="Server name",
)
version: str = Field(
default_factory=lambda: _pkg_version("openscad-mcp"),
description="Server version",
)
transport: TransportType = Field(
TransportType.STDIO,
description="Transport type",
)
host: str = Field(
"localhost",
description="Host for HTTP/SSE transport",
)
port: int = Field(
8000,
ge=1024,
le=65535,
description="Port for HTTP/SSE transport",
)
class Config(BaseModel):
"""Main configuration class."""
# Paths
openscad_path: Optional[str] = Field(
None,
description="Path to OpenSCAD executable",
)
imagemagick_path: Optional[str] = Field(
None,
description="Path to ImageMagick convert command",
)
temp_dir: Path = Field(
default=None,
validate_default=True,
description="Temporary file directory",
)
# Sub-configurations
server: ServerConfig = Field(default_factory=ServerConfig)
rendering: RenderingConfig = Field(default_factory=RenderingConfig)
cache: CacheConfig = Field(default_factory=CacheConfig)
security: SecurityConfig = Field(default_factory=SecurityConfig)
logging: LoggingConfig = Field(default_factory=LoggingConfig)
@field_validator("temp_dir", mode="before")
@classmethod
def set_temp_dir_default(cls, v: Optional[Path]) -> Path:
"""Set temp_dir default at validation time instead of class definition time.
Using Path.cwd() in a Field default evaluates once at import time,
which can produce incorrect results. This validator defers the call
to when the Config instance is actually created.
"""
if v is None:
return Path("/tmp/openscad-mcp")
return Path(v)
@classmethod
def from_env(cls, env_file: Optional[str] = None) -> "Config":
"""
Load configuration from environment variables.
Args:
env_file: Optional path to .env file
Returns:
Configured Config instance
"""
if env_file:
load_dotenv(env_file)
else:
load_dotenv() # Load from default .env
# Build configuration from environment
config_dict = {}
# Paths
if openscad := os.getenv("OPENSCAD_PATH"):
config_dict["openscad_path"] = openscad
if imagemagick := os.getenv("IMAGEMAGICK_PATH"):
config_dict["imagemagick_path"] = imagemagick
if temp_dir := os.getenv("MCP_TEMP_DIR"):
config_dict["temp_dir"] = Path(temp_dir)
# Server configuration
server_config = {}
if transport := os.getenv("MCP_TRANSPORT"):
server_config["transport"] = transport
if host := os.getenv("MCP_HOST"):
server_config["host"] = host
if port := os.getenv("MCP_PORT"):
server_config["port"] = int(port)
if server_config:
config_dict["server"] = ServerConfig(**server_config)
# Rendering configuration
rendering_config = {}
if max_concurrent := os.getenv("MCP_MAX_CONCURRENT_RENDERS"):
rendering_config["max_concurrent"] = int(max_concurrent)
if timeout := os.getenv("MCP_RENDER_TIMEOUT"):
rendering_config["timeout_seconds"] = int(timeout)
if max_width := os.getenv("MCP_MAX_IMAGE_WIDTH"):
rendering_config["max_image_width"] = int(max_width)
if max_height := os.getenv("MCP_MAX_IMAGE_HEIGHT"):
rendering_config["max_image_height"] = int(max_height)
if rendering_config:
config_dict["rendering"] = RenderingConfig(**rendering_config)
# Cache configuration
cache_config = {}
if cache_enabled := os.getenv("MCP_CACHE_ENABLED"):
cache_config["enabled"] = cache_enabled.lower() == "true"
if cache_size := os.getenv("MCP_CACHE_SIZE_MB"):
cache_config["max_size_mb"] = int(cache_size)
if cache_ttl := os.getenv("MCP_CACHE_TTL_HOURS"):
cache_config["ttl_hours"] = int(cache_ttl)
if cache_config:
config_dict["cache"] = CacheConfig(**cache_config)
# Security configuration
security_config = {}
if rate_limit := os.getenv("MCP_RATE_LIMIT"):
security_config["rate_limit"] = int(rate_limit)
if max_file_size := os.getenv("MCP_MAX_FILE_SIZE_MB"):
security_config["max_file_size_mb"] = int(max_file_size)
if security_config:
config_dict["security"] = SecurityConfig(**security_config)
# Logging configuration
logging_config = {}
if log_level := os.getenv("MCP_LOG_LEVEL"):
logging_config["level"] = log_level
if log_file := os.getenv("MCP_LOG_FILE"):
logging_config["file"] = Path(log_file)
if logging_config:
config_dict["logging"] = LoggingConfig(**logging_config)
return cls(**config_dict)
@classmethod
def from_yaml(cls, yaml_file: str) -> "Config":
"""
Load configuration from YAML file.
Args:
yaml_file: Path to YAML configuration file
Returns:
Configured Config instance
"""
with open(yaml_file, "r") as f:
config_dict = yaml.safe_load(f)
return cls(**config_dict)
def to_yaml(self, yaml_file: str) -> None:
"""
Save configuration to YAML file.
Args:
yaml_file: Path to save YAML configuration
"""
with open(yaml_file, "w") as f:
yaml.dump(
self.model_dump(mode="json"),
f,
default_flow_style=False,
)
def setup_logging(logging_config: Optional[LoggingConfig] = None) -> None:
"""Configure the root logger based on LoggingConfig settings.
Sets the root logger level and optionally adds a rotating file handler
if a log file path is configured.
Args:
logging_config: Logging configuration. If None, uses defaults from
the global config instance.
"""
if logging_config is None:
logging_config = get_config().logging
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, logging_config.level, logging.INFO))
# Add a rotating file handler if a log file is configured
if logging_config.file is not None:
# Ensure the parent directory exists
logging_config.file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.handlers.RotatingFileHandler(
filename=str(logging_config.file),
maxBytes=logging_config.max_size_mb * 1024 * 1024,
backupCount=logging_config.rotate_count,
)
file_handler.setLevel(getattr(logging, logging_config.level, logging.INFO))
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
file_handler.setFormatter(formatter)
root_logger.addHandler(file_handler)
# Module-level concurrency semaphore for rendering.
# Initialized lazily via get_render_semaphore() so that the semaphore is
# created inside a running event loop and respects the configured max_concurrent.
_render_semaphore: Optional[asyncio.Semaphore] = None
def get_render_semaphore() -> asyncio.Semaphore:
"""Return the module-level rendering concurrency semaphore.
Creates the semaphore on first call, using the current global config's
``rendering.max_concurrent`` value. The semaphore is cached for the
lifetime of the process.
Returns:
An asyncio.Semaphore that limits concurrent rendering operations.
"""
global _render_semaphore
if _render_semaphore is None:
config = get_config()
_render_semaphore = asyncio.Semaphore(config.rendering.max_concurrent)
return _render_semaphore
# Global configuration instance
_config: Optional[Config] = None
def get_config() -> Config:
"""
Get the global configuration instance.
Returns:
The global Config instance
"""
global _config
if _config is None:
_config = Config.from_env()
return _config
def set_config(config: Config) -> None:
"""
Set the global configuration instance.
Args:
config: Config instance to set globally
"""
global _config, _render_semaphore
_config = config
# Reset semaphore so it picks up the new max_concurrent on next access
_render_semaphore = None
