API 예제 - ComfyUI

이 페이지에서는 간단한 HTTP 제출부터 실시간 이미지 출력과의 완전한 WebSocket 통합까지, ComfyUI 서버 API와 상호작용하는 세 가지 방법을 보여줍니다. 모든 예제는 설명을 위해 기본 SD1.5 워크플로우를 사용합니다. API를 사용하기 전에 API 형식으로 워크플로우를 내보내야 합니다.

이 예제들은 표준 라이브러리와 websocket-client 패키지를 사용한 파이썬 코드입니다(pip install websocket-client). 언어에 관계없이 기본 API 프로토콜은 동일합니다—TypeScript 및 curl equivalent에 대한 클라우드 API 참조를 확인하세요.

방법 1: 제출 후 잊어버리기 (HTTP 전용)

소스: basic_api_example.py 가장 간단한 방식: 워크플로우를 제출하고 결과를 기다리지 않습니다. 나중에 출력물을 확인하는 일회성 작업에 유용합니다.

"""basic_api_example.py — HTTP만을 통해 워크플로우를 제출합니다."""

import json
from urllib import request

SERVER_ADDRESS = "127.0.0.1:8188"


def queue_prompt(prompt):
    p = {"prompt": prompt}
    data = json.dumps(p).encode("utf-8")
    req = request.Request(
        f"http://{SERVER_ADDRESS}/prompt", data=data
    )
    request.urlopen(req)


if __name__ == "__main__":
    # API 형식으로 내보낸 워크플로우를 로드합니다
    prompt_text = """{
        "3": {
            "class_type": "KSampler",
            "inputs": {
                "cfg": 8, "denoise": 1,
                "latent_image": ["5", 0],
                "model": ["4", 0],
                "negative": ["7", 0],
                "positive": ["6", 0],
                "sampler_name": "euler",
                "scheduler": "normal",
                "seed": 8566257, "steps": 20
            }
        },
        "4": {
            "class_type": "CheckpointLoaderSimple",
            "inputs": {"ckpt_name": "v1-5-pruned-emaonly.safetensors"}
        },
        "5": {
            "class_type": "EmptyLatentImage",
            "inputs": {"batch_size": 1, "height": 512, "width": 512}
        },
        "6": {
            "class_type": "CLIPTextEncode",
            "inputs": {"clip": ["4", 1], "text": "masterpiece best quality man"}
        },
        "7": {
            "class_type": "CLIPTextEncode",
            "inputs": {"clip": ["4", 1], "text": "bad hands"}
        },
        "8": {
            "class_type": "VAEDecode",
            "inputs": {"samples": ["3", 0], "vae": ["4", 2]}
        },
        "9": {
            "class_type": "SaveImage",
            "inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]}
        }
    }"""

    prompt = json.loads(prompt_text)
    # 제출 전 입력값 수정
    prompt["3"]["inputs"]["seed"] = 5
    prompt["6"]["inputs"]["text"] = "masterpiece best quality man"

    queue_prompt(prompt)
    print("프롬프트가 성공적으로 제출되었습니다.")

이 방법은 SaveImage 노드를 사용하며, 이 노드는 이미지를 서버의 디스크에 저장합니다. 이를 가져오려면 /view?filename=... 호출을 추가로 실행해야 합니다.

방법 2: WebSocket + 히스토리 (완료 모니터링)

소스: websockets_api_example.py WebSocket을 사용해 실행이 완료될 때까지 기다린 후, /history 엔드포인트를 통해 출력물을 가져옵니다. 대부분의 사용 사례에 권장되는 방식입니다.

"""websockets_api_example.py: WebSocket을 통해 실행 상태를 모니터링하고, /history를 통해 다운로드합니다."""

import websocket  # pip install websocket-client
import uuid
import json
import urllib.request
import urllib.parse

SERVER_ADDRESS = "127.0.0.1:8188"
client_id = str(uuid.uuid4())


def queue_prompt(prompt, prompt_id):
    p = {"prompt": prompt, "client_id": client_id, "prompt_id": prompt_id}
    data = json.dumps(p).encode("utf-8")
    req = urllib.request.Request(
        f"http://{SERVER_ADDRESS}/prompt", data=data
    )
    urllib.request.urlopen(req)


def get_image(filename, subfolder, folder_type):
    params = urllib.parse.urlencode({
        "filename": filename,
        "subfolder": subfolder,
        "type": folder_type,
    })
    with urllib.request.urlopen(
        f"http://{SERVER_ADDRESS}/view?{params}"
    ) as response:
        return response.read()


def get_history(prompt_id):
    with urllib.request.urlopen(
        f"http://{SERVER_ADDRESS}/history/{prompt_id}"
    ) as response:
        return json.loads(response.read())


def get_images(ws, prompt):
    prompt_id = str(uuid.uuid4())
    queue_prompt(prompt, prompt_id)

    while True:
        out = ws.recv()
        if isinstance(out, str):
            message = json.loads(out)
            if message["type"] == "executing":
                data = message["data"]
                if data["node"] is None and data["prompt_id"] == prompt_id:
                    break  # 실행 완료
        # 이진 프레임은 미리보기 이미지이므로 여기서 건너뜁니다
        continue

    history = get_history(prompt_id)[prompt_id]
    output_images = {}
    for node_id in history["outputs"]:
        node_output = history["outputs"][node_id]
        images_output = []
        if "images" in node_output:
            for image in node_output["images"]:
                image_data = get_image(
                    image["filename"], image["subfolder"], image["type"]
                )
                images_output.append(image_data)
        output_images[node_id] = images_output
    return output_images


if __name__ == "__main__":
    prompt_text = """{
        "3": { ... }, "4": { ... }, "5": { ... },
        "6": { ... }, "7": { ... }, "8": { ... },
        "9": { "class_type": "SaveImage", "inputs": { ... } }
    }"""
    prompt = json.loads(prompt_text)
    prompt["3"]["inputs"]["seed"] = 5
    prompt["6"]["inputs"]["text"] = "masterpiece best quality man"

    ws = websocket.WebSocket()
    ws.connect(f"ws://{SERVER_ADDRESS}/ws?clientId={client_id}")
    images = get_images(ws, prompt)
    ws.close()

    print(f"{len(images)}개의 출력 노드에서 이미지를 받았습니다.")

    # 이미지 표시(필요한 경우 Pillow 사용):
    # for node_id in images:
    #     for image_data in images[node_id]:
    #         from PIL import Image
    #         import io
    #         img = Image.open(io.BytesIO(image_data))
    #         img.show()

WebSocket 이진 프레임에는 생성 중인 미리보기 이미지가 포함됩니다. 이를 디코딩하여 실시간 미리보기를 볼 수 있습니다(서버 메시지 페이지에서 이진 형식을 확인하세요).

방법 3: SaveImageWebsocket을 사용한 WebSocket (실시간 이미지)

소스: websockets_api_example_ws_images.py 이미지를 디스크에 저장하지 않고 싶은 경우, SaveImageWebsocket 노드를 사용하세요. 이미지는 WebSocket 이진 프레임을 통해 바로 전송됩니다.

"""websockets_api_example_ws_images.py: WebSocket을 통해 직접 이미지를 수신합니다."""

import websocket  # pip install websocket-client
import uuid
import json
import urllib.request
import urllib.parse

SERVER_ADDRESS = "127.0.0.1:8188"
client_id = str(uuid.uuid4())


def queue_prompt(prompt):
    p = {"prompt": prompt, "client_id": client_id}
    data = json.dumps(p).encode("utf-8")
    req = urllib.request.Request(
        f"http://{SERVER_ADDRESS}/prompt", data=data
    )
    return json.loads(urllib.request.urlopen(req).read())


def get_images(ws, prompt):
    prompt_id = queue_prompt(prompt)["prompt_id"]
    output_images = {}
    current_node = ""

    while True:
        out = ws.recv()
        if isinstance(out, str):
            message = json.loads(out)
            if message["type"] == "executing":
                data = message["data"]
                if data["prompt_id"] == prompt_id:
                    if data["node"] is None:
                        break  # 실행 완료
                    current_node = data["node"]
        else:
            # 이진 프레임; SaveImageWebsocket에서 온 이미지 데이터
            if current_node == "save_image_websocket_node":
                images_output = output_images.get(current_node, [])
                # 처음 8바이트는 타입/메타데이터, 나머지는 이미지 데이터
                images_output.append(out[8:])
                output_images[current_node] = images_output

    return output_images


if __name__ == "__main__":
    prompt_text = """{
        "3": { "class_type": "KSampler", "inputs": { ... } },
        ...
        "save_image_websocket_node": {
            "class_type": "SaveImageWebsocket",
            "inputs": {"images": ["8", 0]}
        }
    }"""
    prompt = json.loads(prompt_text)
    prompt["3"]["inputs"]["seed"] = 5

    ws = websocket.WebSocket()
    ws.connect(f"ws://{SERVER_ADDRESS}/ws?clientId={client_id}")
    images = get_images(ws, prompt)
    ws.close()

    print(f"{len(images)}개의 이미지를 WebSocket을 통해 받았습니다.")

    # 표시(필요한 경우 Pillow 사용):
    # for image_data in images.get("save_image_websocket_node", []):
    #     from PIL import Image
    #     import io
    #     img = Image.open(io.BytesIO(image_data))
    #     img.show()

워크플로우는 일반 SaveImage 노드 대신 class_type: "SaveImageWebsocket"를 가진 노드를 사용해야 합니다.

어떤 방법을 사용해야 할까요?

방법 1: HTTP 전용

즉시 처리 후 잊어버리기. 즉각적인 출력이 필요하지 않거나, 나중에 결과를 가져오는 것이 괜찮을 때 사용하세요.

방법 2: WebSocket + 히스토리

권장. 실행이 완료될 때까지 기다렸다가 출력물을 다운로드하세요. 간편함과 신뢰성의 최적 조화입니다.

방법 3: SaveImageWebsocket

실시간 이미지. 디스크 쓰기 없이 이미지를 전달받고자 하는 인터랙티브 앱에 가장 적합합니다.

전체 API 참조(엔드포인트, 페이로드 형식, 오류 처리)는 서버 라우트 및 서버 메시지 페이지를 참고하세요.

​방법 1: 제출 후 잊어버리기 (HTTP 전용)

​방법 2: WebSocket + 히스토리 (완료 모니터링)

​방법 3: SaveImageWebsocket을 사용한 WebSocket (실시간 이미지)

​어떤 방법을 사용해야 할까요?

방법 1: HTTP 전용

방법 2: WebSocket + 히스토리

방법 3: SaveImageWebsocket

방법 1: 제출 후 잊어버리기 (HTTP 전용)

방법 2: WebSocket + 히스토리 (완료 모니터링)

방법 3: SaveImageWebsocket을 사용한 WebSocket (실시간 이미지)

어떤 방법을 사용해야 할까요?