Code Examples

cURL

Simple Text-to-Video

curl -X POST https://platform.runblob.io/v1/sora/generate \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A beautiful sunset over the ocean with waves crashing on the shore"
  }'

Landscape Video (15 seconds)

curl -X POST https://platform.runblob.io/v1/sora/generate \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "City streets at night with neon lights and busy traffic",
    "orientation": "16:9",
    "duration": 15
  }'

Image-to-Video with URL

curl -X POST https://platform.runblob.io/v1/sora/generate \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "Make the clouds move slowly across the sky",
    "image_url": "https://example.com/landscape.jpg",
    "orientation": "16:9",
    "duration": 10
  }'

Check Status

curl -X GET https://platform.runblob.io/v1/sora/generations/GENERATION_ID/ \
  -H "Authorization: Bearer YOUR_TOKEN"

Get All Generations

curl -X GET "https://platform.runblob.io/v1/sora/generations?limit=20&offset=0" \
  -H "Authorization: Bearer YOUR_TOKEN"

Python

Simple Client

import requests
import time

class SoraClient:
    def __init__(self, token):
        self.token = token
        self.base_url = "https://platform.runblob.io"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
    
    def generate(self, prompt, **kwargs):
        response = requests.post(
            f"{self.base_url}/v1/sora/generate",
            headers=self.headers,
            json={"prompt": prompt, **kwargs}
        )
        response.raise_for_status()
        return response.json()
    
    def generate_from_image(self, prompt, image_url, **kwargs):
        response = requests.post(
            f"{self.base_url}/v1/sora/generate",
            headers=self.headers,
            json={
                "prompt": prompt,
                "image_url": image_url,
                **kwargs
            }
        )
        response.raise_for_status()
        return response.json()
    
    def status(self, generation_id):
        response = requests.get(
            f"{self.base_url}/v1/sora/generations/{generation_id}/",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()
    
    def list_generations(self, limit=50, offset=0):
        response = requests.get(
            f"{self.base_url}/v1/sora/generations",
            headers=self.headers,
            params={"limit": limit, "offset": offset}
        )
        response.raise_for_status()
        return response.json()
    
    def wait_for_completion(self, generation_id, poll_interval=5, timeout=900):
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            result = self.status(generation_id)
            
            if result["status"] == "completed":
                return result["video_url"]
            elif result["status"] == "failed":
                raise Exception(f"Generation failed: {result['error_message']}")
            elif result["status"] == "processing":
                print(f"Progress: {result['progress_pct']}%")
            elif result["status"] == "pending":
                print(f"Position in queue: {result['progress_pos_in_queue']}")
            
            time.sleep(poll_interval)
        
        raise Exception("Generation timeout")

# Text-to-Video
client = SoraClient("YOUR_TOKEN")
result = client.generate("A sunset over mountains")
video_url = client.wait_for_completion(result["generation_id"])
print(f"Video ready: {video_url}")

# Image-to-Video
result = client.generate_from_image(
    "Animate this landscape",
    image_url="https://example.com/image.jpg",
    orientation="16:9",
    duration=15
)
video_url = client.wait_for_completion(result["generation_id"])

Image-to-Video with Base64

import base64
import requests

# Read and encode image
with open("input_image.jpg", "rb") as f:
    image_data = base64.b64encode(f.read()).decode("utf-8")

# Create request
response = requests.post(
    "https://platform.runblob.io/v1/sora/generate",
    headers={
        "Authorization": "Bearer YOUR_TOKEN",
        "Content-Type": "application/json"
    },
    json={
        "prompt": "Animate this portrait with subtle movements",
        "image_base64": f"data:image/jpeg;base64,{image_data}",
        "orientation": "9:16",
        "duration": 10
    }
)

result = response.json()
print(f"Generation ID: {result['generation_id']}")

Advanced Usage with Error Handling

import requests
import time
from typing import Optional

class SoraAPI:
    def __init__(self, token: str):
        self.token = token
        self.base_url = "https://platform.runblob.io/v1/sora"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
    
    def generate(
        self,
        prompt: str,
        orientation: str = "9:16",
        duration: int = 10,
        **kwargs
    ) -> dict:
        """Create a new video generation."""
        payload = {
            "prompt": prompt,
            "orientation": orientation,
            "duration": duration,
            **kwargs
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/generate",
                headers=self.headers,
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 402:
                raise Exception("Insufficient credits")
            elif e.response.status_code == 401:
                raise Exception("Invalid token")
            elif e.response.status_code == 422:
                raise Exception(f"Validation error: {e.response.json()}")
            else:
                raise Exception(f"API Error: {e.response.text}")
    
    def get_status(self, generation_id: str) -> dict:
        """Get generation status."""
        try:
            response = requests.get(
                f"{self.base_url}/generations/{generation_id}/",
                headers=self.headers
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 404:
                raise Exception("Generation not found")
            raise Exception(f"API Error: {e.response.text}")
    
    def wait_for_completion(
        self,
        generation_id: str,
        poll_interval: int = 5,
        timeout: int = 900,
        callback = None
    ) -> Optional[str]:
        """Wait for generation to complete with progress callback."""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            result = self.get_status(generation_id)
            
            # Call progress callback if provided
            if callback:
                callback(result)
            
            if result["status"] == "completed":
                return result["video_url"]
            elif result["status"] == "failed":
                raise Exception(f"Generation failed: {result['error_message']}")
            
            time.sleep(poll_interval)
        
        raise Exception("Generation timeout")
    
    def get_balance(self) -> float:
        """Check account balance."""
        response = requests.get(
            "https://platform.runblob.io/v1/data/balance",
            headers=self.headers
        )
        response.raise_for_status()
        return float(response.json()["balance"])

# Usage with progress callback
def progress_callback(status_data):
    if status_data["status"] == "processing":
        print(f"Progress: {status_data['progress_pct']}%")
    elif status_data["status"] == "pending":
        print(f"Queue position: {status_data['progress_pos_in_queue']}")

try:
    client = SoraAPI("YOUR_TOKEN")
    
    # Check balance first
    balance = client.get_balance()
    print(f"Current balance: ${balance}")
    
    if balance < 0.12:
        print("Insufficient balance for generation")
    else:
        # Generate video
        result = client.generate(
            prompt="A beautiful sunset over the ocean",
            orientation="9:16",
            duration=10
        )
        
        print(f"Generation started: {result['generation_id']}")
        print(f"Price: ${result['price']}")
        
        # Wait for completion with progress updates
        video_url = client.wait_for_completion(
            result['generation_id'],
            callback=progress_callback
        )
        
        print(f"Video ready: {video_url}")
        
except Exception as e:
    print(f"Error: {str(e)}")

JavaScript / TypeScript

React Hook for Sora Generation

import { useState, useCallback } from 'react';

interface GenerationStatus {
  generation_id: string;
  status: 'pending' | 'processing' | 'completed' | 'failed';
  video_url: string | null;
  progress_pct: number | null;
  progress_pos_in_queue: number | null;
  error_message: string | null;
}

interface UseSoraOptions {
  token: string;
  pollInterval?: number;
}

export function useSora({ token, pollInterval = 5000 }: UseSoraOptions) {
  const [status, setStatus] = useState<GenerationStatus | null>(null);
  const [loading, setLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);

  const generate = useCallback(async (
    prompt: string,
    options?: {
      orientation?: '9:16' | '16:9';
      duration?: 10 | 15;
      image_url?: string;
      callback_url?: string;
    }
  ) => {
    setLoading(true);
    setError(null);

    try {
      const response = await fetch('https://platform.runblob.io/v1/sora/generate', {
        method: 'POST',
        headers: {
          'Authorization': `Bearer ${token}`,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({
          prompt,
          orientation: '9:16',
          duration: 10,
          ...options,
        }),
      });

      if (!response.ok) {
        if (response.status === 402) {
          throw new Error('Insufficient credits');
        }
        throw new Error('Failed to create generation');
      }

      const data = await response.json();
      
      // Start polling
      pollStatus(data.generation_id);

      return data;
    } catch (err) {
      setError(err instanceof Error ? err.message : 'Unknown error');
      setLoading(false);
      throw err;
    }
  }, [token, pollInterval]);

  const pollStatus = useCallback((generationId: string) => {
    const interval = setInterval(async () => {
      try {
        const response = await fetch(
          `https://platform.runblob.io/v1/sora/generations/${generationId}/`,
          {
            headers: {
              'Authorization': `Bearer ${token}`,
            },
          }
        );

        if (!response.ok) {
          throw new Error('Failed to get status');
        }

        const data: GenerationStatus = await response.json();
        setStatus(data);

        // Stop polling if completed or failed
        if (data.status === 'completed' || data.status === 'failed') {
          clearInterval(interval);
          setLoading(false);
        }
      } catch (err) {
        setError(err instanceof Error ? err.message : 'Unknown error');
        clearInterval(interval);
        setLoading(false);
      }
    }, pollInterval);

    return () => clearInterval(interval);
  }, [token, pollInterval]);

  return {
    generate,
    status,
    loading,
    error,
  };
}

// Usage in component
function VideoGenerator() {
  const { generate, status, loading, error } = useSora({
    token: 'YOUR_TOKEN',
  });

  const handleGenerate = async () => {
    try {
      await generate('A beautiful sunset over the ocean', {
        orientation: '9:16',
        duration: 10,
      });
    } catch (err) {
      console.error(err);
    }
  };

  return (
    <div>
      <button onClick={handleGenerate} disabled={loading}>
        Generate Video
      </button>

      {loading && status && (
        <div>
          <p>Status: {status.status}</p>
          {status.status === 'pending' && status.progress_pos_in_queue && (
            <p>Position in queue: {status.progress_pos_in_queue}</p>
          )}
          {status.status === 'processing' && status.progress_pct && (
            <progress value={status.progress_pct} max={100}>
              {status.progress_pct}%
            </progress>
          )}
        </div>
      )}

      {status?.status === 'completed' && status.video_url && (
        <video src={status.video_url} controls />
      )}

      {status?.status === 'failed' && (
        <p className="error">{status.error_message}</p>
      )}

      {error && <p className="error">{error}</p>}
    </div>
  );
}

Node.js Client

class SoraAPI {
  constructor(token) {
    this.token = token;
    this.baseUrl = 'https://platform.runblob.io/v1/sora';
  }

  async generate(params) {
    const response = await fetch(`${this.baseUrl}/generate`, {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${this.token}`,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({
        orientation: '9:16',
        duration: 10,
        ...params
      })
    });

    if (!response.ok) {
      throw new Error(`API Error: ${response.statusText}`);
    }

    return response.json();
  }

  async getStatus(generationId) {
    const response = await fetch(
      `${this.baseUrl}/generations/${generationId}/`,
      {
        headers: {
          'Authorization': `Bearer ${this.token}`
        }
      }
    );

    if (!response.ok) {
      throw new Error(`API Error: ${response.statusText}`);
    }

    return response.json();
  }

  async waitForCompletion(generationId, pollInterval = 5000) {
    while (true) {
      const result = await this.getStatus(generationId);
      
      if (result.status === 'completed') {
        return result.video_url;
      } else if (result.status === 'failed') {
        throw new Error(`Generation failed: ${result.error_message}`);
      }
      
      await new Promise(resolve => setTimeout(resolve, pollInterval));
    }
  }
}

// Usage
const client = new SoraAPI('YOUR_TOKEN');

client.generate({
  prompt: 'A beautiful sunset over the ocean'
})
  .then(result => {
    console.log('Generation ID:', result.generation_id);
    return client.waitForCompletion(result.generation_id);
  })
  .then(videoUrl => {
    console.log('Video ready:', videoUrl);
  })
  .catch(error => {
    console.error('Error:', error.message);
  });

Getting Started

Authentication

API Endpoints

cURL

Simple Text-to-Video

Landscape Video (15 seconds)

Image-to-Video with URL

Check Status

Get All Generations

Python

Simple Client

Image-to-Video with Base64

Advanced Usage with Error Handling

JavaScript / TypeScript

React Hook for Sora Generation

Node.js Client

Getting Started

Authentication

API Endpoints

​cURL

​Simple Text-to-Video

​Landscape Video (15 seconds)

​Image-to-Video with URL

​Check Status

​Get All Generations

​Python

​Simple Client

​Image-to-Video with Base64

​Advanced Usage with Error Handling

​JavaScript / TypeScript

​React Hook for Sora Generation

​Node.js Client

cURL

Simple Text-to-Video

Landscape Video (15 seconds)

Image-to-Video with URL

Check Status

Get All Generations

Python

Simple Client

Image-to-Video with Base64

Advanced Usage with Error Handling

JavaScript / TypeScript

React Hook for Sora Generation

Node.js Client