Runway API Quickstart

{
  "prompt": "The character starts walking forward with confidence",
  "imageUrl": "https://example.com/character-image.jpg",
  "duration": 5,
  "quality": "720p",
  "aspectRatio": "16:9"
}

{
  "taskId": "ee603959-debb-48d1-98c4-a6d1c717eba6",
  "prompt": "The cat continues dancing with more energy and spins around",
  "quality": "720p"
}

import requests
import time

class RunwayAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://api.kie.ai/api/v1/runway'
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json'
        }
    
    def generate_video(self, **options):
        response = requests.post(f'{self.base_url}/generate', 
                               headers=self.headers, json=options)
        result = response.json()
        
        if not response.ok or result.get('code') != 200:
            raise Exception(f"Generation failed: {result.get('msg', 'Unknown error')}")
        
        return result['data']['taskId']
    
    def wait_for_completion(self, task_id, max_wait_time=600):
        start_time = time.time()
        
        while time.time() - start_time < max_wait_time:
            status = self.get_task_status(task_id)
            state = status['state']
            
            if state == 'wait':
                print("Task is waiting, continue waiting...")
            elif state == 'queueing':
                print("Task is queueing, continue waiting...")
            elif state == 'generating':
                print("Task is generating, continue waiting...")
            elif state == 'success':
                print("Generation completed successfully!")
                return status
            elif state == 'fail':
                error = status.get('failMsg', 'Task generation failed')
                print(f"Task generation failed: {error}")
                raise Exception(error)
            else:
                print(f"Unknown status: {state}")
                if status.get('failMsg'):
                    print(f"Error message: {status['failMsg']}")
            
            time.sleep(30)  # Wait 30 seconds
        
        raise Exception('Generation timeout')
    
    def get_task_status(self, task_id):
        response = requests.get(f'{self.base_url}/record-detail?taskId={task_id}',
                              headers={'Authorization': f'Bearer {self.api_key}'})
        result = response.json()
        
        if not response.ok or result.get('code') != 200:
            raise Exception(f"Status check failed: {result.get('msg', 'Unknown error')}")
        
        return result['data']
    
    def extend_video(self, task_id, prompt, quality='720p'):
        data = {
            'taskId': task_id,
            'prompt': prompt,
            'quality': quality
        }
        
        response = requests.post(f'{self.base_url}/extend', 
                               headers=self.headers, json=data)
        result = response.json()
        
        if not response.ok or result.get('code') != 200:
            raise Exception(f"Extension failed: {result.get('msg', 'Unknown error')}")
        
        return result['data']['taskId']

# Usage example
def main():
    api = RunwayAPI('YOUR_API_KEY')
    
    try:
        # Text-to-video generation
        print('Starting video generation...')
        task_id = api.generate_video(
            prompt='A fluffy orange cat dancing energetically in a colorful room with disco lights',
            duration=5,
            quality='720p',
            aspectRatio='16:9',
            waterMark=''
        )
        
        # Wait for completion
        print(f'Task ID: {task_id}. Waiting for completion...')
        result = api.wait_for_completion(task_id)
        
        print('Video generation successful!')
        print(f'Video URL: {result["videoInfo"]["videoUrl"]}')
        print(f'Thumbnail URL: {result["videoInfo"]["imageUrl"]}')
        
        # Extend video
        print('\nStarting video extension...')
        extend_task_id = api.extend_video(task_id, 'The cat spins around with increasing energy and excitement', '720p')
        
        extend_result = api.wait_for_completion(extend_task_id)
        print('Video extension successful!')
        print(f'Extended video URL: {extend_result["videoInfo"]["videoUrl"]}')
        
    except Exception as error:
        print(f'Error: {error}')

if __name__ == '__main__':
    main()