mirrors/ComfyUI
1
0
Fork 0
mirror of https://github.com/comfyanonymous/ComfyUI.git synced 2025-09-10 19:46:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

[V3] convert Runway API nodes to the V3 schema (#9487)

Browse Source 
* convert RunAway API nodes to the V3 schema

* fixed small typo

* fix: add tooltip for "seed" input
This commit is contained in:
Alexander Piskun
2025-09-03 23:18:27 +03:00
committed by GitHub
parent 50333f1715
commit 22da0a83e9
1 changed files with 357 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

744
comfy_api_nodes/nodes_runway.py
View File

@@ -12,6 +12,7 @@ User Guides:
""" """
from typing import Union, Optional, Any from typing import Union, Optional, Any
from typing_extensions import override
from enum import Enum from enum import Enum
import torch import torch
@@ -46,9 +47,9 @@ from comfy_api_nodes.apinode_utils import (
validate_string, validate_string,
download_url_to_image_tensor, download_url_to_image_tensor,
) )
from comfy_api_nodes.mapper_utils import model_field_to_node_input
from comfy_api.input_impl import VideoFromFile from comfy_api.input_impl import VideoFromFile
from comfy.comfy_types.node_typing import IO, ComfyNodeABC from comfy_api.latest import ComfyExtension, io as comfy_io
from comfy_api_nodes.util.validation_utils import validate_image_dimensions, validate_image_aspect_ratio
PATH_IMAGE_TO_VIDEO = "/proxy/runway/image_to_video" PATH_IMAGE_TO_VIDEO = "/proxy/runway/image_to_video"
PATH_TEXT_TO_IMAGE = "/proxy/runway/text_to_image" PATH_TEXT_TO_IMAGE = "/proxy/runway/text_to_image"
@@ -85,20 +86,11 @@ class RunwayGen3aAspectRatio(str, Enum):
def get_video_url_from_task_status(response: TaskStatusResponse) -> Union[str, None]: def get_video_url_from_task_status(response: TaskStatusResponse) -> Union[str, None]:
"""Returns the video URL from the task status response if it exists.""" """Returns the video URL from the task status response if it exists."""
if response.output and len(response.output) > 0: if hasattr(response, "output") and len(response.output) > 0:
return response.output[0] return response.output[0]
return None return None
# TODO: replace with updated image validation utils (upstream)
def validate_input_image(image: torch.Tensor) -> bool:
"""
Validate the input image is within the size limits for the Runway API.
See: https://docs.dev.runwayml.com/assets/inputs/#common-error-reasons
"""
return image.shape[2] < 8000 and image.shape[1] < 8000
async def poll_until_finished( async def poll_until_finished(
auth_kwargs: dict[str, str], auth_kwargs: dict[str, str],
api_endpoint: ApiEndpoint[Any, TaskStatusResponse], api_endpoint: ApiEndpoint[Any, TaskStatusResponse],
@@ -134,458 +126,438 @@ def extract_progress_from_task_status(
def get_image_url_from_task_status(response: TaskStatusResponse) -> Union[str, None]: def get_image_url_from_task_status(response: TaskStatusResponse) -> Union[str, None]:
"""Returns the image URL from the task status response if it exists.""" """Returns the image URL from the task status response if it exists."""
if response.output and len(response.output) > 0: if hasattr(response, "output") and len(response.output) > 0:
return response.output[0] return response.output[0]
return None return None
class RunwayVideoGenNode(ComfyNodeABC): async def get_response(
"""Runway Video Node Base.""" task_id: str, auth_kwargs: dict[str, str], node_id: Optional[str] = None, estimated_duration: Optional[int] = None
) -> TaskStatusResponse:
"""Poll the task status until it is finished then get the response."""
return await poll_until_finished(
auth_kwargs,
ApiEndpoint(
path=f"{PATH_GET_TASK_STATUS}/{task_id}",
method=HttpMethod.GET,
request_model=EmptyRequest,
response_model=TaskStatusResponse,
),
estimated_duration=estimated_duration,
node_id=node_id,
)
RETURN_TYPES = ("VIDEO",)
FUNCTION = "api_call"
CATEGORY = "api node/video/Runway"
API_NODE = True
def validate_task_created(self, response: RunwayImageToVideoResponse) -> bool: async def generate_video(
""" request: RunwayImageToVideoRequest,
Validate the task creation response from the Runway API matches auth_kwargs: dict[str, str],
expected format. node_id: Optional[str] = None,
""" estimated_duration: Optional[int] = None,
if not bool(response.id): ) -> VideoFromFile:
raise RunwayApiError("Invalid initial response from Runway API.") initial_operation = SynchronousOperation(
return True endpoint=ApiEndpoint(
path=PATH_IMAGE_TO_VIDEO,
method=HttpMethod.POST,
request_model=RunwayImageToVideoRequest,
response_model=RunwayImageToVideoResponse,
),
request=request,
auth_kwargs=auth_kwargs,
)
def validate_response(self, response: RunwayImageToVideoResponse) -> bool: initial_response = await initial_operation.execute()
"""
Validate the successful task status response from the Runway API
matches expected format.
"""
if not response.output or len(response.output) == 0:
raise RunwayApiError(
"Runway task succeeded but no video data found in response."
)
return True
async def get_response( final_response = await get_response(initial_response.id, auth_kwargs, node_id, estimated_duration)
self, task_id: str, auth_kwargs: dict[str, str], node_id: Optional[str] = None if not final_response.output:
) -> RunwayImageToVideoResponse: raise RunwayApiError("Runway task succeeded but no video data found in response.")
"""Poll the task status until it is finished then get the response."""
return await poll_until_finished( video_url = get_video_url_from_task_status(final_response)
auth_kwargs, return await download_url_to_video_output(video_url)
ApiEndpoint(
path=f"{PATH_GET_TASK_STATUS}/{task_id}",
method=HttpMethod.GET, class RunwayImageToVideoNodeGen3a(comfy_io.ComfyNode):
request_model=EmptyRequest,
response_model=TaskStatusResponse, @classmethod
), def define_schema(cls):
estimated_duration=AVERAGE_DURATION_FLF_SECONDS, return comfy_io.Schema(
node_id=node_id, node_id="RunwayImageToVideoNodeGen3a",
display_name="Runway Image to Video (Gen3a Turbo)",
category="api node/video/Runway",
description="Generate a video from a single starting frame using Gen3a Turbo model. "
"Before diving in, review these best practices to ensure that "
"your input selections will set your generation up for success: "
"https://help.runwayml.com/hc/en-us/articles/33927968552339-Creating-with-Act-One-on-Gen-3-Alpha-and-Turbo.",
inputs=[
comfy_io.String.Input(
"prompt",
multiline=True,
default="",
tooltip="Text prompt for the generation",
),
comfy_io.Image.Input(
"start_frame",
tooltip="Start frame to be used for the video",
),
comfy_io.Combo.Input(
"duration",
options=[model.value for model in Duration],
),
comfy_io.Combo.Input(
"ratio",
options=[model.value for model in RunwayGen3aAspectRatio],
),
comfy_io.Int.Input(
"seed",
default=0,
min=0,
max=4294967295,
step=1,
control_after_generate=True,
display_mode=comfy_io.NumberDisplay.number,
tooltip="Random seed for generation",
),
],
outputs=[
comfy_io.Video.Output(),
],
hidden=[
comfy_io.Hidden.auth_token_comfy_org,
comfy_io.Hidden.api_key_comfy_org,
comfy_io.Hidden.unique_id,
],
is_api_node=True,
) )
async def generate_video( @classmethod
self, async def execute(
request: RunwayImageToVideoRequest, cls,
auth_kwargs: dict[str, str], prompt: str,
node_id: Optional[str] = None, start_frame: torch.Tensor,
) -> tuple[VideoFromFile]: duration: str,
initial_operation = SynchronousOperation( ratio: str,
endpoint=ApiEndpoint( seed: int,
path=PATH_IMAGE_TO_VIDEO, ) -> comfy_io.NodeOutput:
method=HttpMethod.POST, validate_string(prompt, min_length=1)
request_model=RunwayImageToVideoRequest, validate_image_dimensions(start_frame, max_width=7999, max_height=7999)
response_model=RunwayImageToVideoResponse, validate_image_aspect_ratio(start_frame, min_aspect_ratio=0.5, max_aspect_ratio=2.0)
),
request=request, auth_kwargs = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
download_urls = await upload_images_to_comfyapi(
start_frame,
max_images=1,
mime_type="image/png",
auth_kwargs=auth_kwargs, auth_kwargs=auth_kwargs,
) )
initial_response = await initial_operation.execute() return comfy_io.NodeOutput(
self.validate_task_created(initial_response) await generate_video(
task_id = initial_response.id RunwayImageToVideoRequest(
promptText=prompt,
final_response = await self.get_response(task_id, auth_kwargs, node_id) seed=seed,
self.validate_response(final_response) model=Model("gen3a_turbo"),
duration=Duration(duration),
video_url = get_video_url_from_task_status(final_response) ratio=AspectRatio(ratio),
return (await download_url_to_video_output(video_url),) promptImage=RunwayPromptImageObject(
root=[
RunwayPromptImageDetailedObject(
uri=str(download_urls[0]), position="first"
)
]
),
),
auth_kwargs=auth_kwargs,
node_id=cls.hidden.unique_id,
)
)
class RunwayImageToVideoNodeGen3a(RunwayVideoGenNode): class RunwayImageToVideoNodeGen4(comfy_io.ComfyNode):
"""Runway Image to Video Node using Gen3a Turbo model."""
DESCRIPTION = "Generate a video from a single starting frame using Gen3a Turbo model. Before diving in, review these best practices to ensure that your input selections will set your generation up for success: https://help.runwayml.com/hc/en-us/articles/33927968552339-Creating-with-Act-One-on-Gen-3-Alpha-and-Turbo."
@classmethod @classmethod
def INPUT_TYPES(s): def define_schema(cls):
return { return comfy_io.Schema(
"required": { node_id="RunwayImageToVideoNodeGen4",
"prompt": model_field_to_node_input( display_name="Runway Image to Video (Gen4 Turbo)",
IO.STRING, RunwayImageToVideoRequest, "promptText", multiline=True category="api node/video/Runway",
description="Generate a video from a single starting frame using Gen4 Turbo model. "
"Before diving in, review these best practices to ensure that "
"your input selections will set your generation up for success: "
"https://help.runwayml.com/hc/en-us/articles/37327109429011-Creating-with-Gen-4-Video.",
inputs=[
comfy_io.String.Input(
"prompt",
multiline=True,
default="",
tooltip="Text prompt for the generation",
), ),
"start_frame": ( comfy_io.Image.Input(
IO.IMAGE, "start_frame",
{"tooltip": "Start frame to be used for the video"}, tooltip="Start frame to be used for the video",
), ),
"duration": model_field_to_node_input( comfy_io.Combo.Input(
IO.COMBO, RunwayImageToVideoRequest, "duration", enum_type=Duration "duration",
options=[model.value for model in Duration],
), ),
"ratio": model_field_to_node_input( comfy_io.Combo.Input(
IO.COMBO,
RunwayImageToVideoRequest,
"ratio", "ratio",
enum_type=RunwayGen3aAspectRatio, options=[model.value for model in RunwayGen4TurboAspectRatio],
), ),
"seed": model_field_to_node_input( comfy_io.Int.Input(
IO.INT,
RunwayImageToVideoRequest,
"seed", "seed",
default=0,
min=0,
max=4294967295,
step=1,
control_after_generate=True, control_after_generate=True,
display_mode=comfy_io.NumberDisplay.number,
tooltip="Random seed for generation",
), ),
}, ],
"hidden": { outputs=[
"auth_token": "AUTH_TOKEN_COMFY_ORG", comfy_io.Video.Output(),
"comfy_api_key": "API_KEY_COMFY_ORG", ],
"unique_id": "UNIQUE_ID", hidden=[
}, comfy_io.Hidden.auth_token_comfy_org,
} comfy_io.Hidden.api_key_comfy_org,
comfy_io.Hidden.unique_id,
],
is_api_node=True,
)
async def api_call( @classmethod
self, async def execute(
cls,
prompt: str, prompt: str,
start_frame: torch.Tensor, start_frame: torch.Tensor,
duration: str, duration: str,
ratio: str, ratio: str,
seed: int, seed: int,
unique_id: Optional[str] = None, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
# Validate inputs
validate_string(prompt, min_length=1) validate_string(prompt, min_length=1)
validate_input_image(start_frame) validate_image_dimensions(start_frame, max_width=7999, max_height=7999)
validate_image_aspect_ratio(start_frame, min_aspect_ratio=0.5, max_aspect_ratio=2.0)
auth_kwargs = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
# Upload image
download_urls = await upload_images_to_comfyapi( download_urls = await upload_images_to_comfyapi(
start_frame, start_frame,
max_images=1, max_images=1,
mime_type="image/png", mime_type="image/png",
auth_kwargs=kwargs, auth_kwargs=auth_kwargs,
) )
if len(download_urls) != 1:
raise RunwayApiError("Failed to upload one or more images to comfy api.")
return await self.generate_video( return comfy_io.NodeOutput(
RunwayImageToVideoRequest( await generate_video(
promptText=prompt, RunwayImageToVideoRequest(
seed=seed, promptText=prompt,
model=Model("gen3a_turbo"), seed=seed,
duration=Duration(duration), model=Model("gen4_turbo"),
ratio=AspectRatio(ratio), duration=Duration(duration),
promptImage=RunwayPromptImageObject( ratio=AspectRatio(ratio),
root=[ promptImage=RunwayPromptImageObject(
RunwayPromptImageDetailedObject( root=[
uri=str(download_urls[0]), position="first" RunwayPromptImageDetailedObject(
) uri=str(download_urls[0]), position="first"
] )
]
),
), ),
), auth_kwargs=auth_kwargs,
auth_kwargs=kwargs, node_id=cls.hidden.unique_id,
node_id=unique_id, estimated_duration=AVERAGE_DURATION_FLF_SECONDS,
)
) )
class RunwayImageToVideoNodeGen4(RunwayVideoGenNode): class RunwayFirstLastFrameNode(comfy_io.ComfyNode):
"""Runway Image to Video Node using Gen4 Turbo model."""
DESCRIPTION = "Generate a video from a single starting frame using Gen4 Turbo model. Before diving in, review these best practices to ensure that your input selections will set your generation up for success: https://help.runwayml.com/hc/en-us/articles/37327109429011-Creating-with-Gen-4-Video."
@classmethod @classmethod
def INPUT_TYPES(s): def define_schema(cls):
return { return comfy_io.Schema(
"required": { node_id="RunwayFirstLastFrameNode",
"prompt": model_field_to_node_input( display_name="Runway First-Last-Frame to Video",
IO.STRING, RunwayImageToVideoRequest, "promptText", multiline=True category="api node/video/Runway",
description="Upload first and last keyframes, draft a prompt, and generate a video. "
"More complex transitions, such as cases where the Last frame is completely different "
"from the First frame, may benefit from the longer 10s duration. "
"This would give the generation more time to smoothly transition between the two inputs. "
"Before diving in, review these best practices to ensure that your input selections "
"will set your generation up for success: "
"https://help.runwayml.com/hc/en-us/articles/34170748696595-Creating-with-Keyframes-on-Gen-3.",
inputs=[
comfy_io.String.Input(
"prompt",
multiline=True,
default="",
tooltip="Text prompt for the generation",
), ),
"start_frame": ( comfy_io.Image.Input(
IO.IMAGE, "start_frame",
{"tooltip": "Start frame to be used for the video"}, tooltip="Start frame to be used for the video",
), ),
"duration": model_field_to_node_input( comfy_io.Image.Input(
IO.COMBO, RunwayImageToVideoRequest, "duration", enum_type=Duration "end_frame",
tooltip="End frame to be used for the video. Supported for gen3a_turbo only.",
), ),
"ratio": model_field_to_node_input( comfy_io.Combo.Input(
IO.COMBO, "duration",
RunwayImageToVideoRequest, options=[model.value for model in Duration],
),
comfy_io.Combo.Input(
"ratio", "ratio",
enum_type=RunwayGen4TurboAspectRatio, options=[model.value for model in RunwayGen3aAspectRatio],
), ),
"seed": model_field_to_node_input( comfy_io.Int.Input(
IO.INT,
RunwayImageToVideoRequest,
"seed", "seed",
default=0,
min=0,
max=4294967295,
step=1,
control_after_generate=True, control_after_generate=True,
display_mode=comfy_io.NumberDisplay.number,
tooltip="Random seed for generation",
), ),
}, ],
"hidden": { outputs=[
"auth_token": "AUTH_TOKEN_COMFY_ORG", comfy_io.Video.Output(),
"comfy_api_key": "API_KEY_COMFY_ORG", ],
"unique_id": "UNIQUE_ID", hidden=[
}, comfy_io.Hidden.auth_token_comfy_org,
} comfy_io.Hidden.api_key_comfy_org,
comfy_io.Hidden.unique_id,
async def api_call( ],
self, is_api_node=True,
prompt: str,
start_frame: torch.Tensor,
duration: str,
ratio: str,
seed: int,
unique_id: Optional[str] = None,
**kwargs,
) -> tuple[VideoFromFile]:
# Validate inputs
validate_string(prompt, min_length=1)
validate_input_image(start_frame)
# Upload image
download_urls = await upload_images_to_comfyapi(
start_frame,
max_images=1,
mime_type="image/png",
auth_kwargs=kwargs,
)
if len(download_urls) != 1:
raise RunwayApiError("Failed to upload one or more images to comfy api.")
return await self.generate_video(
RunwayImageToVideoRequest(
promptText=prompt,
seed=seed,
model=Model("gen4_turbo"),
duration=Duration(duration),
ratio=AspectRatio(ratio),
promptImage=RunwayPromptImageObject(
root=[
RunwayPromptImageDetailedObject(
uri=str(download_urls[0]), position="first"
)
]
),
),
auth_kwargs=kwargs,
node_id=unique_id,
)
class RunwayFirstLastFrameNode(RunwayVideoGenNode):
"""Runway First-Last Frame Node."""
DESCRIPTION = "Upload first and last keyframes, draft a prompt, and generate a video. More complex transitions, such as cases where the Last frame is completely different from the First frame, may benefit from the longer 10s duration. This would give the generation more time to smoothly transition between the two inputs. Before diving in, review these best practices to ensure that your input selections will set your generation up for success: https://help.runwayml.com/hc/en-us/articles/34170748696595-Creating-with-Keyframes-on-Gen-3."
async def get_response(
self, task_id: str, auth_kwargs: dict[str, str], node_id: Optional[str] = None
) -> RunwayImageToVideoResponse:
return await poll_until_finished(
auth_kwargs,
ApiEndpoint(
path=f"{PATH_GET_TASK_STATUS}/{task_id}",
method=HttpMethod.GET,
request_model=EmptyRequest,
response_model=TaskStatusResponse,
),
estimated_duration=AVERAGE_DURATION_FLF_SECONDS,
node_id=node_id,
) )
@classmethod @classmethod
def INPUT_TYPES(s): async def execute(
return { cls,
"required": {
"prompt": model_field_to_node_input(
IO.STRING, RunwayImageToVideoRequest, "promptText", multiline=True
),
"start_frame": (
IO.IMAGE,
{"tooltip": "Start frame to be used for the video"},
),
"end_frame": (
IO.IMAGE,
{
"tooltip": "End frame to be used for the video. Supported for gen3a_turbo only."
},
),
"duration": model_field_to_node_input(
IO.COMBO, RunwayImageToVideoRequest, "duration", enum_type=Duration
),
"ratio": model_field_to_node_input(
IO.COMBO,
RunwayImageToVideoRequest,
"ratio",
enum_type=RunwayGen3aAspectRatio,
),
"seed": model_field_to_node_input(
IO.INT,
RunwayImageToVideoRequest,
"seed",
control_after_generate=True,
),
},
"hidden": {
"auth_token": "AUTH_TOKEN_COMFY_ORG",
"unique_id": "UNIQUE_ID",
"comfy_api_key": "API_KEY_COMFY_ORG",
},
}
async def api_call(
self,
prompt: str, prompt: str,
start_frame: torch.Tensor, start_frame: torch.Tensor,
end_frame: torch.Tensor, end_frame: torch.Tensor,
duration: str, duration: str,
ratio: str, ratio: str,
seed: int, seed: int,
unique_id: Optional[str] = None, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[VideoFromFile]:
# Validate inputs
validate_string(prompt, min_length=1) validate_string(prompt, min_length=1)
validate_input_image(start_frame) validate_image_dimensions(start_frame, max_width=7999, max_height=7999)
validate_input_image(end_frame) validate_image_dimensions(end_frame, max_width=7999, max_height=7999)
validate_image_aspect_ratio(start_frame, min_aspect_ratio=0.5, max_aspect_ratio=2.0)
validate_image_aspect_ratio(end_frame, min_aspect_ratio=0.5, max_aspect_ratio=2.0)
auth_kwargs = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
# Upload images
stacked_input_images = image_tensor_pair_to_batch(start_frame, end_frame) stacked_input_images = image_tensor_pair_to_batch(start_frame, end_frame)
download_urls = await upload_images_to_comfyapi( download_urls = await upload_images_to_comfyapi(
stacked_input_images, stacked_input_images,
max_images=2, max_images=2,
mime_type="image/png", mime_type="image/png",
auth_kwargs=kwargs, auth_kwargs=auth_kwargs,
) )
if len(download_urls) != 2: if len(download_urls) != 2:
raise RunwayApiError("Failed to upload one or more images to comfy api.") raise RunwayApiError("Failed to upload one or more images to comfy api.")
return await self.generate_video( return comfy_io.NodeOutput(
RunwayImageToVideoRequest( await generate_video(
promptText=prompt, RunwayImageToVideoRequest(
seed=seed, promptText=prompt,
model=Model("gen3a_turbo"), seed=seed,
duration=Duration(duration), model=Model("gen3a_turbo"),
ratio=AspectRatio(ratio), duration=Duration(duration),
promptImage=RunwayPromptImageObject( ratio=AspectRatio(ratio),
root=[ promptImage=RunwayPromptImageObject(
RunwayPromptImageDetailedObject( root=[
uri=str(download_urls[0]), position="first" RunwayPromptImageDetailedObject(
), uri=str(download_urls[0]), position="first"
RunwayPromptImageDetailedObject( ),
uri=str(download_urls[1]), position="last" RunwayPromptImageDetailedObject(
), uri=str(download_urls[1]), position="last"
] ),
]
),
), ),
), auth_kwargs=auth_kwargs,
auth_kwargs=kwargs, node_id=cls.hidden.unique_id,
node_id=unique_id, estimated_duration=AVERAGE_DURATION_FLF_SECONDS,
)
) )
class RunwayTextToImageNode(ComfyNodeABC): class RunwayTextToImageNode(comfy_io.ComfyNode):
"""Runway Text to Image Node."""
RETURN_TYPES = ("IMAGE",)
FUNCTION = "api_call"
CATEGORY = "api node/image/Runway"
API_NODE = True
DESCRIPTION = "Generate an image from a text prompt using Runway's Gen 4 model. You can also include reference images to guide the generation."
@classmethod @classmethod
def INPUT_TYPES(s): def define_schema(cls):
return { return comfy_io.Schema(
"required": { node_id="RunwayTextToImageNode",
"prompt": model_field_to_node_input( display_name="Runway Text to Image",
IO.STRING, RunwayTextToImageRequest, "promptText", multiline=True category="api node/image/Runway",
description="Generate an image from a text prompt using Runway's Gen 4 model. "
"You can also include reference image to guide the generation.",
inputs=[
comfy_io.String.Input(
"prompt",
multiline=True,
default="",
tooltip="Text prompt for the generation",
), ),
"ratio": model_field_to_node_input( comfy_io.Combo.Input(
IO.COMBO,
RunwayTextToImageRequest,
"ratio", "ratio",
enum_type=RunwayTextToImageAspectRatioEnum, options=[model.value for model in RunwayTextToImageAspectRatioEnum],
), ),
}, comfy_io.Image.Input(
"optional": { "reference_image",
"reference_image": ( tooltip="Optional reference image to guide the generation",
IO.IMAGE, optional=True,
{"tooltip": "Optional reference image to guide the generation"}, ),
) ],
}, outputs=[
"hidden": { comfy_io.Image.Output(),
"auth_token": "AUTH_TOKEN_COMFY_ORG", ],
"comfy_api_key": "API_KEY_COMFY_ORG", hidden=[
"unique_id": "UNIQUE_ID", comfy_io.Hidden.auth_token_comfy_org,
}, comfy_io.Hidden.api_key_comfy_org,
} comfy_io.Hidden.unique_id,
],
def validate_task_created(self, response: RunwayTextToImageResponse) -> bool: is_api_node=True,
"""
Validate the task creation response from the Runway API matches
expected format.
"""
if not bool(response.id):
raise RunwayApiError("Invalid initial response from Runway API.")
return True
def validate_response(self, response: TaskStatusResponse) -> bool:
"""
Validate the successful task status response from the Runway API
matches expected format.
"""
if not response.output or len(response.output) == 0:
raise RunwayApiError(
"Runway task succeeded but no image data found in response."
)
return True
async def get_response(
self, task_id: str, auth_kwargs: dict[str, str], node_id: Optional[str] = None
) -> TaskStatusResponse:
"""Poll the task status until it is finished then get the response."""
return await poll_until_finished(
auth_kwargs,
ApiEndpoint(
path=f"{PATH_GET_TASK_STATUS}/{task_id}",
method=HttpMethod.GET,
request_model=EmptyRequest,
response_model=TaskStatusResponse,
),
estimated_duration=AVERAGE_DURATION_T2I_SECONDS,
node_id=node_id,
) )
async def api_call( @classmethod
self, async def execute(
cls,
prompt: str, prompt: str,
ratio: str, ratio: str,
reference_image: Optional[torch.Tensor] = None, reference_image: Optional[torch.Tensor] = None,
unique_id: Optional[str] = None, ) -> comfy_io.NodeOutput:
**kwargs,
) -> tuple[torch.Tensor]:
# Validate inputs
validate_string(prompt, min_length=1) validate_string(prompt, min_length=1)
auth_kwargs = {
"auth_token": cls.hidden.auth_token_comfy_org,
"comfy_api_key": cls.hidden.api_key_comfy_org,
}
# Prepare reference images if provided # Prepare reference images if provided
reference_images = None reference_images = None
if reference_image is not None: if reference_image is not None:
validate_input_image(reference_image) validate_image_dimensions(reference_image, max_width=7999, max_height=7999)
validate_image_aspect_ratio(reference_image, min_aspect_ratio=0.5, max_aspect_ratio=2.0)
download_urls = await upload_images_to_comfyapi( download_urls = await upload_images_to_comfyapi(
reference_image, reference_image,
max_images=1, max_images=1,
mime_type="image/png", mime_type="image/png",
auth_kwargs=kwargs, auth_kwargs=auth_kwargs,
) )
if len(download_urls) != 1:
raise RunwayApiError("Failed to upload reference image to comfy api.")
reference_images = [ReferenceImage(uri=str(download_urls[0]))] reference_images = [ReferenceImage(uri=str(download_urls[0]))]
# Create request
request = RunwayTextToImageRequest( request = RunwayTextToImageRequest(
promptText=prompt, promptText=prompt,
model=Model4.gen4_image, model=Model4.gen4_image,
@@ -593,7 +565,6 @@ class RunwayTextToImageNode(ComfyNodeABC):
referenceImages=reference_images, referenceImages=reference_images,
) )
# Execute initial request
initial_operation = SynchronousOperation( initial_operation = SynchronousOperation(
endpoint=ApiEndpoint( endpoint=ApiEndpoint(
path=PATH_TEXT_TO_IMAGE, path=PATH_TEXT_TO_IMAGE,
@@ -602,34 +573,33 @@ class RunwayTextToImageNode(ComfyNodeABC):
response_model=RunwayTextToImageResponse, response_model=RunwayTextToImageResponse,
), ),
request=request, request=request,
auth_kwargs=kwargs, auth_kwargs=auth_kwargs,
) )
initial_response = await initial_operation.execute() initial_response = await initial_operation.execute()
self.validate_task_created(initial_response)
task_id = initial_response.id
# Poll for completion # Poll for completion
final_response = await self.get_response( final_response = await get_response(
task_id, auth_kwargs=kwargs, node_id=unique_id initial_response.id,
auth_kwargs=auth_kwargs,
node_id=cls.hidden.unique_id,
estimated_duration=AVERAGE_DURATION_T2I_SECONDS,
) )
self.validate_response(final_response) if not final_response.output:
raise RunwayApiError("Runway task succeeded but no image data found in response.")
# Download and return image return comfy_io.NodeOutput(await download_url_to_image_tensor(get_image_url_from_task_status(final_response)))
image_url = get_image_url_from_task_status(final_response)
return (await download_url_to_image_tensor(image_url),)
NODE_CLASS_MAPPINGS = { class RunwayExtension(ComfyExtension):
"RunwayFirstLastFrameNode": RunwayFirstLastFrameNode, @override
"RunwayImageToVideoNodeGen3a": RunwayImageToVideoNodeGen3a, async def get_node_list(self) -> list[type[comfy_io.ComfyNode]]:
"RunwayImageToVideoNodeGen4": RunwayImageToVideoNodeGen4, return [
"RunwayTextToImageNode": RunwayTextToImageNode, RunwayFirstLastFrameNode,
} RunwayImageToVideoNodeGen3a,
RunwayImageToVideoNodeGen4,
RunwayTextToImageNode,
]
NODE_DISPLAY_NAME_MAPPINGS = { async def comfy_entrypoint() -> RunwayExtension:
"RunwayFirstLastFrameNode": "Runway First-Last-Frame to Video", return RunwayExtension()
"RunwayImageToVideoNodeGen3a": "Runway Image to Video (Gen3a Turbo)",
"RunwayImageToVideoNodeGen4": "Runway Image to Video (Gen4 Turbo)",
"RunwayTextToImageNode": "Runway Text to Image",
}