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 API workflow

ComfyUI server needs to be running in order to send CLI requests.

Command to execute the script below:

python3 comfy_api.py --workflow comfyui_workflow_api.json --source-image /absolute/path/to/image.jpg --prompt "prompt to be sent to comfyui"

Script

import os
import sys
import argparse
import time
from urllib import request, parse, error
import json
import random
import uuid
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")
parser.add_argument("--workflow", required=True, help="path of the API workflow")
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(args.workflow))

# --- 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"
    spinner_frames = ["|", "/", "-", "\\"]
    spinner_index = 0
    spinner_trail = ""
    last_check_time = time.time()

    print("\n📡 Polling queue status... (updates every 5s)")
    while True:
        # --- Spinner animation (updates every ~0.1s) ---
        sys.stdout.write("\r" + spinner_trail + spinner_frames[spinner_index])
        sys.stdout.flush()
        spinner_index = (spinner_index + 1) % len(spinner_frames)
        time.sleep(0.1)

        # --- Every 5 seconds, check queue ---
        if time.time() - last_check_time >= 5:
            last_check_time = time.time()
            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 not queue_running and not queue_pending:
                        sys.stdout.write("\r" + spinner_trail + "✅ Done!\n")
                        sys.stdout.flush()
                        break
                    else:
                        # Replace last spinner with X and start new spinner next to it
                        spinner_trail += "X"
            except Exception as e:
                sys.stdout.write(f"\r⚠️ Error polling queue: {e}\n")
                sys.stdout.flush()
                break

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