ComfyUI MCP Server

from __future__ import annotations import os from typing import Any, Dict, List, Optional from mcp.server.fastmcp import FastMCP from .comfy_bridge import ComfyBridge, ComfyImportError mcp = FastMCP("ComfyUI MCP (direct import)", json_response=True) def _get_bridge() -> ComfyBridge: root = os.environ.get("COMFYUI_ROOT", "").strip() if not root: raise ComfyImportError("COMFYUI_ROOT is not set.") enable_execute = os.environ.get("COMFYUI_ENABLE_EXECUTE", "0").strip() == "1" return ComfyBridge(comfyui_root=root, enable_execute=enable_execute) @mcp.tool() def comfy_list_nodes() -> List[str]: """ List installed ComfyUI node class types. """ bridge = _get_bridge() return bridge.list_nodes() @mcp.tool() def comfy_node_schema(node_type: str) -> Dict[str, Any]: """ Get schema for a node_type: INPUT_TYPES() + RETURN_TYPES + misc metadata. """ bridge = _get_bridge() schema = bridge.node_schema(node_type) return { "node_type": schema.node_type, "input": schema.input, "output": schema.output, "output_is_list": schema.output_is_list, "output_name": schema.output_name, "category": schema.category, "description": schema.description, } @mcp.tool() def comfy_validate_workflow(workflow_json: Dict[str, Any]) -> Dict[str, Any]: """ Validate a ComfyUI workflow/prompt JSON using ComfyUI's internal validate_prompt. Accepts either: - {"prompt": {...}} or - raw prompt dict {...} """ bridge = _get_bridge() return bridge.validate_workflow(workflow_json) @mcp.tool() def comfy_dry_run(workflow_json: Dict[str, Any], steps: Optional[int] = 2) -> Dict[str, Any]: """ Optional: best-effort execution attempt after validation. Disabled unless COMFYUI_ENABLE_EXECUTE=1. """ bridge = _get_bridge() return bridge.dry_run(workflow_json, steps=steps) def main() -> None: # Claude Code commonly uses stdio transport for local tools. mcp.run(transport="stdio") if __name__ == "__main__": main()