使用Python调用ComfyUI API实现图像生成

 关键词: ComfyUI API, Stable Diffusion, 图像生成, WebSocket, Python自动化

目录

1、执行效果

2、环境准备

2.1 安装依赖库

2.2 启动ComfyUI服务器

3、请求代码编写

3.1 抓取请求信息

3.2 官方示例脚本

3.3 编写请求代码

4、执行流程解析

4.1 建立WebSocket连接

4.2 提交任务并获取结果

4.3 显示生成图像

---

1、Python执行效果

2、环境准备

2.1 安装依赖库

确保已安装以下Python库：

pip install websocket-client uuid Pillow

• ​websocket-client: 用于与ComfyUI服务器建立WebSocket长连接
• ​uuid: 生成唯一客户端标识符
• ​Pillow: 图像处理库（用于显示生成的图片）

2.2 启动ComfyUI服务器

启动ComfyUI

在设置的服务器配置里查看网址和监听的端口。

浏览器访问服务地址。

3、请求代码编写

3.1 抓取请求信息

在网页端中，选择要运行的工作流，按下F12，运行一次，抓取请求信息。

我们只需要请求参数中的prompt字段。这里相当于每次请求使用的工作流。

3.2 官方示例脚本

参考官方代码仓库的请求脚本：ComfyUI/script_examples/websockets_api_example.py at master · comfyanonymous/ComfyUI · GitHub

3.3 编写请求代码

# 导入所需库
import websocket  # 用于WebSocket通信
import uuid       # 生成唯一客户端ID
import json       # 处理JSON数据
import urllib.request  # 发送HTTP请求
import urllib.parse    # 构造URL参数# 定义服务器地址和生成客户端唯一ID
server_address = "127.0.0.1:8000"
client_id = str(uuid.uuid4())def queue_prompt(prompt):"""提交提示到服务器并返回执行ID"""# 构造包含提示和客户端ID的请求体p = {"prompt": prompt, "client_id": client_id}# 发送POST请求到/prompt端点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_image(filename, subfolder, folder_type):"""根据参数获取生成的图像二进制数据"""# 构造URL查询参数data = {"filename": filename, "subfolder": subfolder, "type": folder_type}url_values = urllib.parse.urlencode(data)# 发送GET请求获取图像with urllib.request.urlopen(f"http://{server_address}/view?{url_values}") as response:return response.read()def get_history(prompt_id):"""获取指定提示ID的执行历史记录"""with urllib.request.urlopen(f"http://{server_address}/history/{prompt_id}") as response:return json.loads(response.read())def get_images(ws, prompt):"""通过WebSocket监听执行过程并获取最终生成的图像"""# 提交提示并获取执行IDprompt_id = queue_prompt(prompt)['prompt_id']output_images = {}# 持续监听WebSocket消息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  # 执行完成时退出循环else:# 二进制数据流（预览图）处理逻辑（已跳过）continue# 获取执行历史并提取生成的图像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_outputreturn output_images# 定义工作流提示模板
prompt = {"6": {"inputs": {"text": "cute anime girl with massive fluffy fennec ears and a big fluffy tail blonde messy long hair blue eyes wearing a maid outfit with a long black gold leaf pattern dress and a white apron mouth open placing a fancy black forest cake with candles on top of a dinner table of an old dark Victorian mansion lit by candlelight with a bright window to the foggy forest and very expensive stuff everywhere there are paintings on the walls","clip": ["30", 1]},"class_type": "CLIPTextEncode","_meta": {"title": "CLIP文本编码器"}},"8": {"inputs": {"samples": ["31", 0], "vae": ["30", 2]},"class_type": "VAEDecode","_meta": {"title": "VAE解码"}},"9": {"inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]},"class_type": "SaveImage","_meta": {"title": "保存图像"}},"27": {"inputs": {"width": 1024, "height": 1024, "batch_size": 1},"class_type": "EmptySD3LatentImage","_meta": {"title": "空Latent_SD3"}},"30": {"inputs": {"ckpt_name": "flux1-dev-fp8.safetensors"},"class_type": "CheckpointLoaderSimple","_meta": {"title": "Checkpoint加载器(简易)"}},"31": {"inputs": {"seed": 972054013131369,"steps": 20,"cfg": 1,"sampler_name": "euler","scheduler": "simple","denoise": 1,"model": ["30", 0],"positive": ["35", 0],"negative": ["33", 0],"latent_image": ["27", 0]},"class_type": "KSampler","_meta": {"title": "K采样器"}},"33": {"inputs": {"text": "", "clip": ["30", 1]},"class_type": "CLIPTextEncode","_meta": {"title": "CLIP文本编码器"}},"35": {"inputs": {"guidance": 3.5, "conditioning": ["6", 0]},"class_type": "FluxGuidance","_meta": {"title": "Flux引导"}}
}# 建立WebSocket连接并获取图像
ws = websocket.WebSocket()
ws.connect(f"ws://{server_address}/ws?clientId={client_id}")
images = get_images(ws, prompt)
ws.close()  # 显式关闭连接# 以下代码可用于显示生成的图像（需取消注释）
for node_id in images:for image_data in images[node_id]:from PIL import Imageimport ioimage = Image.open(io.BytesIO(image_data))image.save("output.jpg")  # 保存到文件image.show()

注意，代码中的请求地址和端口要和你ComfyUI的地址要保持一致。通过Python调用api接口的方式，可以实现批量，自动化等操作，使用起来更灵活。

3.4 代码详细解析

1. 导入库及作用

• ​websocket​：用于与服务器建立WebSocket通信，实时接收生成状态。
• ​uuid​：生成唯一客户端ID，标识当前会话。
• ​json​：处理请求和响应中的JSON数据。
• ​urllib.request​：发送HTTP请求（GET/POST）。
• ​urllib.parse​：构造URL查询参数。

2. 全局变量定义

• ​**server_address = "127.0.0.1:8000"**​：本地服务器地址，指向ComfyUI或其他AI服务。
• ​**client_id = str(uuid.uuid4())**​：生成唯一客户端ID，确保会话隔离。

3. 核心函数解析

(1) queue_prompt(prompt)

• ​功能​：提交生成任务到服务器，返回任务ID。
• ​步骤​：
  1. 构造包含提示和客户端ID的请求体。
  2. 发送POST请求到/prompt端点。
  3. 解析响应中的prompt_id，用于后续查询。

(2) get_image(filename, subfolder, folder_type)

• ​功能​：通过文件名、子文件夹和类型获取图像二进制数据。
• ​步骤​：
  1. 构造URL查询参数（如filename）。
  2. 发送GET请求到/view端点。
  3. 返回图像二进制数据。

(3) get_history(prompt_id)

• ​功能​：获取指定任务ID的历史执行记录。
• ​步骤​：
  1. 请求/history/{prompt_id}端点。
  2. 返回包含输出节点数据的JSON。

(4) get_images(ws, prompt)

• ​功能​：主流程函数，监听任务完成并提取图像。
• ​步骤​：
  1. 提交提示，获取prompt_id。
  2. 通过WebSocket循环监听消息，直到任务完成。
  3. 查询历史记录，遍历所有输出节点。
  4. 调用get_image下载图像数据，按节点ID分类存储。

4. 工作流提示（prompt变量）

• ​结构​：字典定义多个节点，每个节点包含输入、类型和元数据。
• ​关键节点​：
  • ​节点30​：加载模型检查点（CheckpointLoaderSimple）。
  • ​节点6/33​：CLIP文本编码器，处理正/负面提示。
  • ​节点27​：生成空白潜在图像（分辨率1024x1024）。
  • ​节点31​：K采样器（KSampler），配置生成参数（如步数、采样器）。
  • ​节点35​：Flux引导，调整生成条件。
  • ​节点8​：VAE解码，将潜在数据转为图像。
  • ​节点9​：保存图像到文件。

5. WebSocket通信流程

• ​连接建立​：ws.connect()使用客户端ID连接WebSocket端点。
• ​消息监听​：循环接收消息，检测executing类型消息判断任务完成。
• ​资源释放​：显式调用ws.close()关闭连接。

6. 图像保存与展示（注释部分）

• ​步骤​：
  1. 使用PIL.Image读取二进制数据。
  2. 保存为output.jpg并显示图像（需取消注释）。

7. 整体执行流程

1. 提交提示，启动生成任务。
2. 通过WebSocket监听任务状态。
3. 任务完成后，从历史记录提取图像信息。
4. 下载图像数据并保存到本地文件。

代码总结

该脚本实现了一个完整的图像生成流程：

1. ​配置工作流​：通过节点定义生成步骤（模型加载、文本编码、采样、解码等）。
2. ​任务提交与监听​：使用HTTP和WebSocket与服务器交互。
3. ​结果获取​：按节点提取生成的图像数据，最终保存到本地。

4、执行流程解析

4.1 建立WebSocket连接

ws = websocket.WebSocket()
ws.connect(f"ws://{server_address}/ws?clientId={client_id}")

4.2 提交任务并获取结果

images = get_images(ws, prompt)

4.3 显示生成图像

for node_id in images:for image_data in images[node_id]:from PIL import Imageimport ioimage = Image.open(io.BytesIO(image_data))image.save("output.jpg")  # 保存到文件image.show()             # 弹出预览窗口