Ejemplos de la API de ComfyUI Server: HTTP, WebSocket e historial

Esta página muestra tres formas de interactuar con la API del servidor de ComfyUI, desde un envío HTTP simple hasta una integración completa con WebSocket y salida de imágenes en tiempo real. Todos los ejemplos usan el flujo de trabajo predeterminado de SD1.5 a modo de ilustración. Antes de usar la API, debes exportar tu flujo de trabajo en formato API.

Estos ejemplos usan Python con la biblioteca estándar y el paquete websocket-client (pip install websocket-client). El protocolo subyacente de la API es el mismo independientemente del idioma. Consulta la Referencia de la API de Cloud para ver los equivalentes en TypeScript y curl.

Método 1: Enviar y olvidar (solo HTTP)

Fuente: basic_api_example.py El enfoque más sencillo: envía un flujo de trabajo y no esperes los resultados. Es útil para tareas en las que envías y te olvidas, y revisas las salidas más tarde.

"""basic_api_example.py: envía un flujo de trabajo solo por 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__":
    # Carga un flujo de trabajo exportado en formato 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)
    # Modifica las entradas antes de enviar
    prompt["3"]["inputs"]["seed"] = 5
    prompt["6"]["inputs"]["text"] = "masterpiece best quality man"

    queue_prompt(prompt)
    print("Prompt submitted successfully.")

Este método usa el nodo SaveImage, que guarda las imágenes en el disco del servidor. Para recuperarlas, tendrías que hacer después una llamada a GET /view?filename=....

Método 2: WebSocket + historial (supervisar la finalización)

Fuente: websockets_api_example.py Usa WebSocket para esperar a que finalice la ejecución y luego obtén las salidas a través del endpoint /history. Este es el patrón recomendado para la mayoría de los casos de uso.

"""websockets_api_example.py: supervisa la ejecución por WebSocket y descarga mediante /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  # Ejecución terminada
        # Los frames binarios son imágenes de vista previa, omítelos aquí
        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"Got {len(images)} output node(s) with images.")

    # Mostrar las imágenes (requiere 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()

Los frames binarios del WebSocket contienen imágenes de vista previa del progreso durante la generación. Puedes decodificarlos para obtener vistas previas en vivo (consulta la página Server Messages para conocer el formato binario).

Método 3: WebSocket con SaveImageWebsocket (imágenes en tiempo real)

Fuente: websockets_api_example_ws_images.py Para casos en los que no quieres que las imágenes se guarden en el disco, usa el nodo SaveImageWebsocket. Las imágenes llegan directamente mediante frames binarios de WebSocket.

"""websockets_api_example_ws_images.py: recibe imágenes directamente por 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  # Ejecución terminada
                    current_node = data["node"]
        else:
            # Frame binario: datos de imagen de SaveImageWebsocket
            if current_node == "save_image_websocket_node":
                images_output = output_images.get(current_node, [])
                # Los primeros 8 bytes son tipo/meta, el resto son datos de imagen
                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"Received {len(images)} image(s) via WebSocket.")

    # Mostrar (requiere 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()

El flujo de trabajo debe usar un nodo con class_type: "SaveImageWebsocket" (un nodo integrado) en lugar del nodo SaveImage habitual.

¿Qué método debería usar?

Método 1: solo HTTP

Envía y olvídate. Úsalo cuando no necesites la salida de inmediato o cuando obtener los resultados más tarde sea aceptable.

Método 2: WebSocket + History

Recomendado. Espera a que termine y luego descarga las salidas. Es el mejor equilibrio entre simplicidad y fiabilidad.

Método 3: SaveImageWebsocket

Imágenes en tiempo real. Ideal para aplicaciones interactivas en las que quieres recibir las imágenes sin escribir en disco.

Para consultar la referencia completa de la API (endpoints, formatos de payload y manejo de errores), revisa las páginas Server Routes y Server Messages.

Desarrollar con Comfy

Desplegar ComfyUI

Comfy CLI

Specifications

Nodos personalizados

Ejemplos de la API de ComfyUI Server: HTTP, WebSocket e historial

Método 1: Enviar y olvidar (solo HTTP)

Método 2: WebSocket + historial (supervisar la finalización)

Método 3: WebSocket con SaveImageWebsocket (imágenes en tiempo real)

¿Qué método debería usar?

Método 1: solo HTTP

Método 2: WebSocket + History

Método 3: SaveImageWebsocket