Add Minimax Video Generation + Async Task queue polling example (#6)

.github/workflows/update-api-stubs.yml

@@ -0,0 +1,49 @@
+name: Generate API Models
+
+on:
+  schedule:
+    # Run weekly on Monday at 00:00 UTC
+    - cron: '0 0 * * 1'
+  workflow_dispatch:
+    # Allow manual triggering
+
+jobs:
+  generate-models:
+    runs-on: ubuntu-latest
+    
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+      
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.10'
+      
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install 'datamodel-code-generator[http]'
+      
+      - name: Generate API models
+        run: |
+          datamodel-codegen  --use-subclass-enum --url https://stagingapi.comfy.org/openapi --output comfy_api_nodes/apis/stubs.py
+      
+      - name: Check for changes
+        id: git-check
+        run: |
+          git diff --exit-code comfy_extras/apis/stubs.py || echo "changes=true" >> $GITHUB_OUTPUT
+      
+      - name: Create Pull Request
+        if: steps.git-check.outputs.changes == 'true'
+        uses: peter-evans/create-pull-request@v5
+        with:
+          commit-message: 'chore: update API models from OpenAPI spec'
+          title: 'Update API models from OpenAPI spec'
+          body: |
+            This PR updates the API models based on the latest OpenAPI specification.
+            
+            Generated automatically by the Generate API Models workflow.
+          branch: update-api-models
+          delete-branch: true
+          base: main

comfy_api_nodes/apis/client.py

@@ -0,0 +1,457 @@
+import logging
+
+"""
+API Client Framework for ComfyUI
+
+This module provides a flexible framework for making API requests from ComfyUI nodes.
+It supports both synchronous and asynchronous API operations with proper type validation.
+
+Key Components:
+--------------
+1. ApiClient - Handles HTTP requests with authentication and error handling
+2. ApiEndpoint - Defines a single HTTP endpoint with its request/response models
+3. ApiOperation - Executes a single synchronous API operation
+4. PollingOperation - Executes an asynchronous operation with polling for completion
+
+Usage Examples:
+--------------
+
+# Example 1: Synchronous API Operation
+# ------------------------------------
+# For a simple API call that returns the result immediately:
+
+# 1. Create the API client
+api_client = ApiClient(
+    base_url="https://api.example.com",
+    api_key="your_api_key_here",
+    timeout=30.0,
+    verify_ssl=True
+)
+
+# 2. Define the endpoint
+user_info_endpoint = ApiEndpoint(
+    path="/v1/users/me",
+    method=HttpMethod.GET,
+    request_model=EmptyRequest,  # No request body needed
+    response_model=UserProfile,   # Pydantic model for the response
+    query_params=None
+)
+
+# 3. Create the request object
+request = EmptyRequest()
+
+# 4. Create and execute the operation
+operation = ApiOperation(
+    endpoint=user_info_endpoint,
+    request=request
+)
+user_profile = operation.execute(client=api_client)  # Returns immediately with the result
+
+
+# Example 2: Asynchronous API Operation with Polling
+# -------------------------------------------------
+# For an API that starts a task and requires polling for completion:
+
+# 1. Define the endpoints (initial request and polling)
+generate_image_endpoint = ApiEndpoint(
+    path="/v1/images/generate",
+    method=HttpMethod.POST,
+    request_model=ImageGenerationRequest,
+    response_model=TaskCreatedResponse,
+    query_params=None
+)
+
+check_task_endpoint = ApiEndpoint(
+    path="/v1/tasks/{task_id}",
+    method=HttpMethod.GET,
+    request_model=EmptyRequest,
+    response_model=ImageGenerationResult,
+    query_params=None
+)
+
+# 2. Create the request object
+request = ImageGenerationRequest(
+    prompt="a beautiful sunset over mountains",
+    width=1024,
+    height=1024,
+    num_images=1
+)
+
+# 3. Create and execute the polling operation
+operation = PollingOperation(
+    initial_endpoint=generate_image_endpoint,
+    initial_request=request,
+    poll_endpoint=check_task_endpoint,
+    task_id_field="task_id",
+    status_field="status",
+    completed_statuses=["completed"],
+    failed_statuses=["failed", "error"]
+)
+
+# This will make the initial request and then poll until completion
+result = operation.execute(client=api_client)  # Returns the final ImageGenerationResult when done
+"""
+
+from typing import (
+    Dict,
+    Type,
+    Optional,
+    Any,
+    TypeVar,
+    Generic,
+    Callable,
+)
+from pydantic import BaseModel
+from enum import Enum
+import time
+import json
+import requests
+from urllib.parse import urljoin
+
+# Import models from your generated stubs
+
+T = TypeVar("T", bound=BaseModel)
+R = TypeVar("R", bound=BaseModel)
+P = TypeVar("P", bound=BaseModel)  # For poll response
+
+
+class EmptyRequest(BaseModel):
+    """Base class for empty request bodies.
+    For GET requests, fields will be sent as query parameters."""
+
+    pass
+
+
+class HttpMethod(str, Enum):
+    GET = "GET"
+    POST = "POST"
+    PUT = "PUT"
+    DELETE = "DELETE"
+    PATCH = "PATCH"
+
+
+class ApiClient:
+    """
+    Client for making HTTP requests to an API with authentication and error handling.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        api_key: Optional[str] = None,
+        timeout: float = 30.0,
+        verify_ssl: bool = True,
+    ):
+        self.base_url = base_url
+        self.api_key = api_key
+        self.timeout = timeout
+        self.verify_ssl = verify_ssl
+
+    def get_headers(self) -> Dict[str, str]:
+        """Get headers for API requests, including authentication if available"""
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+
+        if self.api_key:
+            headers["Authorization"] = f"Bearer {self.api_key}"
+
+        return headers
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Dict[str, Any]] = None,
+        files: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Make an HTTP request to the API
+
+        Args:
+            method: HTTP method (GET, POST, etc.)
+            path: API endpoint path (will be joined with base_url)
+            params: Query parameters
+            json: JSON body data
+            files: Files to upload
+            headers: Additional headers
+
+        Returns:
+            Parsed JSON response
+
+        Raises:
+            requests.RequestException: If the request fails
+        """
+        url = urljoin(self.base_url, path)
+        self.check_auth_token(self.api_key)
+        # Combine default headers with any provided headers
+        request_headers = self.get_headers()
+        if headers:
+            request_headers.update(headers)
+        try:
+            response = requests.request(
+                method=method,
+                url=url,
+                params=params,
+                json=json,
+                files=files,
+                headers=request_headers,
+                timeout=self.timeout,
+                verify=self.verify_ssl,
+            )
+
+            # Raise exception for error status codes
+            response.raise_for_status()
+        except requests.ConnectionError:
+            raise Exception(
+                f"Unable to connect to the API server at {self.base_url}. Please check your internet connection or verify the service is available."
+            )
+
+        except requests.Timeout:
+            raise Exception(
+                f"Request timed out after {self.timeout} seconds. The server might be experiencing high load or the operation is taking longer than expected."
+            )
+
+        except requests.HTTPError as e:
+            status_code = e.response.status_code if hasattr(e, "response") else None
+            error_message = f"HTTP Error: {str(e)}"
+            logging.debug(f"[DEBUG] API Error: {error_message} (Status: {status_code})")
+            if status_code == 401:
+                error_message = "Unauthorized: Please login first to use this node."
+            if status_code == 402:
+                error_message = "Payment Required: Please add credits to your account to use this node."
+            raise Exception(error_message)
+
+        # Parse and return JSON response
+        if response.content:
+            return response.json()
+        return {}
+
+    def check_auth_token(self, auth_token):
+        """Verify that an auth token is present."""
+        if auth_token is None:
+            raise Exception("Please login first to use this node.")
+        return auth_token
+
+
+class ApiEndpoint(Generic[T, R]):
+    """Defines an API endpoint with its request and response types"""
+
+    def __init__(
+        self,
+        path: str,
+        method: HttpMethod,
+        request_model: Type[T],
+        response_model: Type[R],
+        query_params: Optional[Dict[str, Any]] = None,
+    ):
+        """Initialize an API endpoint definition.
+
+        Args:
+            path: The URL path for this endpoint, can include placeholders like {id}
+            method: The HTTP method to use (GET, POST, etc.)
+            request_model: Pydantic model class that defines the structure and validation rules for API requests to this endpoint
+            response_model: Pydantic model class that defines the structure and validation rules for API responses from this endpoint
+            query_params: Optional dictionary of query parameters to include in the request
+        """
+        self.path = path
+        self.method = method
+        self.request_model = request_model
+        self.response_model = response_model
+        self.query_params = query_params or {}
+
+
+class SynchronousOperation(Generic[T, R]):
+    """
+    Represents a single synchronous API operation.
+    """
+
+    def __init__(
+        self,
+        endpoint: ApiEndpoint[T, R],
+        request: T,
+        api_base: str = "https://stagingapi.comfy.org",
+        auth_token: Optional[str] = None,
+        timeout: float = 30.0,
+        verify_ssl: bool = True,
+    ):
+        self.endpoint = endpoint
+        self.request = request
+        self.response = None
+        self.error = None
+        self.api_base = api_base
+        self.auth_token = auth_token
+        self.timeout = timeout
+        self.verify_ssl = verify_ssl
+
+    def execute(self, client: Optional[ApiClient] = None) -> R:
+        """Execute the API operation using the provided client or create one"""
+        try:
+            # Create client if not provided
+            if client is None:
+                if self.api_base is None:
+                    raise ValueError("Either client or api_base must be provided")
+                client = ApiClient(
+                    base_url=self.api_base,
+                    api_key=self.auth_token,
+                    timeout=self.timeout,
+                    verify_ssl=self.verify_ssl,
+                )
+
+            # Convert request model to dict
+            request_dict = self.request.model_dump(exclude_none=True)
+
+            # Debug log for request
+            logging.debug(f"[DEBUG] API Request: {self.endpoint.method.value} {self.endpoint.path}")
+            logging.debug(f"[DEBUG] Request Data: {json.dumps(request_dict, indent=2)}")
+            logging.debug(f"[DEBUG] Query Params: {self.endpoint.query_params}")
+
+            # Make the request
+            resp = client.request(
+                method=self.endpoint.method.value,
+                path=self.endpoint.path,
+                json=request_dict,
+                params=self.endpoint.query_params,
+            )
+
+            # Debug log for response
+            logging.debug(f"[DEBUG] API Response: {json.dumps(resp, indent=2)}")
+
+            # Parse and return the response
+            return self._parse_response(resp)
+
+        except Exception as e:
+            logging.debug(f"[DEBUG] API Exception: {str(e)}")
+            raise Exception(str(e))
+
+    def _parse_response(self, resp):
+        """Parse response data - can be overridden by subclasses"""
+        # The response is already the complete object, don't extract just the "data" field
+        # as that would lose the outer structure (created timestamp, etc.)
+
+        # Parse response using the provided model
+        self.response = self.endpoint.response_model.model_validate(resp)
+        logging.debug(f"[DEBUG] Parsed Response: {self.response}")
+        return self.response
+
+
+class TaskStatus(str, Enum):
+    """Enum for task status values"""
+
+    COMPLETED = "completed"
+    FAILED = "failed"
+    PENDING = "pending"
+
+
+class PollingOperation(Generic[T, R]):
+    """
+    Represents an asynchronous API operation that requires polling for completion.
+    """
+
+    def __init__(
+        self,
+        poll_endpoint: ApiEndpoint[EmptyRequest, R],
+        completed_statuses: list,
+        failed_statuses: list,
+        status_extractor: Callable[[R], str],
+        request: Optional[T] = None,
+        api_base: str = "https://stagingapi.comfy.org",
+        auth_token: Optional[str] = None,
+        poll_interval: float = 1.0,
+    ):
+        self.poll_endpoint = poll_endpoint
+        self.request = request
+        self.api_base = api_base
+        self.auth_token = auth_token
+        self.poll_interval = poll_interval
+
+        # Polling configuration
+        self.status_extractor = status_extractor or (
+            lambda x: getattr(x, "status", None)
+        )
+        self.completed_statuses = completed_statuses
+        self.failed_statuses = failed_statuses
+
+        # For storing response data
+        self.final_response = None
+        self.error = None
+
+    def execute(self, client: Optional[ApiClient] = None) -> R:
+        """Execute the polling operation using the provided client. If failed, raise an exception."""
+        try:
+            if client is None:
+                client = ApiClient(
+                    base_url=self.api_base,
+                    api_key=self.auth_token,
+                )
+            return self._poll_until_complete(client)
+        except Exception as e:
+            raise Exception(f"Error during polling: {str(e)}")
+
+    def _check_task_status(self, response: R) -> TaskStatus:
+        """Check task status using the status extractor function"""
+        try:
+            status = self.status_extractor(response)
+            if status in self.completed_statuses:
+                return TaskStatus.COMPLETED
+            elif status in self.failed_statuses:
+                return TaskStatus.FAILED
+            return TaskStatus.PENDING
+        except Exception as e:
+            logging.debug(f"Error extracting status: {e}")
+            return TaskStatus.PENDING
+
+    def _poll_until_complete(self, client: ApiClient) -> R:
+        """Poll until the task is complete"""
+        poll_count = 0
+        while True:
+            try:
+                poll_count += 1
+                logging.debug(f"[DEBUG] Polling attempt #{poll_count}")
+
+                request_dict = (
+                    self.request.model_dump(exclude_none=True)
+                    if self.request is not None
+                    else None
+                )
+
+                if poll_count == 1:
+                    logging.debug(
+                        f"[DEBUG] Poll Request: {self.poll_endpoint.method.value} {self.poll_endpoint.path}"
+                    )
+                    logging.debug(
+                      f"[DEBUG] Poll Request Data: {json.dumps(request_dict, indent=2) if request_dict else 'None'}"
+                    )
+
+                # Query task status
+                resp = client.request(
+                    method=self.poll_endpoint.method.value,
+                    path=self.poll_endpoint.path,
+                    params=self.poll_endpoint.query_params,
+                    json=request_dict,
+                )
+
+                # Parse response
+                response_obj = self.poll_endpoint.response_model.model_validate(resp)
+
+                # Check if task is complete
+                status = self._check_task_status(response_obj)
+                logging.debug(f"[DEBUG] Task Status: {status}")
+
+                if status == TaskStatus.COMPLETED:
+                    logging.debug("[DEBUG] Task completed successfully")
+                    self.final_response = response_obj
+                    return self.final_response
+                elif status == TaskStatus.FAILED:
+                    logging.debug(f"[DEBUG] Task failed: {json.dumps(resp)}")
+                    raise Exception(f"Task failed: {json.dumps(resp)}")
+                else:
+                    logging.debug("[DEBUG] Task still pending, continuing to poll...")
+
+                # Wait before polling again
+                logging.debug(f"[DEBUG] Waiting {self.poll_interval} seconds before next poll")
+                time.sleep(self.poll_interval)
+
+            except Exception as e:
+                logging.debug(f"[DEBUG] Polling error: {str(e)}")
+                raise Exception(f"Error while polling: {str(e)}")

comfy_api_nodes/apis/stubs.py

@@ -0,0 +1,513 @@
+# generated by datamodel-codegen:
+#   filename:  http://localhost:8080/openapi
+#   timestamp: 2025-04-18T21:35:21+00:00
+
+from __future__ import annotations
+
+from datetime import datetime
+from enum import Enum
+from typing import Any, Dict, List, Optional
+from uuid import UUID
+
+from pydantic import BaseModel, Field, conint, constr
+
+
+class ComfyNode(BaseModel):
+    category: Optional[str] = Field(
+        None,
+        description='UI category where the node is listed, used for grouping nodes.',
+    )
+    comfy_node_name: Optional[str] = Field(
+        None, description='Unique identifier for the node'
+    )
+    deprecated: Optional[bool] = Field(
+        None,
+        description='Indicates if the node is deprecated. Deprecated nodes are hidden in the UI.',
+    )
+    description: Optional[str] = Field(
+        None, description="Brief description of the node's functionality or purpose."
+    )
+    experimental: Optional[bool] = Field(
+        None,
+        description='Indicates if the node is experimental, subject to changes or removal.',
+    )
+    function: Optional[str] = Field(
+        None, description='Name of the entry-point function to execute the node.'
+    )
+    input_types: Optional[str] = Field(None, description='Defines input parameters')
+    output_is_list: Optional[List[bool]] = Field(
+        None, description='Boolean values indicating if each output is a list.'
+    )
+    return_names: Optional[str] = Field(
+        None, description='Names of the outputs for clarity in workflows.'
+    )
+    return_types: Optional[str] = Field(
+        None, description='Specifies the types of outputs produced by the node.'
+    )
+
+
+class ComfyNodeCloudBuildInfo(BaseModel):
+    build_id: Optional[str] = None
+    location: Optional[str] = None
+    project_id: Optional[str] = None
+    project_number: Optional[str] = None
+
+
+class Customer(BaseModel):
+    createdAt: Optional[datetime] = Field(
+        None, description='The date and time the user was created'
+    )
+    email: Optional[str] = Field(None, description='The email address for this user')
+    id: str = Field(..., description='The firebase UID of the user')
+    name: Optional[str] = Field(None, description='The name for this user')
+    updatedAt: Optional[datetime] = Field(
+        None, description='The date and time the user was last updated'
+    )
+
+
+class Error(BaseModel):
+    details: Optional[List[str]] = Field(
+        None,
+        description='Optional detailed information about the error or hints for resolving it.',
+    )
+    message: Optional[str] = Field(
+        None, description='A clear and concise description of the error.'
+    )
+
+
+class ErrorResponse(BaseModel):
+    error: str
+    message: str
+
+
+class GitCommitSummary(BaseModel):
+    author: Optional[str] = Field(None, description='The author of the commit')
+    branch_name: Optional[str] = Field(
+        None, description='The branch where the commit was made'
+    )
+    commit_hash: Optional[str] = Field(None, description='The hash of the commit')
+    commit_name: Optional[str] = Field(None, description='The name of the commit')
+    status_summary: Optional[Dict[str, str]] = Field(
+        None, description='A map of operating system to status pairs'
+    )
+    timestamp: Optional[datetime] = Field(
+        None, description='The timestamp when the commit was made'
+    )
+
+
+class ImageRequest(BaseModel):
+    aspect_ratio: Optional[str] = Field(
+        None,
+        description="Optional. The aspect ratio (e.g., 'ASPECT_16_9', 'ASPECT_1_1'). Cannot be used with resolution. Defaults to 'ASPECT_1_1' if unspecified.",
+    )
+    color_palette: Optional[Dict[str, Any]] = Field(
+        None, description='Optional. Color palette object. Only for V_2, V_2_TURBO.'
+    )
+    magic_prompt_option: Optional[str] = Field(
+        None, description="Optional. MagicPrompt usage ('AUTO', 'ON', 'OFF')."
+    )
+    model: Optional[str] = Field(
+        None,
+        description="Optional. The model used (e.g., 'V_2', 'V_2A_TURBO'). Defaults to 'V_2' if unspecified.",
+    )
+    negative_prompt: Optional[str] = Field(
+        None,
+        description='Optional. Description of what to exclude. Only for V_1, V_1_TURBO, V_2, V_2_TURBO.',
+    )
+    num_images: Optional[conint(ge=1, le=8)] = Field(
+        1, description='Optional. Number of images to generate (1-8). Defaults to 1.'
+    )
+    prompt: str = Field(
+        ..., description='Required. The prompt to use to generate the image.'
+    )
+    resolution: Optional[str] = Field(
+        None,
+        description="Optional. Resolution (e.g., 'RESOLUTION_1024_1024'). Only for model V_2. Cannot be used with aspect_ratio.",
+    )
+    seed: Optional[conint(ge=0, le=2147483647)] = Field(
+        None, description='Optional. A number between 0 and 2147483647.'
+    )
+    style_type: Optional[str] = Field(
+        None,
+        description="Optional. Style type ('AUTO', 'GENERAL', 'REALISTIC', 'DESIGN', 'RENDER_3D', 'ANIME'). Only for models V_2 and above.",
+    )
+
+
+class IdeogramGenerateRequest(BaseModel):
+    image_request: ImageRequest = Field(
+        ..., description='The image generation request parameters.'
+    )
+
+
+class Datum(BaseModel):
+    is_image_safe: Optional[bool] = Field(
+        None, description='Indicates whether the image is considered safe.'
+    )
+    prompt: Optional[str] = Field(
+        None, description='The prompt used to generate this image.'
+    )
+    resolution: Optional[str] = Field(
+        None, description="The resolution of the generated image (e.g., '1024x1024')."
+    )
+    seed: Optional[int] = Field(
+        None, description='The seed value used for this generation.'
+    )
+    style_type: Optional[str] = Field(
+        None,
+        description="The style type used for generation (e.g., 'REALISTIC', 'ANIME').",
+    )
+    url: Optional[str] = Field(None, description='URL to the generated image.')
+
+
+class IdeogramGenerateResponse(BaseModel):
+    created: Optional[datetime] = Field(
+        None, description='Timestamp when the generation was created.'
+    )
+    data: Optional[List[Datum]] = Field(
+        None, description='Array of generated image information.'
+    )
+
+
+class MachineStats(BaseModel):
+    cpu_capacity: Optional[str] = Field(None, description='Total CPU on the machine.')
+    disk_capacity: Optional[str] = Field(
+        None, description='Total disk capacity on the machine.'
+    )
+    gpu_type: Optional[str] = Field(
+        None, description='The GPU type. eg. NVIDIA Tesla K80'
+    )
+    initial_cpu: Optional[str] = Field(
+        None, description='Initial CPU available before the job starts.'
+    )
+    initial_disk: Optional[str] = Field(
+        None, description='Initial disk available before the job starts.'
+    )
+    initial_ram: Optional[str] = Field(
+        None, description='Initial RAM available before the job starts.'
+    )
+    machine_name: Optional[str] = Field(None, description='Name of the machine.')
+    memory_capacity: Optional[str] = Field(
+        None, description='Total memory on the machine.'
+    )
+    os_version: Optional[str] = Field(
+        None, description='The operating system version. eg. Ubuntu Linux 20.04'
+    )
+    pip_freeze: Optional[str] = Field(None, description='The pip freeze output')
+    vram_time_series: Optional[Dict[str, Any]] = Field(
+        None, description='Time series of VRAM usage.'
+    )
+
+
+class MinimaxBaseResponse(BaseModel):
+    status_code: int = Field(
+        ...,
+        description='Status code. 0 indicates success, other values indicate errors.',
+    )
+    status_msg: str = Field(
+        ..., description='Specific error details or success message.'
+    )
+
+
+class File(BaseModel):
+    bytes: Optional[int] = Field(None, description='File size in bytes')
+    created_at: Optional[int] = Field(
+        None, description='Unix timestamp when the file was created, in seconds'
+    )
+    download_url: Optional[str] = Field(
+        None, description='The URL to download the video'
+    )
+    file_id: Optional[int] = Field(None, description='Unique identifier for the file')
+    filename: Optional[str] = Field(None, description='The name of the file')
+    purpose: Optional[str] = Field(None, description='The purpose of using the file')
+
+
+class MinimaxFileRetrieveResponse(BaseModel):
+    base_resp: MinimaxBaseResponse
+    file: File
+
+
+class Status(str, Enum):
+    Queueing = 'Queueing'
+    Preparing = 'Preparing'
+    Processing = 'Processing'
+    Success = 'Success'
+    Fail = 'Fail'
+
+
+class MinimaxTaskResultResponse(BaseModel):
+    base_resp: MinimaxBaseResponse
+    file_id: Optional[str] = Field(
+        None,
+        description='After the task status changes to Success, this field returns the file ID corresponding to the generated video.',
+    )
+    status: Status = Field(
+        ...,
+        description="Task status: 'Queueing' (in queue), 'Preparing' (task is preparing), 'Processing' (generating), 'Success' (task completed successfully), or 'Fail' (task failed).",
+    )
+    task_id: str = Field(..., description='The task ID being queried.')
+
+
+class Model(str, Enum):
+    T2V_01_Director = 'T2V-01-Director'
+    I2V_01_Director = 'I2V-01-Director'
+    S2V_01 = 'S2V-01'
+    I2V_01 = 'I2V-01'
+    I2V_01_live = 'I2V-01-live'
+    T2V_01 = 'T2V-01'
+
+
+class SubjectReferenceItem(BaseModel):
+    image: Optional[str] = Field(
+        None, description='URL or base64 encoding of the subject reference image.'
+    )
+    mask: Optional[str] = Field(
+        None,
+        description='URL or base64 encoding of the mask for the subject reference image.',
+    )
+
+
+class MinimaxVideoGenerationRequest(BaseModel):
+    callback_url: Optional[str] = Field(
+        None,
+        description='Optional. URL to receive real-time status updates about the video generation task.',
+    )
+    first_frame_image: Optional[str] = Field(
+        None,
+        description='URL or base64 encoding of the first frame image. Required when model is I2V-01, I2V-01-Director, or I2V-01-live.',
+    )
+    model: Model = Field(
+        ...,
+        description='Required. ID of model. Options: T2V-01-Director, I2V-01-Director, S2V-01, I2V-01, I2V-01-live, T2V-01',
+    )
+    prompt: Optional[constr(max_length=2000)] = Field(
+        None,
+        description='Description of the video. Should be less than 2000 characters. Supports camera movement instructions in [brackets].',
+    )
+    prompt_optimizer: Optional[bool] = Field(
+        True,
+        description='If true (default), the model will automatically optimize the prompt. Set to false for more precise control.',
+    )
+    subject_reference: Optional[List[SubjectReferenceItem]] = Field(
+        None,
+        description='Only available when model is S2V-01. The model will generate a video based on the subject uploaded through this parameter.',
+    )
+
+
+class MinimaxVideoGenerationResponse(BaseModel):
+    base_resp: MinimaxBaseResponse
+    task_id: str = Field(
+        ..., description='The task ID for the asynchronous video generation task.'
+    )
+
+
+class NodeStatus(str, Enum):
+    NodeStatusActive = 'NodeStatusActive'
+    NodeStatusDeleted = 'NodeStatusDeleted'
+    NodeStatusBanned = 'NodeStatusBanned'
+
+
+class NodeVersionStatus(str, Enum):
+    NodeVersionStatusActive = 'NodeVersionStatusActive'
+    NodeVersionStatusDeleted = 'NodeVersionStatusDeleted'
+    NodeVersionStatusBanned = 'NodeVersionStatusBanned'
+    NodeVersionStatusPending = 'NodeVersionStatusPending'
+    NodeVersionStatusFlagged = 'NodeVersionStatusFlagged'
+
+
+class NodeVersionUpdateRequest(BaseModel):
+    changelog: Optional[str] = Field(
+        None, description='The changelog describing the version changes.'
+    )
+    deprecated: Optional[bool] = Field(
+        None, description='Whether the version is deprecated.'
+    )
+
+
+class PersonalAccessToken(BaseModel):
+    createdAt: Optional[datetime] = Field(
+        None, description='[Output Only]The date and time the token was created.'
+    )
+    description: Optional[str] = Field(
+        None,
+        description="Optional. A more detailed description of the token's intended use.",
+    )
+    id: Optional[UUID] = Field(None, description='Unique identifier for the GitCommit')
+    name: Optional[str] = Field(
+        None,
+        description='Required. The name of the token. Can be a simple description.',
+    )
+    token: Optional[str] = Field(
+        None,
+        description='[Output Only]. The personal access token. Only returned during creation.',
+    )
+
+
+class PublisherStatus(str, Enum):
+    PublisherStatusActive = 'PublisherStatusActive'
+    PublisherStatusBanned = 'PublisherStatusBanned'
+
+
+class PublisherUser(BaseModel):
+    email: Optional[str] = Field(None, description='The email address for this user.')
+    id: Optional[str] = Field(None, description='The unique id for this user.')
+    name: Optional[str] = Field(None, description='The name for this user.')
+
+
+class StorageFile(BaseModel):
+    file_path: Optional[str] = Field(None, description='Path to the file in storage')
+    id: Optional[UUID] = Field(
+        None, description='Unique identifier for the storage file'
+    )
+    public_url: Optional[str] = Field(None, description='Public URL')
+
+
+class User(BaseModel):
+    email: Optional[str] = Field(None, description='The email address for this user.')
+    id: Optional[str] = Field(None, description='The unique id for this user.')
+    isAdmin: Optional[bool] = Field(
+        None, description='Indicates if the user has admin privileges.'
+    )
+    isApproved: Optional[bool] = Field(
+        None, description='Indicates if the user is approved.'
+    )
+    name: Optional[str] = Field(None, description='The name for this user.')
+
+
+class WorkflowRunStatus(str, Enum):
+    WorkflowRunStatusStarted = 'WorkflowRunStatusStarted'
+    WorkflowRunStatusFailed = 'WorkflowRunStatusFailed'
+    WorkflowRunStatusCompleted = 'WorkflowRunStatusCompleted'
+
+
+class ActionJobResult(BaseModel):
+    action_job_id: Optional[str] = Field(
+        None, description='Identifier of the job this result belongs to'
+    )
+    action_run_id: Optional[str] = Field(
+        None, description='Identifier of the run this result belongs to'
+    )
+    author: Optional[str] = Field(None, description='The author of the commit')
+    avg_vram: Optional[int] = Field(
+        None, description='The average VRAM used by the job'
+    )
+    branch_name: Optional[str] = Field(
+        None, description='Name of the relevant git branch'
+    )
+    comfy_run_flags: Optional[str] = Field(
+        None, description='The comfy run flags. E.g. `--low-vram`'
+    )
+    commit_hash: Optional[str] = Field(None, description='The hash of the commit')
+    commit_id: Optional[str] = Field(None, description='The ID of the commit')
+    commit_message: Optional[str] = Field(None, description='The message of the commit')
+    commit_time: Optional[int] = Field(
+        None, description='The Unix timestamp when the commit was made'
+    )
+    cuda_version: Optional[str] = Field(None, description='CUDA version used')
+    end_time: Optional[int] = Field(
+        None, description='The end time of the job as a Unix timestamp.'
+    )
+    git_repo: Optional[str] = Field(None, description='The repository name')
+    id: Optional[UUID] = Field(None, description='Unique identifier for the job result')
+    job_trigger_user: Optional[str] = Field(
+        None, description='The user who triggered the job.'
+    )
+    machine_stats: Optional[MachineStats] = None
+    operating_system: Optional[str] = Field(None, description='Operating system used')
+    peak_vram: Optional[int] = Field(None, description='The peak VRAM used by the job')
+    pr_number: Optional[str] = Field(None, description='The pull request number')
+    python_version: Optional[str] = Field(None, description='PyTorch version used')
+    pytorch_version: Optional[str] = Field(None, description='PyTorch version used')
+    start_time: Optional[int] = Field(
+        None, description='The start time of the job as a Unix timestamp.'
+    )
+    status: Optional[WorkflowRunStatus] = None
+    storage_file: Optional[StorageFile] = None
+    workflow_name: Optional[str] = Field(None, description='Name of the workflow')
+
+
+class NodeVersion(BaseModel):
+    changelog: Optional[str] = Field(
+        None, description='Summary of changes made in this version'
+    )
+    comfy_node_extract_status: Optional[str] = Field(
+        None, description='The status of comfy node extraction process.'
+    )
+    createdAt: Optional[datetime] = Field(
+        None, description='The date and time the version was created.'
+    )
+    dependencies: Optional[List[str]] = Field(
+        None, description='A list of pip dependencies required by the node.'
+    )
+    deprecated: Optional[bool] = Field(
+        None, description='Indicates if this version is deprecated.'
+    )
+    downloadUrl: Optional[str] = Field(
+        None, description='[Output Only] URL to download this version of the node'
+    )
+    id: Optional[str] = None
+    node_id: Optional[str] = Field(
+        None, description='The unique identifier of the node.'
+    )
+    status: Optional[NodeVersionStatus] = None
+    status_reason: Optional[str] = Field(
+        None, description='The reason for the status change.'
+    )
+    version: Optional[str] = Field(
+        None,
+        description='The version identifier, following semantic versioning. Must be unique for the node.',
+    )
+
+
+class PublisherMember(BaseModel):
+    id: Optional[str] = Field(
+        None, description='The unique identifier for the publisher member.'
+    )
+    role: Optional[str] = Field(
+        None, description='The role of the user in the publisher.'
+    )
+    user: Optional[PublisherUser] = None
+
+
+class Publisher(BaseModel):
+    createdAt: Optional[datetime] = Field(
+        None, description='The date and time the publisher was created.'
+    )
+    description: Optional[str] = None
+    id: Optional[str] = Field(
+        None,
+        description="The unique identifier for the publisher. It's akin to a username. Should be lowercase.",
+    )
+    logo: Optional[str] = Field(None, description="URL to the publisher's logo.")
+    members: Optional[List[PublisherMember]] = Field(
+        None, description='A list of members in the publisher.'
+    )
+    name: Optional[str] = None
+    source_code_repo: Optional[str] = None
+    status: Optional[PublisherStatus] = None
+    support: Optional[str] = None
+    website: Optional[str] = None
+
+
+class Node(BaseModel):
+    author: Optional[str] = None
+    category: Optional[str] = Field(None, description='The category of the node.')
+    description: Optional[str] = None
+    downloads: Optional[int] = Field(
+        None, description='The number of downloads of the node.'
+    )
+    icon: Optional[str] = Field(None, description="URL to the node's icon.")
+    id: Optional[str] = Field(None, description='The unique identifier of the node.')
+    latest_version: Optional[NodeVersion] = None
+    license: Optional[str] = Field(
+        None, description="The path to the LICENSE file in the node's repository."
+    )
+    name: Optional[str] = Field(None, description='The display name of the node.')
+    publisher: Optional[Publisher] = None
+    rating: Optional[float] = Field(None, description='The average rating of the node.')
+    repository: Optional[str] = Field(None, description="URL to the node's repository.")
+    status: Optional[NodeStatus] = None
+    status_detail: Optional[str] = Field(
+        None, description='The status detail of the node.'
+    )
+    tags: Optional[List[str]] = None
+    translations: Optional[Dict[str, Dict[str, Any]]] = None

comfy_api_nodes/nodes_api.py

@@ -1,8 +1,8 @@
-# Add API base URL at the top of the file
-API_BASE = "https://stagingapi.comfy.org"
-
 from inspect import cleandoc
 from comfy.comfy_types.node_typing import ComfyNodeABC, InputTypeDict, IO
+from comfy_api_nodes.apis.client import ApiEndpoint, SynchronousOperation, HttpMethod, PollingOperation, EmptyRequest
+from comfy_api_nodes.apis.stubs import IdeogramGenerateRequest, IdeogramGenerateResponse, ImageRequest, MinimaxVideoGenerationRequest, MinimaxVideoGenerationResponse, MinimaxFileRetrieveResponse, MinimaxTaskResultResponse, Model
+import logging
 
 def check_auth_token(auth_token):
     """Verify that an auth token is present."""
@@ -98,45 +98,44 @@ class IdeogramTextToImage(ComfyNodeABC):
     def api_call(self, prompt, model, aspect_ratio=None, resolution=None,
                  magic_prompt_option="AUTO", seed=0, style_type="NONE",
                  negative_prompt="", num_images=1, color_palette="", auth_token=None):
-        import requests
         import torch
         from PIL import Image
         import io
         import numpy as np
+        import requests
 
-        check_auth_token(auth_token)
+        operation = SynchronousOperation(
-
+            endpoint=ApiEndpoint(
-        # Build payload with all available parameters
+                path="/proxy/ideogram/generate",
-        payload = {
+                method=HttpMethod.POST,
-            "image_request": {
+                request_model=IdeogramGenerateRequest,
-                "prompt": prompt,
+                response_model=IdeogramGenerateResponse
-                "model": model,
+            ),
-                "num_images": num_images,
+            request=IdeogramGenerateRequest(
-                "seed": seed,
+                image_request=ImageRequest(
-            }
+                  prompt=prompt,
-        }
+                  model=model,
-
+                  num_images=num_images,
-        # Make API request
+                  seed=seed,
-        headers = {
+                  aspect_ratio=aspect_ratio if aspect_ratio != "ASPECT_1_1" else None,
-            "Authorization": f"Bearer {auth_token}",
+                  resolution=resolution if resolution != "1024x1024" else None,
-            "Content-Type": "application/json"
+                  magic_prompt_option=magic_prompt_option if magic_prompt_option != "AUTO" else None,
-        }
+                  style_type=style_type if style_type != "NONE" else None,
-
+                  negative_prompt=negative_prompt if negative_prompt else None,
-        response = requests.post(
+                  color_palette=None
-            f"{API_BASE}/proxy/ideogram/generate",
+                )
-            headers=headers,
+            ),
-            json=payload
+            auth_token=auth_token
         )
 
-        if response.status_code != 200:
+        response = operation.execute()
-            raise Exception(f"API request failed: {response.text}")
 
-        # Parse response
+        if not response.data or len(response.data) == 0:
-        response_data = response.json()
+            raise Exception("No images were generated in the response")
-
+        image_url = response.data[0].url
-        # Get the image URL from the response
-        image_url = response_data["data"][0]["url"]
 
+        if not image_url:
+            raise Exception("No image URL was generated in the response")
         img_response = requests.get(image_url)
         if img_response.status_code != 200:
             raise Exception("Failed to download the image")
@@ -165,9 +164,9 @@ class IdeogramTextToImage(ComfyNodeABC):
     #    return ""
 
 
-class RunwayVideoNode:
+class MinimaxVideoNode:
     """
-    Generates videos synchronously based on a given image, prompt, and optional parameters using Runway's API.
+    Generates videos synchronously based on a prompt, and optional parameters using Minimax's API.
     """
     def __init__(self):
         pass
@@ -176,41 +175,15 @@ class RunwayVideoNode:
     def INPUT_TYPES(s):
         return {
             "required": {
-                "prompt_image": ("IMAGE",),  # Will need to handle image URL conversion
                 "prompt_text": ("STRING", {
                     "multiline": True,
                     "default": "",
                     "tooltip": "Text prompt to guide the video generation"
                 }),
-            },
+                "model": (["T2V-01", "I2V-01-Director", "S2V-01", "I2V-01", "I2V-01-live", "T2V-01"], {
-            "optional": {
+                    "default": "T2V-01",
-                "seed": ("INT", {
-                    "default": 0,
-                    "min": 0,
-                    "max": 4294967295,
-                    "step": 1,
-                    "display": "number"
-                }),
-                "model": (["gen3a_turbo"], {
-                    "default": "gen3a_turbo",
                     "tooltip": "Model to use for video generation"
                 }),
-                "duration": ("FLOAT", {
-                    "default": 5.0,
-                    "min": 1.0,
-                    "max": 10.0,
-                    "step": 0.1,
-                    "display": "number",
-                    "tooltip": "Duration of the generated video in seconds"
-                }),
-                "ratio": (["1280:768", "768:1280"], {
-                    "default": "1280:768",
-                    "tooltip": "Aspect ratio of the output video"
-                }),
-                "watermark": ("BOOLEAN", {
-                    "default": False,
-                    "tooltip": "Whether to include watermark in the output"
-                }),
             },
             "hidden": {
                 "auth_token": "AUTH_TOKEN_COMFY_ORG"
@@ -218,62 +191,83 @@ class RunwayVideoNode:
         }
 
     RETURN_TYPES = ("VIDEO",)
-    DESCRIPTION = "Generates videos from images using Runway's API"
+    DESCRIPTION = "Generates videos from prompts using Minimax's API"
     FUNCTION = "generate_video"
     CATEGORY = "video"
     API_NODE = True
+    OUTPUT_NODE = True
 
-    def generate_video(self, prompt_image, prompt_text, seed=0, model="gen3a_turbo",
+    def generate_video(self, prompt_text, seed=0, model="T2V-01", auth_token=None):
-                      duration=5.0, ratio="1280:768", watermark=False, auth_token=None):
+        video_generate_operation = SynchronousOperation(
-        import requests
+            endpoint=ApiEndpoint(
-        check_auth_token(auth_token)
+                path="/proxy/minimax/video_generation",
-        # Convert torch tensor image to URL (you'll need to implement this part)
+                method=HttpMethod.POST,
-        # This is a placeholder - you'll need to either save the image temporarily
+                request_model=MinimaxVideoGenerationRequest,
-        # or upload it to a service that can host it
+                response_model=MinimaxVideoGenerationResponse,
-        image_url = "http://example.com"  # Placeholder
+            ),
-
+            request=MinimaxVideoGenerationRequest(
-        # Build payload
+                model=Model(model),
-        payload = {
+                prompt=prompt_text,
-            "promptImage": image_url,
+                callback_url=None,
-            "promptText": prompt_text,
+                first_frame_image=None,
-            "seed": seed,
+                subject_reference=None,
-            "model": model,
+                prompt_optimizer=None
-            "watermark": watermark,
+            ),
-            "duration": duration,
+            auth_token=auth_token
-            "ratio": ratio
-        }
-
-        # Make API request
-        headers = {
-            "Authorization": f"Bearer {auth_token}",
-            "Content-Type": "application/json",
-        }
-
-        response = requests.post(
-            f"{API_BASE}/proxy/runway/image_to_video",
-            headers=headers,
-            json=payload
         )
+        response = video_generate_operation.execute()
 
-        if response.status_code != 200:
+        task_id = response.task_id
-            raise Exception(f"API request failed: {response.text}")
 
-        # Parse response
+        video_generate_operation = PollingOperation(
-        # response_data = response.json()
+            poll_endpoint=ApiEndpoint(
+                path="/proxy/minimax/query/video_generation",
+                method=HttpMethod.GET,
+                request_model=EmptyRequest,
+                response_model=MinimaxTaskResultResponse,
+                query_params={
+                    "task_id": task_id
+                }
+            ),
+            completed_statuses=["Success"],
+            failed_statuses=["Fail"],
+            status_extractor=lambda x: x.status.value,
+            auth_token=auth_token
+        )
+        task_result = video_generate_operation.execute()
+
+        file_id = task_result.file_id
+
+        file_retrieve_operation = SynchronousOperation(
+            endpoint=ApiEndpoint(
+                path="/proxy/minimax/files/retrieve",
+                method=HttpMethod.GET,
+                request_model=EmptyRequest,
+                response_model=MinimaxFileRetrieveResponse,
+                query_params={
+                    "file_id": file_id
+                }
+            ),
+            request=EmptyRequest(),
+            auth_token=auth_token
+        )
+        file_result = file_retrieve_operation.execute()
+
+        file_url = file_result.file.download_url
+
+        logging.info(f"Generated video URL: {file_url}")
 
-        # Note: You'll need to implement the actual video handling here
-        # This is a placeholder return
         return (None,)
 
 # A dictionary that contains all nodes you want to export with their names
 # NOTE: names should be globally unique
 NODE_CLASS_MAPPINGS = {
     "IdeogramTextToImage": IdeogramTextToImage,
-    "RunwayVideoNode": RunwayVideoNode
+    "MinimaxVideoNode": MinimaxVideoNode
 }
 
 # A dictionary that contains the friendly/humanly readable titles for the nodes
 NODE_DISPLAY_NAME_MAPPINGS = {
     "IdeogramTextToImage": "Ideogram Text to Image",
-    "RunwayVideoNode": "Runway Video Generator"
+    "MinimaxVideoNode": "Minimax Video Generator"
 }

nodes.py

@@ -2258,6 +2258,10 @@ def init_builtin_extra_nodes():
         "nodes_optimalsteps.py",
         "nodes_hidream.py",
         "nodes_fresca.py",
+    ]
+
+    api_nodes_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy_api_nodes")
+    api_nodes_files = [
         "nodes_api.py",
     ]
 
@@ -2266,6 +2270,10 @@ def init_builtin_extra_nodes():
         if not load_custom_node(os.path.join(extras_dir, node_file), module_parent="comfy_extras"):
             import_failed.append(node_file)
 
+    for node_file in api_nodes_files:
+        if not load_custom_node(os.path.join(api_nodes_dir, node_file), module_parent="comfy_api_nodes"):
+            import_failed.append(node_file)
+
     return import_failed
 
 

uv.lock

@@ -0,0 +1,7 @@
+version = 1
+requires-python = ">=3.9"
+
+[[package]]
+name = "comfyui"
+version = "0.3.26"
+source = { virtual = "." }