execute_workflow
Run programmatically built ComfyUI workflows to generate images from workflow dictionaries and specified output nodes.
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 primary handler function for the 'execute_workflow' tool, registered via @mcp.tool(). It validates the workflow 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, ) -> Image | str: """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 execution logic: submits the workflow to ComfyUI via POST /prompt, polls for completion and result using poll_for_result, handles errors, optionally returns image URL or embedded Image object.def _execute_workflow(workflow: dict, output_node_id: str, ctx: Context | None) -> Image | str: """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:23-29 (registration)Central registration point for all tools, including execution tools via register_execution_tools(mcp). This function is called from src/comfy_mcp_server/__init__.py.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-93 (registration)Invocation of tool registration during server initialization.register_all_tools(mcp)