Build a Conversational Video Clip Service with Gemini Omni Flash and FastAP
On June 30, 2026, Google opened developer access to Gemini Omni Flash, its video generation model, in public preview through the Gemini API at $0.10 per second of output.
Introduction
On June 30, 2026, Google opened developer access to Gemini Omni Flash, its video generation model, in public preview through the Gemini API at $0.10 per second of output. The headline feature is not the generation, it is the refinement: Omni Flash accepts text, images, and video as input, so you can generate a clip, watch it, and fix it with an instruction instead of re-rolling from a rewritten prompt. This tutorial builds a clip service around that loop: a FastAPI app with generate, refine, and cost-ledger endpoints, where every clip keeps its version history and every second of output is metered before it is spent.
What Gemini Omni Flash Is
Omni Flash is Google's unified video model in the Gemini API, model ID
gemini-omni-flash-preview. Three properties shape the architecture of anything you build on
it. Input is multimodal: text, image, and video can be combined in one request, which is what
makes iterative editing possible. Output is capped at 10 seconds per generation while the model
is in preview. And pricing is per second of output, roughly $0.10, so a 10-second clip costs about
a dollar and a sloppy afternoon of retries costs lunch.
Video generation is asynchronous in the Gemini API. You submit a request, get a long-running operation back, and poll it until the video is ready. That asynchrony has to surface in your service design, which is why the job store below exists.
Why Refinement Changes the Pipeline Shape
One-shot video pipelines treat the prompt as source code: if the clip is wrong, you edit the prompt and regenerate from zero, paying full price for a new roll of the dice and hoping nothing else changes. Omni Flash's conversational editing flips that: the previous output video goes back in as input alongside a short instruction ("same shot, slow the camera push, warmer light"), so iteration converges instead of resetting. Version two is built from version one, not from a fresh interpretation of a longer prompt.
That means the unit your service manages is not a video file. It is a clip lineage: an ordered chain of versions, each one a refinement of the last, each with its own cost. Get that object right and the endpoints write themselves
Generate the First Clip
Install the SDK, set your API key from Google AI Studio, and generate one clip end to end.
pip install google-genai fastapi uvicorn
export GEMINI_API_KEY="your-key-here"import time
from google import genai
client = genai.Client()
operation = client.models.generate_videos(
model="gemini-omni-flash-preview",
prompt="Product clip: a matte black ceramic pour-over dripper on an oak counter, "
"slow camera push-in, steam rising, soft morning light, 8 seconds",
)
while not operation.done:
time.sleep(5)
operation = client.operations.get(operation)
video = operation.response.generated_videos[0]
client.files.download(file=video.video)
video.video.save("dripper-v1.mp4")
print("saved dripper-v1.mp4")The poll loop is the part people underestimate. Video operations take noticeably longer than image calls, so nothing here belongs in a request handler; in the service it moves to a background worker.
Wrap Generation in a Job Store
Clips need identity, status, and lineage before anything else works. A small in-memory store is enough to make the pattern visible; swap it for Postgres before two users share it.
import uuid
from dataclasses import dataclass, field
@dataclass
class Version:
video_path: str
instruction: str
seconds: float
cost_usd: float
@dataclass
class Clip:
id: str
status: str = "queued" # queued | rendering | ready | failed
error: str | None = None
versions: list[Version] = field(default_factory=list)
CLIPS: dict[str, Clip] = {}
def new_clip() -> Clip:
clip = Clip(id=uuid.uuid4().hex[:12])
CLIPS[clip.id] = clip
return clipEvery render appends a Version instead of overwriting. When the client asks "why does version four look like that," the answer is in the instruction chain, not in someone's memory of what they typed.
Add the Refine Endpoint
Refinement sends the latest version's video back to the model with a short instruction. This is the endpoint that justifies the whole service. Write render once, with optional sources, and every later capability becomes a kwarg.
def render(clip: Clip, instruction: str,
source_video: str | None = None, source_image: str | None = None):
clip.status = "rendering"
try:
kwargs = {"model": "gemini-omni-flash-preview", "prompt": instruction}
if source_video:
kwargs["video"] = client.files.upload(file=source_video)
if source_image:
kwargs["image"] = client.files.upload(file=source_image)
operation = client.models.generate_videos(**kwargs)
while not operation.done:
time.sleep(5)
operation = client.operations.get(operation)
video = operation.response.generated_videos[0]
path = f"clips/{clip.id}-v{len(clip.versions) + 1}.mp4"
client.files.download(file=video.video)
video.video.save(path)
seconds = getattr(video.video, "duration_seconds", 10.0) or 10.0
clip.versions.append(Version(path, instruction, seconds, seconds * 0.10))
clip.status = "ready"
except Exception as exc:
clip.status, clip.error = "failed", str(exc)The instruction for a refine turn stays short on purpose. You are not re-describing the clip; you are describing the delta. "Same shot, slower push, hold on the steam for the last two seconds" outperforms a second full scene description, because the source video is already carrying the scene.
Seed the First Frame from a Product Still
Omni Flash also takes image input, which means a real product photo (or one you generated in an image pipeline) can anchor the clip so the video matches the asset your catalog already shows. Because render already accepts an image source, seeding from a still is a call site, not new machinery:
clip = new_clip()
render(
clip,
"Animate this product photo: slow push-in, steam rising from the dripper, warm morning light",
source_image="assets/dripper-hero.png",
)Image-seeded clips are the honest option for physical products: the geometry and color come from a real photograph, and the model animates rather than invents. Pure text-to-video is for mood and motion, not for representing a SKU a customer will unbox.
Meter the Seconds Before You Spend Them
Per-second pricing with a refinement loop has a failure mode: an enthusiastic reviewer iterating a clip nine times has spent nine dollars before anyone notices the meter. (Nobody in the approval thread thinks of themselves as the person who bought the tenth version.) The guard belongs in code, and per clip, not per account.
MAX_SECONDS_PER_CLIP = 60.0
PRICE_PER_SECOND = 0.10
def preflight(clip: Clip, requested_seconds: float = 10.0):
spent = sum(v.seconds for v in clip.versions)
if spent + requested_seconds > MAX_SECONDS_PER_CLIP:
raise ValueError(
f"clip {clip.id} has consumed {spent:.0f}s "
f"(${spent * PRICE_PER_SECOND:.2f}); cap is {MAX_SECONDS_PER_CLIP:.0f}s"
)Sixty seconds per clip is six full renders at the preview cap. If a clip is not right after six versions, the problem is the brief, and no budget fixes a brief.
One-Shot Generation vs Conversational Refinement
| | One-shot (Veo-class flagship) | Gemini Omni Flash refinement | |---|---|---| | Iteration unit | New prompt, full re-roll | Instruction + previous video | | Cost of a fix | Full clip price, fresh dice | One more metered render, converging | | Consistency between takes | Not guaranteed | Anchored by the input video | | Quality ceiling | Higher (flagship models) | Preview model, 10s cap | | Best for | Hero spots, final masters | Drafts, product clips, iteration-heavy work |
The comparison is a workflow choice more than a model ranking. Teams will reasonably draft in Omni Flash and re-shoot the approved cut on a flagship model, the way editors rough-cut in proxies.
Limitations
The 10-second preview cap rules out anything longer than a product beat or a social clip; stitching segments is your job, downstream. The model ID carries -preview, so treat it like weather: check it before you ship, and expect the exact API surface (parameter names, duration fields) to move. Per-second pricing punishes idle experimentation at scale even while individual clips stay cheap. Refinement converges on composition, but fine text, logos, and exact brand colors remain unreliable in generated video, so overlay those in your edit layer. And clip lineage lives in your store, not Google's: lose the version chain and you are back to archaeology by filename.
The Final Working Example
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from fastapi import FastAPI, HTTPException, BackgroundTasks
from google import genai
from pydantic import BaseModel
client = genai.Client()
app = FastAPI(title="omni-flash-clips")
Path("clips").mkdir(exist_ok=True)
MODEL = "gemini-omni-flash-preview"
PRICE_PER_SECOND = 0.10
MAX_SECONDS_PER_CLIP = 60.0
@dataclass
class Version:
video_path: str
instruction: str
seconds: float
cost_usd: float
@dataclass
class Clip:
id: str
status: str = "queued"
error: str | None = None
versions: list[Version] = field(default_factory=list)
CLIPS: dict[str, Clip] = {}
class GenerateReq(BaseModel):
prompt: str
class FromImageReq(BaseModel):
image_path: str
instruction: str
class RefineReq(BaseModel):
instruction: str
def spent_seconds(clip: Clip) -> float:
return sum(v.seconds for v in clip.versions)
def render(clip: Clip, instruction: str,
source_video: str | None = None, source_image: str | None = None):
clip.status = "rendering"
try:
kwargs = {"model": MODEL, "prompt": instruction}
if source_video:
kwargs["video"] = client.files.upload(file=source_video)
if source_image:
kwargs["image"] = client.files.upload(file=source_image)
operation = client.models.generate_videos(**kwargs)
while not operation.done:
time.sleep(5)
operation = client.operations.get(operation)
video = operation.response.generated_videos[0]
path = f"clips/{clip.id}-v{len(clip.versions) + 1}.mp4"
client.files.download(file=video.video)
video.video.save(path)
seconds = getattr(video.video, "duration_seconds", 10.0) or 10.0
clip.versions.append(Version(path, instruction, seconds, seconds * PRICE_PER_SECOND))
clip.status = "ready"
except Exception as exc:
clip.status, clip.error = "failed", str(exc)
@app.post("/clips")
def create_clip(req: GenerateReq, background: BackgroundTasks):
clip = Clip(id=uuid.uuid4().hex[:12])
CLIPS[clip.id] = clip
background.add_task(render, clip, req.prompt)
return {"id": clip.id, "status": clip.status}
@app.post("/clips/from-image")
def create_clip_from_image(req: FromImageReq, background: BackgroundTasks):
clip = Clip(id=uuid.uuid4().hex[:12])
CLIPS[clip.id] = clip
background.add_task(render, clip, req.instruction, None, req.image_path)
return {"id": clip.id, "status": clip.status}
@app.post("/clips/{clip_id}/refine")
def refine_clip(clip_id: str, req: RefineReq, background: BackgroundTasks):
clip = CLIPS.get(clip_id)
if clip is None:
raise HTTPException(404, "unknown clip")
if clip.status != "ready":
raise HTTPException(409, f"clip is {clip.status}")
if spent_seconds(clip) + 10.0 > MAX_SECONDS_PER_CLIP:
raise HTTPException(402, f"clip budget exhausted at {spent_seconds(clip):.0f}s")
background.add_task(render, clip, req.instruction, clip.versions[-1].video_path, None)
return {"id": clip.id, "version": len(clip.versions) + 1}
@app.get("/clips/{clip_id}")
def get_clip(clip_id: str):
clip = CLIPS.get(clip_id)
if clip is None:
raise HTTPException(404, "unknown clip")
return {"id": clip.id, "status": clip.status, "error": clip.error,
"spent_usd": round(spent_seconds(clip) * PRICE_PER_SECOND, 2),
"versions": [{"path": v.video_path, "instruction": v.instruction,
"seconds": v.seconds, "cost_usd": round(v.cost_usd, 2)}
for v in clip.versions]}Run it with uvicorn app:app --reload. Create a clip by POSTing a prompt to /clips (or a photo path plus instruction to /clips/from-image), poll /clips/{id} until ready, then send instructions to /clips/{id}/refine and watch the version chain and the ledger grow together.
When to Use It
Build on Omni Flash when iteration is the workload: product clips, social variants, storyboard drafts, anything where the second version matters more than the first. The refinement loop rewards services that preserve lineage and meter spend, which is exactly what you built here. When the deliverable is a hero spot with a soundtrack and a director, hand the approved draft to a flagship one-shot model and keep Omni Flash as the room where the drafts got cheap.
You might also like
Keep reading from the journal.
June 23, 2026AgenticAI
Too accurate to be comfortable
When personalization crosses into surveillance
July 15, 2026Engineering
Replace Tool Loops with GPT-5.6 Native Tool Calling
On July 9, 2026, OpenAI moved the GPT-5.6 family (Sol, Terra, and Luna) to general availability across ChatGPT, Codex, and the API, and the Responses API picked up the feature that matters most for builders: Programmatic Tool Calling.
July 13, 2026DocumentProcessing
The deal dies in the security review
Purpose-based access control makes minimum-necessary enforceable at query time