ComfyUI MCP Server

Instructions

Implementation Reference

• comfy_mcp/mcp/server.py:118-182 (handler)

  The core handler function for the 'write_workflow' tool. It is decorated with @mcp.tool for registration and implements the logic to write DSL workflows to disk as JSON or DSL files, including conversion using DSLParser and DslToJsonConverter.

  @mcp.tool
  async def write_workflow(
      ctx: Context,
      filepath: str,
      dsl: str,
      format: str = "json"
  ) -> dict:
      """Write a workflow to disk.
  
      Takes DSL content and writes it to disk. By default, converts to JSON format.
      Can optionally save as .dsl format directly.
  
      Args:
          filepath: Destination file path
          dsl: Workflow content in DSL format
          format: Output format ("json" or "dsl", default: "json")
  
      Returns:
          Status dict with path, size, and format info
  
      Examples:
          write_workflow("workflows/new_workflow.json", dsl_content)
          write_workflow("workflows/backup.dsl", dsl_content, format="dsl")
      """
      await ctx.info(f"Writing workflow to {filepath}")
  
      try:
          path = validate_path(filepath)
  
          # Check if file exists
          if path.exists():
              await ctx.info(f"⚠️  File {filepath} already exists, will overwrite")
  
          if format == "dsl":
              # Write DSL directly
              path.write_text(dsl)
              await ctx.info(f"✓ Wrote DSL to {filepath}")
  
          elif format == "json":
              # Convert DSL to JSON
              await ctx.info("Converting DSL to JSON...")
  
              parser = DSLParser()
              workflow_ast = parser.parse(dsl)
  
              converter = DslToJsonConverter()
              json_workflow = converter.convert(workflow_ast)
  
              json_content = json.dumps(json_workflow, indent=2)
              path.write_text(json_content)
  
              await ctx.info(f"✓ Wrote JSON to {filepath}")
  
          else:
              raise ToolError(f"Unsupported format: {format}. Use 'json' or 'dsl'")
  
          return {
              "status": "success",
              "path": str(path),
              "size": path.stat().st_size,
              "format": format
          }
  
      except Exception as e:
          raise ToolError(f"Error writing workflow: {e}")