Writing

Generating Video with Genkit Python: Veo's New First-Class generate() API

The normal Genkit model path is ai.generate() calling a model action, waiting for a ModelResponse, and returning it. That path assumes the model responds in mil

Video generation is genuinely different from text generation. A Gemini call finishes in two seconds. A Veo call takes 30 to 120 seconds, and the API does not block waiting for it. You get an operation handle back immediately, then poll until it’s done. That pattern is fine at the SDK level, but it’s awkward to wire by hand every time, and the raw google-genai SDK leaves the polling, error handling, and response extraction entirely to you. PR #5692 lands a first-class Veo interface in Genkit Python that makes the generate-and-poll pattern as simple to call as ai.generate(), with ai.generate_operation() and ai.check_operation() handling the machinery.

This post walks through the full workflow: how Veo is registered as a background model, how to start a generation, how the poll loop works, how to extract the video URL, and the traps that get you in production.

How background models work in Genkit

The normal Genkit model path is ai.generate() calling a model action, waiting for a ModelResponse, and returning it. That path assumes the model responds in milliseconds to seconds.

Veo does not fit that assumption. Under the hood, calling generateVideos on the Google GenAI SDK starts a long-running operation (LRO) and returns a job name like operations/12345. You call operations.get() repeatedly until done=True, then extract the video URIs from the response.

Genkit exposes this through the BackgroundAction class. When the GoogleAI plugin loads, it registers each Veo model under the background-model action type rather than the standard model action type. A BackgroundAction wraps three operations: start (kick off the job), check (poll status), and an optional cancel. The action keys look like /background-model/googleai/veo-3.1-generate-preview.

The new generate_operation() method on Genkit provides a single entry point that hides which model type you’re dealing with. For models that support long-running operations, it starts the background action and returns the initial Operation. For everything else, it raises an error explaining the mismatch. check_operation() takes an operation and returns an updated one, routing to the right check action via the operation.action field the framework sets automatically.

Starting a generation

The two model families behave differently:

For GoogleAI (Gemini API), Veo is registered as a background model and generation is always asynchronous:

import asyncio
import time

from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
from genkit._core._background import lookup_background_action
from genkit._core._typing import Operation, Part, Role, TextPart
from genkit.model import Message, ModelRequest

ai = Genkit(plugins=[GoogleAI()])

async def generate_video(prompt: str, model: str = 'googleai/veo-3.1-generate-preview') -> Operation:
    action = await lookup_background_action(ai.registry, f'/background-model/{model}')
    if action is None:
        raise ValueError(f'Model not found: {model}')

    operation = await action.start(
        ModelRequest(
            messages=[Message(role=Role.USER, content=[Part(root=TextPart(text=prompt))])],
            config={
                'aspectRatio': '16:9',
                'durationSeconds': 8,
            },
        )
    )
    return operation

The start() call returns almost immediately. operation.done will be False. operation.id holds the operation name (e.g., operations/xyz) you need to poll with.

For VertexAI, the Veo model is registered differently and its generate() method blocks until the LRO completes. The Vertex path wraps the blocking wait internally, so you use the standard ai.generate() call and get a completed ModelResponse back:

from genkit.plugins.google_genai import VertexAI

ai_vertex = Genkit(plugins=[VertexAI()])

# This blocks until Veo finishes (can take 30-120 seconds).
response = await ai_vertex.generate(
    model='vertexai/veo-3.0-generate-001',
    prompt='A paper airplane gliding through a bright classroom',
    config={'aspectRatio': '16:9', 'durationSeconds': 5},
)

The article focuses on the GoogleAI asynchronous path because that’s where the poll loop complexity lives.

The poll loop

Once you have an Operation, you check it on a timer until done=True. The practical timeout for Veo is 180 seconds; short videos (5s at 720p) tend to finish in 40 to 60 seconds. 8-second 1080p videos take closer to 90 seconds.

async def poll_operation(operation: Operation, timeout_seconds: int = 180) -> Operation:
    action = await lookup_background_action(ai.registry, f'/background-model/{model_name}')
    if action is None:
        raise ValueError('Background action disappeared during poll')

    deadline = time.monotonic() + timeout_seconds
    while not operation.done:
        if time.monotonic() > deadline:
            raise TimeoutError(f'Veo operation timed out after {timeout_seconds}s')
        await asyncio.sleep(5)
        operation = await action.check(operation)

    if operation.error:
        raise RuntimeError(f'Veo failed: {operation.error.message}')

    return operation

A few things to notice here. The sleep before the first check is important: polling immediately after start() always returns done=False, wasting a round-trip. Five seconds is a reasonable floor. You also need to preserve the operation object across iterations because each check() call returns a fresh object with the updated state; do not mutate the original.

The operation.error check matters too. done=True does not mean success. The operation can complete with an error (quota exceeded, content policy rejection, invalid prompt). Check operation.error before trying to extract the video URL.

Extracting the video URL

When the operation finishes successfully, the video URI lives inside operation.output:

def extract_video_url(operation: Operation) -> str | None:
    if not isinstance(operation.output, dict):
        return None
    message = operation.output.get('message', {})
    content = message.get('content', [])
    if not content:
        return None
    media = content[0].get('media', {})
    return media.get('url')

The URL format is a Google Cloud Storage URI: https://generativelanguage.googleapis.com/v1beta/files/some-id. The file is available for 48 hours. If you need it longer, download it to your own storage before that window closes.

A complete end-to-end flow pulling it all together:

@ai.flow(name='generate_video')
async def veo_flow(prompt: str) -> dict:
    model = 'googleai/veo-3.1-generate-preview'
    action = await lookup_background_action(ai.registry, f'/background-model/{model}')
    if action is None:
        raise ValueError(f'Veo not found: {model}')

    operation = await action.start(
        ModelRequest(
            messages=[Message(role=Role.USER, content=[Part(root=TextPart(text=prompt))])],
            config={'aspectRatio': '16:9', 'durationSeconds': 5},
        )
    )

    deadline = time.monotonic() + 180
    while not operation.done:
        if time.monotonic() > deadline:
            raise TimeoutError('Veo timed out')
        await asyncio.sleep(5)
        operation = await action.check(operation)

    if operation.error:
        raise RuntimeError(f'Generation failed: {operation.error.message}')

    url = extract_video_url(operation)
    return {'operation_id': operation.id, 'video_url': url}

VeoConfig options

VeoConfigSchema (aliased as VeoConfig) controls the generation:

from genkit.plugins.google_genai.models.veo import VeoConfig

config = VeoConfig(
    aspect_ratio='16:9',       # '16:9', '9:16', '1:1', '4:3', '3:4'
    duration_seconds=8,        # 5 to 8 seconds depending on model
    resolution='1080p',        # '720p', '1080p'; model-dependent
    seed=42,                   # Deterministic output for a given prompt
    negative_prompt='blurry, low quality',
    enhance_prompt=True,       # Let the model improve your prompt
    person_generation='allow_adult',
)

The config fields use camelCase aliases when passed as a dict (aspectRatio, durationSeconds, etc.) and snake_case when using the Pydantic model. Both work via model_dump(by_alias=True).

A note on resolution: resolution is only honored by Veo 3.0 and above. Veo 2.0 ignores it. Similarly, enhance_prompt is Veo 3.x only. If you pass unsupported options to an older model, they are silently dropped.

Veo 3.1 capabilities

The VeoVersion enum in the plugin covers eight model variants:

from genkit.plugins.google_genai.models.veo import VeoVersion

# Stable models
VeoVersion.VEO_2_0             # 'veo-2.0-generate-001' — original, 5-second video
VeoVersion.VEO_3_0             # 'veo-3.0-generate-001' — adds audio generation
VeoVersion.VEO_3_0_FAST        # 'veo-3.0-fast-generate-001' — faster, lower cost
VeoVersion.VEO_3_1             # 'veo-3.1-generate-001' — image-to-video + audio
VeoVersion.VEO_3_1_FAST        # 'veo-3.1-fast-generate-001'

# Preview models
VeoVersion.VEO_3_1_PREVIEW     # 'veo-3.1-generate-preview'
VeoVersion.VEO_3_1_FAST_PREVIEW

Veo 3.1 supports image-to-video: you provide a reference image as part of the request content. The current Python plugin notes this in the TODO comment in VeoModel._build_prompt() (line: “Support image input if Veo supports it (e.g. for image-to-video)”). The code path is not yet wired in this file; check the plugin’s __init__.py for any updates when building against the latest main.

The response shape regression

One thing worth understanding: _from_veo_operation() handles two different shapes for the response body.

When action.start() returns, the operation dict uses a plain dict structure:

{ 'response': { 'generateVideoResponse': { 'generatedSamples': [...] } } }

When action.check() returns, the SDK gives back a Pydantic GenerateVideosResponse object directly. Before the fix in this file, passing the Pydantic object to response extraction would raise AttributeError: 'GenerateVideosResponse' object has no attribute 'get'. The test suite (veo_test.py) now covers both paths explicitly with TestFromVeoOperation.test_pydantic_response_with_videos. If you’re using an older version of the plugin (before this fix), your poll loop may crash on the first check that returns data.

Contrast with the raw SDK

Without Genkit, the same flow using google-genai directly looks like this:

from google import genai
from google.genai import types

client = genai.Client(api_key='...')

# Start
op = await client.aio.models.generate_videos(
    model='veo-3.1-generate-preview',
    prompt='A paper airplane...',
    config=types.GenerateVideosConfig(aspect_ratio='16:9', number_of_videos=1),
)

# Poll
while not op.done:
    await asyncio.sleep(5)
    op = await client.aio.operations.get(operation=op)

# Extract
for gv in op.result().generated_videos:
    print(gv.video.uri)

That is roughly the same amount of code. What Genkit adds on top: the operation is tracked through the registry, operation.action carries the action key so you can resume polling across process restarts, error objects are normalized to Operation.error with a consistent shape, and the whole flow composes with Genkit’s observability and middleware layers. For a one-off script the SDK is fine. For a production flow that runs inside a Cloud Run job and needs tracing, the background model pattern earns its keep.

Production traps

A few things that bite people:

The operation TTL is 24 hours from when you started it. If your polling process crashes and you restart it hours later, the operation might be expired. Store operation.id durably (Cloud Firestore works) before the first poll, and handle the case where check() returns a 404.

Veo does not stream progress. There is no percentage-complete. The only states are in-progress, done-success, and done-error. A 90-second wall-clock wait with no feedback is normal.

The durationSeconds option is a requested duration, not a guarantee. Veo may generate a shorter video if the content completes sooner. Plan for this when you’re building UIs that tell users how long to wait.

Content policy errors come back as operation.done=True with operation.error set, not as HTTP errors. Your client code needs to check operation.error after the loop, not just assume success when done=True.

The video_url in the output is a temporary signed URI. Fetching it works immediately after generation completes, but the URL expires. If you need the video for more than 48 hours, download and store it yourself before serving it to users.