基于本地的ComfyUI的API调用指南

基于本地的ComfyUI API调用指南

ComfyUI作为文生图、文生视频、图生图、图生视频的各类模型的工作流都非常好用，可组合各种模型和Lora。但是想要通过API调用相对比较复杂，且无完善的文档。
本文阐述了ComfyUI API调用完整指南,涵盖工作流导出、API调用流程、参数说明及Python代码实现。
零、代码

不想看太多字，直接下载Demo查看示例即可
https://github.com/zer0Black/ComfyUI-Api-Demo
一、概述

ComfyUI提供了强大的API接口,允许开发者通过编程方式调用ComfyUI工作流,实现AI图像生成的自动化。通过API调用,可以批量生成图像、集成到现有系统中,大幅提升工作效率。
二、准备工作

2.1 搭建工作流

在ComfyUI WebUI中按照需求搭建工作流:

• 打开ComfyUI界面,拖拽所需的节点
  
• 连接节点之间的数据流
  
• 配置节点参数(模型、提示词、采样器等)
  
• 点击"Queue Prompt"测试工作流是否能正常生成图像
  
• 确认工作流运行无误后,准备导出
  

2.2 导出API格式的JSON文件

导出步骤:

• 在ComfyUI界面点击顶部菜单栏的工作流 (Workflow)
  
• 选择导出(API) (Export API)
  
• 浏览器会自动下载一个JSON文件,通常命名为workflow_api.json
  

重要提示: 导出的JSON文件包含了工作流的完整配置,包括所有节点的ID、参数和连接关系。这个文件是API调用的核心输入。
三、JSON文件详解

3.1 JSON文件结构

导出的workflow_api.json文件是一个字典,键为节点ID,值为节点配置:

1. {
2.   "60": {
3.     "inputs": {
4.       "filename_prefix": "Qwen-Image-2512",
5.       "images": [
6.         "86:8",
7.         0
8.       ]
9.     },
10.     "class_type": "SaveImage",
11.     "_meta": {
12.       "title": "保存图像"
13.     }
14.   },
15.   "91": {
16.     "inputs": {
17.       "value": "提示词写在这"
18.     },
19.     "class_type": "PrimitiveStringMultiline",
20.     "_meta": {
21.       "title": "Prompt"
22.     }
23.   },
24.   "86:39": {
25.     "inputs": {
26.       "vae_name": "qwen_image_vae.safetensors"
27.     },
28.     "class_type": "VAELoader",
29.     "_meta": {
30.       "title": "加载VAE"
31.     }
32.   },
33.   "86:38": {
34.     "inputs": {
35.       "clip_name": "qwen_2.5_vl_7b_fp8_scaled.safetensors",
36.       "type": "qwen_image",
37.       "device": "default"
38.     },
39.     "class_type": "CLIPLoader",
40.     "_meta": {
41.       "title": "加载CLIP"
42.     }
43.   },
44.   "86:37": {
45.     "inputs": {
46.       "unet_name": "qwen_image_2512_fp8_e4m3fn.safetensors",
47.       "weight_dtype": "default"
48.     },
49.     "class_type": "UNETLoader",
50.     "_meta": {
51.       "title": "UNet加载器"
52.     }
53.   },
54.   "86:3": {
55.     "inputs": {
56.       "seed": 4,
57.       "steps": 50,
58.       "cfg": 4,
59.       "sampler_name": "euler",
60.       "scheduler": "simple",
61.       "denoise": 1,
62.       "model": [
63.         "86:66",
64.         0
65.       ],
66.       "positive": [
67.         "86:81",
68.         0
69.       ],
70.       "negative": [
71.         "86:7",
72.         0
73.       ],
74.       "latent_image": [
75.         "86:58",
76.         0
77.       ]
78.     },
79.     "class_type": "KSampler",
80.     "_meta": {
81.       "title": "K采样器"
82.     }
83.   },
84.   "86:58": {
85.     "inputs": {
86.       "width": 1664,
87.       "height": 928,
88.       "batch_size": 1
89.     },
90.     "class_type": "EmptySD3LatentImage",
91.     "_meta": {
92.       "title": "空Latent图像（SD3）"
93.     }
94.   },
95.   "86:81": {
96.     "inputs": {
97.       "text": [
98.         "91",
99.         0
100.       ],
101.       "clip": [
102.         "86:38",
103.         0
104.       ]
105.     },
106.     "class_type": "CLIPTextEncode",
107.     "_meta": {
108.       "title": "CLIP Text Encode (Positive Prompt)"
109.     }
110.   },
111.   "86:8": {
112.     "inputs": {
113.       "samples": [
114.         "86:3",
115.         0
116.       ],
117.       "vae": [
118.         "86:39",
119.         0
120.       ]
121.     },
122.     "class_type": "VAEDecode",
123.     "_meta": {
124.       "title": "VAE解码"
125.     }
126.   },
127.   "86:66": {
128.     "inputs": {
129.       "shift": 3.1000000000000005,
130.       "model": [
131.         "86:37",
132.         0
133.       ]
134.     },
135.     "class_type": "ModelSamplingAuraFlow",
136.     "_meta": {
137.       "title": "采样算法（AuraFlow）"
138.     }
139.   },
140.   "86:7": {
141.     "inputs": {
142.       "text": "低分辨率，低画质，肢体畸形，手指畸形，画面过饱和，蜡像感，人脸无细节，过度光滑，画面具有AI感。构图混乱。文字模糊，扭曲",
143.       "clip": [
144.         "86:38",
145.         0
146.       ]
147.     },
148.     "class_type": "CLIPTextEncode",
149.     "_meta": {
150.       "title": "CLIP Text Encode (Negative Prompt)"
151.     }
152.   }
153. }

复制代码

3.2 JSON文件的来源

JSON文件是通过ComfyUI WebUI的"工作流 > 导出(API)"功能自动生成的,它准确反映了当前可视化工作流的状态。
3.3 需要修改的参数

通常情况下,你需要修改以下参数来实现动态调用:
节点类型参数名称说明修改建议KSamplerseed随机种子设置为-1可自动生成随机数,或每次请求指定不同的值KSamplersteps采样步数根据质量需求调整,通常20-50KSamplercfg提示词相关性通常7-12,数值越高越遵循提示词CLIPTextEncodetext提示词修改为你要生成的图像描述EmptyLatentImagewidth图像宽度根据需求调整(如512, 768, 1024)EmptyLatentImageheight图像高度根据需求调整CheckpointLoaderSimpleckpt_name模型文件名切换不同的基础模型修改示例:
由于每个工作流配置出的节点名称不一样，需要根据到处的json格式进行修改

1. # 加载原始JSON
2. with open("workflow_api.json", "r", encoding="utf-8") as f:
3.     workflow = json.load(f)
5. # 修改提示词
6. workflow["91"]["inputs"]["text"] = "一只可爱的猫咪在花园里玩耍,高清摄影,8K"
8. # 修改图像尺寸
9. workflow["86:58"]["inputs"]["width"] = 1024
10. workflow["86:58"]["inputs"]["height"] = 1024
12. # 随机种子
13. workflow["86:3"]["inputs"]["seed"] = -1

复制代码

四、API调用流程

4.1 调用流程图

1. 1. 搭建ComfyUI工作流
2.        ↓
3. 2. 在WebUI中测试并确认工作正常
4.        ↓
5. 3. 导出API格式JSON文件
6.        ↓
7. 4. 在代码中加载JSON
8.        ↓
9. 5. 修改需要动态调整的参数
10.        ↓
11. 6. 发送POST请求到ComfyUI服务器
12.        ↓
13. 7. 等待并接收生成结果
14.        ↓
15. 8. 处理返回的图像数据

复制代码

4.2 获取ComfyUI服务器地址

• 本地部署: 通常为 http://127.0.0.1:8188/api
  
• 云服务器: 替换为你的公网IP或域名
  

五、Python代码实现

5.1 基础同步调用

最简单的调用方式,直接等待结果返回:

1. import requests
2. import json
4. # ComfyUI服务器地址
5. COMFYUI_ADDRESS = "http://127.0.0.1:8188/api"
7. def generate_image_sync(prompt_text, width=512, height=512):
8.     """
9.     同步调用ComfyUI API生成图像
10.     """
11.     # 1. 加载工作流JSON
12.     with open("workflow_api.json", "r", encoding="utf-8") as f:
13.         workflow = json.load(f)
15.     # 2. 修改参数
16.     workflow["91"]["inputs"]["text"] = prompt_text  # 修改提示词
17.     workflow["86:58"]["inputs"]["width"] = width
18.     workflow["86:58"]["inputs"]["height"] = height
19.     workflow["86:3"]["inputs"]["seed"] = -1  # 随机种子
21.     # 3. 发送POST请求
22.     endpoint = f"{COMFYUI_ADDRESS}/prompt"
23.     response = requests.post(
24.         endpoint,
25.         json={"prompt": workflow},
26.         headers={"Content-Type": "application/json"}
27.     )
29.     # 4. 获取prompt_id
30.     result = response.json()
31.     prompt_id = result["prompt_id"]
32.     print(f"任务已提交,prompt_id: {prompt_id}")
34.     # 5. 等待生成完成
35.     history_endpoint = f"{COMFYUI_ADDRESS}/history/{prompt_id}"
36.     import time
37.     while True:
38.         history_response = requests.get(history_endpoint)
39.         history_data = history_response.json()
41.         if prompt_id in history_data:
42.             # 获取生成的图像
43.             outputs = history_data[prompt_id]["outputs"]
44.             for node_id, output_data in outputs.items():
45.                 if "images" in output_data:
46.                     for img in output_data["images"]:
47.                         image_filename = img["filename"]
48.                         print(f"图像已生成: {image_filename}")
49.                         return {
50.                             "prompt_id": prompt_id,
51.                             "filename": image_filename,
52.                             "full_path": f"{COMFYUI_ADDRESS}/view?filename={image_filename}"
53.                         }
54.             break
56.         time.sleep(1)
58.     return None
60. # 使用示例
61. if __name__ == "__main__":
62.     result = generate_image_sync(
63.         prompt_text="一只可爱的猫咪在花园里玩耍,高清摄影,8K,细节丰富",
64.         width=768,
65.         height=768
66.     )
67.     print(result)

复制代码

5.2 使用WebSocket实时监听

获取实时进度,适合需要显示生成过程的场景:

1. import websocket
2. import json
3. import uuid
4. import requests
5. import time
7. class ComfyUIWebSocketClient:
8.     def __init__(self, server_address="127.0.0.1:8188"):
9.         self.server_address = server_address
10.         self.client_id = str(uuid.uuid4())
11.         self.ws = None
12.         self.result = None
14.     def send_prompt(self, workflow):
15.         """
16.         通过WebSocket发送工作流
17.         """
18.         # 连接WebSocket
19.         ws_url = f"ws://{self.server_address}/ws?clientId={self.client_id}"
20.         self.ws = websocket.create_connection(ws_url)
22.         # 发送POST请求
23.         prompt_endpoint = f"http://{self.server_address}/api/prompt"
24.         payload = {
25.             "prompt": workflow,
26.             "client_id": self.client_id
27.         }
29.         response = requests.post(
30.             prompt_endpoint,
31.             json=payload,
32.             headers={"Content-Type": "application/json"}
33.         )
35.         return response.json()
37.     def listen_progress(self, prompt_id):
38.         """
39.         监听生成进度
40.         """
41.         try:
42.             while True:
43.                 message = self.ws.recv()
44.                 data = json.loads(message)
46.                 # 处理不同类型的消息
47.                 if data["type"] == "status":
48.                     print(f"状态更新: {data['data']}")
50.                 elif data["type"] == "executing":
51.                     node_id = data["data"].get("node")
52.                     print(f"正在执行节点: {node_id}")
54.                 elif data["type"] == "progress":
55.                     value = data["data"]["value"]
56.                     max_value = data["data"]["max"]
57.                     print(f"进度: {value}/{max_value}")
59.                 elif data["type"] == "execution_start":
60.                     print("开始执行")
62.                 elif data["type"] == "execution_success":
63.                     print("执行成功!")
64.                     self.result = data
65.                     break
67.                 elif data["type"] == "execution_cached":
68.                     print("使用了缓存")
70.         except KeyboardInterrupt:
71.             print("监听中断")
72.         finally:
73.             self.ws.close()
75. def generate_image_with_websocket(prompt_text):
76.     """
77.     使用WebSocket方式生成图像
78.     """
79.     client = ComfyUIWebSocketClient()
81.     # 加载并修改工作流
82.     with open("workflow_api.json", "r", encoding="utf-8") as f:
83.         workflow = json.load(f)
85.     workflow["91"]["inputs"]["text"] = prompt_text  # 修改提示词
86.     workflow["86:3"]["inputs"]["seed"] = -1  # 随机种子
88.     # 发送工作流
89.     result = client.send_prompt(workflow)
90.     prompt_id = result["prompt_id"]
91.     print(f"Prompt ID: {prompt_id}")
93.     # 监听进度
94.     client.listen_progress(prompt_id)
96.     return client.result
98. # 使用示例
99. if __name__ == "__main__":
100.     result = generate_image_with_websocket("日落时分的海边风光,金黄色的阳光洒在沙滩上")
101.     print(f"生成完成: {result}")

复制代码

5.3 批量生成图像

自动化批量生成多个图像:

1. import requests
2. import json
3. import time
4. import concurrent.futures
6. COMFYUI_ADDRESS = "http://127.0.0.1:8188"
8. def generate_single_image(prompt_text, seed=None, width=512, height=512):
9.     """
10.     生成单张图像
11.     """
12.     with open("workflow_api.json", "r", encoding="utf-8") as f:
13.         workflow = json.load(f)
15.     workflow["91"]["inputs"]["text"] = prompt_text  # 修改提示词
16.     workflow["86:58"]["inputs"]["width"] = width
17.     workflow["86:58"]["inputs"]["height"] = height
18.     workflow["86:3"]["inputs"]["seed"] = -1  # 随机种子
19.    
21.     if seed is None:
22.         workflow["86:3"]["inputs"]["seed"] = -1  # 随机种子
23.     else:
24.         workflow["86:3"]["inputs"]["seed"] = seed
26.     endpoint = f"{COMFYUI_ADDRESS}/prompt"
27.     response = requests.post(
28.         endpoint,
29.         json={"prompt": workflow},
30.         headers={"Content-Type": "application/json"}
31.     )
33.     result = response.json()
34.     prompt_id = result["prompt_id"]
36.     # 等待完成
37.     while True:
38.         history_response = requests.get(f"{COMFYUI_ADDRESS}/history/{prompt_id}")
39.         history_data = history_response.json()
41.         if prompt_id in history_data:
42.             outputs = history_data[prompt_id]["outputs"]
43.             for node_id, output_data in outputs.items():
44.                 if "images" in output_data:
45.                     filename = output_data["images"][0]["filename"]
46.                     return {
47.                         "prompt": prompt_text,
48.                         "seed": workflow["86:3"]["inputs"]["seed"],
49.                         "filename": filename,
50.                         "url": f"{COMFYUI_ADDRESS}/view?filename={filename}"
51.                     }
52.         time.sleep(0.5)
54. def batch_generate(prompts, max_workers=2):
55.     """
56.     批量生成图像(并发)
57.     """
58.     results = []
59.     with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
60.         future_to_prompt = {
61.             executor.submit(generate_single_image, prompt): prompt
62.             for prompt in prompts
63.         }
65.         for future in concurrent.futures.as_completed(future_to_prompt):
66.             prompt = future_to_prompt[future]
67.             try:
68.                 result = future.result()
69.                 results.append(result)
70.                 print(f"✓ 完成: {prompt}")
71.             except Exception as e:
72.                 print(f"✗ 失败: {prompt}, 错误: {e}")
74.     return results
76. # 使用示例
77. if __name__ == "__main__":
78.     prompts = [
79.         "一只可爱的猫咪在花园里玩耍",
80.         "美丽的日落风景,金色阳光",
81.         "高山湖泊,倒影清晰",
82.         "城市夜景,霓虹闪烁",
83.         "森林小径,阳光斑驳"
84.     ]
86.     print("开始批量生成...")
87.     results = batch_generate(prompts, max_workers=2)
89.     print(f"\n生成完成,共 {len(results)} 张图像:")
90.     for i, result in enumerate(results, 1):
91.         print(f"{i}. {result['prompt']}")
92.         print(f"   文件: {result['filename']}")
93.         print(f"   链接: {result['url']}\n")

复制代码

5.4 下载生成的图像

将生成的图像保存到本地:

1. import requests
2. import os
4. def download_image(image_url, save_path, filename=None):
5.     """
6.     下载生成的图像到本地
7.     """
8.     try:
9.         response = requests.get(image_url)
10.         response.raise_for_status()
12.         if filename is None:
13.             filename = image_url.split("filename=")[-1]
15.         full_path = os.path.join(save_path, filename)
16.         os.makedirs(save_path, exist_ok=True)
18.         with open(full_path, "wb") as f:
19.             f.write(response.content)
21.         print(f"图像已保存: {full_path}")
22.         return full_path
24.     except Exception as e:
25.         print(f"下载失败: {e}")
26.         return None
28. # 使用示例
29. if __name__ == "__main__":
30.     image_url = "http://127.0.0.1:8188/view?filename=ComfyUI_00001.png"
31.     save_path = "./generated_images"
32.     download_image(image_url, save_path)

复制代码

六、传参要求详解

6.1 请求参数

参数类型必填说明promptobject是工作流JSON对象client_idstring否客户端标识符,用于区分不同请求6.2 响应参数

参数类型说明prompt_idstring任务的唯一标识符numbernumber队列中排队的任务数node_errorsobject节点错误信息6.3 常见HTTP状态码

状态码说明200请求成功400请求参数错误500服务器内部错误七、常见问题

Q1: 提示词不生效怎么办?

检查节点ID是否正确。导出的JSON中,CLIPTextEncode节点的ID可能与你的不同,需要根据实际JSON调整。
Q2: 如何获取生成的图像文件?

方法1: 通过/view端点直接访问

1. image_url = f"{COMFYUI_ADDRESS}/view?filename={filename}"

复制代码

方法2: 从ComfyUI的output目录直接复制文件
Q3: 批量生成速度慢怎么办?

• 使用多线程并发请求
  
• 优化工作流,减少不必要的节点
  
• 使用更强大的GPU
  
• 考虑使用云平台的Serverless API,支持多实例并发
  

Q4: API调用返回错误?

检查:

• ComfyUI服务器是否正常运行
  
• 工作流JSON格式是否正确
  
• 节点连接关系是否完整
  
• 模型文件是否存在
  

八、进阶技巧

8.1 动态替换模型

1. def change_model(workflow, model_name):
2.     """
3.     切换模型
4.     """
5.     # 假设节点4是CheckpointLoaderSimple
6.     workflow["4"]["inputs"]["ckpt_name"] = model_name
7.     return workflow
9. # 使用
10. workflow = change_model(workflow, "sd_xl_base_1.0.safetensors")

复制代码

8.2 添加ControlNet

导出的JSON如果包含ControlNet,需要确保ControlNet模型已加载:

1. # ControlNet预处理器节点通常需要额外的参数
2. workflow["controlnet_node"]["inputs"]["image"] = ["load_image_node", 0]

复制代码

8.3 Lora调用

1. # 在KSampler之前添加Lora加载节点
2. # 修改model的连接,从CheckpointLoader改为LoraLoader
3. workflow["3"]["inputs"]["model"] = ["lora_loader_node", 0]

复制代码

九、总结

通过ComfyUI API,我们可以实现:

• ✓ 工作流自动化执行
  
• ✓ 批量图像生成
  
• ✓ 集成到现有系统
  
• ✓ 实时进度监控
  
• ✓ 云端部署和扩展
  

关键步骤:

• 在ComfyUI WebUI中搭建并测试工作流
  
• 导出API格式JSON文件
  
• 在代码中加载JSON并修改参数
  
• 通过HTTP/WebSocket发送请求
  
• 处理返回结果
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

这个有用。