基于本地的ComfyUI API调用指南
一、概述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,值为节点配置:{ 60: { inputs: { filename_prefix: Qwen-Image-2512, images: [ 86:8, 0 ] }, class_type: SaveImage, _meta: { title: 保存图像 } }, 91: { inputs: { value: 提示词写在这 }, class_type: PrimitiveStringMultiline, _meta: { title: Prompt } }, 86:39: { inputs: { vae_name: qwen_image_vae.safetensors }, class_type: VAELoader, _meta: { title: 加载VAE } }, 86:38: { inputs: { clip_name: qwen_2.5_vl_7b_fp8_scaled.safetensors, type: qwen_image, device: default }, class_type: CLIPLoader, _meta: { title: 加载CLIP } }, 86:37: { inputs: { unet_name: qwen_image_2512_fp8_e4m3fn.safetensors, weight_dtype: default }, class_type: UNETLoader, _meta: { title: UNet加载器 } }, 86:3: { inputs: { seed: 4, steps: 50, cfg: 4, sampler_name: euler, scheduler: simple, denoise: 1, model: [ 86:66, 0 ], positive: [ 86:81, 0 ], negative: [ 86:7, 0 ], latent_image: [ 86:58, 0 ] }, class_type: KSampler, _meta: { title: K采样器 } }, 86:58: { inputs: { width: 1664, height: 928, batch_size: 1 }, class_type: EmptySD3LatentImage, _meta: { title: 空Latent图像SD3 } }, 86:81: { inputs: { text: [ 91, 0 ], clip: [ 86:38, 0 ] }, class_type: CLIPTextEncode, _meta: { title: CLIP Text Encode (Positive Prompt) } }, 86:8: { inputs: { samples: [ 86:3, 0 ], vae: [ 86:39, 0 ] }, class_type: VAEDecode, _meta: { title: VAE解码 } }, 86:66: { inputs: { shift: 3.1000000000000005, model: [ 86:37, 0 ] }, class_type: ModelSamplingAuraFlow, _meta: { title: 采样算法AuraFlow } }, 86:7: { inputs: { text: 低分辨率低画质肢体畸形手指畸形画面过饱和蜡像感人脸无细节过度光滑画面具有AI感。构图混乱。文字模糊扭曲, clip: [ 86:38, 0 ] }, class_type: CLIPTextEncode, _meta: { title: CLIP Text Encode (Negative Prompt) } } }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格式进行修改# 加载原始JSON with open(workflow_api.json, r, encodingutf-8) as f: workflow json.load(f) # 修改提示词 workflow[91][inputs][text] 一只可爱的猫咪在花园里玩耍,高清摄影,8K # 修改图像尺寸 workflow[86:58][inputs][width] 1024 workflow[86:58][inputs][height] 1024 # 随机种子 workflow[86:3][inputs][seed] -1四、API调用流程4.1 调用流程图1. 搭建ComfyUI工作流 ↓ 2. 在WebUI中测试并确认工作正常 ↓ 3. 导出API格式JSON文件 ↓ 4. 在代码中加载JSON ↓ 5. 修改需要动态调整的参数 ↓ 6. 发送POST请求到ComfyUI服务器 ↓ 7. 等待并接收生成结果 ↓ 8. 处理返回的图像数据4.2 获取ComfyUI服务器地址本地部署: 通常为http://127.0.0.1:8188/api云服务器: 替换为你的公网IP或域名五、Python代码实现5.1 基础同步调用最简单的调用方式,直接等待结果返回:import requests import json # ComfyUI服务器地址 COMFYUI_ADDRESS http://127.0.0.1:8188/api def generate_image_sync(prompt_text, width512, height512): 同步调用ComfyUI API生成图像 # 1. 加载工作流JSON with open(workflow_api.json, r, encodingutf-8) as f: workflow json.load(f) # 2. 修改参数 workflow[91][inputs][text] prompt_text # 修改提示词 workflow[86:58][inputs][width] width workflow[86:58][inputs][height] height workflow[86:3][inputs][seed] -1 # 随机种子 # 3. 发送POST请求 endpoint f{COMFYUI_ADDRESS}/prompt response requests.post( endpoint, json{prompt: workflow}, headers{Content-Type: application/json} ) # 4. 获取prompt_id result response.json() prompt_id result[prompt_id] print(f任务已提交,prompt_id: {prompt_id}) # 5. 等待生成完成 history_endpoint f{COMFYUI_ADDRESS}/history/{prompt_id} import time while True: history_response requests.get(history_endpoint) history_data history_response.json() if prompt_id in history_data: # 获取生成的图像 outputs history_data[prompt_id][outputs] for node_id, output_data in outputs.items(): if images in output_data: for img in output_data[images]: image_filename img[filename] print(f图像已生成: {image_filename}) return { prompt_id: prompt_id, filename: image_filename, full_path: f{COMFYUI_ADDRESS}/view?filename{image_filename} } break time.sleep(1) return None # 使用示例 if __name__ __main__: result generate_image_sync( prompt_text一只可爱的猫咪在花园里玩耍,高清摄影,8K,细节丰富, width768, height768 ) print(result)5.2 使用WebSocket实时监听获取实时进度,适合需要显示生成过程的场景:import websocket import json import uuid import requests import time class ComfyUIWebSocketClient: def __init__(self, server_address127.0.0.1:8188): self.server_address server_address self.client_id str(uuid.uuid4()) self.ws None self.result None def send_prompt(self, workflow): 通过WebSocket发送工作流 # 连接WebSocket ws_url fws://{self.server_address}/ws?clientId{self.client_id} self.ws websocket.create_connection(ws_url) # 发送POST请求 prompt_endpoint fhttp://{self.server_address}/api/prompt payload { prompt: workflow, client_id: self.client_id } response requests.post( prompt_endpoint, jsonpayload, headers{Content-Type: application/json} ) return response.json() def listen_progress(self, prompt_id): 监听生成进度 try: while True: message self.ws.recv() data json.loads(message) # 处理不同类型的消息 if data[type] status: print(f状态更新: {data[data]}) elif data[type] executing: node_id data[data].get(node) print(f正在执行节点: {node_id}) elif data[type] progress: value data[data][value] max_value data[data][max] print(f进度: {value}/{max_value}) elif data[type] execution_start: print(开始执行) elif data[type] execution_success: print(执行成功!) self.result data break elif data[type] execution_cached: print(使用了缓存) except KeyboardInterrupt: print(监听中断) finally: self.ws.close() def generate_image_with_websocket(prompt_text): 使用WebSocket方式生成图像 client ComfyUIWebSocketClient() # 加载并修改工作流 with open(workflow_api.json, r, encodingutf-8) as f: workflow json.load(f) workflow[91][inputs][text] prompt_text # 修改提示词 workflow[86:3][inputs][seed] -1 # 随机种子 # 发送工作流 result client.send_prompt(workflow) prompt_id result[prompt_id] print(fPrompt ID: {prompt_id}) # 监听进度 client.listen_progress(prompt_id) return client.result # 使用示例 if __name__ __main__: result generate_image_with_websocket(日落时分的海边风光,金黄色的阳光洒在沙滩上) print(f生成完成: {result})5.3 批量生成图像自动化批量生成多个图像:import requests import json import time import concurrent.futures COMFYUI_ADDRESS http://127.0.0.1:8188 def generate_single_image(prompt_text, seedNone, width512, height512): 生成单张图像 with open(workflow_api.json, r, encodingutf-8) as f: workflow json.load(f) workflow[91][inputs][text] prompt_text # 修改提示词 workflow[86:58][inputs][width] width workflow[86:58][inputs][height] height workflow[86:3][inputs][seed] -1 # 随机种子 if seed is None: workflow[86:3][inputs][seed] -1 # 随机种子 else: workflow[86:3][inputs][seed] seed endpoint f{COMFYUI_ADDRESS}/prompt response requests.post( endpoint, json{prompt: workflow}, headers{Content-Type: application/json} ) result response.json() prompt_id result[prompt_id] # 等待完成 while True: history_response requests.get(f{COMFYUI_ADDRESS}/history/{prompt_id}) history_data history_response.json() if prompt_id in history_data: outputs history_data[prompt_id][outputs] for node_id, output_data in outputs.items(): if images in output_data: filename output_data[images][0][filename] return { prompt: prompt_text, seed: workflow[86:3][inputs][seed], filename: filename, url: f{COMFYUI_ADDRESS}/view?filename{filename} } time.sleep(0.5) def batch_generate(prompts, max_workers2): 批量生成图像(并发) results [] with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_prompt { executor.submit(generate_single_image, prompt): prompt for prompt in prompts } for future in concurrent.futures.as_completed(future_to_prompt): prompt future_to_prompt[future] try: result future.result() results.append(result) print(f✓ 完成: {prompt}) except Exception as e: print(f✗ 失败: {prompt}, 错误: {e}) return results # 使用示例 if __name__ __main__: prompts [ 一只可爱的猫咪在花园里玩耍, 美丽的日落风景,金色阳光, 高山湖泊,倒影清晰, 城市夜景,霓虹闪烁, 森林小径,阳光斑驳 ] print(开始批量生成...) results batch_generate(prompts, max_workers2) print(f\n生成完成,共 {len(results)} 张图像:) for i, result in enumerate(results, 1): print(f{i}. {result[prompt]}) print(f 文件: {result[filename]}) print(f 链接: {result[url]}\n)5.4 下载生成的图像将生成的图像保存到本地:import requests import os def download_image(image_url, save_path, filenameNone): 下载生成的图像到本地 try: response requests.get(image_url) response.raise_for_status() if filename is None: filename image_url.split(filename)[-1] full_path os.path.join(save_path, filename) os.makedirs(save_path, exist_okTrue) with open(full_path, wb) as f: f.write(response.content) print(f图像已保存: {full_path}) return full_path except Exception as e: print(f下载失败: {e}) return None # 使用示例 if __name__ __main__: image_url http://127.0.0.1:8188/view?filenameComfyUI_00001.png save_path ./generated_images 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端点直接访问image_url f{COMFYUI_ADDRESS}/view?filename{filename}方法2: 从ComfyUI的output目录直接复制文件Q3: 批量生成速度慢怎么办?使用多线程并发请求优化工作流,减少不必要的节点使用更强大的GPU考虑使用云平台的Serverless API,支持多实例并发Q4: API调用返回错误?检查:ComfyUI服务器是否正常运行工作流JSON格式是否正确节点连接关系是否完整模型文件是否存在八、进阶技巧8.1 动态替换模型def change_model(workflow, model_name): 切换模型 # 假设节点4是CheckpointLoaderSimple workflow[4][inputs][ckpt_name] model_name return workflow # 使用 workflow change_model(workflow, sd_xl_base_1.0.safetensors)8.2 添加ControlNet导出的JSON如果包含ControlNet,需要确保ControlNet模型已加载:# ControlNet预处理器节点通常需要额外的参数 workflow[controlnet_node][inputs][image] [load_image_node, 0]8.3 Lora调用# 在KSampler之前添加Lora加载节点 # 修改model的连接,从CheckpointLoader改为LoraLoader workflow[3][inputs][model] [lora_loader_node, 0]九、总结通过ComfyUI API,我们可以实现:✓ 工作流自动化执行✓ 批量图像生成✓ 集成到现有系统✓ 实时进度监控