ComfyUI MCP Server

Instructions

    Args:
        workflow: Workflow dict to save (in API format)
        name: Filename (with or without .json extension).
              If not provided, generates a random funny name like 'cosmic-penguin'.
        format: Output format:
            - 'api': Raw API format for execution (flat dict)
            - 'ui': Litegraph format for ComfyUI editor (default)

    Returns path to saved file or error message.
    

Implementation Reference

• src/comfy_mcp_server/tools/workflow.py:267-323 (handler)

  The core handler function for the 'save_workflow' tool. This @mcp.tool()-decorated function implements the logic to save workflows in either 'api' or 'ui' format to the configured directory, handling name generation, format conversion via api_to_ui_workflow, and file I/O.

  def save_workflow(
      workflow: dict = Field(description="Workflow to save"),
      name: str = Field(
          default=None,
          description="Filename (without .json). If not provided, generates a funny name.",
      ),
      format: str = Field(
          default="ui",
          description="Output format: 'api' (execution) or 'ui' (editor-compatible)",
      ),
      ctx: Context = None,
  ) -> str:
      """Save a workflow to the workflows directory.
  
      Args:
          workflow: Workflow dict to save (in API format)
          name: Filename (with or without .json extension).
                If not provided, generates a random funny name like 'cosmic-penguin'.
          format: Output format:
              - 'api': Raw API format for execution (flat dict)
              - 'ui': Litegraph format for ComfyUI editor (default)
  
      Returns path to saved file or error message.
      """
      # Determine target directory based on format
      if format == "ui":
          target_dir = settings.workflows_ui_dir or settings.workflows_dir
      else:
          target_dir = settings.workflows_dir
  
      if not target_dir:
          return "Error: COMFY_WORKFLOWS_DIR not configured"
  
      # Auto-generate funny name if not provided
      if not name:
          name = generate_slug(2)
  
      if not name.endswith(".json"):
          name = f"{name}.json"
  
      path = Path(target_dir) / name
  
      if ctx:
          ctx.info(f"Saving to: {path} (format: {format})")
  
      try:
          path.parent.mkdir(parents=True, exist_ok=True)
  
          # Convert to UI format if requested
          output_data = api_to_ui_workflow(workflow) if format == "ui" else workflow
  
          with open(path, "w") as f:
              json.dump(output_data, f, indent=2)
          return f"Saved: {path}"
      except Exception as e:
          return f"Error: {e}"

• src/comfy_mcp_server/tools/__init__.py:23-29 (registration)

  The register_all_tools function calls register_workflow_tools(mcp), which registers the save_workflow tool among others.

  def register_all_tools(mcp):
      """Register all tools with the MCP server."""
      register_system_tools(mcp)
      register_discovery_tools(mcp)
      register_workflow_tools(mcp)
      register_execution_tools(mcp)

• src/comfy_mcp_server/__init__.py:92-92 (registration)

  Invocation of register_all_tools(mcp) in the main server initialization, which indirectly registers the save_workflow tool.

  register_all_tools(mcp)

• src/comfy_mcp_server/tools/workflow.py:50-182 (helper)

  Helper function used by save_workflow to convert API workflows to UI/Litegraph format when format='ui'. Performs node ID mapping, layout computation, link creation, and serialisation.

  def api_to_ui_workflow(api_workflow: dict) -> dict:
      """Convert API format workflow to SerialisableGraph format (version 1).
  
      API format: { "node_id": {"class_type": "...", "inputs": {...}}, ... }
      UI format: { "version": 1, "nodes": [...], "links": [...], ... }
  
      Based on: ComfyUI_frontend/src/lib/litegraph/src/types/serialisation.ts
  
      Args:
          api_workflow: Workflow in API format
  
      Returns:
          Workflow in SerialisableGraph format compatible with ComfyUI editor
      """
      ui_nodes: list[UINode] = []
      ui_links: list[SerialisableLLink] = []
  
      # Map string node IDs to integer IDs
      node_id_map: dict[str, int] = {}
      for idx, node_id in enumerate(api_workflow.keys(), start=1):
          node_id_map[node_id] = idx
  
      # Track link ID
      link_id = 0
  
      # Compute node sizes for layout
      node_sizes = {
          node_id: get_node_size(data.get("class_type", "")) for node_id, data in api_workflow.items()
      }
  
      # Graph-aware layout using topological layers
      positions = compute_workflow_layout(api_workflow, node_sizes)
  
      for idx, (node_id, node_data) in enumerate(api_workflow.items()):
          int_id = node_id_map[node_id]
          class_type = node_data.get("class_type", "Unknown")
          inputs = node_data.get("inputs", {})
  
          # Get position from graph layout
          pos_x, pos_y = positions.get(node_id, (100, 100))
          size = node_sizes[node_id]
  
          # Separate connection inputs from widget values
          node_inputs: list[UINodeSlot] = []
          node_outputs: list[UINodeSlot] = []
          widgets_values: list = []
  
          for input_name, value in inputs.items():
              if isinstance(value, list) and len(value) == 2:
                  # This is a connection: [source_node_id, output_index]
                  source_node_str = str(value[0])
                  source_slot = value[1]
  
                  if source_node_str in node_id_map:
                      link_id += 1
                      source_int_id = node_id_map[source_node_str]
  
                      # Add input slot
                      node_inputs.append(
                          UINodeSlot(
                              name=input_name,
                              type="*",  # Wildcard type
                              link=link_id,
                              slot_index=len(node_inputs),
                          )
                      )
  
                      # Add link (object format for SerialisableGraph)
                      ui_links.append(
                          SerialisableLLink(
                              id=link_id,
                              origin_id=source_int_id,
                              origin_slot=source_slot,
                              target_id=int_id,
                              target_slot=len(node_inputs) - 1,
                              type="*",
                          )
                      )
              else:
                  # This is a widget value
                  widgets_values.append(value)
  
          # Add a default output for nodes that might be sources
          # (detected by checking if other nodes reference them)
          has_outgoing = any(
              isinstance(inp, list) and len(inp) == 2 and str(inp[0]) == node_id
              for n in api_workflow.values()
              for inp in n.get("inputs", {}).values()
          )
          if has_outgoing or "SaveImage" not in class_type:
              node_outputs.append(UINodeSlot(name="output", type="*", links=[], slot_index=0))
  
          # Create UI node
          ui_node = UINode(
              id=int_id,
              type=class_type,
              pos=(pos_x, pos_y),
              size=size,
              flags=UINodeFlags(),
              order=idx,
              mode=0,
              inputs=node_inputs,
              outputs=node_outputs,
              properties={"Node name for S&R": class_type},
              widgets_values=widgets_values if widgets_values else None,
          )
          ui_nodes.append(ui_node)
  
      # Update output links on source nodes
      for link in ui_links:
          for node in ui_nodes:
              if node.id == link.origin_id and node.outputs:
                  if node.outputs[link.origin_slot].links is None:
                      node.outputs[link.origin_slot].links = []
                  node.outputs[link.origin_slot].links.append(link.id)
  
      # Build UI workflow (SerialisableGraph version 1 format)
      max_node_id = max(node_id_map.values()) if node_id_map else 0
      ui_workflow = UIWorkflow(
          version=1,
          state=UIWorkflowState(
              lastNodeId=max_node_id,
              lastLinkId=link_id,
              lastGroupId=0,
              lastRerouteId=0,
          ),
          nodes=ui_nodes,
          links=ui_links,
          groups=[],
          extra=UIWorkflowExtra(),
      )
  
      return ui_workflow.to_dict()

• src/comfy_mcp_server/tools/workflow.py:217-218 (registration)

  The register_workflow_tools function defines and registers the save_workflow tool using @mcp.tool() decorator.

  def register_workflow_tools(mcp):
      """Register workflow management tools."""