ComfyUI MCP Server

Instructions

    This is a simplified interface for quick image generation.
    Requires COMFY_WORKFLOW_JSON_FILE, PROMPT_NODE_ID, and OUTPUT_NODE_ID
    to be configured.

    For more control, use run_workflow() or execute_workflow().

    Args:
        prompt: Text description of the image to generate
    

Implementation Reference

• src/comfy_mcp_server/tools/execution.py:110-141 (handler)

  The primary handler for the generate_image tool. Decorated with @mcp.tool() for automatic registration and schema definition. Loads the default workflow, injects the user prompt into the configured prompt node, and delegates execution to _execute_workflow.

  @mcp.tool()
  def generate_image(
      prompt: str = Field(description="Text prompt for image generation"),
      ctx: Context = None,
  ):
      """Generate an image using the default workflow.
  
      This is a simplified interface for quick image generation.
      Requires COMFY_WORKFLOW_JSON_FILE, PROMPT_NODE_ID, and OUTPUT_NODE_ID
      to be configured.
  
      For more control, use run_workflow() or execute_workflow().
  
      Args:
          prompt: Text description of the image to generate
      """
      if not settings.workflow_json_file:
          return "Error: COMFY_WORKFLOW_JSON_FILE not configured"
      if not settings.prompt_node_id:
          return "Error: PROMPT_NODE_ID not configured"
      if not settings.output_node_id:
          return "Error: OUTPUT_NODE_ID not configured"
  
      with open(settings.workflow_json_file) as f:
          workflow = json.load(f)
  
      workflow[settings.prompt_node_id]["inputs"]["text"] = prompt
  
      if ctx:
          ctx.info(f"Generating: {prompt[:50]}...")
  
      return _execute_workflow(workflow, settings.output_node_id, ctx)

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

  Key helper function called by generate_image to handle workflow submission to ComfyUI, polling for completion, image retrieval, and returning either Image object or URL based on settings.

  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:23-28 (registration)

  Top-level registration function that invokes register_execution_tools(mcp), which defines and registers the generate_image tool.

  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/settings.py:65-72 (helper)

  Pydantic settings fields required by generate_image for the default workflow path, prompt input node ID, and output node ID.

  workflow_json_file: str | None = Field(
      default=None,
      alias="comfy_workflow_json_file",
      description="Default workflow file for generate_image",
  )
  prompt_node_id: str | None = Field(default=None, description="Default prompt node ID")
  output_node_id: str | None = Field(default=None, description="Default output node ID")
  output_mode: str = Field(

• src/comfy_mcp_server/__init__.py:92-92 (registration)

  Initializes the MCP server and calls register_all_tools(mcp), starting the tool registration process for generate_image.

  register_all_tools(mcp)