実験的 API: この API は実験的であり、変更される可能性があります。エンドポイント、リクエスト/レスポンス形式、および動作は予告なく変更される場合があります。
Comfy Cloud API
Comfy Cloud API は、Comfy Cloud インフラストラクチャ上でワークフローを実行するためのプログラムによるアクセスを提供します。この API はローカルの ComfyUI の API と互換性があり、既存の統合を簡単に移行できます。サブスクリプションが必要: API へのアクセスは Standard、Creator および Pro ティアで利用可能です。Free ティアには API アクセスは含まれません。詳細は料金プランをご覧ください。
クレジットと使用量
API リクエストは、Comfy Cloud のウェブ UI と同じ月間クレジット枠から消費されます。API 専用のクレジットプールはありません。各ティアに含まれるクレジット、追加購入オプション、ワークフロー単位のランタイム上限は、API ジョブにも UI ジョブと同じ条件で適用されます。Creator および Pro ティアの月間クレジット数については、料金プランを参照してください。月の途中でクレジットが不足した場合、アカウントダッシュボードから追加購入できます。ベース 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 は API 形式 のワークフローを受け付けます(ノード ID をキーとし、class_type、inputs などを含む)。この形式は、ComfyUI フロントエンドの「Export Workflow (API)」オプションによって生成されます。ジョブ
ワークフローを送信すると、ジョブが作成されます。ジョブは非同期で実行されます:POST /api/promptを介してワークフローを送信prompt_id(ジョブ ID)を受け取る- WebSocket を介して進捗を監視するか、ステータスをポーリング
- 完了時に出力を取得
並列実行(同時ジョブ)
API ユーザーは、前のジョブの完了を待たずに、複数のワークフローを同時に送信できます。複数のPOST /api/prompt リクエストを送信するだけです。特別なヘッダーやパラメーターは必要ありません。ディスパッチャーは、サブスクリプションティアの制限まで、それらを並列で実行します。
| サブスクリプションティア | 同時ジョブ数 |
|---|---|
| Standard | 1 |
| Creator | 3 |
| Pro | 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"Job submitted: {pid}")
# 各ジョブをポーリングまたは 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(`Job submitted: ${pid}`);
}
// 各ジョブをポーリングまたは 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(`Job submitted: {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"Job submitted: {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} 秒以内に完了しませんでした`);
}
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} 秒以内に完了しませんでした")
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} 秒以内に完了しませんでした`));
}, 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} 秒以内に完了しませんでした")
# 完了を待機し、出力を収集
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 または他の出力ノードに対応します。これらの ID は、ワークフロー JSON ファイルを開き、class_type が SaveImage、VHS_VideoCombine などのノードを探すことで見つけることができます。
# 単一の出力ファイルをダウンロード(-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")
/api/view エンドポイントは、一時署名付き URL への 302 リダイレクトを返します。ファイルをダウンロードするには、HTTP クライアントがリダイレクトに従う必要があります。完全な例
以下は、3 つのすべてのステップを組み合わせた完全なエンドツーエンドの例です: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 = "a beautiful sunset";
// 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"] = "a beautiful sunset"
# 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()
利用可能なエンドポイント
次のステップ
上記のクイックスタートでは、ワークフローの送信と結果の取得の基礎をカバーしています。より高度なユースケースについては、Cloud API リファレンスを参照してください:- 入力ファイルのアップロード - 外部入力を必要とするワークフローのために、画像、マスク、またはその他のユーザー提供コンテンツをアップロード
- ワークフロー入力の修正 - 送信前にプロンプト、シード、またはノード設定などのワークフローパラメーターを動的に変更
- パートナーノードの使用 - 追加の API キー設定を必要とする外部 AI サービス(Flux Pro、Ideogram など)を呼び出す
- キュー管理 - キューのステータスの監視、ジョブのキャンセル、または実行中の実行の中断
- エラー処理 - HTTP エラー、実行失敗の処理、および例外タイプの理解
- OpenAPI 仕様 - コード生成用の機械可読 API 仕様