03.01.01.ComfyUI Ubuntu:环境搭建篇(集成 调用ComfyUI的API 使用Pythin 中间件方式)

📅 2026/7/9 2:00:09
03.01.01.ComfyUI Ubuntu:环境搭建篇(集成 调用ComfyUI的API 使用Pythin 中间件方式)
总操作流程1、下载安装2、写代码3、测试下载安装pipinstallwebsocket-client uuid Pillowapi: https://docs.comfy.org/zh/development/comfyui-server/api-examples写代码方法一提交即忘仅 HTTPcat/usr/local/software/ComfyUI/script_examples/test_basic_api.pyEOF 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, datadata) 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.pyEOF 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), datadata) 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.pyEOF 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), datadata) 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/outputWebSocket 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