ComfyUI with API

ComfyUI allows to execute workflows from the command line using API workflows

Enable API workflows

You have to activate dev mode. In version 0.3.59 you have to click on comfyUI icon at top left, settings. Search for "dev mode" and activate it

Export a workflow in API version

Click on comfyUI icon at top left, File, Export (API)

Script for executing the workflow

import json
import os
import random
import argparse
import uuid
import time
from urllib import request, parse, error
from PIL import Image

# --- Argument parsing ---
parser = argparse.ArgumentParser(description="Run ComfyUI workflow via API")
parser.add_argument("--source-image", required=True, help="Absolute path to the source image")
parser.add_argument("--prompt", required=True, help="Positive text prompt for generation")
args = parser.parse_args()

# --- Check source image path validity ---
if not args.source_image.startswith("/"):
    print("Error: --source-image must be an absolute path (it should start with '/')")
    exit(1)

if not os.path.isfile(args.source_image):
    print(f"Error: Source image not found at '{args.source_image}'")
    exit(1)

# --- Load the workflow from file ---
prompt_workflow = json.load(open("workflow.json"))

# --- Node references (adjust if your JSON IDs differ) ---
loadImageNode = prompt_workflow["14"]
upscaleImageNode = prompt_workflow["71"]
wanImageToVideoNode = prompt_workflow["10"]
promptPositiveNode = prompt_workflow["13"]
KSamplerAdvancedNode = prompt_workflow["3"]

# --- Set workflow values ---
loadImageNode["inputs"]["image"] = args.source_image

with Image.open(args.source_image) as img:
    sourceImageWidth, sourceImageHeight = img.size

upscaleImageNode["inputs"]["width"] = sourceImageWidth
upscaleImageNode["inputs"]["height"] = sourceImageHeight
wanImageToVideoNode["inputs"]["width"] = sourceImageWidth
wanImageToVideoNode["inputs"]["height"] = sourceImageHeight

promptPositiveNode["inputs"]["text"] = args.prompt
KSamplerAdvancedNode["inputs"]["noise_seed"] = random.randint(1, 18446744073709551614)

# --- Function: queue the workflow ---
def queue_prompt(prompt_workflow):
    payload = {
        "prompt": prompt_workflow,
        "client_id": str(uuid.uuid4())
    }

    data = json.dumps(payload).encode("utf-8")
    req = request.Request(
        "http://127.0.0.1:8188/prompt",
        data=data,
        headers={"Content-Type": "application/json"}
    )

    try:
        with request.urlopen(req) as response:
            res = json.loads(response.read().decode("utf-8"))
            print("Workflow queued:", res)
            return payload["client_id"]
    except error.HTTPError as e:
        print("HTTP error:", e.code, e.reason)
        print(e.read().decode())
        exit(1)
    except Exception as e:
        print("Unexpected error:", e)
        exit(1)

# --- Function: poll the queue every 5 seconds ---
def poll_queue_status():
    queue_url = "http://127.0.0.1:8188/queue"
    print("\nPolling queue status every 5 seconds...")
    printed_header="false"
    while True:
        try:
            with request.urlopen(queue_url) as res:
                data = json.loads(res.read().decode("utf-8"))
                queue_running = data.get("queue_running", [])
                queue_pending = data.get("queue_pending", [])
                if printed_header=="false":
                    print(f"Running: {len(queue_running)} | Pending: {len(queue_pending)} ")
                    printed_header="true"
                else: 
                    print(".")
                if not queue_running and not queue_pending:
                    print("Processing complete!")
                    break
        except Exception as e:
            print("Error polling queue:", e)

        time.sleep(5)

# --- Run ---
client_id = queue_prompt(prompt_workflow)
poll_queue_status()