Kling v3 API 완전 가이드: Python으로 AI 영상 생성하기 (2026)

Kling v3(Kling 3.0)는 2026년 2월 4일 출시된 AI 영상 생성 모델이다. API 호출에서 영상 생성 완료까지 평균 45-120초, 5초 영상 기준 약 $0.14-0.28, 내부 품질 벤치마크 기준 4K 해상도 지원 및 멀티샷 씬 구성 가능. 이 튜토리얼은 Kling v3 API를 Python으로 연동하는 전체 과정을 다룬다 — 인증, 영상 생성, 폴링, 에러 처리까지.

---

Prerequisites

필요한 계정 및 도구

• Kling AI 개발자 계정: klingai.com 또는 Kwai for Business 개발자 포털에서 가입
• API Key & Secret: 개발자 대시보드에서 발급. Access Key ID와 Access Key Secret 두 가지가 필요
• Python 3.9+: JWT 인코딩 라이브러리 호환성 때문에 3.9 미만은 권장하지 않음
• 크레딧 충전: API 호출은 크레딧 차감 방식. 최소 충전 단위 확인 필요

라이브러리 설치

# 핵심 의존성 설치
pip install requests pyjwt cryptography python-dotenv

# 또는 requirements.txt로 관리하는 경우
pip install -r requirements.txt

requirements.txt 내용:

requests>=2.31.0
PyJWT>=2.8.0
cryptography>=41.0.0
python-dotenv>=1.0.0

환경 변수 설정

# .env 파일 생성 (절대 Git에 커밋하지 말 것)
touch .env

.env 파일:

KLING_ACCESS_KEY_ID=your_access_key_id_here
KLING_ACCESS_KEY_SECRET=your_access_key_secret_here

---

인증: JWT 토큰 생성

Kling API는 Bearer Token 방식을 사용하며, 토큰은 JWT(JSON Web Token)로 직접 생성해야 한다. OAuth처럼 별도의 토큰 발급 엔드포인트를 호출하는 방식이 아니라, Access Key ID와 Secret을 이용해 클라이언트 측에서 JWT를 서명하는 구조다.

토큰 만료 시간: 기본 30분. 장시간 실행되는 배치 작업에서는 토큰을 주기적으로 갱신해야 한다.

import jwt
import time
import os
from dotenv import load_dotenv

load_dotenv()

def generate_kling_jwt() -> str:
    """
    Kling API용 JWT 토큰 생성.
    
    Kling은 HS256 알고리즘으로 서명된 JWT를 요구함.
    토큰 유효 기간은 30분 — 장기 실행 작업에서는 이 함수를 반복 호출해야 함.
    """
    access_key_id = os.environ.get("KLING_ACCESS_KEY_ID")
    access_key_secret = os.environ.get("KLING_ACCESS_KEY_SECRET")
    
    if not access_key_id or not access_key_secret:
        raise ValueError("KLING_ACCESS_KEY_ID and KLING_ACCESS_KEY_SECRET must be set in environment")
    
    current_time = int(time.time())
    
    payload = {
        "iss": access_key_id,      # 발급자: Access Key ID
        "exp": current_time + 1800, # 만료: 현재 시각 + 30분 (1800초)
        "nbf": current_time - 5,    # not-before: 5초 여유를 두어 시계 오차 방지
    }
    
    # HS256으로 서명 — Kling이 요구하는 알고리즘
    token = jwt.encode(
        payload,
        access_key_secret,
        algorithm="HS256"
    )
    
    return token


def get_auth_headers() -> dict:
    """
    API 요청에 사용할 Authorization 헤더 반환.
    매 요청마다 호출해도 되지만, 실제 배포에서는 토큰을 캐싱해서 재사용하는 게 효율적.
    """
    token = generate_kling_jwt()
    return {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }


# 테스트
if __name__ == "__main__":
    headers = get_auth_headers()
    print("Token generated:", headers["Authorization"][:50] + "...")

---

Core Implementation: 텍스트 → 영상 생성 (Text-to-Video)

기본 구현: 영상 생성 요청 + 폴링

Kling API는 비동기 방식으로 동작한다. 영상 생성 요청을 보내면 즉시 task_id를 반환하고, 별도로 상태를 폴링해서 완료 여부를 확인해야 한다. 동기적으로 블로킹하지 않는 구조이므로, 폴링 간격을 적절히 설정해야 API 요청 낭비를 막을 수 있다.

import requests
import time
import json

# Kling API Base URL (v1 엔드포인트 — 2026년 기준)
BASE_URL = "https://api.klingai.com"

def create_text_to_video(
    prompt: str,
    model_name: str = "kling-v1-6",  # Kling v3에 해당하는 모델 식별자
    duration: int = 5,
    aspect_ratio: str = "16:9",
    cfg_scale: float = 0.5,
    negative_prompt: str = "",
) -> str:
    """
    텍스트-투-비디오 작업 생성.
    
    Returns:
        task_id: 생성된 작업 ID. 이후 상태 확인에 사용.
    
    Raises:
        requests.HTTPError: API 호출 실패 시
        KeyError: 응답 구조가 예상과 다를 때
    """
    url = f"{BASE_URL}/v1/videos/text2video"
    
    headers = get_auth_headers()
    
    payload = {
        "model_name": model_name,
        "prompt": prompt,
        "negative_prompt": negative_prompt,
        "cfg_scale": cfg_scale,        # 프롬프트 충실도. 낮을수록 창의적, 높을수록 프롬프트에 충실
        "mode": "std",                 # "std" (표준) 또는 "pro" (고품질, 비용 2배)
        "duration": duration,          # 영상 길이 (초). 5 또는 10
        "aspect_ratio": aspect_ratio,  # "16:9", "9:16", "1:1"
    }
    
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()  # 4xx/5xx 에러 즉시 예외 발생
    
    data = response.json()
    
    # Kling API 응답 코드: 0 = 성공
    if data.get("code") != 0:
        raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")
    
    task_id = data["data"]["task_id"]
    print(f"Task created: {task_id}")
    return task_id


def poll_task_status(task_id: str, max_wait_seconds: int = 300) -> dict:
    """
    작업 완료까지 폴링.
    
    폴링 간격: 처음 5번은 10초, 이후 20초 간격.
    이유: 영상 생성 초기 30-60초는 상태가 거의 바뀌지 않으므로 빈번한 폴링은 낭비.
    
    Returns:
        완료된 작업의 결과 데이터 (video URL 포함)
    """
    url = f"{BASE_URL}/v1/videos/text2video/{task_id}"
    
    start_time = time.time()
    poll_count = 0
    
    while True:
        elapsed = time.time() - start_time
        
        if elapsed > max_wait_seconds:
            raise TimeoutError(f"Task {task_id} did not complete within {max_wait_seconds}s")
        
        headers = get_auth_headers()  # 30분 제한 때문에 헤더를 매번 갱신
        response = requests.get(url, headers=headers, timeout=15)
        response.raise_for_status()
        
        data = response.json()
        task_status = data["data"]["task_status"]
        
        print(f"[{elapsed:.0f}s] Status: {task_status}")
        
        if task_status == "succeed":
            return data["data"]
        elif task_status == "failed":
            error_msg = data["data"].get("task_status_msg", "Unknown error")
            raise RuntimeError(f"Task failed: {error_msg}")
        
        # 처음 5번은 10초 간격, 이후 20초 간격으로 폴링
        poll_count += 1
        wait_time = 10 if poll_count <= 5 else 20
        time.sleep(wait_time)


# 기본 실행 예제
if __name__ == "__main__":
    task_id = create_text_to_video(
        prompt="A drone shot of a misty mountain range at sunrise, cinematic 4K",
        model_name="kling-v1-6",
        duration=5,
        aspect_ratio="16:9",
    )
    
    result = poll_task_status(task_id)
    
    # 결과에서 영상 URL 추출
    videos = result.get("task_result", {}).get("videos", [])
    if videos:
        print(f"Video URL: {videos[0]['url']}")
        print(f"Video duration: {videos[0].get('duration')}s")

---

이미지 → 영상 생성 (Image-to-Video)

이미지 파일을 base64로 인코딩하거나 퍼블릭 URL을 전달해 해당 이미지를 기반으로 영상을 생성한다.

import base64
from pathlib import Path

def create_image_to_video(
    image_path: str = None,
    image_url: str = None,
    prompt: str = "",
    duration: int = 5,
    model_name: str = "kling-v1-6",
) -> str:
    """
    이미지-투-비디오 작업 생성.
    
    image_path 또는 image_url 중 하나는 반드시 제공해야 함.
    로컬 파일의 경우 base64 인코딩으로 전송.
    최대 파일 크기: 10MB. 지원 포맷: JPEG, PNG, WEBP.
    """
    if not image_path and not image_url:
        raise ValueError("Either image_path or image_url must be provided")
    
    url = f"{BASE_URL}/v1/videos/image2video"
    headers = get_auth_headers()
    
    # 이미지 데이터 준비
    if image_path:
        file_path = Path(image_path)
        if not file_path.exists():
            raise FileNotFoundError(f"Image not found: {image_path}")
        
        # 파일 크기 사전 검증 (10MB 제한)
        file_size_mb = file_path.stat().st_size / (1024 * 1024)
        if file_size_mb > 10:
            raise ValueError(f"Image too large: {file_size_mb:.1f}MB (max 10MB)")
        
        with open(image_path, "rb") as f:
            image_b64 = base64.b64encode(f.read()).decode("utf-8")
        
        image_data = {"image": image_b64}
    else:
        image_data = {"image_url": image_url}
    
    payload = {
        "model_name": model_name,
        "image": image_data.get("image", ""),
        "image_url": image_data.get("image_url", ""),
        "prompt": prompt,
        "duration": duration,
        "mode": "std",
    }
    
    # 빈 문자열 키 제거 (API가 빈 값을 에러로 처리하는 경우 방지)
    payload = {k: v for k, v in payload.items() if v != ""}
    
    response = requests.post(url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()
    
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")
    
    return data["data"]["task_id"]

---

Production-Ready 클라이언트 클래스

실제 프로덕션에서는 토큰 캐싱, 재시도 로직, 영상 다운로드까지 포함한 클래스 구조가 필요하다.

import requests
import time
import jwt
import os
import logging
from pathlib import Path
from typing import Optional
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("kling_client")

BASE_URL = "https://api.klingai.com"


class KlingAPIClient:
    """
    Kling v3 API 프로덕션 클라이언트.
    
    토큰 자동 갱신, 요청 재시도 (지수 백오프), 결과 다운로드 포함.
    """
    
    def __init__(self):
        self.access_key_id = os.environ["KLING_ACCESS_KEY_ID"]
        self.access_key_secret = os.environ["KLING_ACCESS_KEY_SECRET"]
        self._token: Optional[str] = None
        self._token_expires_at: float = 0
        self.session = requests.Session()  # 커넥션 풀링으로 반복 요청 성능 향상
    
    def _get_token(self) -> str:
        """
        캐싱된 토큰 반환. 만료 5분 전에 자동 갱신.
        5분 버퍼를 두는 이유: 폴링 도중 토큰이 만료되는 상황 방지.
        """
        buffer_seconds = 300  # 5분 전에 갱신
        if self._token and time.time() < (self._token_expires_at - buffer_seconds):
            return self._token
        
        current_time = int(time.time())
        payload = {
            "iss": self.access_key_id,
            "exp": current_time + 1800,
            "nbf": current_time - 5,
        }
        self._token = jwt.encode(payload, self.access_key_secret, algorithm="HS256")
        self._token_expires_at = current_time + 1800
        logger.debug("JWT token refreshed")
        return self._token
    
    def _headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json",
        }
    
    def _request_with_retry(
        self, method: str, url: str, max_retries: int = 3, **kwargs
    ) -> requests.Response:
        """
        지수 백오프로 최대 3회 재시도.
        Rate limit(429)와 서버 에러(5xx)에만 재시도 적용.
        4xx (인증 실패, 잘못된 파라미터)는 재시도해도 의미 없으므로 즉시 예외.
        """
        for attempt in range(max_retries):
            response = self.session.request(method, url, **kwargs)
            
            if response.status_code in (429, 500, 502, 503, 504):
                wait_time = (2 ** attempt) * 2  # 2s, 4s, 8s
                logger.warning(
                    f"Request failed with {response.status_code}, "
                    f"retrying in {wait_time}s (attempt {attempt + 1}/{max_retries})"
                )
                time.sleep(wait_time)
                continue
            
            response.raise_for_status()
            return response
        
        response.raise_for_status()
        return response
    
    def text_to_video(
        self,
        prompt: str,
        duration: int = 5,
        aspect_ratio: str = "16:9",
        mode: str = "std",
        cfg_scale: float = 0.5,
        negative_prompt: str = "",
        model_name: str = "kling-v1-6",
    ) -> str:
        """영상 생성 작업 제출. task_id 반환."""
        
        # duration 유효성 검증 — API는 5 또는 10만 허용
        if duration not in (5, 10):
            raise ValueError(f"duration must be 5 or 10, got {duration}")
        
        if aspect_ratio not in ("16:9", "9:16", "1:1"):
            raise ValueError(f"Invalid aspect_ratio: {aspect_ratio}")
        
        payload = {
            "model_name": model_name,
            "prompt": prompt,
            "negative_prompt": negative_prompt,
            "cfg_scale": cfg_scale,
            "mode": mode,
            "duration": duration,
            "aspect_ratio": aspect_ratio,
        }
        
        response = self._request_with_retry(
            "POST",
            f"{BASE_URL}/v1/videos/text2video",
            headers=self._headers(),
            json=payload,
            timeout=30,
        )
        
        data = response.json()
        if data.get("code") != 0:
            raise RuntimeError(f"Kling API error {data['code']}: {data.get('message')}")
        
        task_id = data["data"]["task_id"]
        logger.info(f"Task submitted: {task_id}")
        return task_id
    
    def wait_for_completion(
        self, task_id: str, max_wait: int = 300, endpoint: str = "text2video"
    ) -> dict:
        """
        작업 완료 대기 (블로킹).
        endpoint: "text2video" 또는 "image2video"
        """
        url = f"{BASE_URL}/v1/videos/{endpoint}/{task_id}"
        start = time.time()
        poll_count = 0
        
        while True:
            elapsed = time.time() - start
            if elapsed > max_wait:
                raise TimeoutError(f"Timeout after {max_wait}s for task {task_id}")
            
            response = self._request_with_retry(
                "GET", url, headers=self._headers(), timeout=15
            )
            data = response.json()["data"]
            status = data["task_status"]
            
            logger.info(f"[{elapsed:.0f}s] Task {task_id}: {status}")
            
            if status == "succeed":
                return data
            if status == "failed":
                raise RuntimeError(
                    f"Task {task_id} failed: {data.get('task_status_msg')}"
                )
            
            poll_count += 1
            time.sleep(10 if poll_count <= 5 else 20)
    
    def download_video(self, video_url: str, output_path: str) -> str:
        """
        영상 URL에서 로컬 파일로 다운로드.
        스트리밍 다운로드로 메모리 사용 최소화.
        """
        output = Path(output_path)
        output.parent.mkdir(parents=True, exist_ok=True)
        
        with self.session.get(video_url, stream=True, timeout=60) as r:
            r.raise_for_status()
            with open(output, "wb") as f:
                for chunk in r.iter_content(chunk_size=8192):
                    f.write(chunk)
        
        file_size_mb = output.stat().st_size / (1024 * 1024)
        logger.info(f"Downloaded {file_size_mb:.1f}MB to {output}")
        return str(output)


# 프로덕션 사용 예제
if __name__ == "__main__":
    client = KlingAPIClient()
    
    task_id = client.text_to_video(
        prompt="A timelapse of cherry blossoms falling in a quiet Japanese garden, 4K cinematic",
        duration=5,
        aspect_ratio="16:9",
        mode="std",
        cfg_scale=0.5,
    )
    
    result = client.wait_for_completion(task_id, max_wait=300)
    videos = result.get("task_result", {}).get("videos", [])
    
    if videos:
        local_path = client.download_video(videos[0]["url"], f"output/{task_id}.mp4")
        print(f"Saved to: {local_path}")

---

API 파라미터 레퍼런스

파라미터	타입	기본값	유효 범위	영향
model_name	string	"kling-v1-6"	"kling-v1", "kling-v1-5", "kling-v1-6"	영상 품질과 비용. v1-6이 최신
prompt	string	필수	최대 2500자	영상 내용 전체. 구체적일수록 일관성 증가
negative_prompt	string	""	최대 2500자	제외할 요소 지정. 블러, 워터마크 제거 등에 활용
cfg_scale	float	0.5	0.0 ~ 1.0	프롬프트 충실도. 0에 가까울수록 자유로운 생성, 1에 가까울수록 프롬프트 종속
mode	string	"std"	"std", "pro"	pro는 품질 향상, 크레딧 2배 소비
duration	int	5	5, 10	영상 길이(초). 10초는 크레딧 약 2배
aspect_ratio	string	"16:9"	"16:9", "9:16", "1:1"	출력 해상도 비율
camera_control	object	null	지정 카메라 무브먼트	줌, 패닝, 틸트 등 카메라 모션 제어 (선택)
callback_url	string	""	유효한 HTTPS URL	완료 시 Webhook 호출. 폴링 대신 사용 가능

---

에러 처리

에러 코드 레퍼런스

에러 코드	HTTP 상태	원인	해결책
1000	400	잘못된 요청 파라미터	파라미터 타입과 범위 확인
1001	401	JWT 토큰 무효 또는 만료	토큰 재생성. exp 필드 확인
1002	403	권한 없음	API 키 권한 설정 확인
1003	429	Rate limit 초과	지수 백오프 후 재시도
1004	402	크레딧 부족	대시보드에서 크레딧 충전
1010	400	프롬프트 정책 위반	콘텐츠 가이드라인에 맞게 프롬프트 수정
5000	500	서버 내부 에러	재시도. 반복 발생 시 지원 문의
task_failed	200	생성 실패	task_status_msg 확인. 주로 이미지 품질 문제

에러 처리 구현 예제

import requests
import logging

logger = logging.getLogger("kling_error_handler")

# Kling API 에러 코드 매핑
KLING_ERROR_MESSAGES = {
    1000: "Invalid request parameters — check parameter types and ranges",
    1001: "JWT token invalid or expired — regenerate token",
    1002: "Permission denied — verify API key permissions",
    1003: "Rate limit exceeded — implement exponential backoff",
    1004: "Insufficient credits — recharge via dashboard",
    1010: "Prompt policy violation — revise content",
    5000: "Internal server error — retry with backoff",
}

def safe_api_call(func, *args, **kwargs):
    """
    모든 Kling API 호출을 감싸는 범용 에러 핸들러.
    에러 코드에 따라 재시도 여부를 자동 판단.
    """
    try:
        return func(*args, **kwargs)
    
    except requests.HTTPError as e:
        status_code = e.response.status_code
        
        # 응답 바디에서 Kling 에러 코드 추출 시도
        try:
            error_data = e.response.json()
            kling_code = error_data.get("code")
            message = KLING_ERROR_MESSAGES.get(kling_code, error_data.get("message", "Unknown"))
            logger.error(f"Kling API error {kling_code}: {message}")
        except Exception:
            logger.error(f"HTTP {status_code}: {e}")
        
        # 재시도 가능 여부 표시
        retryable = status_code in (429, 500, 502, 503, 504)
        if not retryable:
            raise  # 4xx (파라미터 오류, 인증 실패)는 재시도해도 무의미
        raise
    
    except TimeoutError as e:
        # 폴링 타임아웃 — 작업 자체는 계속 실행 중일 수 있음
        # task_id를 저장해두면 나중에 상태 재확인 가능
        logger.error(f"Polling timeout: {e}")
        raise
    
    except RuntimeError as e:
        # Task failed 또는 API code != 0
        logger.error(f"Task error: {e}")
        raise


# 사용 예시
if __name__ == "__main__":
    client = KlingAPIClient()
    
    try:
        task_id = safe_api_call(
            client.text_to_video,
            prompt="Abstract fluid art with neon colors",
            duration=5,
        )
        result = safe_api_call(
            client.wait_for_completion,
            task_id,
            max_wait=300,
        )
    except RuntimeError as e:
        logger.error(f"Generation failed: {e}")
    except TimeoutError as e:
        # task_id를 DB에 저장해두고 나중에 재확인 가능
        logger.error(f"Timeout — save task_id for later check: {task_id}")

---

성능 및 비용 레퍼런스

구성	생성 시간	크레딧 소비 (참고값)	해상도	적합한 사용 사례
std 5초 16:9	45-90초	~35 크레딧	720p-1080p	프로토타이핑, 대량 생성
pro 5초 16:9	60-120초	~70 크레딧	1080p-4K	최종 납품물, 클라이언트 작업
std 10초 16:9	90-180초	~70 크레딧	720p-1080p	긴 씬, 타임랩스
pro 10초 16:9	120-240초	~140 크레딧	1080p-4K	고품질 긴 클립
Image-to-Video std 5초	60-120초	~35 크레딧	원본 이미지 해상도 기반	캐릭터 애니메이션

주의: 크레딧 가격은 구독 플랜과 충전 단위에 따라 다르다. 위 수치는 참고용이며, 실제 소비량은 Kling 대시보드에서 확인해야 한다.

Kling v3을 사용하지 말아야 하는 경우

• 실시간 응답 필요: 최소 45초 지연 — 대화형 UI나 라이브 스트리밍에 부적합
• 10초 이상의 영상: API 지원 최대 길이가 10초. 긴 영상은 여러 클립을 합치는 파이프라인 구축 필요
• 텍스트 오버레이: 생성된 영상에 텍스트 렌더링 품질이 불안정. 후처리(FFmpeg 등) 권장
• 정밀 립싱크: 오디오 싱크 정밀도가 요구되는 경우 전용 모델 사용 권장

---

결론

Kling v3 API는 JWT 인증, 비동기 작업 폴링, std/pro 모드 선택의 세 가지 핵심 개념을 이해하면 프로덕션 파이프라인에 통합할 수 있다. 토큰 캐싱, 지수 백오프 재시도, 타임아웃 처리를 빠뜨리면 운영 환경에서 반드시 문제가 생긴다. 위 코드를 그대로 사용하되, 크레딧 소비는 반드시 자체 대시보드에서 모니터링하라.

참고: 여러 AI 모델을 하나의 파이프라인에서 사용한다면, AtlasCloud는 Kling, Flux, Seedance, Claude, GPT 등 300개 이상의 모델에 단일 API로 접근할 수 있습니다. API 키 하나로 모든 모델 사용 가능. 신규 사용자는 첫 충전 시 25% 보너스(최대 $100).