ComfyUI API로 이미지 일괄 생성을 자동화하는 방법

안정적으로 동작하는 ComfyUI workflow가 있고 200개 상품마다 대표 이미지 4장을 만들어야 한다고 가정합니다. 매번 GUI에서 prompt와 seed를 바꾸는 방식은 확장하기 어렵습니다. 첫 /prompt POST는 API format이 아닌 파일 때문에 node_errors를 반환할 수 있습니다. 작업이 끝나도 /history에는 filename과 subfolder만 있으며 실제 파일은 /view로 받아야 합니다.

GUI에서 스크립트 생산으로 옮길 때는 올바른 API format 내보내기, 무작정 polling하지 않는 WebSocket 완료 대기, OOM을 막는 동시 실행 및 VRAM 제어가 필요합니다. 아래에서 최소 스크립트, 매개변수화, 큐 전략, 백엔드 체크리스트를 다룹니다.

로컬 ComfyUI Server API 시작점

CLI 실행과 기본 포트

ComfyUI는 기본적으로 127.0.0.1:8188에서 수신합니다. 같은 컴퓨터의 프로세스만 접근한다면 일반 실행으로 충분합니다. LAN의 다른 장치가 연결할 때만 수신 주소를 지정합니다.

# Local access only
python main.py --port 8188

# Allow LAN access; also configure a firewall, authentication, or a reverse proxy
python main.py --listen 0.0.0.0 --port 8188

터미널에 Starting server가 표시되면 http://127.0.0.1:8188을 엽니다. ComfyUI 프런트엔드가 나타나면 준비된 상태입니다.

VRAM이 부족하면 현재 버전이 지원하는 --lowvram, --novram, 마지막 수단인 느린 --cpu를 검토합니다. 여유가 크면 --highvram도 시험할 수 있습니다. GPU 용량만으로 고정하지 마세요. 모델, 정밀도, VAE, workflow가 최대 사용량을 바꿉니다. python main.py --help와 최신 공식 문서를 확인합니다.

주요 API Route

로컬 Route는 server.py에 정의됩니다. 스크립트에서 주로 쓰는 endpoint는 다음과 같습니다.

Route	용도	매개변수/응답
/prompt	workflow 검증 후 큐에 추가	POST {"prompt": workflow_dict, "client_id": "..."}. 성공 시 prompt_id, 실패 시 node_errors
/history/{prompt_id}	작업 기록과 출력 metadata 조회	GET. filename, subfolder, type이 있는 outputs
/view?filename=...&subfolder=...&type=...	출력 파일 다운로드	GET. 바이너리 데이터
/ws	WebSocket 실행 상태 수신	ws://127.0.0.1:8188/ws?clientId=...
/queue	현재 큐 조회	GET. 대기 작업
/interrupt	현재 실행 중단	POST. timeout 처리
/upload/image	입력 이미지 업로드	POST multipart/form-data
/object_info	사용 가능한 노드와 매개변수 조회	GET. 노드 존재 확인

Route는 버전에 따라 달라질 수 있습니다. 동작이 다르면 최신 문서나 server.py를 확인합니다.

WebSocket 메시지 유형

작업을 기다릴 때 다음 메시지를 봅니다.

• status: queue_remaining을 포함한 큐 상태
• execution_start: prompt_id와 실행 시작
• execution_cached: 캐시에서 재사용한 노드
• executing: 현재 노드. node is None이고 prompt_id가 같으면 완료
• progress: 현재 단계와 전체 단계
• executed: 노드 완료와 출력 metadata

완료 판단은 type == "executing"을 받고 data.node가 None이며 data.prompt_id가 제출 작업과 같은지 확인하는 방식입니다.

API Format Workflow 내보내기

Export Workflow (API) 단계

프런트엔드에서 저장한 workflow JSON은 API가 기대하는 형식과 다릅니다. 일반 workflow를 POST하면 node_errors가 날 수 있으므로 API format으로 내보냅니다.

다음 순서로 진행합니다.

1. ComfyUI 프런트엔드에서 정상 이미지를 생성하는 workflow를 불러옵니다
2. File -> Export Workflow (API)를 선택합니다. 버전에 따라 Save (API Format)일 수 있습니다
3. workflow_api.json 같은 .json으로 저장합니다
4. "3", "6" 같은 숫자 node ID와 각 노드의 class_type, inputs를 확인합니다

메뉴 이름은 바뀔 수 있으므로 사용 중인 버전의 동등한 API 내보내기 기능을 선택합니다.

API format과 Save format 차이

Save format에는 프런트엔드 레이아웃 정보가 있습니다. API format은 실행에 필요한 정보만 유지합니다.

형식	포함 정보	용도
Save format	노드 위치, 색상, 그룹, 크기, 시각적 연결	프런트엔드에서 다시 편집
API format	숫자 node ID, class_type, inputs, 선택적 _meta	스크립트나 API로 실행

Save format을 /prompt에 바로 보내면 node_errors 또는 error가 반환될 수 있습니다. 프런트엔드에서 불러와 다시 내보냅니다.

Node ID 관리

매개변수화하려면 prompt, seed, 크기, 출력 노드 ID가 필요합니다. 다음 항목을 찾습니다.

• Prompt node: CLIPTextEncode. 단순 예시에서는 "6"인 경우가 많음
• Seed node: KSampler. 단순 예시에서는 "3"인 경우가 많음
• Width/height node: EmptyLatentImage 또는 workflow별 노드의 inputs
• Output node: SaveImage 또는 SaveImageWebsocket

프런트엔드의 Note와 Group으로 역할을 기록할 수 있지만 스크립트는 내보낸 JSON을 확인해야 합니다. ID는 특정 그래프에 속하며 다시 내보내면 달라질 수 있습니다.

최소 스크립트: 제출, 대기, 다운로드

API 흐름은 workflow를 /prompt에 제출하고 완료를 기다린 다음 /history에서 metadata를 읽어 /view로 다운로드하는 세 단계입니다.

대기하지 않는 HTTP 제출

가장 작은 클라이언트는 /prompt에 POST하고 기다리지 않습니다. 별도 worker가 작업을 조회하는 구성에 적합합니다.

import json
import requests

# Load an API-format workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# Build the request payload
payload = {
    "prompt": workflow,
    "client_id": "my-script-client"  # Optional; associates the job with WebSocket events
}

# Add the job to the queue
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)

if response.status_code == 200:
    result = response.json()
    prompt_id = result["prompt_id"]
    print(f"Submitted: {prompt_id}")
else:
    error = response.json()
    print(f"Submission failed: {error}")
    # node_errors contains node-level validation details

성공은 {"prompt_id": "...", "number": ...}, 실패는 {"error": {...}, "node_errors": {...}}를 반환합니다. 제출은 큐에 추가할 뿐 완료를 기다리지 않습니다.

WebSocket으로 완료 대기

WebSocket 이벤트는 잦은 polling을 줄입니다. 먼저 연결하고 작업을 제출한 뒤 마지막 executing 이벤트를 기다립니다.

import json
import uuid
import requests
import websocket

# Load the workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# Generate a client ID
client_id = str(uuid.uuid4())

# Connect to WebSocket
ws = websocket.create_connection(f"ws://127.0.0.1:8188/ws?clientId={client_id}")

# Submit the job
payload = {"prompt": workflow, "client_id": client_id}
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
prompt_id = response.json()["prompt_id"]

# Wait for completion
while True:
    message = ws.recv()
    data = json.loads(message)

    if data["type"] == "executing":
        # node is None with the same prompt_id when the job is complete
        if data["data"]["node"] is None and data["data"]["prompt_id"] == prompt_id:
            print("Job complete")
            break

ws.close()

300초 같은 최대 대기 시간을 둡니다. /interrupt는 현재 실행을 중단하므로 주의해서 사용합니다.

History와 View로 이미지 다운로드

완료 후 /history/{prompt_id}에서 출력 metadata를 받고 실제 파일은 /view로 다운로드합니다.

import requests

# Retrieve job history
history_url = f"http://127.0.0.1:8188/history/{prompt_id}"
history = requests.get(history_url).json()

# Traverse output nodes
outputs = history[prompt_id]["outputs"]
for node_id, node_output in outputs.items():
    if "images" in node_output:
        for image in node_output["images"]:
            filename = image["filename"]
            subfolder = image.get("subfolder", "")
            type = image.get("type", "output")

            # Build the download URL
            view_url = f"http://127.0.0.1:8188/view?filename={filename}&subfolder={subfolder}&type={type}"

            # Download the binary image
            img_data = requests.get(view_url).content
            with open(f"output_{filename}", "wb") as f:
                f.write(img_data)
            print(f"Saved: output_{filename}")

outputs 구조는 {node_id: {"images": [{"filename": "...", "subfolder": "...", "type": "..."}]}}입니다. History는 metadata, /view는 바이너리를 반환합니다.

일괄 작업 매개변수화

안정적인 workflow를 일괄 템플릿으로 사용합니다. 각 반복에서는 prompt, seed, 크기 등 선택한 입력만 바꿉니다.

Prompt, Seed, 크기 매개변수화

노드의 inputs 값을 바꿉니다.

import json
import random

# Load the workflow
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# Change the prompt; use the IDs from your own workflow
workflow["6"]["inputs"]["text"] = "a beautiful landscape, sunset, mountains"

# Generate a random seed
workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

# Change the dimensions
workflow["5"]["inputs"]["width"] = 1024
workflow["5"]["inputs"]["height"] = 768

# Submit the job...

실제 내보낸 workflow에서 "6", "3", "5"를 확인합니다. 이후 루프에서 제출 전 prompt와 seed를 변경할 수 있습니다.

prompts = [
    "product photo, white background",
    "product photo, outdoor scene",
    "product photo, studio lighting"
]

for i, prompt_text in enumerate(prompts):
    workflow["6"]["inputs"]["text"] = prompt_text
    workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

    # Submit the job
    payload = {"prompt": workflow, "client_id": client_id}
    response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
    prompt_id = response.json()["prompt_id"]

    # Wait over WebSocket...
    # Download outputs...

공식 예시도 KSampler.seed와 CLIPTextEncode.text를 바꾸므로 매개변수 위치를 찾는 출발점으로 사용할 수 있습니다.

일괄 큐 전략

200장을 만들 때 workflow의 batch size를 키우거나 많은 prompt를 제출할 수 있습니다. 안전한 선택은 모델과 측정한 VRAM 여유에 달려 있습니다.

전략	특징	적합한 경우
제출당 한 장	작업별 VRAM을 제어하기 쉬움	여유가 적거나 큰 모델, 복잡한 workflow
Batch size	한 prompt에서 여러 장, 일반적으로 높은 VRAM 피크	측정된 여유가 크고 작은 모델, 단순 workflow
동시 요청	활성 수를 제한하며 여러 prompt 제출	용량을 측정했고 worker를 제어하는 환경

20개 prompt를 동시에 보내면 큐가 쌓이고 OOM 또는 서비스 중단이 발생할 수 있습니다.

보수적인 시작은 순차 제출, WebSocket 완료, VRAM 모니터링입니다. 이전 작업이 끝난 뒤 다음 작업을 보냅니다. 메모리가 부족하면 부하를 줄이거나 현재 옵션을 검증합니다.

동시 실행 제어 예시

여러 GPU나 격리된 cloud worker에서는 semaphore로 활성 작업 수를 제한할 수 있습니다.

import copy
import threading

# Allow at most two active jobs
semaphore = threading.Semaphore(2)

def submit_and_wait(prompt_text, seed):
    with semaphore:
        # Give each task its own copy instead of mutating shared state
        job_workflow = copy.deepcopy(workflow)
        job_workflow["6"]["inputs"]["text"] = prompt_text
        job_workflow["3"]["inputs"]["seed"] = seed

        # Submit and wait...
        # WebSocket loop...

    # The permit is returned for the next task

# Submit the batch
threads = []
for i in range(50):
    t = threading.Thread(target=submit_and_wait, args=(prompts[i], seeds[i]))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

동시 실행 1부터 시작합니다. 실제 workflow에서 메모리 피크, 지연, 실패를 측정한 뒤 늘립니다. 정밀도, VAE, 후처리, worker 격리가 한도를 바꿉니다.

VRAM 모니터링

일괄 실행 중 시스템 통계를 조회하고 선택한 임계값을 넘으면 새 제출을 멈춥니다.

import time

def check_vram(threshold=0.8):
    stats = requests.get("http://127.0.0.1:8188/system_stats").json()
    vram_used = stats["system_stats"]["devices"][0]["vram_used"]
    vram_total = stats["system_stats"]["devices"][0]["vram_total"]
    return (vram_used / vram_total) > threshold

# Check memory before each submission
for prompt_text in prompts:
    while check_vram(0.85):
        print("VRAM pressure is high; waiting 30 seconds...")
        time.sleep(30)

    # Submit the next job...
    # Wait over WebSocket...

메모리 피크는 KSampler 실행 중 자주 발생하고 완료 후 낮아질 수 있습니다. 10~30초 간격으로 확인합니다.

일괄 출력 보관

출력에는 예측 가능한 구조가 필요합니다. {prompt_id}_{seed}_{timestamp}.png는 작업 ID, seed, 생성 시간을 보존합니다.

최소한 다음을 기록합니다.

• prompt_id: ComfyUI 작업 ID
• seed: 난수 seed
• prompt_text: 사용한 prompt
• width/height: 출력 크기
• timestamp: 생성 시간
• business_id: 상품 SKU나 주문 번호 같은 업무 ID

outputs/20260624/batch_001/처럼 날짜나 batch별로 정리합니다. 파일만으로 부족하면 SQLite나 PostgreSQL로 metadata와 경로를 연결합니다.

백엔드 엔지니어링 체크리스트

백엔드 wrapper에는 request ID, timeout, 큐 제한, 메모리 모니터링, 명시적 오류 처리가 필요합니다. 동작하는 스크립트만으로 안정적인 서비스가 되지는 않습니다.

Request ID와 멱등성

UUID나 주문 번호 같은 고유 request_id를 만들고 ComfyUI prompt_id와 연결합니다. 완료된 ID가 다시 오면 재생성하지 않고 기존 결과를 반환합니다.

import uuid

# Business request ID
request_id = str(uuid.uuid4())

# Store the mapping in a database or cache
request_prompt_map[request_id] = prompt_id

# Return an existing result for duplicate requests
if request_id in completed_requests:
    return get_cached_result(request_id)

ComfyUI가 만든 prompt_id는 애플리케이션의 request_id와 다릅니다. 매핑을 영구 저장합니다.

Timeout과 취소

300초 같은 최대 실행 시간을 정합니다. 초과하면 /interrupt로 현재 실행을 중단합니다.

import time

timeout = 300  # Five minutes
start_time = time.time()

# Wait for WebSocket messages...
while True:
    elapsed = time.time() - start_time
    if elapsed > timeout:
        # Interrupt the current execution
        requests.post("http://127.0.0.1:8188/interrupt")
        print("Job timed out and was interrupted")
        break

    # Process normal messages...

WebSocket이 끊기거나 일정 시간 응답이 없으면 재연결하거나 history를 조회합니다. 중단 후 /queue를 확인하고 대기 작업 처리 방식을 정합니다.

큐 제한과 VRAM 모니터링

대기 작업 5개처럼 명시적 상한을 두고 초과 요청은 거부하거나 미룹니다. 활성 worker 수도 제한합니다.

/system_stats로 VRAM을 조회합니다.

stats = requests.get("http://127.0.0.1:8188/system_stats").json()
vram_used = stats["system_stats"]["devices"][0]["vram_used"]
vram_total = stats["system_stats"]["devices"][0]["vram_total"]
vram_percent = vram_used / vram_total

if vram_percent > 0.8:
    print("VRAM pressure is high; rejecting a new request")

부하가 높으면 새 작업을 거부하거나 큐가 줄 때까지 기다립니다. 반복 전에 batch size와 workflow 복잡도를 낮춥니다.

오류 분류와 재시도

오류별로 다른 대응이 필요합니다.

오류	가능한 원인	처리
node_errors	모델/노드 없음, 잘못된 매개변수, 입력 누락	workflow와 환경 수정, 재시도하지 않음
OOM	VRAM 부족	batch/부하를 낮추고 검증한 옵션 변경, 같은 조건 재시도 금지
WebSocket 끊김	네트워크 문제	다시 연결하고 /history/{prompt_id} 조회
작업 timeout	느린 모델 로드 또는 복잡한 workflow	timeout 연장 또는 단순화, 한 번만 재시도
서비스 중단	메모리 고갈 또는 GPU 오류	로그 확인, 재시작 후 신중히 재시도

node_errors는 없는 노드 클래스와 맞지 않는 입력 형식을 알려 줍니다. workflow, custom nodes, 모델을 먼저 고칩니다. OOM도 부하 변경이 필요합니다.

Cloud API와 로컬 API 비교

Comfy Cloud는 hosted workflow 실행을 제공하지만 인증, 작업 상태, WebSocket 주소, 동시 실행이 다릅니다. API format은 재사용할 수 있으나 endpoint는 최신 참조를 따라야 합니다.

Route 차이

주요 차이는 다음과 같습니다.

기능	로컬 API	Cloud API
작업 제출	/prompt	/api/prompt
상태/결과	/history/{prompt_id}	/api/job/{prompt_id}/status와 /api/jobs/{job_id}
출력 다운로드	/view	/api/view
WebSocket	/ws?clientId=...	/ws?clientId=...&token=...

Cloud에는 X-API-Key와 구독이 필요합니다. 로컬 API는 기본적으로 127.0.0.1에서만 수신하지만 --listen, proxy, port forwarding을 쓰면 인증과 접근 제어를 직접 추가해야 합니다. Cloud API는 experimental이며 /api/history_v2/{prompt_id}는 deprecated되어 /api/jobs/{job_id}가 권장됩니다.

동시 실행과 구독 제한

Cloud 동시 실행 수는 요금제에 따라 달라지고 초과 작업은 큐에서 기다립니다. 출력은 cloud storage에 있으며 /api/view는 임시 서명 URL을 반환합니다.

요금제, 제한, 실행 시간, 가격은 바뀔 수 있어 숫자를 고정하지 않습니다. Cloud는 로컬 GPU 관리를 줄이고, 로컬에서는 큐, VRAM, 보안, 안정성을 직접 관리합니다.

다음 단계

시리즈 내 문서

이 페이지는 동작하는 workflow에서 프로그래밍 가능한 일괄 생산으로 이어지는 단계입니다.

• ComfyUI workflow 재사용: 가져오기, 누락 노드, 모델 경로
• ComfyUI 저 VRAM 최적화: 메모리 옵션, batch, OOM, Tiled VAE
• ComfyUI 영상 생성: 이미지 일괄 처리와 영상 workflow의 경계
• ComfyUI 유지보수: 누락 노드, 시작 실패, 버전 충돌

관련 주제

더 넓은 자동화 패턴은 다음 내용을 참고합니다.

• n8n AI workflow 자동화: ComfyUI와 여러 도구 연결
• Ollama API 실전: 프로그램 호출, 큐, 구조화 출력
• LLM 구조화 출력: 신뢰할 수 있는 데이터 추출과 API 연동

결론

ComfyUI를 스크립트 생산으로 옮기는 순서는 API workflow 내보내기, WebSocket 대기, /history와 /view 다운로드, 선택 입력 매개변수화, request ID, timeout, 큐 제한, VRAM 모니터링 추가입니다.

먼저 workflow 하나와 이미지 한 장을 실행합니다. 이후 매개변수를 바꾼 10장을 만들고 메모리와 큐를 관찰합니다. 작은 batch가 안정된 뒤 멱등성, 취소, 모니터링, 결과 기록을 추가합니다.