> ## Documentation Index
> Fetch the complete documentation index at: https://docs.comfy.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Kling Text to Video - ComfyUI 組み込みノード

> KlingのAI技術を用いてテキスト記述を動画に変換するノード

<img src="https://mintcdn.com/dripart/5003JSxULDwNImme/images/built-in-nodes/api_nodes/kwai_vgi/kling-text-to-video.jpg?fit=max&auto=format&n=5003JSxULDwNImme&q=85&s=7ddc3d180854e9a1e836154b3f36a58d" alt="ComfyUI 組み込み Kling Text to Video ノード" width="1731" height="1754" data-path="images/built-in-nodes/api_nodes/kwai_vgi/kling-text-to-video.jpg" />

Kling Text to Video ノードは、Kling の API サービスに接続して、テキスト記述から動画を生成します。ユーザーは、目的の動画内容を記述したテキストを入力するだけで、対応する動画コンテンツを作成できます。

## パラメータ

### 必須パラメータ

| パラメータ            | 型      | デフォルト値            | 説明                      |
| ---------------- | ------ | ----------------- | ----------------------- |
| prompt           | 文字列    | ""                | 生成したい動画内容を記述するプロンプトテキスト |
| negative\_prompt | 文字列    | ""                | 動画に含めたくない要素を指定するプロンプト   |
| cfg\_scale       | 浮動小数点数 | 7.0               | プロンプトへの従い具合を制御するパラメータ   |
| model\_name      | 選択     | "kling-v2-master" | 使用する動画生成モデル             |
| aspect\_ratio    | 選択     | AspectRatio 列挙型   | 出力動画のアスペクト比             |
| duration         | 選択     | Duration 列挙型      | 生成される動画の長さ              |
| mode             | 選択     | Mode 列挙型          | 動画生成モード                 |

### 出力

| 出力             | 型   | 説明         |
| -------------- | --- | ---------- |
| VIDEO          | 動画  | 生成された動画    |
| Kling ID       | 文字列 | タスク識別子     |
| Duration (sec) | 文字列 | 動画の長さ（秒単位） |

## 動作原理

このノードは、テキストプロンプトを Kling の API サーバーに送信し、サーバー側で処理された後に生成された動画を返却します。処理には、初期リクエストとタスク状態のポーリングが含まれます。完了後、ノードは動画をダウンロードして出力します。

ユーザーは、ネガティブプロンプトや CFG スケール、動画の各種属性などのパラメータを調整することで、生成結果を制御できます。また、システムはプロンプトの長さを検証し、API の要件を満たすことを保証します。

## ソースコード

\[ノードのソースコード（2025-05-03 更新）]

```python theme={null}

class KlingTextToVideoNode(KlingNodeBase):
    """Kling Text to Video Node"""

    @staticmethod
    def poll_for_task_status(task_id: str, auth_token: str) -> KlingText2VideoResponse:
        """Polls the Kling API endpoint until the task reaches a terminal state."""
        polling_operation = PollingOperation(
            poll_endpoint=ApiEndpoint(
                path=f"{PATH_TEXT_TO_VIDEO}/{task_id}",
                method=HttpMethod.GET,
                request_model=EmptyRequest,
                response_model=KlingText2VideoResponse,
            ),
            completed_statuses=[
                TaskStatus.succeed.value,
            ],
            failed_statuses=[TaskStatus.failed.value],
            status_extractor=lambda response: (
                response.data.task_status.value
                if response.data and response.data.task_status
                else None
            ),
            auth_token=auth_token,
        )
        return polling_operation.execute()

    @classmethod
    def INPUT_TYPES(s):
        return {
            "required": {
                "prompt": model_field_to_node_input(
                    IO.STRING, KlingText2VideoRequest, "prompt", multiline=True
                ),
                "negative_prompt": model_field_to_node_input(
                    IO.STRING, KlingText2VideoRequest, "negative_prompt", multiline=True
                ),
                "model_name": model_field_to_node_input(
                    IO.COMBO,
                    KlingText2VideoRequest,
                    "model_name",
                    enum_type=ModelName,
                    default="kling-v2-master",
                ),
                "cfg_scale": model_field_to_node_input(
                    IO.FLOAT, KlingText2VideoRequest, "cfg_scale"
                ),
                "mode": model_field_to_node_input(
                    IO.COMBO, KlingText2VideoRequest, "mode", enum_type=Mode
                ),
                "duration": model_field_to_node_input(
                    IO.COMBO, KlingText2VideoRequest, "duration", enum_type=Duration
                ),
                "aspect_ratio": model_field_to_node_input(
                    IO.COMBO,
                    KlingText2VideoRequest,
                    "aspect_ratio",
                    enum_type=AspectRatio,
                ),
            },
            "hidden": {"auth_token": "AUTH_TOKEN_COMFY_ORG"},
        }

    RETURN_TYPES = ("VIDEO", "STRING", "STRING")
    RETURN_NAMES = ("VIDEO", "Kling ID", "Duration (sec)")
    DESCRIPTION = "Kling Text to Video Node"

    def api_call(
        self,
        prompt: str,
        negative_prompt: str,
        model_name: str,
        cfg_scale: float,
        mode: str,
        duration: int,
        aspect_ratio: str,
        camera_control: Optional[CameraControl] = None,
        auth_token: Optional[str] = None,
    ) -> tuple[VideoFromFile, str, str]:
        validate_prompts(prompt, negative_prompt, MAX_PROMPT_LENGTH_T2V)
        initial_operation = SynchronousOperation(
            endpoint=ApiEndpoint(
                path=PATH_TEXT_TO_VIDEO,
                method=HttpMethod.POST,
                request_model=KlingText2VideoRequest,
                response_model=KlingText2VideoResponse,
            ),
            request=KlingText2VideoRequest(
                prompt=prompt if prompt else None,
                negative_prompt=negative_prompt if negative_prompt else None,
                duration=Duration(duration),
                mode=Mode(mode),
                model_name=ModelName(model_name),
                cfg_scale=cfg_scale,
                aspect_ratio=AspectRatio(aspect_ratio),
                camera_control=camera_control,
            ),
            auth_token=auth_token,
        )

        initial_response = initial_operation.execute()
        if not is_valid_initial_response(initial_response):
            error_msg = f"Kling initial request failed. Code: {initial_response.code}, Message: {initial_response.message}, Data: {initial_response.data}"
            logging.error(error_msg)
            raise KlingApiError(error_msg)

        task_id = initial_response.data.task_id
        final_response = self.poll_for_task_status(task_id, auth_token)
        if not is_valid_video_response(final_response):
            error_msg = (
                f"Kling task {task_id} succeeded but no video data found in response."
            )
            logging.error(error_msg)
            raise KlingApiError(error_msg)

        video = final_response.data.task_result.videos[0]
        logging.debug("Kling task %s succeeded. Video URL: %s", task_id, video.url)
        return (
            download_url_to_video_output(video.url),
            str(video.id),
            str(video.duration),
        )
```