실험적 API: 이 API는 실험적인 것으로, 변경될 수 있습니다. 엔드포인트, 요청/응답 형식 및 동작은 사전 통지 없이 수정될 수 있습니다.
Comfy Cloud API
Comfy Cloud API는 Comfy Cloud 인프라에서 워크플로우를 실행하기 위한 프로그래밍 방식 접근을 제공합니다. 이 API는 로컬 ComfyUI의 API와 호환되므로 기존 통합을 쉽게 마이그레이션할 수 있습니다.구독 필요: API 접근은 스탠다드, 크리에이터, 프로 계층에서 이용 가능합니다. 무료 계층에는 API 접근이 포함되지 않습니다. 자세한 내용은 가격 정책을 참조하십시오.
크레딧 및 사용법
API 요청은 Comfy Cloud 웹 UI와 동일한 월별 크레딧 할당량을 사용합니다 — 별도의 API 크레딧 풀은 없습니다. 각 계층에 포함된 크레딧, 추가 옵션 및 워크플로우당 런타임 제한은 UI 작업과 정확히 동일하게 API 작업에도 적용됩니다. 가격 페이지에서 스탠다드, 크리에이터 및 프로 계층의 월별 크레딧 수치를 확인하십시오. 한 달 중간에 크레딧이 부족해지면 계정 대시보드에서 추가 구매가 가능합니다.기본 URL
https://cloud.comfy.org
인증
모든 API 요청에는X-API-Key 헤더를 통해 전달되는 API 키가 필요합니다.
API 키 얻기
API 키 생성 및 관리에 대한 지침은 API 키 얻기를 참조하십시오.API 키 사용하기
모든 요청 시X-API-Key 헤더에 API 키를 전달하십시오:
curl -X GET "https://cloud.comfy.org/api/user" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY"
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
const response = await fetch("https://cloud.comfy.org/api/user", {
headers: { "X-API-Key": API_KEY },
});
const user = await response.json();
import os
import requests
API_KEY = os.environ["COMFY_CLOUD_API_KEY"]
headers = {"X-API-Key": API_KEY}
response = requests.get(
"https://cloud.comfy.org/api/user",
headers=headers
)
핵심 개념
워크플로우
ComfyUI 워크플로우는 노드 그래프를 설명하는 JSON 객체입니다. API는 ComfyUI 프론트엔드의 “저장(API 형식)” 옵션으로 생성된 “API 형식”(노드 ID를 키로 하며 class_type, inputs 등이 포함된) 워크플로우를 수락합니다.작업
워크플로우를 제출하면 작업이 생성됩니다. 작업은 비동기적으로 실행됩니다:POST /api/prompt을 통해 워크플로우 제출prompt_id(작업 ID) 받기- WebSocket을 통해 진행 상태 모니터링 또는 상태 조회
- 완료 시 출력물 가져오기
병렬 실행(동시 작업)
API 사용자는 이전 작업이 완료될 때까지 기다릴 필요 없이 여러 워크플로우를 동시에 제출할 수 있습니다. 단순히 여러POST /api/prompt 요청을 보내기만 하면 되며, 특별한 헤더나 매개변수는 필요하지 않습니다. 디스패처는 구독 계층의 제한 내에서 병렬로 실행합니다.
| 구독 계층 | 동시 작업 |
|---|---|
| 스탠다드 | 1 |
| 크리에이터 | 3 |
| 프로 | 5 |
현재 병렬 실행은 API를 통해서만 이용 가능합니다. 구독 상세 정보는 가격 정책을 참조하십시오.
예제: 병렬로 여러 작업 제출하기
import os
import json
import asyncio
import aiohttp
BASE_URL = "https://cloud.comfy.org"
API_KEY = os.environ["COMFY_CLOUD_API_KEY"]
async def submit_workflow(session, workflow):
"""단일 워크플로우를 제출하고 prompt_id 반환"""
async with session.post(
f"{BASE_URL}/api/prompt",
headers={"X-API-Key": API_KEY, "Content-Type": "application/json"},
json={"prompt": workflow},
) as response:
result = await response.json()
return result["prompt_id"]
async def main():
with open("workflow_api.json") as f:
base_workflow = json.load(f)
# 시드를 변경해 변형 생성
workflows = []
for seed in [42, 123, 456]:
workflow = json.loads(json.dumps(base_workflow))
workflow["3"]["inputs"]["seed"] = seed
workflows.append(workflow)
# 모든 워크플로우를 동시에 제출
async with aiohttp.ClientSession() as session:
prompt_ids = await asyncio.gather(
*[submit_workflow(session, wf) for wf in workflows]
)
for pid in prompt_ids:
print(f"작업 제출됨: {pid}")
# 각 작업 모니터링을 위해 polling 또는 WebSocket 사용...
asyncio.run(main())
const BASE_URL = "https://cloud.comfy.org";
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
async function submitWorkflow(
workflow: Record<string, any>
): Promise<string> {
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: { "X-API-Key": API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ prompt: workflow }),
});
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return (await response.json()).prompt_id;
}
async function main() {
const base = JSON.parse(
await Deno.readTextFile("workflow_api.json")
);
// 시드를 변경해 변형 생성
const seeds = [42, 123, 456];
const workflows = seeds.map((seed) => {
const wf = structuredClone(base);
wf["3"].inputs.seed = seed;
return wf;
});
// 모든 워크플로우를 동시에 제출
const promptIds = await Promise.all(
workflows.map((wf) => submitWorkflow(wf))
);
for (const pid of promptIds) {
console.log(`작업 제출됨: ${pid}`);
}
// 각 작업 모니터링을 위해 polling 또는 WebSocket 사용...
}
main();
출력물
생성된 콘텐츠(이미지, 동영상, 오디오)는 클라우드 스토리지에 저장됩니다. 출력 파일은/api/view 엔드포인트를 통해 다운로드할 수 있으며, 이 엔드포인트는 임시 서명된 URL로 302 리다이렉트를 반환합니다.
빠른 시작
워크플로우를 제출하고 진행 상태를 모니터링하며 출력물을 가져오는 방법을 보여주는 전체 예제입니다:1단계: 워크플로우 제출
curl -X POST "https://cloud.comfy.org/api/prompt" \
-H "X-API-Key: $COMFY_CLOUD_API_KEY" \
-H "Content-Type: application/json" \
-d '{"prompt": '"$(cat workflow_api.json)"'"}'
const BASE_URL = "https://cloud.comfy.org";
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
// ComfyUI에서 API 형식으로 내보낸 워크플로우 로드
const workflow = JSON.parse(await Deno.readTextFile("workflow_api.json"));
// 워크플로우 제출
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: { "X-API-Key": API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ prompt: workflow }),
});
const result = await response.json();
const promptId = result.prompt_id;
console.log(`작업 제출됨: ${promptId}`);
import os
import requests
import json
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"}
# ComfyUI에서 API 형식으로 내보낸 워크플로우 로드
with open("workflow_api.json") as f:
workflow = json.load(f)
# 워크플로우 제출
response = requests.post(
f"{BASE_URL}/api/prompt",
headers=get_headers(),
json={"prompt": workflow}
)
result = response.json()
prompt_id = result["prompt_id"]
print(f"작업 제출됨: {prompt_id}")
2단계: 작업 진행 상태 모니터링
폴링 또는 WebSocket을 사용해 작업 완료 여부를 모니터링할 수 있습니다.옵션 A: 폴링(간단함)
작업 상태 값: 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("작업 완료!")
옵션 B: WebSocket(실시간 진행 상태)
실시간 진행 상태 업데이트 및 출력 메타데이터 수집을 위해: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 참조를 참조해 세부 메시지 유형 및 바이너리 미리보기 이미지 처리 방법을 확인하십시오.
3단계: 출력물 다운로드
작업이 완료되면 생성된 파일을 다운로드하세요. WebSocket에서 반환되거나 히스토리 엔드포인트를 통해 이용 가능한outputs 객체는 노드 ID별로 정리된 출력 데이터를 포함합니다. 각 노드의 출력에는 images, video 또는 audio 배열과 파일 메타데이터가 포함될 수 있습니다.
출력물 구조 예시:
{
"9": {
"images": [
{
"filename": "ComfyUI_00001_.png",
"subfolder": "",
"type": "output"
}
]
}
}
"9" 예시)는 워크플로우의 SaveImage 또는 기타 출력 노드에 해당합니다. 워크플로우 JSON 파일을 열어 class_type이 SaveImage, VHS_VideoCombine 등의 노드를 찾으면 이 ID를 확인할 수 있습니다.
# 단일 출력 파일 다운로드하기 (302 리디렉션은 -L 옵션으로 따라가기)
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: 작업에서 가져온 출력 사전 (노드 ID -> 출력 데이터)
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")
/api/view 엔드포인트는 임시 서명된 URL로 302 리다이렉트를 반환합니다. HTTP 클라이언트는 리다이렉트를 따라야만 파일을 다운로드할 수 있습니다.전체 예제
세 단계를 모두 결합한 전체 엔드투엔드 예제입니다:import { readFile, writeFile } from "fs/promises";
const BASE_URL = "https://cloud.comfy.org";
const API_KEY = process.env.COMFY_CLOUD_API_KEY!;
async function main() {
// 1. 워크플로우 로드 및 수정
const workflow = JSON.parse(await readFile("workflow_api.json", "utf-8"));
workflow["3"].inputs.seed = 42;
workflow["6"].inputs.text = "아름다운 일몰";
// 2. 워크플로우 제출
const response = await fetch(`${BASE_URL}/api/prompt`, {
method: "POST",
headers: { "X-API-Key": API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ prompt: workflow }),
});
const { prompt_id } = await response.json();
console.log(`작업 제출됨: ${prompt_id}`);
// 3. 완료 여부 휠링
while (true) {
const statusRes = await fetch(`${BASE_URL}/api/job/${prompt_id}/status`, {
headers: { "X-API-Key": API_KEY },
});
const { status } = await statusRes.json();
if (status === "completed") break;
if (["failed", "cancelled"].includes(status)) {
throw new Error(`작업 ${status}`);
}
await new Promise((resolve) => setTimeout(resolve, 2000));
}
// 4. 작업 세부 정보 엔드포인트를 통해 출력물 가져오기
const jobRes = await fetch(`${BASE_URL}/api/jobs/${prompt_id}`, {
headers: { "X-API-Key": API_KEY },
});
const job = await jobRes.json();
const outputs = job.outputs;
// 5. 출력 파일 다운로드
for (const nodeOutputs of Object.values(outputs)) {
for (const fileInfo of (nodeOutputs as any).images ?? []) {
const params = new URLSearchParams({
filename: fileInfo.filename,
subfolder: fileInfo.subfolder ?? "",
type: "output",
});
const viewRes = await fetch(`${BASE_URL}/api/view?${params}`, {
headers: { "X-API-Key": API_KEY },
redirect: "manual",
});
const signedUrl = viewRes.headers.get("location")!;
const fileRes = await fetch(signedUrl);
await writeFile(`./${fileInfo.filename}`, Buffer.from(await fileRes.arrayBuffer()));
console.log(`다운로드 완료: ${fileInfo.filename}`);
}
}
}
main();
import os
import requests
import json
import time
BASE_URL = "https://cloud.comfy.org"
API_KEY = os.environ["COMFY_CLOUD_API_KEY"]
def main():
# 1. 워크플로우 로드 및 수정
with open("workflow_api.json") as f:
workflow = json.load(f)
workflow["3"]["inputs"]["seed"] = 42
workflow["6"]["inputs"]["text"] = "아름다운 일몰"
# 2. 워크플로우 제출
response = requests.post(
f"{BASE_URL}/api/prompt",
headers={"X-API-Key": API_KEY, "Content-Type": "application/json"},
json={"prompt": workflow}
)
prompt_id = response.json()["prompt_id"]
print(f"작업 제출됨: {prompt_id}")
# 3. 완료 여부 휠링
while True:
status_res = requests.get(
f"{BASE_URL}/api/job/{prompt_id}/status",
headers={"X-API-Key": API_KEY}
)
status = status_res.json()["status"]
if status == "completed":
break
if status in ("failed", "cancelled"):
raise RuntimeError(f"작업 {status}")
time.sleep(2)
# 4. 작업 세부 정보 엔드포인트를 통해 출력물 가져오기
job_res = requests.get(
f"{BASE_URL}/api/jobs/{prompt_id}",
headers={"X-API-Key": API_KEY}
)
job = job_res.json()
outputs = job["outputs"]
# 5. 출력 파일 다운로드
for node_outputs in outputs.values():
for file_info in node_outputs.get("images", []):
params = {
"filename": file_info["filename"],
"subfolder": file_info.get("subfolder", ""),
"type": "output"
}
view_res = requests.get(
f"{BASE_URL}/api/view",
headers={"X-API-Key": API_KEY},
params=params
)
with open(file_info["filename"], "wb") as f:
f.write(view_res.content)
print(f"다운로드 완료: {file_info['filename']}")
if __name__ == "__main__":
main()
사용 가능한 엔드포인트
다음 단계
위의 빠른 시작은 워크플로우 제출과 결과 가져오기에 대한 기본 사항을 다룹니다. 더 고급 사용 사례를 위해서는 클라우드 API 참조를 참고하십시오:- 입력 파일 업로드 - 워크플로우에 외부 입력이 필요한 경우 이미지, 마스크 또는 기타 사용자 제공 콘텐츠 업로드
- 워크플로우 입력 수정 - 제출 전에 프롬프트, 시드 또는 노드 설정과 같은 워크플로우 파라미터를 동적으로 변경
- 파트너 노드 사용 - 추가 API 키 구성이 필요한 외부 AI 서비스(Flux Pro, Ideogram 등) 호출
- 대기열 관리 - 대기열 상태 모니터링, 작업 취소 또는 실행 중인 작업 중단
- 오류 처리 - HTTP 오류, 실행 실패 처리 및 예외 유형 이해
- OpenAPI 규격 - 코드 생성을 위한 기계 판독 가능한 API 규격