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
)

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();

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}")

# 轮询任务完成状态
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("任务已完成！")

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)

# 下载单个输出文件（使用 -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")

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()