Google Veo 3 API Guide: Video Generation with Audio for Developers

import google.generativeai as genai
import time

genai.configure(api_key="your-google-api-key")

# Generate video with audio
model = genai.GenerativeModel("veo-3")

response = model.generate_content(
    "A barista making a latte in a cozy coffee shop. The espresso machine hisses, "
    "milk steams, and soft jazz plays in the background. Morning light streams through "
    "the window. Cinematic, shallow depth of field.",
    generation_config={
        "response_modalities": ["video"],
        "video_config": {
            "resolution": "720p",
            "duration_seconds": 6,
            "include_audio": True
        }
    }
)

# Get the generated video
video = response.candidates[0].content.parts[0]
with open("output.mp4", "wb") as f:
    f.write(video.data)

from google.cloud import aiplatform
from google.protobuf import struct_pb2

aiplatform.init(project="your-project", location="us-central1")

endpoint = aiplatform.Endpoint(
    endpoint_name="publishers/google/models/veo-3"
)

instance = struct_pb2.Struct()
instance.update({
    "prompt": "A drone shot flying over a tropical beach at sunset. "
              "Waves crash on the shore, seagulls call overhead. "
              "4K cinematic quality.",
    "video_config": {
        "duration_seconds": 8,
        "resolution": "1080p",
        "fps": 24,
        "include_audio": True,
        "audio_config": {
            "include_dialogue": False,
            "include_sfx": True,
            "include_music": True
        }
    }
})

response = endpoint.predict(instances=[instance])
video_bytes = response.predictions[0]["video"]

from openai import OpenAI
import base64

client = OpenAI(
    api_key="your-crazyrouter-key",
    base_url="https://crazyrouter.com/v1"
)

# Crazyrouter wraps Veo 3 in OpenAI-compatible format
response = client.chat.completions.create(
    model="veo-3",
    messages=[{
        "role": "user",
        "content": "Generate a 6-second video: A cat knocking a glass off a table "
                   "in slow motion. The glass shatters on the floor. Include sound effects."
    }],
    extra_body={
        "video_config": {
            "duration_seconds": 6,
            "resolution": "720p",
            "include_audio": True
        }
    }
)

# Response contains base64 video
video_data = base64.b64decode(response.choices[0].message.content)
with open("cat_video.mp4", "wb") as f:
    f.write(video_data)

[Subject] + [Action] + [Setting] + [Audio Description] + [Style/Camera]

"A chef flambes a pan of shrimp in a professional kitchen. The flame 
whooshes up, oil sizzles, and the chef narrates 'perfectly caramelized.' 
Shot on 35mm film, warm color grading, medium close-up."

"Two friends laughing at a cafe table. One tells a joke, the other 
spits out their coffee. Background chatter and clinking cups. 
Handheld camera, natural lighting, indie film aesthetic."

"A thunderstorm rolling over a wheat field. Lightning cracks, thunder 
rumbles, rain intensifies. Time-lapse clouds, wide angle, dramatic 
contrast."

import time
import random

def generate_with_retry(prompt, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt, ...)
            return response
        except Exception as e:
            if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e):
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Waiting {wait:.1f}s (attempt {attempt + 1})")
                time.sleep(wait)
            else:
                raise
    raise Exception("Max retries exceeded")

import asyncio
from collections import deque

class Veo3Queue:
    def __init__(self, max_concurrent=3, rpm_limit=5):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rpm_limit = rpm_limit
        self.request_times = deque()

    async def generate(self, prompt, config):
        # Rate limit check
        now = time.time()
        while len(self.request_times) >= self.rpm_limit:
            oldest = self.request_times[0]
            if now - oldest > 60:
                self.request_times.popleft()
            else:
                await asyncio.sleep(60 - (now - oldest))
                now = time.time()

        async with self.semaphore:
            self.request_times.append(time.time())
            return await self._call_api(prompt, config)