execute_workflow

Executes programmatically built ComfyUI workflows to generate images from workflow dictionaries, returning the output or error messages for automated image generation.

Instructions

Execute an arbitrary workflow dict.

    Args:
        workflow: Workflow dict in ComfyUI API format
        output_node_id: Node ID that outputs the final image

    Returns the generated image or error message.
    Use this for programmatically built workflows.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`workflow`	Yes	Complete workflow dict
`output_node_id`	Yes	Node ID to get output from

Implementation Reference

src/comfy_mcp_server/tools/execution.py:83-108 (handler)

The main handler for the 'execute_workflow' tool. Validates input format and delegates execution to the internal _execute_workflow helper.

@mcp.tool()
def execute_workflow(
    workflow: dict = Field(description="Complete workflow dict"),
    output_node_id: str = Field(description="Node ID to get output from"),
    ctx: Context = None,
):
    """Execute an arbitrary workflow dict.

    Args:
        workflow: Workflow dict in ComfyUI API format
        output_node_id: Node ID that outputs the final image

    Returns the generated image or error message.
    Use this for programmatically built workflows.
    """
    # Check for UI format workflows
    if is_ui_format(workflow):
        return (
            "Error: Workflow is in UI format (has nodes/widgets_values). "
            "UI format uses positional arrays that can cause parameter misalignment errors. "
            "Please provide workflow in API format with explicit 'class_type' and 'inputs'."
        )

    if ctx:
        ctx.info("Executing custom workflow...")
    return _execute_workflow(workflow, output_node_id, ctx)

src/comfy_mcp_server/tools/execution.py:254-298 (helper)

Core implementation logic for executing the workflow: submits to ComfyUI API, polls for completion, handles output as image or URL.

def _execute_workflow(workflow: dict, output_node_id: str, ctx: Context | None):
    """Internal function to execute workflow and return result."""
    # Submit workflow
    status, resp_data = comfy_post("/prompt", {"prompt": workflow})

    if status != 200:
        error_msg = resp_data.get("error", f"status {status}")
        return f"Failed to submit workflow: {error_msg}"

    prompt_id = resp_data.get("prompt_id")
    if not prompt_id:
        node_errors = resp_data.get("node_errors", {})
        if node_errors:
            return f"Workflow validation failed:\n{json.dumps(node_errors, indent=2)}"
        return "Failed to get prompt_id from response"

    if ctx:
        ctx.info(f"Submitted: {prompt_id}")

    # Poll callback for progress logging
    def on_poll(attempt: int, max_attempts: int):
        if ctx and attempt % 5 == 0:
            ctx.info(f"Waiting... ({attempt}/{max_attempts})")

    # Poll for result
    image_data = poll_for_result(prompt_id, output_node_id, on_poll=on_poll)

    if image_data:
        if ctx:
            ctx.info("Image generated successfully")

        if settings.output_mode.lower() == "url":
            # Return URL instead of image data
            history = comfy_get(f"/history/{prompt_id}")
            if prompt_id in history:
                outputs = history[prompt_id].get("outputs", {})
                if output_node_id in outputs:
                    images = outputs[output_node_id].get("images", [])
                    if images:
                        url_values = urllib.parse.urlencode(images[0])
                        return get_file_url(settings.comfy_url_external, url_values)

        return Image(data=image_data, format="png")

    return "Failed to generate image. Use get_queue_status() and get_history() to debug."

src/comfy_mcp_server/tools/__init__.py:28-28 (registration)
Registration call for execution tools (including execute_workflow) within the register_all_tools function.
```
register_execution_tools(mcp)
```
src/comfy_mcp_server/__init__.py:92-92 (registration)
Top-level registration of all tools, which includes the execute_workflow tool via the execution tools module.
```
register_all_tools(mcp)
```

src/comfy_mcp_server/tools/execution.py:20-23 (helper)

Helper function used by execute_workflow to validate workflow format (API vs UI).

def is_ui_format(workflow: dict) -> bool:
    """Detect if workflow is in UI format (has nodes/links) vs API format (has class_type/inputs)."""
    return "nodes" in workflow or "version" in workflow

ComfyUI MCP Server