Async API

Content-Type: application/json
Authorization: sk-APIKEY

{
  "id": "video_abc123",
  "object": "video",
  "created": 1762181811,
  "status": "queued",
  "model": "veo-3.1"
}

{
  "id": "video_abc123",
  "object": "video",
  "created": 1762181811,
  "status": "completed",
  "model": "veo-3.1",
  "prompt": "A cute kitten playing",
  "url": "https://veo-video.gptkey.asia/assets/flow/xxx.mp4",
  "duration": 8,
  "resolution": "720x1280"
}

import requests
import time

class VEOClient:
    """VEO 3.1 Async API Client"""

    def __init__(self, api_key: str, base_url: str = "https://api.apiyi.com"):
        self.base_url = base_url
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": api_key
        }

    def create_video(self, prompt: str, model: str = "veo-3.1") -> str:
        """Create video task, returns video_id"""
        response = requests.post(
            f"{self.base_url}/v1/videos",
            headers=self.headers,
            json={"prompt": prompt, "model": model}
        )
        response.raise_for_status()
        return response.json()["id"]

    def get_status(self, video_id: str) -> dict:
        """Query task status"""
        response = requests.get(
            f"{self.base_url}/v1/videos/{video_id}",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def get_content(self, video_id: str) -> dict:
        """Get video content"""
        response = requests.get(
            f"{self.base_url}/v1/videos/{video_id}/content",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

    def wait_for_completion(self, video_id: str, timeout: int = 600, interval: int = 5) -> dict:
        """Wait for task completion"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            status_data = self.get_status(video_id)
            status = status_data.get("status")

            if status == "completed":
                return self.get_content(video_id)
            elif status == "failed":
                raise Exception(f"Video generation failed: {status_data}")

            print(f"Task in progress... Status: {status}")
            time.sleep(interval)

        raise TimeoutError("Timeout waiting for completion")


# Usage example
if __name__ == "__main__":
    client = VEOClient("sk-your-api-key")

    try:
        # 1. Create task
        video_id = client.create_video(
            prompt="A kitten walking in a sunny garden",
            model="veo-3.1-fast"
        )
        print(f"Task created, ID: {video_id}")

        # 2. Wait for completion and get result
        result = client.wait_for_completion(video_id)
        print(f"Video generated successfully!")
        print(f"Video URL: {result['url']}")
        print(f"Resolution: {result['resolution']}")

    except Exception as e:
        print(f"Error: {e}")

{
  "id": "video_xyz789",
  "object": "video",
  "created": 1762181811,
  "status": "queued",
  "model": "veo-3.1-landscape-fl"
}

{
  "error": {
    "code": "invalid_api_key",
    "message": "Invalid API key provided",
    "type": "authentication_error"
  }
}