write_workflow
Save ComfyUI workflow content to disk by converting DSL to JSON format or storing as DSL directly for persistent storage and reuse.
Instructions
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")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| filepath | Yes | ||
| dsl | Yes | ||
| format | No | json |
Implementation Reference
- comfy_mcp/mcp/server.py:118-182 (handler)The primary handler function for the 'write_workflow' tool. It is registered via the @mcp.tool decorator. Handles writing DSL workflows to disk as either JSON or DSL format, including conversion logic 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}")