config.py•6.48 kB
"""Configuration helpers for the ComfyUI MCP server."""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping, MutableMapping, Optional
import tomllib
@dataclass(slots=True)
class DirectorySettings:
"""Paths pointing to ComfyUI asset directories."""
checkpoints: Optional[Path] = None
loras: Optional[Path] = None
vaes: Optional[Path] = None
text_encoders: Optional[Path] = None
embeddings: Optional[Path] = None
@classmethod
def from_dict(cls, data: Optional[Mapping[str, Any]]) -> "DirectorySettings":
data = data or {}
return cls(
checkpoints=_maybe_path(data.get("checkpoints")),
loras=_maybe_path(data.get("loras")),
vaes=_maybe_path(data.get("vaes")),
text_encoders=_maybe_path(data.get("text_encoders")),
embeddings=_maybe_path(data.get("embeddings")),
)
@dataclass(slots=True)
class ParameterBounds:
"""Default numeric bounds for workflow parameters."""
cfg_min: float = 1.0
cfg_max: float = 30.0
steps_min: int = 1
steps_max: int = 150
width_min: int = 64
width_max: int = 2048
height_min: int = 64
height_max: int = 2048
@classmethod
def from_dict(cls, data: Optional[Mapping[str, Any]]) -> "ParameterBounds":
if not data:
return cls()
filtered = {
key: data[key]
for key in (
"cfg_min",
"cfg_max",
"steps_min",
"steps_max",
"width_min",
"width_max",
"height_min",
"height_max",
)
if key in data
}
return cls(**filtered)
@dataclass(slots=True)
class FeatureToggles:
"""Feature switches that can be enabled or disabled."""
enable_streaming: bool = True
enable_batch_execution: bool = False
watch_workflows: bool = False
@classmethod
def from_dict(cls, data: Optional[Mapping[str, Any]]) -> "FeatureToggles":
if not data:
return cls()
filtered = {key: data[key] for key in ("enable_streaming", "enable_batch_execution", "watch_workflows") if key in data}
return cls(**filtered)
@dataclass(slots=True)
class ComfyUISettings:
"""Top-level settings for the MCP server."""
base_url: str = "http://127.0.0.1:8188"
api_key: Optional[str] = None
default_workflow: Optional[str] = None
workflows_path: Path = Path("workflows")
directories: DirectorySettings = field(default_factory=DirectorySettings)
default_bounds: ParameterBounds = field(default_factory=ParameterBounds)
feature_toggles: FeatureToggles = field(default_factory=FeatureToggles)
@classmethod
def from_dict(cls, data: Optional[Mapping[str, Any]]) -> "ComfyUISettings":
data = dict(data or {})
directories = DirectorySettings.from_dict(data.get("directories"))
bounds = ParameterBounds.from_dict(data.get("default_bounds"))
toggles = FeatureToggles.from_dict(data.get("feature_toggles"))
workflows_path = _maybe_path(data.get("workflows_path")) or Path("workflows")
return cls(
base_url=data.get("base_url", cls.base_url),
api_key=data.get("api_key"),
default_workflow=data.get("default_workflow"),
workflows_path=workflows_path,
directories=directories,
default_bounds=bounds,
feature_toggles=toggles,
)
def resolve_path(self, *paths: str | os.PathLike[str]) -> Path:
"""Resolve a path relative to the repository root."""
base = Path.cwd()
return base.joinpath(*paths)
def workflows_directory(self) -> Path:
"""Return the absolute path to the workflows directory."""
path = self.workflows_path
if not path.is_absolute():
path = Path.cwd() / path
return path
DEFAULT_CONFIG_PATH_ENV = "COMFYUI_MCP_CONFIG"
ENV_PREFIX = "COMFYUI_MCP_"
def load_settings(config_path: Optional[os.PathLike[str]] = None, env: Optional[Mapping[str, str]] = None) -> ComfyUISettings:
"""Load settings from a TOML file and environment variables."""
env = dict(env or os.environ)
config_path = Path(config_path or env.get(DEFAULT_CONFIG_PATH_ENV, "")).expanduser().resolve() if (
config_path or env.get(DEFAULT_CONFIG_PATH_ENV)
) else None
data: MutableMapping[str, Any] = {}
if config_path:
with open(config_path, "rb") as fp:
data.update(tomllib.load(fp))
apply_environment_overrides(data, env)
return ComfyUISettings.from_dict(data)
def apply_environment_overrides(config: MutableMapping[str, Any], env: Mapping[str, str]) -> None:
"""Apply environment overrides using the ``COMFYUI_MCP_*`` prefix."""
for key, value in env.items():
if not key.startswith(ENV_PREFIX):
continue
field = key.removeprefix(ENV_PREFIX).lower()
if field == "base_url":
config["base_url"] = value
elif field == "api_key":
config["api_key"] = value
elif field == "default_workflow":
config["default_workflow"] = value
elif field == "workflows_path":
config["workflows_path"] = value
elif field.startswith("directories_"):
directories = config.setdefault("directories", {})
directories[field.replace("directories_", "")] = value
elif field.startswith("bounds_"):
bounds = config.setdefault("default_bounds", {})
bounds[field.replace("bounds_", "")] = _coerce_numeric(value)
elif field.startswith("features_"):
features = config.setdefault("feature_toggles", {})
features[field.replace("features_", "")] = _coerce_bool(value)
def _maybe_path(value: Any) -> Optional[Path]:
if value is None or value == "":
return None
return Path(str(value)).expanduser()
def _coerce_numeric(value: str) -> Any:
try:
if "." in value:
return float(value)
return int(value)
except ValueError:
return value
def _coerce_bool(value: str) -> bool:
return value.lower() in {"1", "true", "yes", "on"}
__all__ = [
"ComfyUISettings",
"DirectorySettings",
"FeatureToggles",
"ParameterBounds",
"load_settings",
]