Add Comfy Cloud API nodes

Amp-Thread-ID: https://ampcode.com/threads/T-019fd9e0-653f-74cb-a744-9d35f9264778
Co-authored-by: Amp <amp@ampcode.com>

comfy_api_nodes/apis/comfy_cloud.py

@@ -0,0 +1,120 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+
+ComfyCloudWorkflow = Literal[
+    "text-to-image",
+    "text-to-video",
+    "image-to-video",
+    "image-edit",
+    "image.ideogram-4-design.v1",
+    "image.krea-2-creative-image.v1",
+    "image.mage-flow-image.v1",
+    "image.flux-2-reference-edit.v1",
+    "image.qwen-image-edit-2511.v1",
+    "image.seedvr2-image-upscale.v1",
+    "video.minimax-h3-text-sound.v1",
+    "video.minimax-h3-image-sound.v1",
+    "video.ltx-2-3-image-audio-performance.v1",
+    "video.ltx-2-3-first-last-frame.v1",
+    "video.wan-2-2-14b-first-last-frame.v1",
+    "video.scail-2-character-replacement.v1",
+    "audio.ace-step-1-5-xl-turbo.v1",
+    "audio.stable-audio-3-medium.v1",
+    "audio.chatterbox-multilingual-voice-clone.v1",
+    "audio.chatterbox-dialogue.v1",
+    "audio.chatterbox-voice-conversion.v1",
+    "audio.melbandroformer-stem-separation.v1",
+    "3d.triposplat-image-to-gaussian-splat.v1",
+    "3d.hunyuan3d-2-1-image-to-3d.v1",
+    "3d.hunyuan3d-multiview-to-3d.v1",
+    "3d.moge-2-photo-to-textured-mesh.v1",
+    "3d.moge-2-panorama-to-3d-scene.v1",
+]
+
+
+class ComfyCloudWorkflowInputs(BaseModel):
+    prompt: str | None = Field(None)
+    image_url: str | None = Field(None)
+    assets: dict[str, "ComfyCloudAssetInput"] | None = Field(None)
+    audio_url: str | None = Field(None)
+    first_frame_url: str | None = Field(None)
+    last_frame_url: str | None = Field(None)
+    reference_character_url: str | None = Field(None)
+    driving_video_url: str | None = Field(None)
+    instruction: str | None = Field(None)
+    prompt_enhance: bool | None = Field(None)
+    enhance_prompt: bool | None = Field(None)
+    negative_prompt: str | None = Field(None)
+    aspect_ratio: str | None = Field(None)
+    duration_seconds: float | None = Field(None)
+    guidance: float | None = Field(None)
+    quality_mode: str | None = Field(None)
+    seed: int | None = Field(None, ge=0, le=0xFFFFFFFFFFFFFFFF)
+    scale: str | None = Field(None)
+    scene_prompt: str | None = Field(None)
+    driving_subject: str | None = Field(None)
+    reference_subject: str | None = Field(None)
+    style_prompt: str | None = Field(None)
+    lyrics: str | None = Field(None)
+    bpm: int | None = Field(None)
+    time_signature: str | None = Field(None)
+    language: str | None = Field(None)
+    key: str | None = Field(None)
+    expand_prompt: bool | None = Field(None)
+    category: str | None = Field(None)
+    text: str | None = Field(None)
+    exaggeration: float | None = Field(None)
+    cfg_weight: float | None = Field(None)
+    temperature: float | None = Field(None)
+    script: str | None = Field(None)
+    remove_background: bool | None = Field(None)
+    gaussian_count: int | None = Field(None)
+    fov_degrees: float | None = Field(None)
+    detail: int | None = Field(None)
+    mesh_decimation: int | None = Field(None)
+    gap_threshold: float | None = Field(None)
+    texture: bool | None = Field(None)
+    split_resolution: int | None = Field(None)
+    merge_resolution: int | None = Field(None)
+
+
+class ComfyCloudAssetInput(BaseModel):
+    type: Literal["IMAGE", "VIDEO", "AUDIO"] = Field(...)
+    url: str = Field(...)
+
+
+class ComfyCloudGenerateRequest(BaseModel):
+    workflow: ComfyCloudWorkflow = Field(...)
+    inputs: ComfyCloudWorkflowInputs = Field(...)
+
+
+class ComfyCloudGenerateResponse(BaseModel):
+    task_id: str = Field(..., min_length=1)
+    status: str = Field(...)
+    polling_url: str = Field(...)
+    cancel_url: str = Field(...)
+
+    @field_validator("task_id")
+    @classmethod
+    def task_id_must_not_be_blank(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("task_id must not be blank")
+        return value
+
+
+class ComfyCloudStatusResponse(BaseModel):
+    task_id: str = Field(..., min_length=1)
+    status: str = Field(...)
+    progress: float | None = Field(None, ge=0, le=100)
+    output_url: str | None = Field(None)
+    output_urls: dict[str, str] | None = Field(None)
+    error: str | None = Field(None)
+
+    @field_validator("task_id")
+    @classmethod
+    def task_id_must_not_be_blank(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("task_id must not be blank")
+        return value

comfy_api_nodes/util/__init__.py

@@ -32,6 +32,7 @@ from .conversions import (
 )
 from .download_helpers import (
     download_url_as_bytesio,
+    download_url_to_audio_input,
     download_url_to_bytesio,
     download_url_to_file_3d,
     download_url_to_image_tensor,
@@ -76,6 +77,7 @@ __all__ = [
     "upload_video_to_comfyapi",
     # Download helpers
     "download_url_as_bytesio",
+    "download_url_to_audio_input",
     "download_url_to_bytesio",
     "download_url_to_file_3d",
     "download_url_to_image_tensor",

comfy_api_nodes/util/download_helpers.py

@@ -11,7 +11,7 @@ import torch
 from aiohttp.client_exceptions import ClientError, ContentTypeError
 
 from comfy_api.latest import IO as COMFY_IO
-from comfy_api.latest import InputImpl, Types
+from comfy_api.latest import Input, InputImpl, Types
 from folder_paths import get_output_directory
 
 from . import request_logger
@@ -24,9 +24,10 @@ from ._helpers import (
 )
 from .client import _diagnose_connectivity
 from .common_exceptions import ApiServerError, LocalNetworkError, ProcessingInterrupted
-from .conversions import bytesio_to_image_tensor
+from .conversions import audio_bytes_to_audio_input, bytesio_to_image_tensor
 
 _RETRY_STATUS = {408, 429, 500, 502, 503, 504}
+_MAX_IN_MEMORY_DOWNLOAD_BYTES = 512 * 1024 * 1024
 
 
 async def download_url_to_bytesio(
@@ -38,6 +39,7 @@ async def download_url_to_bytesio(
     retry_delay: float = 1.0,
     retry_backoff: float = 2.0,
     cls: type[COMFY_IO.ComfyNode] = None,
+    allow_redirects: bool = True,
 ) -> None:
     """Stream-download a URL to `dest`.
 
@@ -58,6 +60,8 @@ async def download_url_to_bytesio(
     attempt = 0
     delay = retry_delay
     headers: dict[str, str] = {}
+    is_path_sink = isinstance(dest, (str, Path))
+    can_reset_sink = is_path_sink or (callable(getattr(dest, "seek", None)) and callable(getattr(dest, "truncate", None)))
 
     parsed_url = urlparse(url)
     if not parsed_url.scheme and not parsed_url.netloc:  # is URL relative?
@@ -68,10 +72,12 @@ async def download_url_to_bytesio(
 
     while True:
         attempt += 1
+        if not is_path_sink and can_reset_sink:
+            dest.seek(0)
+            dest.truncate(0)
         op_id = _generate_operation_id("GET", url, attempt)
         timeout_cfg = aiohttp.ClientTimeout(total=timeout)
 
-        is_path_sink = isinstance(dest, (str, Path))
         fhandle = None
         session: aiohttp.ClientSession | None = None
         stop_evt: asyncio.Event | None = None
@@ -96,7 +102,9 @@ async def download_url_to_bytesio(
 
             monitor_task = asyncio.create_task(_monitor())
 
-            req_task = asyncio.create_task(session.get(to_aiohttp_url(url), headers=headers))
+            req_task = asyncio.create_task(
+                session.get(to_aiohttp_url(url), headers=headers, allow_redirects=allow_redirects)
+            )
             done, pending = await asyncio.wait({req_task, monitor_task}, return_when=asyncio.FIRST_COMPLETED)
 
             if monitor_task in done and req_task in pending:
@@ -111,7 +119,7 @@ async def download_url_to_bytesio(
                 raise ProcessingInterrupted("Task cancelled") from None
 
             async with resp:
-                if resp.status >= 400:
+                if resp.status >= 300:
                     with contextlib.suppress(Exception):
                         try:
                             body = await resp.json()
@@ -129,11 +137,17 @@ async def download_url_to_bytesio(
                         )
 
                     if resp.status in _RETRY_STATUS and attempt <= max_retries:
+                        if not can_reset_sink:
+                            raise Exception(f"Failed to download (HTTP {resp.status}); destination cannot be reset for retry.")
                         await sleep_with_interrupt(delay, cls, None, None, None)
                         delay *= retry_backoff
                         continue
                     raise Exception(f"Failed to download (HTTP {resp.status}).")
 
+                max_bytes = None if is_path_sink else _MAX_IN_MEMORY_DOWNLOAD_BYTES
+                if max_bytes is not None and resp.content_length is not None and resp.content_length > max_bytes:
+                    raise ValueError(f"Download exceeds the {max_bytes}-byte in-memory limit.")
+
                 if is_path_sink:
                     p = Path(str(dest))
                     with contextlib.suppress(Exception):
@@ -160,10 +174,12 @@ async def download_url_to_bytesio(
                             break
                         continue
 
-                    sink.write(chunk)
                     written += len(chunk)
+                    if max_bytes is not None and written > max_bytes:
+                        raise ValueError(f"Download exceeds the {max_bytes}-byte in-memory limit.")
+                    sink.write(chunk)
 
-                if isinstance(dest, BytesIO):
+                if not is_path_sink and hasattr(dest, "seek"):
                     with contextlib.suppress(Exception):
                         dest.seek(0)
 
@@ -180,6 +196,8 @@ async def download_url_to_bytesio(
             raise ProcessingInterrupted("Task cancelled") from None
         except (ClientError, OSError) as e:
             if attempt <= max_retries:
+                if not can_reset_sink:
+                    raise ApiServerError("The download failed and its destination cannot be reset for retry.") from e
                 request_logger.log_request_response(
                     operation_id=op_id,
                     request_method="GET",
@@ -221,10 +239,11 @@ async def download_url_to_image_tensor(
     *,
     timeout: float = None,
     cls: type[COMFY_IO.ComfyNode] = None,
+    allow_redirects: bool = True,
 ) -> torch.Tensor:
     """Downloads an image from a URL and returns a [B, H, W, C] tensor."""
     result = BytesIO()
-    await download_url_to_bytesio(url, result, timeout=timeout, cls=cls)
+    await download_url_to_bytesio(url, result, timeout=timeout, cls=cls, allow_redirects=allow_redirects)
     return bytesio_to_image_tensor(result)
 
 
@@ -234,13 +253,42 @@ async def download_url_to_video_output(
     timeout: float = None,
     max_retries: int = 5,
     cls: type[COMFY_IO.ComfyNode] = None,
+    allow_redirects: bool = True,
 ) -> InputImpl.VideoFromFile:
     """Downloads a video from a URL and returns a `VIDEO` output."""
     result = BytesIO()
-    await download_url_to_bytesio(video_url, result, timeout=timeout, max_retries=max_retries, cls=cls)
+    await download_url_to_bytesio(
+        video_url,
+        result,
+        timeout=timeout,
+        max_retries=max_retries,
+        cls=cls,
+        allow_redirects=allow_redirects,
+    )
     return InputImpl.VideoFromFile(result)
 
 
+async def download_url_to_audio_input(
+    audio_url: str,
+    *,
+    timeout: float = None,
+    max_retries: int = 5,
+    cls: type[COMFY_IO.ComfyNode] = None,
+    allow_redirects: bool = True,
+) -> Input.Audio:
+    """Downloads audio from a URL and decodes it into a Comfy AUDIO input."""
+    result = BytesIO()
+    await download_url_to_bytesio(
+        audio_url,
+        result,
+        timeout=timeout,
+        max_retries=max_retries,
+        cls=cls,
+        allow_redirects=allow_redirects,
+    )
+    return audio_bytes_to_audio_input(result.getvalue())
+
+
 async def download_url_as_bytesio(
     url: str,
     *,
@@ -270,6 +318,7 @@ async def download_url_to_file_3d(
     timeout: float | None = None,
     max_retries: int = 5,
     cls: type[COMFY_IO.ComfyNode] = None,
+    allow_redirects: bool = True,
 ) -> Types.File3D:
     """Downloads a 3D model file from a URL into memory as BytesIO.
 
@@ -284,6 +333,7 @@ async def download_url_to_file_3d(
         timeout=timeout,
         max_retries=max_retries,
         cls=cls,
+        allow_redirects=allow_redirects,
     )
 
     if task_id is not None: