save_workflow
Save ComfyUI workflows as JSON files for execution or editing. Choose between API format for running workflows or UI format for visual editing in ComfyUI.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| workflow | Yes | Workflow to save | |
| name | No | Filename (without .json). If not provided, generates a funny name. | |
| format | No | Output format: 'api' (execution) or 'ui' (editor-compatible) | ui |
Implementation Reference
- 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) - 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."""