calesthio/OpenMontage · error · KlingAPIError

Kling Classic task failed

Error message

Kling Classic task failed

What it means

Raised when a Kling Classic (official Kling API) async task poll returns a terminal failure status (CLASSIC_FAILURE_STATUS). The message is taken from the API's task_status_msg or message field, defaulting to 'Kling Classic task failed' when the API returns neither. The full poll response is attached to the exception via the response kwarg for diagnostics.

Source

Thrown at tools/_kling/client.py:97

        task_id: str,
        result_key: str,
        timeout_seconds: int = 900,
        poll_interval: float = 5.0,
    ) -> list[dict[str, Any]]:
        deadline = time.time() + timeout_seconds
        while time.time() < deadline:
            data = self.get(f"{path.rstrip('/')}/{task_id}")
            payload = data.get("data") or {}
            status = payload.get("task_status") or payload.get("status")
            if status == CLASSIC_SUCCESS_STATUS:
                task_result = payload.get("task_result") or {}
                outputs = task_result.get(result_key) or []
                if not isinstance(outputs, list):
                    raise KlingAPIError(f"Kling Classic result path data.task_result.{result_key} is not a list")
                return outputs
            if status == CLASSIC_FAILURE_STATUS:
                message = payload.get("task_status_msg") or payload.get("message") or "Kling Classic task failed"
                raise KlingAPIError(str(message), code=payload.get("task_status"), response=data)
            if status not in CLASSIC_PENDING_STATUSES:
                raise KlingAPIError(f"Unexpected Kling Classic task status {status!r}", response=data)
            time.sleep(min(poll_interval, max(0.0, deadline - time.time())))
        raise TimeoutError(f"Kling Classic task {task_id} timed out after {timeout_seconds}s")

    def create_turbo(self, path: str, payload: dict[str, Any]) -> str:
        data = self.post(path, payload)
        task_id = ((data.get("data") or {}).get("id"))
        if not task_id:
            raise KlingAPIError(f"Kling Turbo create response missing data.id: {data}")
        return str(task_id)

    def poll_turbo(
        self,
        task_id: str,
        timeout_seconds: int = 900,
        poll_interval: float = 5.0,
    ) -> list[dict[str, Any]]:

View on GitHub (pinned to 95e1c3d0ab)

Solutions

1. Catch KlingAPIError and inspect error.response -> data.task_status_msg for the server's actual failure reason
2. Check account credits/quota on the Kling console and top up if exhausted
3. Soften or rewrite the prompt if the message indicates content moderation (e.g. 'sensitive content')
4. Verify the payload parameters (duration, aspect_ratio, mode) against the Kling docs for the specific endpoint
5. Retry once — transient upstream failures do occur; use is_retryable_kling_error to gate the retry

Example fix

// before
outputs = client.poll_classic('/v1/videos/text2video', task_id, 'videos')

// after
from tools._kling.errors import KlingAPIError
try:
    outputs = client.poll_classic('/v1/videos/text2video', task_id, 'videos')
except KlingAPIError as e:
    detail = (e.response or {}).get('data', {}).get('task_status_msg') if e.response else None
    raise RuntimeError(f'generation failed: {detail or e}') from e

from tools._kling.errors import KlingAPIError, is_retryable_kling_error

# no pre-API validation possible for server-side failure, but gate the retry:
def safe_poll(client, path, task_id, result_key):
    try:
        return client.poll_classic(path, task_id, result_key)
    except KlingAPIError as e:
        if is_retryable_kling_error(e):
            return client.poll_classic(path, task_id, result_key)
        raise

def is_classic_failure(exc: Exception) -> bool:
    return (
        isinstance(exc, KlingAPIError)
        and getattr(exc, 'response', None) is not None
        and (exc.response.get('data') or {}).get('task_status') == 'failed'
    )

try:
    outputs = client.poll_classic(path, task_id, 'videos')
except KlingAPIError as e:
    detail = ((e.response or {}).get('data') or {}).get('task_status_msg', str(e))
    logger.error('kling classic failed: %s (task %s)', detail, task_id)
    raise

When it happens

Trigger: Calling poll_classic() after create_classic_task() and the task transitions to the failure status. The server-side generation failed: content moderation rejection, invalid generation parameters (e.g. unsupported duration/resolution combo), insufficient credits/quota, or upstream model error.

Common situations: Prompts with sensitive content rejected by Kling moderation; account out of credits mid-generation; passing a parameter combination the endpoint does not support (e.g. 10s duration on a model that only allows 5s); malformed image reference causing the render to fail server-side.

Related errors

• Kling Classic task {task_id} timed out after {timeout_second
• Kling Turbo task failed
• Unexpected Kling Classic task status {status!r}
• Kling Turbo poll response missing data[0]: {data}
• Unexpected Kling Turbo task status {status!r}

AI-assisted analysis of calesthio/OpenMontage@95e1c3d0ab (2026-08-15). Data as JSON: /api/errors/04384eb51a22ebc0. Report an issue: GitHub.