API 接口 #

┌─────────────────────────────────────────────────────────────┐
│                    API 架构                                  │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  客户端                      ComfyUI 服务器                  │
│  ┌──────────┐              ┌──────────┐                    │
│  │ Python   │              │          │                    │
│  │ 脚本     │──HTTP/WS────→│ ComfyUI  │                    │
│  │          │              │ Server   │                    │
│  └──────────┘              │          │                    │
│                            └──────────┘                    │
│  ┌──────────┐                    │                         │
│  │ 其他应用  │                    │                         │
│  │ Web/App  │────────────────────┘                         │
│  └──────────┘                                              │
│                                                             │
└─────────────────────────────────────────────────────────────┘

API 主要功能：

工作流管理：
├── 提交工作流执行
├── 查询执行状态
├── 中断执行
└── 清空队列

文件操作：
├── 上传图像
├── 获取生成图像
├── 查看模型列表
└── 管理输出文件

系统信息：
├── 查看系统状态
├── 获取队列信息
└── 查看执行历史

启动 ComfyUI 服务器：

基本命令：
python main.py

指定端口：
python main.py --port 8188

监听所有接口：
python main.py --listen 0.0.0.0

完整配置：
python main.py --listen 0.0.0.0 --port 8188

启动后访问：
http://localhost:8188

远程访问设置：

允许远程访问：
python main.py --listen 0.0.0.0

安全建议：
├── 使用防火墙限制访问
├── 设置反向代理
├── 添加认证（需要额外配置）
└── 使用 HTTPS

反向代理示例（Nginx）：
server {
    listen 80;
    server_name your-domain.com;

    location / {
        proxy_pass http://127.0.0.1:8188;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

API 端点列表：

POST /prompt
├── 功能：提交工作流执行
├── 输入：工作流 JSON
└── 返回：执行 ID

GET /history/{prompt_id}
├── 功能：查询执行历史
├── 输入：执行 ID
└── 返回：执行结果

GET /queue
├── 功能：查看队列状态
└── 返回：队列信息

POST /interrupt
├── 功能：中断当前执行
└── 返回：确认信息

POST /clear
├── 功能：清空队列
└── 返回：确认信息

GET /view
├── 功能：获取图像文件
├── 参数：filename, type, subfolder
└── 返回：图像数据

POST /upload/image
├── 功能：上传图像
├── 输入：图像文件
└── 返回：文件名

GET /object_info
├── 功能：获取节点信息
└── 返回：所有节点定义

GET /object_info/{node_class}
├── 功能：获取特定节点信息
└── 返回：节点定义

基本 API 调用：

import requests
import json

# ComfyUI 服务器地址
server = "http://127.0.0.1:8188"

# 加载工作流
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# 提交工作流
def queue_prompt(workflow):
    response = requests.post(
        f"{server}/prompt",
        json={"prompt": workflow}
    )
    return response.json()

# 查询状态
def get_history(prompt_id):
    response = requests.get(
        f"{server}/history/{prompt_id}"
    )
    return response.json()

# 获取图像
def get_image(filename, subfolder="", folder_type="output"):
    params = {
        "filename": filename,
        "subfolder": subfolder,
        "type": folder_type
    }
    response = requests.get(
        f"{server}/view",
        params=params
    )
    return response.content

# 完整流程
result = queue_prompt(workflow)
prompt_id = result["prompt_id"]

# 等待完成
import time
while True:
    history = get_history(prompt_id)
    if prompt_id in history:
        break
    time.sleep(1)

# 获取结果
outputs = history[prompt_id]["outputs"]
for node_id, output in outputs.items():
    if "images" in output:
        for image in output["images"]:
            image_data = get_image(
                image["filename"],
                image.get("subfolder", ""),
                image.get("type", "output")
            )
            with open(f"output_{image['filename']}", "wb") as f:
                f.write(image_data)

使用 WebSocket 监控执行：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if data["type"] == "executing":
        if data["data"]["node"] is None:
            print("Execution completed")
            ws.close()

def on_error(ws, error):
    print(f"Error: {error}")

def on_open(ws):
    # 提交工作流
    workflow = {...}
    ws.send(json.dumps({
        "type": "execute",
        "data": workflow
    }))

# 连接 WebSocket
ws = websocket.WebSocketApp(
    "ws://127.0.0.1:8188/ws",
    on_message=on_message,
    on_error=on_error,
    on_open=on_open
)
ws.run_forever()

批量生成图像：

import requests
import json
import time

server = "http://127.0.0.1:8188"

def generate_batch(workflow, seeds):
    results = []
    
    for seed in seeds:
        # 修改种子
        workflow["3"]["inputs"]["seed"] = seed
        
        # 提交工作流
        response = requests.post(
            f"{server}/prompt",
            json={"prompt": workflow}
        )
        prompt_id = response.json()["prompt_id"]
        
        # 等待完成
        while True:
            history = requests.get(
                f"{server}/history/{prompt_id}"
            ).json()
            if prompt_id in history:
                break
            time.sleep(0.5)
        
        results.append({
            "seed": seed,
            "prompt_id": prompt_id,
            "outputs": history[prompt_id]["outputs"]
        })
    
    return results

# 使用示例
with open("workflow_api.json") as f:
    workflow = json.load(f)

seeds = range(100, 110)  # 10 个种子
results = generate_batch(workflow, seeds)

获取工作流 API 格式：

方法 1：界面保存
├── 点击 "Save (API Format)"
└── 保存为 .json 文件

方法 2：通过 API
GET /object_info
└── 获取所有节点定义

API 格式结构：
{
    "node_id": {
        "class_type": "NodeClassName",
        "inputs": {
            "input_name": "value",
            ...
        }
    },
    ...
}

基础文生图 API 格式：

{
    "3": {
        "class_type": "KSampler",
        "inputs": {
            "seed": 123456789,
            "steps": 20,
            "cfg": 7,
            "sampler_name": "euler_a",
            "scheduler": "normal",
            "denoise": 1,
            "model": ["4", 0],
            "positive": ["6", 0],
            "negative": ["7", 0],
            "latent_image": ["5", 0]
        }
    },
    "4": {
        "class_type": "CheckpointLoaderSimple",
        "inputs": {
            "ckpt_name": "v1-5-pruned-emaonly.safetensors"
        }
    },
    "5": {
        "class_type": "EmptyLatentImage",
        "inputs": {
            "width": 512,
            "height": 512,
            "batch_size": 1
        }
    },
    "6": {
        "class_type": "CLIPTextEncode",
        "inputs": {
            "text": "a beautiful landscape",
            "clip": ["4", 1]
        }
    },
    "7": {
        "class_type": "CLIPTextEncode",
        "inputs": {
            "text": "ugly, low quality",
            "clip": ["4", 1]
        }
    },
    "8": {
        "class_type": "VAEDecode",
        "inputs": {
            "samples": ["3", 0],
            "vae": ["4", 2]
        }
    },
    "9": {
        "class_type": "SaveImage",
        "inputs": {
            "filename_prefix": "ComfyUI",
            "images": ["8", 0]
        }
    }
}

症状：无法连接到 API

解决：
├── 确认 ComfyUI 正在运行
├── 检查端口是否正确
├── 检查防火墙设置
└── 确认 --listen 设置

症状：API 返回错误

解决：
├── 检查工作流 JSON 格式
├── 确认节点 ID 正确
├── 检查输入参数类型
└── 查看服务器日志

症状：无法获取生成的图像

解决：
├── 确认执行已完成
├── 检查文件名和路径
├── 确认输出目录权限
└── 检查图像格式