メインコンテンツへスキップ
実験的 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)」オプションによって生成されます。

​
ジョブ

ワークフローを送信すると、ジョブが作成されます。ジョブは非同期で実行されます：
  1. POST /api/prompt を介してワークフローを送信
  2. prompt_id（ジョブ ID）を受け取る
  3. WebSocket を介して進捗を監視するか、ステータスをポーリング
  4. 完了時に出力を取得

​
並列実行（同時ジョブ）

API ユーザーは、前のジョブの完了を待たずに、複数のワークフローを同時に送信できます。複数の POST /api/prompt リクエストを送信するだけです。特別なヘッダーやパラメーターは必要ありません。ディスパッチャーは、サブスクリプションティアの制限まで、それらを並列で実行します。
サブスクリプションティア同時ジョブ数
Standard1
Creator3
Pro5
同時実行制限を超えて送信されたジョブは、通常通りキューに入れられ、スロットが空くと自動的に実行されます。
並列実行は現在、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"
      }
    ]
  }
}
ノード ID（この例では "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()

​
利用可能なエンドポイント

カテゴリ説明
ワークフローワークフローの送信、ステータスの確認
ジョブジョブのステータスとキューの監視
入力画像、マスク、その他の入力のアップロード
出力生成されたコンテンツのダウンロード
WebSocketリアルタイムの進捗更新
オブジェクト情報利用可能なノードとその定義

​
次のステップ

上記のクイックスタートでは、ワークフローの送信と結果の取得の基礎をカバーしています。より高度なユースケースについては、Cloud API リファレンスを参照してください：
  • 入力ファイルのアップロード - 外部入力を必要とするワークフローのために、画像、マスク、またはその他のユーザー提供コンテンツをアップロード
  • ワークフロー入力の修正 - 送信前にプロンプト、シード、またはノード設定などのワークフローパラメーターを動的に変更
  • パートナーノードの使用 - 追加の API キー設定を必要とする外部 AI サービス（Flux Pro、Ideogram など）を呼び出す
  • キュー管理 - キューのステータスの監視、ジョブのキャンセル、または実行中の実行の中断
  • エラー処理 - HTTP エラー、実行失敗の処理、および例外タイプの理解
追加リソース：