Python에서 Runway ML 텍스트-투-비디오 API 사용하기: 실행 가능한 예제와 hiapi로의 마이그레이션

Python에서 텍스트-투-비디오 모델을 호출하려 할 때 Runway Gen-3는 가장 먼저 떠오르는 선택지입니다. 이 가이드는 Runway 스타일의 비동기 패턴(작업 생성 → 폴링 → 다운로드)을 보여주는 최소한의 실행 가능한 Python 예제를 제시한 뒤, 동일한 코드를 hiapi의 wan2.7-video/text-to-video로 매핑합니다. 이는 같은 /v1/tasks 구조를 사용하는 드롭인 대체方案입니다.

준비물

• Python 3.9+ 및 requests 라이브러리(pip install requests).
• API 키. hiapi의 경우 hiapi 대시보드에서 키를 발급받아 HIAPI_API_KEY로 내보내세요(sk-로 시작합니다).
• 프롬프트. 비디오 모델은 단일 텍스트 프롬프트와 몇 가지 파라미터(종횡비, 재생 시간(초))를 받습니다.

아래 모든 곳의 인증 헤더는 Authorization: Bearer sk-<key>입니다. HTTP 401과 permission_denied를 받았다면 키가 잘못되었거나 선택한 모델에 대한 액세스 권한이 없는 것입니다.

최소 실행 예제(Runway 스타일 비동기 패턴)

텍스트-투-비디오 제공업체는 비디오를 동기적으로 반환하지 않습니다. 생성에 30–120초가 걸립니다. 표준 구조는 다음과 같습니다:

1. POST /tasks에 모델 id + 입력을 담아 보내면 → taskId를 반환합니다.
2. GET /tasks/{id} → status가 success 또는 fail이 될 때까지 폴링합니다.
3. output[0].url을 읽고 MP4를 다운로드합니다(해당 URL은 단기 유효합니다).

아래는 이 루프를 구현한 자체 포함 Python 스크립트입니다:

import os
import time
import requests

API_BASE = "https://api.runwayml.com/v1"  # 또는 같은 구조를 사용하는 모든 제공업체
API_KEY = os.environ["RUNWAY_API_KEY"]

def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
    resp = requests.post(
        f"{API_BASE}/tasks",
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json={
            "model": "gen3a_turbo",
            "input": {
                "prompt": prompt,
                "aspect_ratio": aspect_ratio,
                "duration": duration,
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

def wait_task(task_id: str, timeout_s: int = 600, interval: int = 5) -> dict:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        resp = requests.get(
            f"{API_BASE}/tasks/{task_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            timeout=30,
        )
        resp.raise_for_status()
        task = resp.json()["data"]
        if task["status"] == "success":
            return task
        if task["status"] == "fail":
            raise RuntimeError(f"task failed: {task.get('error')}")
        time.sleep(interval)
    raise TimeoutError(f"task {task_id} not done after {timeout_s}s")

def download(url: str, path: str) -> None:
    with requests.get(url, stream=True, timeout=120) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(chunk_size=1 << 20):
                f.write(chunk)

if __name__ == "__main__":
    tid = create_task("a cat walking on a beach at sunset, cinematic")
    print(f"task {tid} created, polling…")
    task = wait_task(tid)
    download(task["output"][0]["url"], "out.mp4")
    print("wrote out.mp4")

이 스크립트는 의도적으로 간단하게 유지됩니다. SDK도, 비동기 접착 코드도 없이 requests만 사용합니다. 일단 이것이 작동하면, 완전히 동일한 제어 흐름을 유지하면서 두 개의 문자열만 바꿔 제공업체를 전환할 수 있습니다.

hiapi로 마이그레이션(wan2.7-video/text-to-video)

hiapi는 https://api.hiapi.ai/v1/tasks에서 동일한 비동기 작업 계약을 노출합니다. 호출 코드에서 유일한 차이는 다음과 같습니다:

1. API_BASE가 https://api.hiapi.ai/v1로 바뀝니다.
2. 모델 id는 wan2.7-video/text-to-video입니다(URL이 아닌 순수 id).
3. 인증 키는 hiapi의 sk-… 키입니다.

구체적으로, create_task만 변경되며 wait_task와 download는 완전히 동일합니다:

import os
import requests

API_BASE = "https://api.hiapi.ai/v1"
API_KEY = os.environ["HIAPI_API_KEY"]

def create_task(prompt: str, aspect_ratio: str = "16:9", duration: int = 5) -> str:
    resp = requests.post(
        f"{API_BASE}/tasks",
        headers={"Authorization": f"Bearer {API_KEY}",
                 "Content-Type": "application/json"},
        json={
            "model": "wan2.7-video/text-to-video",
            "input": {
                "prompt": prompt,
                "ratio": aspect_ratio,  # wan2.7에서는 ratio라고 부릅니다: 16:9, 9:16, 1:1, 4:3, 3:4
                "duration": duration,   # 2-15초
                "resolution": "720P",   # 720P $0.10/초; 기본값은 1080P, $0.167/초
            },
        },
        timeout=60,
    )
    resp.raise_for_status()
    return resp.json()["data"]["taskId"]

응답 구조는 완전히 동일합니다. 생성 시 {"data": {"taskId": "..."}}를 반환하고, 폴링 시 다음을 반환합니다:

{
  "data": {
    "status": "success",
    "output": [{"url": "https://…/result.mp4", "expireAt": 1750000000}],
    "error": null
  }
}

따라서 이전 섹션의 wait_task(...)와 download(...)는 수정 없이 작동합니다. 전체 마이그레이션은 단 두 줄입니다: API_BASE 변경과 model 변경. 주변의 모든 것은 그대로 유지됩니다.

output[].url은 단기 유효합니다(몇 시간, 응답에 expireAt 포함). 이를 리다이렉트처럼 취급하세요. 즉시 바이트를 다운로드하여 직접 저장하세요(S3, R2, 자체 CDN). 원본 URL을 페이지에 절대 삽입하지 마세요.

프로덕션 레벨 패턴

위의 최소 예제는 데모에는 적합하지만, 출시 전에는 추가로 네 가지가 필요합니다.

가능하면 폴링 대신 콜백 사용

5초마다 폴링하는 것은 요청을 낭비하고 워커를 1~2분간 점유합니다. 환경에 공개 webhook URL이 있다면 API에 전달하여 콜백을 받으세요:

resp = requests.post(
    f"{API_BASE}/tasks",
    headers={"Authorization": f"Bearer {API_KEY}",
             "Content-Type": "application/json"},
    json={
        "model": "wan2.7-video/text-to-video",
        "input": {"prompt": prompt, "ratio": "16:9", "duration": 5, "resolution": "720P"},
        "callback": {
            "url": "https://yourapp.com/hooks/hiapi",
            "when": "final",  # 상태가 success 또는 fail일 때만 콜백합니다
        },
    },
    timeout=60,
)

webhook은 전체 작업 객체(폴링 응답과 동일한 구조)를 받으므로 다운로드 단계는 완전히 동일합니다.

폴링 vs 콜백 사용 시점:

• 폴링: 스크립트, notebook, 배치 작업, 그리고 HTTP 리스너를 노출하고 싶지 않은 모든 경우.
• 콜백: 웹 앱, 큐 워커, 그리고 공개 엔드포인트가 있는 모든 경우. 항상 멱등성(다음 섹션)과 함께 사용하세요. 제공업체가 재시도할 수 있습니다.

제출을 멱등하게 만들기

코드가 create_task를 재시도하면(네트워크 지터, 루프 중간 크래시) 동일한 프롬프트를 중복 제출하여 두 번 과금되기 쉽습니다. 저렴한 해결책: 입력을 안정적인 클라이언트 키로 해시하고 호출 전에 자체 테이블에서 중복 제거하세요:

import hashlib, json

def submit_key(prompt: str, aspect_ratio: str, duration: int) -> str:
    payload = json.dumps({"p": prompt, "a": aspect_ratio, "d": duration},
                         sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

이 키를 Idempotency-Key 요청 헤더로 보내세요(최대 255바이트). hiapi는 계정과 키당 하나의 작업만 생성하며, 동일한 키를 사용한 재시도는 중복 과금 대신 원래 taskId를 반환합니다. 이를 기반으로 (submit_key → task_id)를 자체 데이터베이스에 저장하면 나중에 작업을 조회할 때도 도움이 됩니다.

오류를 정확히 처리하기

hiapi에서 두 가지 오류 래퍼를 보게 됩니다:

{ "error": {
    "code": "permission_denied",
    "message": "This API key cannot use the selected model.",
    "request_id": "20260626…",
    "type": "hiapi_error"
} }

…… 인증/권한 실패 시(HTTP 401/403), 그리고 작업이 최종 실패할 때:

{ "data": { "status": "fail",
            "error": {"code": "...", "message": "..."} } }

이들을 구분하여 처리하세요. 생성 시의 permission_denied는 키가 잘못되었거나 해당 모델을 포함하지 않음을 의미하므로 재시도가 무의미합니다. 폴링 후 작업의 status: "fail"은 일시적 모델 오류일 수 있으므로 동일한 프롬프트로 한 번 재시도할 가치가 있습니다. 지원 담당자가 추적할 수 있도록 항상 request_id를 기록하세요.

가격 및 할당량

비디오 생성은 요청이 아닌 출력 초 단위로 과금되며, 해상도와 모델 티어에 따라 다릅니다. 단위 경제성을 계산하기 전에 hiapi 가격 페이지의 실시간 수치를 확인하세요. 저장소에 달러 값을 하드코딩하지 마세요.

관련 자료

• Wan 2.7 텍스트-투-비디오 모델 페이지 — 모델 카드, 지원되는 aspect_ratio/duration 값, 그리고 동일한 /v1/tasks 호출을 발행하는 playground.
• hiapi 문서 — 비동기 작업 계약의 전체 참조(/v1/tasks 생성, 폴링, 콜백 스키마, 오류 코드).
• 모델 카탈로그 — 동일한 호출 구조를 공유하는 이미지, 비디오, 오디오 모델.

자주 묻는 질문

Q: 비디오 모델과 이미지 모델에 서로 다른 SDK가 필요한가요? A: 아닙니다. 둘 다 POST /v1/tasks에 {model, input}을 담아 보내고, 둘 다 {data:{taskId}}를 반환합니다. 모델별 input 필드만 다릅니다. 비디오는 duration을 받고 이미지는 받지 않습니다. 폴링 루프는 완전히 동일합니다.

Q: 왜 output[].url이 몇 시간 후에 404가 되나요? A: 해당 URL은 서명되었으며 만료됩니다(응답의 expireAt은 Unix 타임스탬프). 즉시 바이트를 다운로드하여 자체 스토리지에 푸시하고 거기서 제공하세요.

Q: 다시 작성하지 않고 Runway Gen-3에서 Wan 2.7로 전환하려면 어떻게 해야 하나요? A: 제공업체가 동일한 비동기 작업 구조(생성 → 폴링 → 출력 URL)를 사용한다면 두 개의 문자열을 변경하세요: API_BASE와 model id. 이를 환경 변수(API_BASE, MODEL_ID)로 감싸면 동일한 코드로 A/B 테스트를 할 수 있습니다.

Q: 키가 올바른 것 같은데 permission_denied가 발생합니다. A: 두 가지를 확인하세요: (1) 키가 특정 모델에 액세스할 수 있는지(일부 모델은 유료 티어 필요), 그리고 (2) Authorization 헤더가 정확히 공백이 있는 Bearer sk-...인지 — Bearer: sk-...나 원본 키가 아닌지. 오류 본문의 전체 request_id는 지원 담당자가 추적하는 데 필요합니다.

Q: 폴링을 해야 하나요, 콜백을 사용해야 하나요? A: 공개 HTTP 엔드포인트가 있는 환경(웹 앱, 큐 워커)에서는 콜백을 사용하세요. 스크립트, cron 작업, notebook처럼 webhook을 설정하는 것이 어색한 모든 곳에서는 폴링을 사용하세요. 어느 쪽이든 taskId를 저장하여 크래시된 워커가 재제출 대신 폴링으로 복구할 수 있도록 하세요.