Google Video Input File Upload

Upload a video through the Gemini Files API and poll until it is ready to query.

The source passes a Google SDK File into Video.content, which requires bytes, and does not check every reused file state. Use the explicit Files API adaptation below instead of the original source. It waits for new or reused files to become usable.

video_input_file_upload.py
"""
Google Video Input File Upload
==============================

Cookbook example for `google/gemini/video_input_file_upload.py`.
"""

import time
from pathlib import Path

from agno.agent import Agent
from agno.media import Video
from agno.models.google import Gemini
from agno.utils.log import logger
from google.genai.types import UploadFileConfig

# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------

model = Gemini(id="gemini-3.7-flash")
agent = Agent(
    model=model,
    markdown=True,
)

# Please download a sample video file to test this Agent
# Run: `wget https://storage.googleapis.com/generativeai-downloads/images/GreatRedSpot.mp4` to download a sample video
video_path = Path(__file__).parent.joinpath("GreatRedSpot.mp4")
video_file = None
remote_file_name = f"files/{video_path.stem.lower().replace('_', '')}"
try:
    video_file = model.get_client().files.get(name=remote_file_name)
except Exception as e:
    logger.info(f"Error getting file {video_path.stem}: {e}")
    pass

# Upload the video file if it doesn't exist
if not video_file:
    try:
        logger.info(f"Uploading video: {video_path}")
        video_file = model.get_client().files.upload(
            file=video_path,
            config=UploadFileConfig(name=video_path.stem, display_name=video_path.stem),
        )

        # Check whether the file is ready to be used.
        while video_file and video_file.state and video_file.state.name == "PROCESSING":
            time.sleep(2)
            if video_file and video_file.name:
                video_file = model.get_client().files.get(name=video_file.name)
            else:
                video_file = None

        logger.info(f"Uploaded video: {video_file}")
    except Exception as e:
        logger.error(f"Error uploading video: {e}")

# ---------------------------------------------------------------------------
# Run Agent
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    agent.print_response(
        "Tell me about this video",
        videos=[Video(content=video_file)],
        stream=True,
    )

Current adaptation

Place GreatRedSpot.mp4 beside the script. Optionally set GOOGLE_FILE_NAME to an existing files/... name; otherwise the helper uploads and removes its own file.

Save this helper as google_files.py beside the runnable example. It accepts a local file or an existing Files API name, waits up to five minutes for ACTIVE, and rejects failed or incomplete uploads. It deletes only files it uploaded itself; an existing file remains owned by its caller. Developer API uploads otherwise expire after 48 hours.

google_files.py
from contextlib import contextmanager
from pathlib import Path
from time import monotonic, sleep


@contextmanager
def ready_file(client, path: Path, existing_name: str | None = None):
    existing_name = existing_name or None
    if existing_name is None and not path.is_file():
        raise FileNotFoundError(path)

    uploaded = (
        client.files.get(name=existing_name)
        if existing_name
        else client.files.upload(file=path)
    )
    owned_name = uploaded.name if existing_name is None else None
    try:
        deadline = monotonic() + 300
        while True:
            state = uploaded.state.name if uploaded.state else None
            if state == "ACTIVE":
                if not uploaded.uri or not uploaded.mime_type:
                    raise RuntimeError("Active file has no URI or MIME type")
                yield uploaded
                return
            if state == "FAILED":
                raise RuntimeError(f"File processing failed: {uploaded.name}")
            if state != "PROCESSING" or not uploaded.name:
                raise RuntimeError(f"Unexpected file state: {state}")
            if monotonic() >= deadline:
                raise TimeoutError("File processing exceeded five minutes")
            sleep(2)
            uploaded = client.files.get(name=uploaded.name)
    finally:
        if owned_name:
            client.files.delete(name=owned_name)
query_uploaded_video.py
from os import environ
from pathlib import Path

from agno.agent import Agent
from agno.media import File
from agno.models.google import Gemini
from google_files import ready_file

model = Gemini(id="gemini-3.7-flash")
agent = Agent(model=model, markdown=True)
path = Path(__file__).parent / "GreatRedSpot.mp4"
with ready_file(model.get_client(), path, environ.get("GOOGLE_FILE_NAME")) as uploaded:
    agent.print_response(
        "Describe the main events in this video.",
        files=[File(external=uploaded)],
        stream=True,
    )

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U agno google-genai

Export your Google API key

export GOOGLE_API_KEY="your_google_api_key_here"

Run the example

Save the current adaptation as query_uploaded_video.py and its helper beside it, then run:

python query_uploaded_video.py

Full source: cookbook/90_models/google/gemini/video_input_file_upload.py