ComfyUI engineering: workflow como código

{
  "1": {
    "class_type": "CheckpointLoaderSimple",
    "inputs": { "ckpt_name": "flux1-dev.safetensors" }
  },
  "2": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "a cyberpunk cat, neon lighting, 35mm film",
      "clip": ["1", 1]
    }
  },
  "3": {
    "class_type": "CLIPTextEncode",
    "inputs": { "text": "", "clip": ["1", 1] }
  },
  "4": {
    "class_type": "EmptyLatentImage",
    "inputs": { "width": 1024, "height": 1024, "batch_size": 1 }
  },
  "5": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 42, "steps": 28, "cfg": 3.5,
      "sampler_name": "euler", "scheduler": "simple", "denoise": 1.0,
      "model": ["1", 0], "positive": ["2", 0], "negative": ["3", 0],
      "latent_image": ["4", 0]
    }
  },
  "6": {
    "class_type": "VAEDecode",
    "inputs": { "samples": ["5", 0], "vae": ["1", 2] }
  },
  "7": {
    "class_type": "SaveImage",
    "inputs": { "images": ["6", 0], "filename_prefix": "out" }
  }
}

# Exemplo: um nó que aplica um filtro de luminância numa imagem
import torch

class LuminanceShift:
    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "image": ("IMAGE",),
                "shift":  ("FLOAT", {"default": 0.1, "min": -1.0, "max": 1.0, "step": 0.01}),
            },
            "optional": {
                "mask": ("MASK",),
            },
        }

    RETURN_TYPES = ("IMAGE",)
    RETURN_NAMES = ("image",)
    FUNCTION = "shift_luminance"
    CATEGORY = "image/postprocess"

    def shift_luminance(self, image, shift, mask=None):
        # image shape: [B, H, W, 3] em [0,1]
        # Luminance: 0.299 R + 0.587 G + 0.114 B
        lum = image[..., 0] * 0.299 + image[..., 1] * 0.587 + image[..., 2] * 0.114
        delta = shift
        if mask is not None:
            delta = shift * mask.unsqueeze(-1)
        out = (image + delta).clamp(0, 1)
        return (out,)

NODE_CLASS_MAPPINGS = {"LuminanceShift": LuminanceShift}
NODE_DISPLAY_NAME_MAPPINGS = {"LuminanceShift": "Luminance Shift"}

# Cliente Python — manda workflow e baixa resultado
import json, time, uuid, requests, websocket

COMFY = "http://localhost:8188"
CLIENT_ID = str(uuid.uuid4())

def submit(workflow_json: dict) -> str:
    r = requests.post(f"{COMFY}/prompt", json={
        "prompt": workflow_json,
        "client_id": CLIENT_ID,
    })
    return r.json()["prompt_id"]

def wait_via_ws(prompt_id: str) -> dict:
    ws = websocket.WebSocket()
    ws.connect(f"ws://localhost:8188/ws?clientId={CLIENT_ID}")
    while True:
        msg = json.loads(ws.recv())
        if msg["type"] == "executing":
            data = msg["data"]
            if data["node"] is None and data["prompt_id"] == prompt_id:
                ws.close()
                break
    # busca outputs no /history
    return requests.get(f"{COMFY}/history/{prompt_id}").json()[prompt_id]

def download_image(filename: str, subfolder: str = "", folder_type: str = "output") -> bytes:
    r = requests.get(f"{COMFY}/view", params={
        "filename": filename, "subfolder": subfolder, "type": folder_type,
    })
    return r.content

# Uso
with open("workflow.json") as f:
    wf = json.load(f)
wf["5"]["inputs"]["seed"] = int(time.time())   # parametrize
pid = submit(wf)
result = wait_via_ws(pid)
img = result["outputs"]["7"]["images"][0]
png_bytes = download_image(img["filename"], img["subfolder"], img["type"])
open("out.png", "wb").write(png_bytes)