ComfyUI MCP Server

Instructions

Implementation Reference

• comfy_mcp/mcp/server.py:56-116 (handler)

  The core handler function for the 'read_workflow' tool, decorated with @mcp.tool for registration. Reads workflow files (.json or .dsl), auto-converts JSON to DSL using JsonToDslConverter, handles path validation, errors, and logging.

  @mcp.tool
  async def read_workflow(ctx: Context, filepath: str) -> str:
      """Read a workflow file and return it as DSL format.
  
      Supports both JSON and DSL input files. Automatically detects format
      and converts JSON to DSL transparently.
  
      Args:
          filepath: Path to workflow file (.json or .dsl)
  
      Returns:
          Workflow content in DSL format
  
      Examples:
          read_workflow("workflows/my_workflow.json")
          read_workflow("../dsl/examples/dsl/simple.dsl")
      """
      await ctx.info(f"Reading workflow from {filepath}")
  
      try:
          path = Path(filepath).resolve()
  
          # Allow access to dsl/examples directory
          if "examples" not in str(path):
              path = validate_path(filepath)
  
          if not path.exists():
              raise ToolError(f"File not found: {filepath}")
  
          content = path.read_text()
  
          # If already DSL, return as-is
          if path.suffix == ".dsl":
              await ctx.info("File is already in DSL format")
              return content
  
          # If JSON, convert to DSL
          if path.suffix == ".json":
              await ctx.info("Converting JSON to DSL...")
  
              workflow = json.loads(content)
  
              # Handle full ComfyUI format
              if is_full_workflow_format(workflow):
                  workflow = full_workflow_to_simplified(workflow)
  
              # Convert to DSL
              converter = JsonToDslConverter()
              workflow_ast = converter.convert(workflow)
              dsl_text = str(workflow_ast)
  
              await ctx.info(f"✓ Converted to DSL ({len(dsl_text)} chars)")
              return dsl_text
  
          raise ToolError(f"Unsupported file format: {path.suffix}")
  
      except json.JSONDecodeError as e:
          raise ToolError(f"Invalid JSON in {filepath}: {e}")
      except Exception as e:
          raise ToolError(f"Error reading workflow: {e}")

• comfy_mcp/mcp/server.py:56-56 (registration)

  The @mcp.tool decorator registers the read_workflow function as an MCP tool.

  @mcp.tool

• comfy_mcp/mcp/server.py:44-52 (helper)

  Helper function validate_path used by read_workflow for security path validation.

  def validate_path(filepath: str, base: Path = WORKFLOWS_BASE) -> Path:
      """Validate file path is within allowed directory"""
      path = Path(filepath).resolve()
      try:
          path.relative_to(base.resolve())
          return path
      except ValueError:
          raise ToolError(f"Access denied: {filepath} is outside allowed directory")