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):
    """Submit a single workflow and return the 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)

    # Create variations by changing the seed
    workflows = []
    for seed in [42, 123, 456]:
        workflow = json.loads(json.dumps(base_workflow))
        workflow["3"]["inputs"]["seed"] = seed
        workflows.append(workflow)

    # Submit all workflows concurrently
    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}")

    # Poll or use WebSocket to monitor each job...

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

  // Create variations by changing the seed
  const seeds = [42, 123, 456];
  const workflows = seeds.map((seed) => {
    const wf = structuredClone(base);
    wf["3"].inputs.seed = seed;
    return wf;
  });

  // Submit all workflows concurrently
  const promptIds = await Promise.all(
    workflows.map((wf) => submitWorkflow(wf))
  );

  for (const pid of promptIds) {
    console.log(`Job submitted: ${pid}`);
  }

  // Poll or use WebSocket to monitor each job...
}

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!;

// Load your workflow (exported from ComfyUI in API format)
const workflow = JSON.parse(await Deno.readTextFile("workflow_api.json"));

// Submit the workflow
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"}

# Load your workflow (exported from ComfyUI in API format)
with open("workflow_api.json") as f:
    workflow = json.load(f)

# Submit the workflow
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}")

# Poll for job completion
curl -X GET "$BASE_URL/api/job/{prompt_id}/status" \
  -H "X-API-Key: $COMFY_CLOUD_API_KEY"

# Response examples:
# {"status": "pending"}      - Job is queued
# {"status": "in_progress"}  - Job is currently running
# {"status": "completed"}    - Job finished successfully
# {"status": "failed"}       - Job encountered an error
# {"status": "cancelled"}    - Job was 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(`Job failed with status: ${status}`);
    }

    await new Promise((resolve) => setTimeout(resolve, pollInterval));
  }

  throw new Error(`Job ${promptId} did not complete within ${timeout}s`);
}

await pollForCompletion(promptId);
console.log("Job completed!");

def get_job_status(prompt_id: str) -> str:
    """Get the current status of a job."""
    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:
    """Poll until job completes or times out."""
    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"Job failed with status: {status}")

        time.sleep(poll_interval)

    raise TimeoutError(f"Job {prompt_id} did not complete within {timeout}s")

poll_for_completion(prompt_id)
print("Job completed!")

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(`Job did not complete within ${timeout / 1000}s`));
    }, timeout);

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      const msgType = data.type;
      const msgData = data.data ?? {};

      // Filter to our job
      if (msgData.prompt_id !== promptId) return;

      if (msgType === "executing") {
        const node = msgData.node;
        if (node) {
          console.log(`Executing node: ${node}`);
        } else {
          console.log("Execution complete");
        }
      } else 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") {
        console.log("Job completed successfully!");
        clearTimeout(timer);
        ws.close();
        resolve(outputs);
      } else if (msgType === "execution_error") {
        const errorMsg = msgData.exception_message ?? "Unknown error";
        clearTimeout(timer);
        ws.close();
        reject(new Error(`Execution error: ${errorMsg}`));
      }
    };

    ws.onerror = (err) => {
      clearTimeout(timer);
      reject(err);
    };
  });
}

// Wait for completion and collect outputs
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:
    """Connect to WebSocket and listen for job completion.

    Returns:
        Final outputs from the job
    """
    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", {})

                        # Filter to our job
                        if msg_data.get("prompt_id") != prompt_id:
                            continue

                        if msg_type == "executing":
                            node = msg_data.get("node")
                            if node:
                                print(f"Executing node: {node}")

                        elif msg_type == "progress":
                            value = msg_data.get("value", 0)
                            max_val = msg_data.get("max", 100)
                            print(f"Progress: {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("Job completed successfully!")
                            return outputs

                        elif msg_type == "execution_error":
                            error_msg = msg_data.get("exception_message", "Unknown error")
                            raise RuntimeError(f"Execution error: {error_msg}")

                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        raise RuntimeError(f"WebSocket error: {ws.exception()}")

            try:
                return await asyncio.wait_for(receive_messages(), timeout=timeout)
            except asyncio.TimeoutError:
                raise TimeoutError(f"Job did not complete within {timeout}s")

# Wait for completion and collect outputs
outputs = await listen_for_completion(prompt_id)

# Download a single output file (follow 302 redirect with -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 });
  // Get the redirect 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")!;

  // Fetch from signed 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(`Saved: ${path}`);
      }
    }
  }
}

// Download all outputs
await saveOutputs(outputs, "./my_outputs");

def download_output(filename: str, subfolder: str = "", output_type: str = "output") -> bytes:
    """Download an output file.

    Args:
        filename: Name of the file
        subfolder: Subfolder path (usually empty)
        output_type: "output" for final outputs, "temp" for previews

    Returns:
        File bytes
    """
    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 = "."):
    """Save all outputs from a job to disk.

    Args:
        outputs: Outputs dict from job (node_id -> output_data)
        output_dir: Directory to save files to
    """
    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"Saved: {output_path}")

# Download all outputs
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. Load and modify workflow
  const workflow = JSON.parse(await readFile("workflow_api.json", "utf-8"));
  workflow["3"].inputs.seed = 42;
  workflow["6"].inputs.text = "a beautiful sunset";

  // 2. Submit workflow
  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(`Job submitted: ${prompt_id}`);

  // 3. Poll for completion
  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(`Job ${status}`);
    }
    await new Promise((resolve) => setTimeout(resolve, 2000));
  }

  // 4. Get outputs via job detail endpoint
  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. Download output files
  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(`Downloaded: ${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. Load and modify workflow
    with open("workflow_api.json") as f:
        workflow = json.load(f)

    workflow["3"]["inputs"]["seed"] = 42
    workflow["6"]["inputs"]["text"] = "a beautiful sunset"

    # 2. Submit workflow
    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"Job submitted: {prompt_id}")

    # 3. Poll for completion
    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"Job {status}")
        time.sleep(2)

    # 4. Get outputs via job detail endpoint
    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. Download output files
    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"Downloaded: {file_info['filename']}")

if __name__ == "__main__":
    main()