"Pika 2.2 API Integration Guide: Build Video Generation Pipelines in 2026"

import requests

PIKA_API_KEY = "your-pika-api-key"
BASE_URL = "https://api.pika.art/v1"

headers = {
    "Authorization": f"Bearer {PIKA_API_KEY}",
    "Content-Type": "application/json"
}

def generate_video(prompt: str, duration: int = 4, resolution: str = "720p"):
    """Generate a video from a text prompt"""
    response = requests.post(
        f"{BASE_URL}/generate",
        headers=headers,
        json={
            "prompt": prompt,
            "duration": duration,
            "resolution": resolution,
            "aspect_ratio": "16:9",
            "motion_strength": 3,  # 1-5, higher = more motion
            "guidance_scale": 12,  # How closely to follow prompt
            "negative_prompt": "blurry, low quality, distorted faces"
        }
    )
    return response.json()

# Example
result = generate_video(
    prompt="A futuristic city at night with flying cars and neon signs, "
           "rain falling, cyberpunk atmosphere, cinematic",
    duration=8,
    resolution="1080p"
)
task_id = result["data"]["task_id"]

import base64

def image_to_video(image_path: str, prompt: str, duration: int = 4):
    """Animate a static image"""
    with open(image_path, "rb") as f:
        image_b64 = base64.b64encode(f.read()).decode()
    
    response = requests.post(
        f"{BASE_URL}/generate/image-to-video",
        headers=headers,
        json={
            "image": f"data:image/jpeg;base64,{image_b64}",
            "prompt": prompt,
            "duration": duration,
            "motion_strength": 2,
            "camera_motion": "slow_zoom_in"
        }
    )
    return response.json()

# Animate a product photo
result = image_to_video(
    "product_hero.jpg",
    "Subtle rotation with floating particles, premium feel",
    duration=6
)

def apply_scene_effect(image_path: str, effect: str, target: str = None):
    """Apply Pika 2.2 scene effects
    
    Effects: explode, melt, inflate, crush, dissolve, 
             freeze, burn, grow, shrink, levitate
    """
    with open(image_path, "rb") as f:
        image_b64 = base64.b64encode(f.read()).decode()
    
    response = requests.post(
        f"{BASE_URL}/generate/effects",
        headers=headers,
        json={
            "image": f"data:image/jpeg;base64,{image_b64}",
            "effect": effect,
            "target_description": target,  # e.g., "the red car"
            "intensity": 0.7,  # 0.0-1.0
            "duration": 4
        }
    )
    return response.json()

# Make a building explode
result = apply_scene_effect(
    "city_scene.jpg",
    effect="explode",
    target="the glass skyscraper in the center"
)

# Melt a chocolate bar
result = apply_scene_effect(
    "chocolate.jpg",
    effect="melt",
    target="the chocolate bar"
)

import time

def wait_for_video(task_id: str, timeout: int = 300):
    """Poll until video is ready"""
    start = time.time()
    
    while time.time() - start < timeout:
        response = requests.get(
            f"{BASE_URL}/tasks/{task_id}",
            headers=headers
        )
        data = response.json()["data"]
        
        if data["status"] == "completed":
            return {
                "video_url": data["video_url"],
                "thumbnail": data["thumbnail_url"],
                "duration": data["duration"],
                "cost_credits": data["credits_used"]
            }
        elif data["status"] == "failed":
            raise Exception(f"Generation failed: {data.get('error', 'Unknown')}")
        
        # Adaptive polling: start fast, slow down
        elapsed = time.time() - start
        sleep_time = 2 if elapsed < 30 else 5 if elapsed < 120 else 10
        time.sleep(sleep_time)
    
    raise TimeoutError(f"Video generation timed out after {timeout}s")

# Usage
video = wait_for_video(task_id)
print(f"Video ready: {video['video_url']}")

import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class VideoResult:
    url: str
    model: str
    cost: float
    duration: float

class VideoGenerationPipeline:
    """Multi-model video generation with automatic fallback"""
    
    def __init__(self, crazyrouter_key: str):
        self.api_key = crazyrouter_key
        self.base_url = "https://crazyrouter.com/v1"
        
        # Model priority: cheapest first, fallback to premium
        self.models = [
            {"name": "pika-2.2", "max_retries": 2},
            {"name": "seedance-2.0", "max_retries": 2},
            {"name": "kling-2.1", "max_retries": 1},
            {"name": "veo-3", "max_retries": 1},  # Premium fallback
        ]
    
    async def generate(self, prompt: str, duration: int = 8,
                       resolution: str = "720p") -> VideoResult:
        """Try each model in order until one succeeds"""
        
        last_error = None
        for model_config in self.models:
            model = model_config["name"]
            retries = model_config["max_retries"]
            
            for attempt in range(retries):
                try:
                    result = await self._call_model(
                        model, prompt, duration, resolution
                    )
                    return result
                except RateLimitError:
                    # Move to next model immediately
                    break
                except Exception as e:
                    last_error = e
                    if attempt < retries - 1:
                        await asyncio.sleep(2 ** attempt)
        
        raise Exception(f"All models failed. Last error: {last_error}")
    
    async def _call_model(self, model: str, prompt: str,
                          duration: int, resolution: str) -> VideoResult:
        """Call a specific model via Crazyrouter"""
        import openai
        
        client = openai.OpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )
        
        response = client.chat.completions.create(
            model=model,
            messages=[{
                "role": "user",
                "content": f"Generate video: {prompt}"
            }]
        )
        
        return VideoResult(
            url=response.choices[0].message.content,
            model=model,
            cost=0.0,  # Tracked by Crazyrouter dashboard
            duration=duration
        )

# Usage
pipeline = VideoGenerationPipeline("sk-cr-your-key")

# Generates with Pika 2.2 first, falls back automatically
video = asyncio.run(pipeline.generate(
    "Product unboxing animation with confetti and sparkles",
    duration=8,
    resolution="720p"
))
print(f"Generated with {video.model}: {video.url}")