read_workflow
Convert ComfyUI workflow files between JSON and DSL formats for AI agents to manage workflows through natural language interactions.
Instructions
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")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| filepath | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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")