总操作流程：
下载安装
pipinstallwebsocket-client uuid Pillow
- api: https://docs.comfy.org/zh/development/comfyui-server/api-examples
写代码
方法一：提交即忘（仅 HTTP）
cat>/usr/local/software/ComfyUI/script_examples/test_basic_api.py<<'EOF' import json from urllib import request 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": "stabilityai/stable-diffusion-3-medium/sd3_medium_incl_clips_t5xxlfp16.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 girl" } }, "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 ] } } } """ def queue_prompt(prompt): p = {"prompt": prompt} data = json.dumps(p).encode('utf-8') req = request.Request("http://10.3.11.173:8188/prompt", data=data) request.urlopen(req) prompt = json.loads(prompt_text) prompt["6"]["inputs"]["text"] = "画一只小猫咪" prompt["3"]["inputs"]["seed"] = 5 queue_prompt(prompt) EOF
方法二：WebSocket + History（监控执行完成）
cat>/usr/local/software/ComfyUI/script_examples/test_websockets_api.py<<'EOF' import websocket import uuid import json import urllib.request import urllib.parse server_address = "10.3.11.173: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("http://{}/prompt".format(server_address), data=data) 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 = str(uuid.uuid4()) queue_prompt(prompt, prompt_id) output_images = {} 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 #Execution is done else: # If you want to be able to decode the binary stream for latent previews, here is how you can do it: # bytesIO = BytesIO(out[8:]) # preview_image = Image.open(bytesIO) # This is your preview in PIL image format, store it in a global continue #previews are binary data history = get_history(prompt_id)[prompt_id] 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 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": "stabilityai/stable-diffusion-3-medium/sd3_medium_incl_clips_t5xxlfp16.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 girl" } }, "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) #set the text prompt for our positive CLIPTextEncode prompt["6"]["inputs"]["text"] = "画一只大闸蟹" prompt["3"]["inputs"]["seed"] = 5 ws = websocket.WebSocket() ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id)) images = get_images(ws, prompt) ws.close() EOF
方法三：WebSocket 配合 SaveImageWebsocket（实时获取图片）
cat>/usr/local/software/ComfyUI/script_examples/test_websockets_api_example_ws_images.py<<'EOF' import websocket import uuid import json import urllib.request import urllib.parse server_address = "10.3.11.173: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 = """ { "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": "stabilityai/stable-diffusion-3-medium/sd3_medium_incl_clips_t5xxlfp16.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 girl" } }, "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["6"]["inputs"]["text"] = "画一只毛毛虫" prompt["3"]["inputs"]["seed"] = 5 ws = websocket.WebSocket() ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id)) images = get_images(ws, prompt) ws.close() EOF
测试
# 本身就有的代码cd/usr/local/software/ComfyUI/script_exampleschmod0777-R*&&chown$USER:$USER-R*
方法一：提交即忘（仅 HTTP）
# 运行时候查看ComfyUI输出的日志python /usr/local/software/ComfyUI/script_examples/test_basic_api.py# 自动生成的图片在如下文件夹ls/usr/local/software/ComfyUI/output
WebSocket + History（监控执行完成）
# 运行时候查看ComfyUI输出的日志python /usr/local/software/ComfyUI/script_examples/test_websockets_api.py# 自动生成的图片在如下文件夹ls/usr/local/software/ComfyUI/output
方法三：WebSocket 配合 SaveImageWebsocket（实时获取图片）
# 运行时候查看ComfyUI输出的日志python /usr/local/software/ComfyUI/script_examples/test_websockets_api_example_ws_images.py# 自动生成的图片在如下文件夹ls/usr/local/software/ComfyUI/output
（支持资料、图片参考_相关定制）_文章底部可以扫码)
