ComfyUI as a Production API: The Complete Guide

curl -fsS http://localhost:8188/system_stats
curl -fsS http://localhost:8188/queue
curl -fsS http://localhost:8188/object_info

curl -fsS -X POST http://localhost:8188/prompt \
  -H 'Content-Type: application/json' \
  -d @workflow_api.json

{
  "1": {
    "class_type": "CheckpointLoaderSimple",
    "_meta": { "title": "Load Checkpoint" },
    "inputs": {
      "ckpt_name": "sd_xl_base_1.0.safetensors"
    }
  },
  "2": {
    "class_type": "CLIPTextEncode",
    "_meta": { "title": "Positive Prompt" },
    "inputs": {
      "clip": ["1", 1],
      "text": "a photograph of an astronaut riding a horse"
    }
  },
  "3": {
    "class_type": "CLIPTextEncode",
    "_meta": { "title": "Negative Prompt" },
    "inputs": {
      "clip": ["1", 1],
      "text": ""
    }
  },
  "4": {
    "class_type": "EmptyLatentImage",
    "_meta": { "title": "Empty Latent Image" },
    "inputs": {
      "width": 1024,
      "height": 1024,
      "batch_size": 1
    }
  },
  "5": {
    "class_type": "VAEDecode",
    "_meta": { "title": "VAE Decode" },
    "inputs": {
      "samples": ["6", 0],
      "vae":     ["1", 2]
    }
  },
  "6": {
    "class_type": "KSampler",
    "_meta": { "title": "KSampler" },
    "inputs": {
      "model":        ["1", 0],
      "positive":     ["2", 0],
      "negative":     ["3", 0],
      "latent_image": ["4", 0],
      "seed": 42,
      "steps": 20,
      "cfg": 7.0,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1.0
    }
  },
  "7": {
    "class_type": "SaveImage",
    "_meta": { "title": "Save Image" },
    "inputs": {
      "images":        ["5", 0],
      "filename_prefix": "output"
    }
  }
}

import uuid
import time
import httpx  # pip install httpx (or use requests)

COMFYUI_URL = "http://localhost:8188"


def run_workflow(workflow: dict) -> dict:
    """Submit a workflow and block until complete. Returns the history entry."""
    client_id = str(uuid.uuid4())

    # Submit the workflow
    resp = httpx.post(f"{COMFYUI_URL}/prompt", json={
        "prompt": workflow,
        "client_id": client_id,
    })
    resp.raise_for_status()
    data = resp.json()

    if data.get("node_errors"):
        raise ValueError(f"Workflow has node errors: {data['node_errors']}")

    prompt_id = data["prompt_id"]
    print(f"Submitted: prompt_id={prompt_id}, queue_position={data['number']}")

    # Poll until complete (history endpoint returns {} while pending)
    while True:
        history = httpx.get(f"{COMFYUI_URL}/history/{prompt_id}").json()
        if prompt_id in history:
            return history[prompt_id]
        time.sleep(0.5)


def get_output_image(filename: str, subfolder: str = "") -> bytes:
    """Fetch a generated image by filename. Returns raw bytes."""
    resp = httpx.get(f"{COMFYUI_URL}/view", params={
        "filename": filename,
        "type": "output",
        "subfolder": subfolder,
    })
    resp.raise_for_status()
    return resp.content


def extract_images(history_entry: dict) -> list[dict]:
    """Pull image references from a completed history entry."""
    images = []
    for node_id, node_output in history_entry.get("outputs", {}).items():
        for img in node_output.get("images", []):
            images.append(img)
    return images


# Usage
if __name__ == "__main__":
    import json

    with open("workflow_api.json") as f:
        workflow = json.load(f)

    result = run_workflow(workflow)
    for img in extract_images(result):
        image_bytes = get_output_image(img["filename"], img.get("subfolder", ""))
        with open(img["filename"], "wb") as f:
            f.write(image_bytes)
        print(f"Saved: {img['filename']}")

const COMFYUI_URL = 'http://localhost:8188'

async function submitWorkflow(workflow: unknown) {
  const response = await fetch(`${COMFYUI_URL}/prompt`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      prompt: workflow,
      client_id: crypto.randomUUID(),
    }),
  })

  if (!response.ok) throw new Error(`Prompt submit failed: ${response.status}`)
  return response.json()
}

async function waitForResult(promptId: string) {
  for (;;) {
    const response = await fetch(`${COMFYUI_URL}/history/${promptId}`)
    const history = await response.json()
    if (history[promptId]) return history[promptId]
    await new Promise((resolve) => setTimeout(resolve, 500))
  }
}

import uuid
import json
import httpx
import websocket  # pip install websocket-client

COMFYUI_URL = "http://localhost:8188"
COMFYUI_WS = "ws://localhost:8188"


def run_with_progress(workflow: dict) -> list[dict]:
    """Submit a workflow and stream progress. Returns list of output image refs."""
    client_id = str(uuid.uuid4())
    ws = websocket.WebSocket()
    ws.connect(f"{COMFYUI_WS}/ws?clientId={client_id}")

    # Submit after connecting (so we don't miss any events)
    resp = httpx.post(f"{COMFYUI_URL}/prompt", json={
        "prompt": workflow,
        "client_id": client_id,
    })
    resp.raise_for_status()
    prompt_id = resp.json()["prompt_id"]
    print(f"Submitted: {prompt_id}")

    output_images = []

    # Listen for events until execution is complete
    while True:
        raw = ws.recv()
        if isinstance(raw, bytes):
            # Binary messages are image previews - skip or handle separately
            continue

        msg = json.loads(raw)
        msg_type = msg.get("type")
        data = msg.get("data", {})

        if msg_type == "status":
            queue_remaining = data.get("status", {}).get("exec_info", {}).get("queue_remaining", "?")
            print(f"Queue remaining: {queue_remaining}")

        elif msg_type == "executing":
            if data.get("prompt_id") != prompt_id:
                continue  # event for a different job
            if data.get("node") is None:
                # node=None signals that this prompt finished executing
                print("Execution complete.")
                break
            print(f"Executing node: {data['node']}")

        elif msg_type == "progress":
            if data.get("prompt_id") != prompt_id:
                continue
            print(f"  Step {data['value']}/{data['max']}")

        elif msg_type == "executed":
            if data.get("prompt_id") != prompt_id:
                continue
            images = data.get("output", {}).get("images", [])
            output_images.extend(images)
            for img in images:
                print(f"  Output ready: {img['filename']}")

    ws.close()
    return output_images

import json
import copy


def load_workflow(path: str) -> dict:
    with open(path) as f:
        return json.load(f)


def set_prompt(
    workflow: dict,
    positive_prompt: str,
    negative_prompt: str,
    seed: int | None = None,
) -> dict:
    """Return a new workflow with updated prompts and optional seed."""
    wf = copy.deepcopy(workflow)  # never mutate the original template

    for node_id, node in wf.items():
        class_type = node.get("class_type", "")
        title = node.get("_meta", {}).get("title", "").lower()

        if class_type == "CLIPTextEncode":
            if "neg" in title or "negative" in title:
                node["inputs"]["text"] = negative_prompt
            else:
                node["inputs"]["text"] = positive_prompt

        if seed is not None and class_type == "KSampler":
            node["inputs"]["seed"] = seed

    return wf


def set_checkpoint(workflow: dict, checkpoint_name: str) -> dict:
    """Swap the checkpoint in a workflow."""
    wf = copy.deepcopy(workflow)
    for node in wf.values():
        if node.get("class_type") == "CheckpointLoaderSimple":
            node["inputs"]["ckpt_name"] = checkpoint_name
    return wf


# Usage
if __name__ == "__main__":
    import random
    from comfyui_client import run_workflow, get_output_image, extract_images

    template = load_workflow("workflow_api.json")
    workflow = set_prompt(
        template,
        positive_prompt="a high-resolution photograph of a red sports car, studio lighting",
        negative_prompt="blurry, low quality, watermark",
        seed=random.randint(0, 2**32 - 1),
    )

    result = run_workflow(workflow)
    for img in extract_images(result):
        data = get_output_image(img["filename"])
        with open(f"out_{img['filename']}", "wb") as f:
            f.write(data)

server {
    listen 443 ssl;
    server_name api.yoursite.com;

    ssl_certificate     /etc/letsencrypt/live/api.yoursite.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/api.yoursite.com/privkey.pem;

    location / {
        # Reject requests without a valid API key
        if ($http_x_api_key != "your-secret-key-here") {
            return 401 "Unauthorized";
        }

        proxy_pass         http://127.0.0.1:8188;
        proxy_http_version 1.1;
        proxy_set_header   Upgrade $http_upgrade;
        proxy_set_header   Connection "upgrade";
        proxy_set_header   Host $host;
    }
}

import time, requests

def queue_with_retry(prompt: dict, api_url: str, api_key: str, max_retries=3):
    headers = {"X-API-Key": api_key, "Content-Type": "application/json"}

    # Submit
    resp = requests.post(f"{api_url}/prompt",
                         json={"prompt": prompt}, headers=headers)
    resp.raise_for_status()
    data = resp.json()

    # Reject if workflow has validation errors
    if data.get("node_errors"):
        raise ValueError(f"Workflow errors: {data['node_errors']}")

    prompt_id = data["prompt_id"]

    # Poll with exponential backoff
    for attempt in range(max_retries):
        time.sleep(2 ** attempt)  # 1s, 2s, 4s
        h = requests.get(f"{api_url}/history/{prompt_id}", headers=headers).json()
        if prompt_id in h:
            return h[prompt_id]   # job complete

    raise TimeoutError(f"Job {prompt_id} did not complete after {max_retries} retries")

def download_outputs(prompt_id: str, api_url: str, api_key: str) -> list[bytes]:
    headers = {"X-API-Key": api_key}
    history = requests.get(f"{api_url}/history/{prompt_id}", headers=headers).json()

    images = []
    for node_output in history[prompt_id]["outputs"].values():
        for img in node_output.get("images", []):
            params = {
                "filename":  img["filename"],
                "subfolder": img["subfolder"],
                "type":      img["type"],
            }
            r = requests.get(f"{api_url}/view", params=params, headers=headers)
            r.raise_for_status()
            images.append(r.content)   # raw JPEG or PNG bytes

    return images

import { Queue, Worker } from 'bullmq'
import Redis from 'ioredis'

const connection = new Redis(process.env.REDIS_URL || 'redis://localhost:6379')

// --- API side: enqueue and return job_id immediately ---
const imageQueue = new Queue('image-generation', { connection })

export async function enqueueJob(workflow: unknown) {
  const job = await imageQueue.add('generate', { workflow }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 5000 },
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 },
  })
  return job.id  // return this to the caller for status polling
}

// --- Worker side: one worker per GPU, calls local ComfyUI ---
const COMFYUI_URL = process.env.COMFYUI_URL || 'http://localhost:8188'

const worker = new Worker('image-generation', async (job) => {
  const { workflow } = job.data

  const res = await fetch(`${COMFYUI_URL}/prompt`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt: workflow, client_id: job.id }),
  })
  const { prompt_id } = await res.json()

  // Poll /history until complete (max 240s)
  for (let i = 0; i < 120; i++) {
    await new Promise(r => setTimeout(r, 2000))
    const hist = await fetch(`${COMFYUI_URL}/history/${prompt_id}`).then(r => r.json())
    if (hist[prompt_id]) {
      return await uploadOutputsToS3(hist[prompt_id].outputs, job.id)
    }
  }
  throw new Error('Timeout: job did not complete in 240s')
}, {
  connection,
  concurrency: 1,  // one job at a time per GPU worker
})

worker.on('failed', (job, err) => {
  console.error(JSON.stringify({ job_id: job?.id, error: err.message }))
})

import json, time

def log_job(prompt_id: str, status: str, duration_ms: int, model: str, error: str = None):
    """status: queued | running | done | failed"""
    print(json.dumps({
        "ts": time.time(),
        "prompt_id": prompt_id,
        "status": status,
        "duration_ms": duration_ms,
        "model": model,
        "error": error,
    }), flush=True)