实验性 API: 此 API 处于实验阶段,可能会发生变化。端点、请求/响应格式和行为可能会在未事先通知的情况下进行修改。部分端点为兼容本地 ComfyUI 而保留,但可能具有不同的语义(例如,某些字段会被忽略)。
需要订阅: 通过 API 运行工作流需要有效的 Comfy Cloud 订阅。请查看定价方案了解详情。
设置
所有示例都使用以下通用的导入和配置:export COMFY_CLOUD_API_KEY="your-api-key"
export BASE_URL="https://cloud.comfy.org"
import { readFile, writeFile } from "fs/promises";
const BASE_URL = "https://cloud.comfy.org";
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
function getHeaders(): HeadersInit {
return {
"X-API-Key": API_KEY,
"Content-Type": "application/json",
};
}
import os
import requests
import json
import time
import asyncio
import aiohttp
BASE_URL = "https://cloud.comfy.org"
API_KEY = os.environ["COMFY_CLOUD_API_KEY"]
def get_headers():
return {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
对象信息
获取可用的节点定义。这对于了解可用的节点及其输入/输出规范非常有用。curl -X GET "$BASE_URL/api/object_info" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY"
async function getObjectInfo(): Promise<Record<string, any>> {
const response = await fetch(`${BASE_URL}/api/object_info`, {
headers: getHeaders(),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
const objectInfo = await getObjectInfo();
console.log(`Available nodes: ${Object.keys(objectInfo).length}`);
const ksampler = objectInfo["KSampler"] ?? {};
console.log(`KSampler inputs: ${Object.keys(ksampler.input?.required ?? {})}`);
def get_object_info():
"""Fetch all available node definitions from cloud."""
response = requests.get(
f"{BASE_URL}/api/object_info",
headers=get_headers()
)
response.raise_for_status()
return response.json()
# Get all nodes
object_info = get_object_info()
print(f"Available nodes: {len(object_info)}")
# Get a specific node's definition
ksampler = object_info.get("KSampler", {})
inputs = list(ksampler.get('input', {}).get('required', {}).keys())
print(f"KSampler inputs: {inputs}")
上传输入
上传图像、遮罩或其他文件以在工作流中使用。直接上传(Multipart)
curl -X POST "$BASE_URL/api/upload/image" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-F "image=@my_image.png" \
-F "type=input" \
-F "overwrite=true"
async function uploadInput(
filePath: string,
inputType: string = "input"
): Promise<{ name: string; subfolder: string }> {
const file = await readFile(filePath);
const formData = new FormData();
formData.append("image", new Blob([file]), filePath.split("/").pop());
formData.append("type", inputType);
formData.append("overwrite", "true");
const response = await fetch(`${BASE_URL}/api/upload/image`, {
method: "POST",
headers: { "X-API-Key": API_KEY },
body: formData,
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
const result = await uploadInput("my_image.png");
console.log(`Uploaded: ${result.name} to ${result.subfolder}`);
def upload_input(file_path: str, input_type: str = "input") -> dict:
"""Upload a file directly to cloud.
Args:
file_path: Path to the file to upload
input_type: "input" for images, "temp" for temporary files
Returns:
Upload response with filename and subfolder
"""
with open(file_path, "rb") as f:
files = {"image": f}
data = {"type": input_type, "overwrite": "true"}
response = requests.post(
f"{BASE_URL}/api/upload/image",
headers={"X-API-Key": API_KEY}, # No Content-Type for multipart
files=files,
data=data
)
response.raise_for_status()
return response.json()
# Upload an image
result = upload_input("my_image.png")
print(f"Uploaded: {result['name']} to {result['subfolder']}")
上传遮罩
subfolder 参数为 API 兼容性而接受,但在云存储中会被忽略。所有文件都存储在扁平的、内容寻址的命名空间中。curl -X POST "$BASE_URL/api/upload/mask" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-F "image=@mask.png" \
-F "type=input" \
-F "subfolder=clipspace" \
-F 'original_ref={"filename":"my_image.png","subfolder":"","type":"input"}'
async function uploadMask(
filePath: string,
originalRef: { filename: string; subfolder: string; type: string }
): Promise<{ name: string; subfolder: string }> {
const file = await readFile(filePath);
const formData = new FormData();
formData.append("image", new Blob([file]), filePath.split("/").pop());
formData.append("type", "input");
formData.append("subfolder", "clipspace");
formData.append("original_ref", JSON.stringify(originalRef));
const response = await fetch(`${BASE_URL}/api/upload/mask`, {
method: "POST",
headers: { "X-API-Key": API_KEY },
body: formData,
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
const maskResult = await uploadMask("mask.png", {
filename: "my_image.png",
subfolder: "",
type: "input",
});
console.log(`Uploaded mask: ${maskResult.name}`);
def upload_mask(file_path: str, original_ref: dict) -> dict:
"""Upload a mask associated with an original image.
Args:
file_path: Path to the mask file
original_ref: Reference to the original image {"filename": "...", "subfolder": "...", "type": "..."}
"""
with open(file_path, "rb") as f:
files = {"image": f}
data = {
"type": "input",
"subfolder": "clipspace",
"original_ref": json.dumps(original_ref)
}
response = requests.post(
f"{BASE_URL}/api/upload/mask",
headers={"X-API-Key": API_KEY},
files=files,
data=data
)
response.raise_for_status()
return response.json()
运行工作流
提交工作流以执行。支持并发提交: 根据您的订阅等级,您可以提交多个工作流而无需等待之前的任务完成。任务会根据您的等级限制并行运行——超出的任务会自动排队。详情及并发限制请参阅并行执行。
提交工作流
curl -X POST "$BASE_URL/api/prompt" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-H "Content-Type: application/json" \
-d '{"prompt": '"$(cat workflow_api.json)"'}'
async function submitWorkflow(workflow: Record<string, any>): Promise<string> {
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: getHeaders(),
body: JSON.stringify({ prompt: workflow }),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
if (result.error) {
throw new Error(`Workflow error: ${result.error}`);
}
return result.prompt_id;
}
const workflow = JSON.parse(await readFile("workflow_api.json", "utf-8"));
const promptId = await submitWorkflow(workflow);
console.log(`Submitted job: ${promptId}`);
def submit_workflow(workflow: dict) -> str:
"""Submit a workflow and return the prompt_id (job ID).
Args:
workflow: ComfyUI workflow in API format
Returns:
prompt_id for tracking the job
"""
response = requests.post(
f"{BASE_URL}/api/prompt",
headers=get_headers(),
json={"prompt": workflow}
)
response.raise_for_status()
result = response.json()
if "error" in result:
raise ValueError(f"Workflow error: {result['error']}")
return result["prompt_id"]
# Load and submit a workflow
with open("workflow_api.json") as f:
workflow = json.load(f)
prompt_id = submit_workflow(workflow)
print(f"Submitted job: {prompt_id}")
使用合作伙伴节点
如果您的工作流包含合作伙伴节点(调用外部 AI 服务的节点,如 Flux Pro、Ideogram 等),您必须在请求体的extra_data 字段中包含您的 Comfy API 密钥。
在浏览器中运行工作流时,ComfyUI 前端会自动将您的 API 密钥打包到
extra_data 中。本节仅适用于直接调用 API 的情况。curl -X POST "$BASE_URL/api/prompt" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"prompt": '"$(cat workflow_api.json)"',
"extra_data": {
"api_key_comfy_org": "your-comfy-api-key"
}
}'
async function submitWorkflowWithPartnerNodes(
workflow: Record<string, any>,
apiKey: string
): Promise<string> {
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: getHeaders(),
body: JSON.stringify({
prompt: workflow,
extra_data: {
api_key_comfy_org: apiKey,
},
}),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const result = await response.json();
return result.prompt_id;
}
// 当工作流包含合作伙伴节点时使用(例如 Flux Pro、Ideogram 等)
const promptId = await submitWorkflowWithPartnerNodes(workflow, API_KEY);
def submit_workflow_with_partner_nodes(workflow: dict, api_key: str) -> str:
"""提交使用合作伙伴节点的工作流。
Args:
workflow: API 格式的 ComfyUI 工作流
api_key: 来自 platform.comfy.org 的 API 密钥
Returns:
用于跟踪任务的 prompt_id
"""
response = requests.post(
f"{BASE_URL}/api/prompt",
headers=get_headers(),
json={
"prompt": workflow,
"extra_data": {
"api_key_comfy_org": api_key
}
}
)
response.raise_for_status()
return response.json()["prompt_id"]
# 当工作流包含合作伙伴节点时使用
prompt_id = submit_workflow_with_partner_nodes(workflow, API_KEY)
在 platform.comfy.org 生成您的 API 密钥。此密钥与 Cloud API 身份验证(
X-API-Key 请求头)使用的是同一个密钥。请参阅获取 API 密钥了解详细步骤。修改工作流输入
function setWorkflowInput(
workflow: Record<string, any>,
nodeId: string,
inputName: string,
value: any
): Record<string, any> {
if (workflow[nodeId]) {
workflow[nodeId].inputs[inputName] = value;
}
return workflow;
}
// Example: Set seed and prompt
let workflow = JSON.parse(await readFile("workflow_api.json", "utf-8"));
workflow = setWorkflowInput(workflow, "3", "seed", 12345);
workflow = setWorkflowInput(workflow, "6", "text", "a beautiful landscape");
def set_workflow_input(workflow: dict, node_id: str, input_name: str, value) -> dict:
"""Modify a workflow input value.
Args:
workflow: The workflow dict
node_id: ID of the node to modify
input_name: Name of the input field
value: New value
Returns:
Modified workflow
"""
if node_id in workflow:
workflow[node_id]["inputs"][input_name] = value
return workflow
# Example: Set seed and prompt
workflow = set_workflow_input(workflow, "3", "seed", 12345)
workflow = set_workflow_input(workflow, "6", "text", "a beautiful landscape")
检查任务状态
任务状态值: API 返回以下状态值之一:| 状态 | 描述 |
|---|---|
pending | 任务已排队,等待开始 |
in_progress | 任务正在执行 |
completed | 任务成功完成 |
failed | 任务遇到错误 |
cancelled | 任务被用户取消 |
# 轮询任务完成状态
curl -X GET "$BASE_URL/api/job/{prompt_id}/status" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY"
# 响应示例:
# {"status": "pending"} - 任务已排队
# {"status": "in_progress"} - 任务正在运行
# {"status": "completed"} - 任务成功完成
# {"status": "failed"} - 任务遇到错误
# {"status": "cancelled"} - 任务被取消
interface JobStatus {
status: string;
}
async function getJobStatus(promptId: string): Promise<JobStatus> {
const response = await fetch(`${BASE_URL}/api/job/${promptId}/status`, {
headers: getHeaders(),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
async function pollForCompletion(
promptId: string,
timeout: number = 300,
pollInterval: number = 2000
): Promise<void> {
const startTime = Date.now();
while (Date.now() - startTime < timeout * 1000) {
const { status } = await getJobStatus(promptId);
if (status === "completed") {
return;
} else if (["failed", "cancelled"].includes(status)) {
throw new Error(`任务失败,状态:${status}`);
}
await new Promise((resolve) => setTimeout(resolve, pollInterval));
}
throw new Error(`任务 ${promptId} 在 ${timeout}s 内未完成`);
}
await pollForCompletion(promptId);
console.log("任务已完成!");
def get_job_status(prompt_id: str) -> str:
"""获取任务的当前状态。"""
response = requests.get(
f"{BASE_URL}/api/job/{prompt_id}/status",
headers=get_headers()
)
response.raise_for_status()
return response.json()["status"]
def poll_for_completion(prompt_id: str, timeout: int = 300, poll_interval: float = 2.0) -> None:
"""轮询直到任务完成或超时。"""
start_time = time.time()
while time.time() - start_time < timeout:
status = get_job_status(prompt_id)
if status == "completed":
return
elif status in ("failed", "cancelled"):
raise RuntimeError(f"任务失败,状态:{status}")
time.sleep(poll_interval)
raise TimeoutError(f"任务 {prompt_id} 在 {timeout}s 内未完成")
poll_for_completion(prompt_id)
print("任务已完成!")
实时进度 WebSocket
连接 WebSocket 以获取实时执行更新。clientId 参数目前会被忽略——同一用户的所有连接都会收到相同的消息。为了向前兼容,仍建议传递唯一的 clientId。async function listenForCompletion(
promptId: string,
timeout: number = 300000
): Promise<Record<string, any>> {
const wsUrl = `wss://cloud.comfy.org/ws?clientId=${crypto.randomUUID()}&token=${API_KEY}`;
const outputs: Record<string, any> = {};
return new Promise((resolve, reject) => {
const ws = new WebSocket(wsUrl);
const timer = setTimeout(() => {
ws.close();
reject(new Error(`任务在 ${timeout / 1000}s 内未完成`));
}, timeout);
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
const msgType = data.type;
const msgData = data.data ?? {};
// 过滤我们的任务
if (msgData.prompt_id !== promptId) return;
if (msgType === "executing") {
const node = msgData.node;
if (node) {
console.log(`正在执行节点:${node}`);
} else {
console.log("执行完成");
}
} else if (msgType === "progress") {
console.log(`进度:${msgData.value}/${msgData.max}`);
} else if (msgType === "executed" && msgData.output) {
outputs[msgData.node] = msgData.output;
} else if (msgType === "execution_success") {
console.log("任务成功完成!");
clearTimeout(timer);
ws.close();
resolve(outputs);
} else if (msgType === "execution_error") {
const errorMsg = msgData.exception_message ?? "未知错误";
clearTimeout(timer);
ws.close();
reject(new Error(`执行错误:${errorMsg}`));
}
};
ws.onerror = (err) => {
clearTimeout(timer);
reject(err);
};
});
}
// 等待完成并收集输出
const outputs = await listenForCompletion(promptId);
import asyncio
import aiohttp
import json
import uuid
async def listen_for_completion(prompt_id: str, timeout: float = 300.0) -> dict:
"""连接到 WebSocket 并监听任务完成。
Returns:
任务的最终输出
"""
ws_url = BASE_URL.replace("https://", "wss://")
client_id = str(uuid.uuid4())
ws_url = f"{ws_url}/ws?clientId={client_id}&token={API_KEY}"
outputs = {}
async with aiohttp.ClientSession() as session:
async with session.ws_connect(ws_url) as ws:
async def receive_messages():
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
msg_type = data.get("type")
msg_data = data.get("data", {})
# 过滤我们的任务
if msg_data.get("prompt_id") != prompt_id:
continue
if msg_type == "executing":
node = msg_data.get("node")
if node:
print(f"正在执行节点:{node}")
elif msg_type == "progress":
value = msg_data.get("value", 0)
max_val = msg_data.get("max", 100)
print(f"进度:{value}/{max_val}")
elif msg_type == "executed":
node_id = msg_data.get("node")
output = msg_data.get("output", {})
if output:
outputs[node_id] = output
elif msg_type == "execution_success":
print("任务成功完成!")
return outputs
elif msg_type == "execution_error":
error_msg = msg_data.get("exception_message", "未知错误")
raise RuntimeError(f"执行错误:{error_msg}")
elif msg.type == aiohttp.WSMsgType.ERROR:
raise RuntimeError(f"WebSocket 错误:{ws.exception()}")
try:
return await asyncio.wait_for(receive_messages(), timeout=timeout)
except asyncio.TimeoutError:
raise TimeoutError(f"任务在 {timeout}s 内未完成")
# 等待完成并收集输出
outputs = await listen_for_completion(prompt_id)
WebSocket 消息类型
消息以 JSON 文本帧的形式发送,除非另有说明。| 类型 | 描述 |
|---|---|
status | 队列状态更新,包含 queue_remaining 计数 |
notification | 用户友好的状态消息(value 字段包含如 “Executing workflow…” 的文本) |
execution_start | 工作流执行已开始 |
executing | 特定节点正在执行(节点 ID 在 node 字段中) |
progress | 节点内的步骤进度(采样步骤的 value/max) |
progress_state | 扩展进度状态,包含节点元数据(嵌套的 nodes 对象) |
executed | 节点完成并输出结果(图像、视频等在 output 字段中) |
execution_cached | 因输出已缓存而跳过的节点(nodes 数组) |
execution_success | 整个工作流成功完成 |
execution_error | 工作流失败(包含 exception_type、exception_message、traceback) |
execution_interrupted | 工作流被用户取消 |
二进制消息(预览图像)
在图像生成过程中,ComfyUI 会发送包含预览图像的二进制 WebSocket 帧。这些是原始二进制数据(不是 JSON):| 二进制类型 | 值 | 描述 |
|---|---|---|
PREVIEW_IMAGE | 1 | 扩散采样期间的进度预览 |
TEXT | 3 | 节点的文本输出(进度文本) |
PREVIEW_IMAGE_WITH_METADATA | 4 | 带有节点上下文元数据的预览图像 |
- PREVIEW_IMAGE (1)
- TEXT (3)
- PREVIEW_WITH_METADATA (4)
| 偏移 | 大小 | 字段 | 描述 |
|---|---|---|---|
| 0 | 4 字节 | type | 0x00000001 |
| 4 | 4 字节 | image_type | 格式代码(1=JPEG, 2=PNG) |
| 8 | 可变 | image_data | 原始图像字节 |
| 偏移 | 大小 | 字段 | 描述 |
|---|---|---|---|
| 0 | 4 字节 | type | 0x00000003 |
| 4 | 4 字节 | node_id_len | node_id 字符串的长度 |
| 8 | N 字节 | node_id | UTF-8 编码的节点 ID |
| 8+N | 可变 | text | UTF-8 编码的进度文本 |
| 偏移 | 大小 | 字段 | 描述 |
|---|---|---|---|
| 0 | 4 字节 | type | 0x00000004 |
| 4 | 4 字节 | metadata_len | 元数据 JSON 的长度 |
| 8 | N 字节 | metadata | UTF-8 JSON(见下文) |
| 8+N | 可变 | image_data | 原始 JPEG/PNG 字节 |
{
"node_id": "3",
"display_node_id": "3",
"real_node_id": "3",
"prompt_id": "abc-123",
"parent_node_id": null
}
请参阅 OpenAPI 规范 了解每种 JSON 消息类型的完整模式定义。
下载输出
在任务完成后检索生成的文件。# 下载单个输出文件(使用 -L 跟随 302 重定向)
curl -L "$BASE_URL/api/view?filename=output.png&subfolder=&type=output" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-o output.png
async function downloadOutput(
filename: string,
subfolder: string = "",
outputType: string = "output"
): Promise<ArrayBuffer> {
const params = new URLSearchParams({ filename, subfolder, type: outputType });
// 获取重定向 URL
const response = await fetch(`${BASE_URL}/api/view?${params}`, {
headers: { "X-API-Key": API_KEY },
redirect: "manual",
});
if (response.status !== 302) throw new Error(`HTTP ${response.status}`);
const signedUrl = response.headers.get("location")!;
// 从签名 URL 获取文件
const fileResponse = await fetch(signedUrl);
if (!fileResponse.ok) throw new Error(`HTTP ${fileResponse.status}`);
return fileResponse.arrayBuffer();
}
async function saveOutputs(
outputs: Record<string, any>,
outputDir: string = "."
): Promise<void> {
for (const nodeOutputs of Object.values(outputs)) {
for (const key of ["images", "video", "audio"]) {
for (const fileInfo of (nodeOutputs as any)[key] ?? []) {
const data = await downloadOutput(
fileInfo.filename,
fileInfo.subfolder ?? "",
fileInfo.type ?? "output"
);
const path = `${outputDir}/${fileInfo.filename}`;
await writeFile(path, Buffer.from(data));
console.log(`已保存:${path}`);
}
}
}
}
// 下载所有输出
await saveOutputs(outputs, "./my_outputs");
def download_output(filename: str, subfolder: str = "", output_type: str = "output") -> bytes:
"""下载输出文件。
Args:
filename: 文件名
subfolder: 子文件夹路径(通常为空)
output_type: "output" 表示最终输出,"temp" 表示预览
Returns:
文件字节
"""
params = {
"filename": filename,
"subfolder": subfolder,
"type": output_type
}
response = requests.get(
f"{BASE_URL}/api/view",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.content
def save_outputs(outputs: dict, output_dir: str = "."):
"""将任务的所有输出保存到磁盘。
Args:
outputs: 任务的输出字典(node_id -> output_data)
output_dir: 保存文件的目录
"""
import os
os.makedirs(output_dir, exist_ok=True)
for node_id, node_outputs in outputs.items():
for key in ("images", "video", "audio"):
for file_info in node_outputs.get(key, []):
filename = file_info["filename"]
subfolder = file_info.get("subfolder", "")
output_type = file_info.get("type", "output")
data = download_output(filename, subfolder, output_type)
output_path = os.path.join(output_dir, filename)
with open(output_path, "wb") as f:
f.write(data)
print(f"已保存:{output_path}")
# 下载所有输出
save_outputs(outputs, "./my_outputs")
完整端到端示例
以下是一个将所有内容整合在一起的完整示例:const BASE_URL = "https://cloud.comfy.org";
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
function getHeaders(): HeadersInit {
return { "X-API-Key": API_KEY, "Content-Type": "application/json" };
}
async function submitWorkflow(workflow: Record<string, any>): Promise<string> {
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: getHeaders(),
body: JSON.stringify({ prompt: workflow }),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return (await response.json()).prompt_id;
}
async function waitForCompletion(
promptId: string,
timeout: number = 300000
): Promise<Record<string, any>> {
const wsUrl = `wss://cloud.comfy.org/ws?clientId=${crypto.randomUUID()}&token=${API_KEY}`;
const outputs: Record<string, any> = {};
return new Promise((resolve, reject) => {
const ws = new WebSocket(wsUrl);
const timer = setTimeout(() => {
ws.close();
reject(new Error("Job timed out"));
}, timeout);
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.data?.prompt_id !== promptId) return;
const msgType = data.type;
const msgData = data.data ?? {};
if (msgType === "progress") {
console.log(`Progress: ${msgData.value}/${msgData.max}`);
} else if (msgType === "executed" && msgData.output) {
outputs[msgData.node] = msgData.output;
} else if (msgType === "execution_success") {
clearTimeout(timer);
ws.close();
resolve(outputs);
} else if (msgType === "execution_error") {
clearTimeout(timer);
ws.close();
reject(new Error(msgData.exception_message ?? "Unknown error"));
}
};
ws.onerror = (err) => {
clearTimeout(timer);
reject(err);
};
});
}
async function downloadOutputs(
outputs: Record<string, any>,
outputDir: string
): Promise<void> {
for (const nodeOutputs of Object.values(outputs)) {
for (const key of ["images", "video", "audio"]) {
for (const fileInfo of (nodeOutputs as any)[key] ?? []) {
const params = new URLSearchParams({
filename: fileInfo.filename,
subfolder: fileInfo.subfolder ?? "",
type: fileInfo.type ?? "output",
});
// Get redirect URL (don't follow to avoid sending auth to storage)
const response = await fetch(`${BASE_URL}/api/view?${params}`, {
headers: { "X-API-Key": API_KEY },
redirect: "manual",
});
if (response.status !== 302) throw new Error(`HTTP ${response.status}`);
const signedUrl = response.headers.get("location")!;
// Fetch from signed URL without auth headers
const fileResponse = await fetch(signedUrl);
if (!fileResponse.ok) throw new Error(`HTTP ${fileResponse.status}`);
const path = `${outputDir}/${fileInfo.filename}`;
await writeFile(path, Buffer.from(await fileResponse.arrayBuffer()));
console.log(`Downloaded: ${path}`);
}
}
}
}
async function main() {
// 1. Load workflow
const workflow = JSON.parse(await readFile("workflow_api.json", "utf-8"));
// 2. Modify workflow parameters
workflow["3"].inputs.seed = 42;
workflow["6"].inputs.text = "a beautiful sunset over mountains";
// 3. Submit workflow
const promptId = await submitWorkflow(workflow);
console.log(`Job submitted: ${promptId}`);
// 4. Wait for completion with progress
const outputs = await waitForCompletion(promptId);
console.log(`Job completed! Found ${Object.keys(outputs).length} output nodes`);
// 5. Download outputs
await downloadOutputs(outputs, "./outputs");
console.log("Done!");
}
main();
import os
import requests
import json
import asyncio
import aiohttp
import uuid
BASE_URL = "https://cloud.comfy.org"
API_KEY = os.environ["COMFY_CLOUD_API_KEY"]
def get_headers():
return {"X-API-Key": API_KEY, "Content-Type": "application/json"}
def upload_image(file_path: str) -> dict:
"""Upload an image and return the reference for use in workflows."""
with open(file_path, "rb") as f:
response = requests.post(
f"{BASE_URL}/api/upload/image",
headers={"X-API-Key": API_KEY},
files={"image": f},
data={"type": "input", "overwrite": "true"}
)
response.raise_for_status()
return response.json()
def submit_workflow(workflow: dict) -> str:
"""Submit workflow and return prompt_id."""
response = requests.post(
f"{BASE_URL}/api/prompt",
headers=get_headers(),
json={"prompt": workflow}
)
response.raise_for_status()
return response.json()["prompt_id"]
async def wait_for_completion(prompt_id: str, timeout: float = 300.0) -> dict:
"""Wait for job completion via WebSocket."""
ws_url = BASE_URL.replace("https://", "wss://") + f"/ws?clientId={uuid.uuid4()}&token={API_KEY}"
outputs = {}
async with aiohttp.ClientSession() as session:
async with session.ws_connect(ws_url) as ws:
start = asyncio.get_event_loop().time()
async for msg in ws:
if asyncio.get_event_loop().time() - start > timeout:
raise TimeoutError("Job timed out")
if msg.type != aiohttp.WSMsgType.TEXT:
continue
data = json.loads(msg.data)
if data.get("data", {}).get("prompt_id") != prompt_id:
continue
msg_type = data.get("type")
msg_data = data.get("data", {})
if msg_type == "progress":
print(f"Progress: {msg_data.get('value')}/{msg_data.get('max')}")
elif msg_type == "executed":
if output := msg_data.get("output"):
outputs[msg_data["node"]] = output
elif msg_type == "execution_success":
return outputs
elif msg_type == "execution_error":
raise RuntimeError(msg_data.get("exception_message", "Unknown error"))
return outputs
def download_outputs(outputs: dict, output_dir: str):
"""Download all output files."""
os.makedirs(output_dir, exist_ok=True)
for node_outputs in outputs.values():
for key in ["images", "video", "audio"]:
for file_info in node_outputs.get(key, []):
params = {
"filename": file_info["filename"],
"subfolder": file_info.get("subfolder", ""),
"type": file_info.get("type", "output")
}
response = requests.get(f"{BASE_URL}/api/view", headers=get_headers(), params=params)
response.raise_for_status()
path = os.path.join(output_dir, file_info["filename"])
with open(path, "wb") as f:
f.write(response.content)
print(f"Downloaded: {path}")
async def main():
# 1. Load workflow
with open("workflow_api.json") as f:
workflow = json.load(f)
# 2. Optionally upload input images
# image_ref = upload_image("input.png")
# workflow["1"]["inputs"]["image"] = image_ref["name"]
# 3. Modify workflow parameters
workflow["3"]["inputs"]["seed"] = 42
workflow["6"]["inputs"]["text"] = "a beautiful sunset over mountains"
# 4. Submit workflow
prompt_id = submit_workflow(workflow)
print(f"Job submitted: {prompt_id}")
# 5. Wait for completion with progress
outputs = await wait_for_completion(prompt_id)
print(f"Job completed! Found {len(outputs)} output nodes")
# 6. Download outputs
download_outputs(outputs, "./outputs")
print("Done!")
if __name__ == "__main__":
asyncio.run(main())
队列管理
获取队列状态
curl -X GET "$BASE_URL/api/queue" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY"
async function getQueue(): Promise<{
queue_running: any[];
queue_pending: any[];
}> {
const response = await fetch(`${BASE_URL}/api/queue`, {
headers: getHeaders(),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json();
}
const queue = await getQueue();
console.log(`Running: ${queue.queue_running.length}`);
console.log(`Pending: ${queue.queue_pending.length}`);
def get_queue():
"""Get current queue status."""
response = requests.get(
f"{BASE_URL}/api/queue",
headers=get_headers()
)
response.raise_for_status()
return response.json()
queue = get_queue()
print(f"Running: {len(queue.get('queue_running', []))}")
print(f"Pending: {len(queue.get('queue_pending', []))}")
取消任务
curl -X POST "$BASE_URL/api/queue" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-H "Content-Type: application/json" \
-d '{"delete": ["PROMPT_ID_HERE"]}'
async function cancelJob(promptId: string): Promise<void> {
const response = await fetch(`${BASE_URL}/api/queue`, {
method: "POST",
headers: getHeaders(),
body: JSON.stringify({ delete: [promptId] }),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
}
def cancel_job(prompt_id: str):
"""Cancel a pending or running job."""
response = requests.post(
f"{BASE_URL}/api/queue",
headers=get_headers(),
json={"delete": [prompt_id]}
)
response.raise_for_status()
return response.json()
中断当前执行
curl -X POST "$BASE_URL/api/interrupt" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY"
async function interrupt(): Promise<void> {
const response = await fetch(`${BASE_URL}/api/interrupt`, {
method: "POST",
headers: getHeaders(),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
}
def interrupt():
"""Interrupt the currently running job."""
response = requests.post(
f"{BASE_URL}/api/interrupt",
headers=get_headers()
)
response.raise_for_status()
错误处理
HTTP 错误
REST API 端点返回标准 HTTP 状态码:| 状态码 | 描述 |
|---|---|
400 | 无效请求(错误的工作流、缺少字段) |
401 | 未授权(无效或缺少 API 密钥) |
402 | 积分不足 |
429 | 订阅未激活 |
500 | 内部服务器错误 |
执行错误
在工作流执行期间,错误通过execution_error WebSocket 消息传递。exception_type 字段标识错误类别:
| 异常类型 | 描述 |
|---|---|
ValidationError | 无效的工作流或输入 |
ModelDownloadError | 所需模型不可用或下载失败 |
ImageDownloadError | 从 URL 下载输入图像失败 |
OOMError | GPU 内存不足 |
InsufficientFundsError | 账户积分不足(用于合作伙伴节点) |
InactiveSubscriptionError | 订阅未激活 |