Ejemplos de API

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 lenguaje. 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 (monitorizar 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: monitorea 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 los casos en los que no quieres que las imágenes se guarden en disco, usa el nodo SaveImageWebsocket. Las imágenes se entregan directamente mediante mensajes binarios del WebSocket.

"""websockets_api_example_ws_images.py — Receive images directly via 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  # Execution done
                    current_node = data["node"]
        else:
            # Binary frame — image data from SaveImageWebsocket
            if current_node == "save_image_websocket_node":
                images_output = output_images.get(current_node, [])
                # The first 8 bytes are type/meta, rest is image data
                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.")

    # Display (requires 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 olvida. Úsalo cuando no necesites resultados inmediatos o cuando puedas recuperar los resultados más tarde.

Método 2: WebSocket + Historial

RECOMENDADO. Espera a que termine y luego descarga las salidas. El mejor equilibrio entre simplicidad y fiabilidad.

Método 3: SaveImageWebsocket

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

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

Desarrollar con Comfy

Desplegar ComfyUI

Comfy CLI

Specifications

Nodos personalizados

Ejemplos de API

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

Método 2: WebSocket + historial (monitorizar 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 + Historial

Método 3: SaveImageWebsocket

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

​Método 2: WebSocket + historial (monitorizar 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 + Historial

Método 3: SaveImageWebsocket

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

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

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

¿Qué método debería usar?