Call ComfyUI APi to generate an image then render inline in terminal using kitty graphics protocol

generate_image_comfy_ui_terminal_render.py

#!/usr/bin/env uv run
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "pillow",
#     "term-image",
#     "websocket-client",
#     "standard-imghdr"
# ]
# ///

import sys

import websocket  #NOTE: websocket-client (https://github.com/websocket-client/websocket-client)
import uuid
import json
import urllib.request
import urllib.parse
import random
from PIL import Image
from term_image.image import AutoImage

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("http://{}/prompt".format(server_address), data=data)
    return json.loads(urllib.request.urlopen(req).read())

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

def get_history(prompt_id):
    with urllib.request.urlopen("http://{}/history/{}".format(server_address, prompt_id)) as response:
        return json.loads(response.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 is done
                    else:
                        current_node = data['node']
        else:
            if current_node == 'save_image_websocket_node':
                images_output = output_images.get(current_node, [])
                images_output.append(out[8:])
                output_images[current_node] = images_output

    return output_images


prompt_text = """
{ 
  "62": {
    "inputs": {
      "clip_name": "qwen_3_4b.safetensors",
      "type": "lumina2",
      "device": "default"
    },
    "class_type": "CLIPLoader",
    "_meta": {
      "title": "Load CLIP"
    }
  },
  "63": {
    "inputs": {
      "vae_name": "ae.safetensors"
    },
    "class_type": "VAELoader",
    "_meta": {
      "title": "Load VAE"
    }
  },
  "65": {
    "inputs": {
      "samples": [
        "70",
        0
      ],
      "vae": [
        "63",
        0
      ]
    },
    "class_type": "VAEDecode",
    "_meta": {
      "title": "VAE Decode"
    }
  },
  "66": {
    "inputs": {
      "unet_name": "z_image_turbo_bf16.safetensors",
      "weight_dtype": "default"
    },
    "class_type": "UNETLoader",
    "_meta": {
      "title": "Load Diffusion Model"
    }
  },
  "67": {
    "inputs": {
      "text": "He walked to the window and swung his arms back and forth to generate a little circulation in his upper body. From his home here on Deck Two Morrow could make out, through the open, multilayered flooring, some details of Deck Three below; he looked down over houses, factories, offices and—looming above all the other buildings—the imposing shoulders of the Planner Temples, scattered across the split levels like blocky clouds. Beyond the buildings and streets stood the walls of the world: sheets of metal, ribbed for strength. And over it all lay the multilevelled sky, a lid of girders and panels, enclosing and oppressive.",
      "clip": [
        "62",
        0
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  },
  "68": {
    "inputs": {
      "width": 1024,
      "height": 1024,
      "batch_size": 1
    },
    "class_type": "EmptySD3LatentImage",
    "_meta": {
      "title": "EmptySD3LatentImage"
    }
  },
  "69": {
    "inputs": {
      "shift": 3,
      "model": [
        "66",
        0
      ]
    },
    "class_type": "ModelSamplingAuraFlow",
    "_meta": {
      "title": "ModelSamplingAuraFlow"
    }
  },
  "70": {
    "inputs": {
      "seed": 464678193752713,
      "steps": 8,
      "cfg": 1,
      "sampler_name": "dpmpp_2m",
      "scheduler": "karras",
      "denoise": 1,
      "model": [
        "69",
        0
      ],
      "positive": [
        "67",
        0
      ],
      "negative": [
        "71",
        0
      ],
      "latent_image": [
        "68",
        0
      ]
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "KSampler"
    }
  },
  "71": {
    "inputs": {
      "text": "low quality, bad anatomy, extra digits, missing digits, extra limbs, missing limbs",
      "clip": [
        "62",
        0
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP Text Encode (Prompt)"
    }
  },
  "save_image_websocket_node": {
    "inputs": {
      "images": [
        "65",
        0
      ]
    },
    "class_type": "SaveImageWebsocket",
    "_meta": {
      "title": "Save Image (Websocket)"
    }
  }
}
"""

prompt = json.loads(prompt_text)


while True:
    print("Paste your multiline prompt (press Enter twice to submit):")

    lines = []
    while True:
        line = input()
        if not line.strip():  # Stop on blank line
            break
        lines.append(line)

    input_text = "\n".join(lines).strip()

    if not input_text:
        print("No text entered, skipping...")
        continue

    #set the text prompt for our positive CLIPTextEncode
    prompt["67"]["inputs"]["text"] =  "Cinematic concept art, futuristic scifi high quality render: " + input_text

    #set the seed for our KSampler node
    prompt["70"]["inputs"]["seed"] = random.randint(0, 999999999999999)

    ws = websocket.WebSocket()
    ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id))
    images = get_images(ws, prompt)
    ws.close() # for in case this example is used in an environment where it will be repeatedly called, like in a Gradio app. otherwise, you'll randomly receive connection timeouts

    #Commented out code to display the output images:

    for node_id in images:
        print("Images from node {}:".format(node_id))
        for image_data in images[node_id]:
            from PIL import Image
            import io
            image = Image.open(io.BytesIO(image_data))
            # image.show()
            auto_image = AutoImage(image)
            auto_image.height = 30
            auto_image.width = 30
            # auto_image.draw()
            print("{:1.1#}".format(auto_image))