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
- 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) - 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) - 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