[ACTION] Extend Vertex AI MCP server with Veo3 support

[MichielMAnalytics]

I just checked it out and Vertex AI supports Veo3 through API. It would be great if you could support this as well through your MCP server. I'm talking about the MCP server with slug google_vertex_ai.

See API Doc: https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/veo-video-generation

I played around with some example code which may be helpful. Below a script on how to use the vertex api to generate Veo3 videos:

################################################################################
import os
import time
import urllib.request
from dotenv import load_dotenv

from google import genai
from google.genai import types

Load environment variables

load_dotenv()

def save_video_locally(video_data, output_dir="./generated_videos", filename=None):
"""
Saves video data (bytes or URL) to a local file.
"""
# Create output directory if it doesn't exist
os.makedirs(output_dir, exist_ok=True)

# Generate filename if not provided
if not filename:
    filename = f"video_{int(time.time())}.mp4"
filepath = os.path.join(output_dir, filename)

if isinstance(video_data, bytes):
    # Save video bytes directly
    with open(filepath, 'wb') as f:
        f.write(video_data)
    print(f"Video saved to: {filepath}")
elif isinstance(video_data, str) and video_data.startswith('http'):
    # Download from URL
    urllib.request.urlretrieve(video_data, filepath)
    print(f"Video downloaded and saved to: {filepath}")
else:
    print(f"Unsupported video data type: {type(video_data)}")
    return None

return filepath

def generate_video_from_text(client, prompt, enhance_prompt=True, generate_audio=True):
"""
Generate videos from a text prompt.

Args:
    client: The genai client instance
    prompt: A detailed description of the video you would like to see
    enhance_prompt: Whether to enhance the provided prompt
    generate_audio: Whether to include audio in the output video

Returns:
    The operation result if successful, None otherwise
"""

video_model = "veo-3.0-generate-preview"

operation = client.models.generate_videos(
    model=video_model,
    prompt=prompt,
    config=types.GenerateVideosConfig(
        aspect_ratio="16:9",
        number_of_videos=1,
        duration_seconds=8,
        person_generation="allow_adult",
        enhance_prompt=enhance_prompt,
        generate_audio=generate_audio,
    ),
)

while not operation.done:
    time.sleep(15)
    operation = client.operations.get(operation)
    print(f"Operation status: {operation.name}")

print(f"Operation completed. Done: {operation.done}")
print(f"Operation object: {operation}")

if hasattr(operation, 'error') and operation.error:
    print(f"Error generating video: {operation.error}")
    return None

if hasattr(operation, 'result'):
    print(f"Has result: {operation.result}")
    if operation.result and hasattr(operation.result, 'generated_videos') and operation.result.generated_videos:
        video = operation.result.generated_videos[0].video
        if hasattr(video, 'uri') and video.uri:
            print(f"Video generated successfully (URI): {video.uri}")
        elif hasattr(video, 'video_bytes') and video.video_bytes:
            print(f"Video generated successfully (bytes): {len(video.video_bytes)} bytes")
        else:
            print(f"Video generated but no URI or bytes found")
        return operation.result

print("No video was generated in the result")
return None

def generate_video_from_image(client, prompt, image_path_or_url, enhance_prompt=True, generate_audio=True):
"""
Generate videos from an image with optional text prompt.

Args:
    client: The genai client instance
    prompt: Optional prompt to direct movement and audio
    image_path_or_url: Local path or URL of the input image
    enhance_prompt: Whether to enhance the provided prompt
    generate_audio: Whether to include audio in the output video

Returns:
    The operation result if successful, None otherwise
"""
video_model = "veo-3.0-generate-preview"




operation = client.models.generate_videos(
    model=video_model,
    prompt=prompt,
    image=types.Image(
        gcs_uri=image_path_or_url if image_path_or_url.startswith("gs://") else None,
        uri=image_path_or_url if image_path_or_url.startswith("http") else None,
        mime_type="image/png",
    ),
    config=types.GenerateVideosConfig(
        aspect_ratio="16:9",
        number_of_videos=1,
        duration_seconds=8,
        person_generation="allow_adult",
        enhance_prompt=enhance_prompt,
        generate_audio=generate_audio,
    ),
)

while not operation.done:
    time.sleep(15)
    operation = client.operations.get(operation)
    print(f"Operation status: {operation.name}")

print(f"Operation completed. Done: {operation.done}")
print(f"Operation object: {operation}")

if hasattr(operation, 'error') and operation.error:
    print(f"Error generating video: {operation.error}")
    return None

if hasattr(operation, 'result'):
    print(f"Has result: {operation.result}")
    if operation.result and hasattr(operation.result, 'generated_videos') and operation.result.generated_videos:
        video = operation.result.generated_videos[0].video
        if hasattr(video, 'uri') and video.uri:
            print(f"Video generated successfully (URI): {video.uri}")
        elif hasattr(video, 'video_bytes') and video.video_bytes:
            print(f"Video generated successfully (bytes): {len(video.video_bytes)} bytes")
        else:
            print(f"Video generated but no URI or bytes found")
        return operation.result

print("No video was generated in the result")
return None

def main():
"""
Main function to demonstrate Veo 3 video generation.
"""
# Set your project configuration from environment variables
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT")
if not PROJECT_ID or PROJECT_ID == "your-project-id":
raise ValueError("Please set GOOGLE_CLOUD_PROJECT in your .env file")

LOCATION = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1")
LOCAL_OUTPUT_DIR = os.environ.get("LOCAL_OUTPUT_DIR", "./generated_videos")

print(f"Using project: {PROJECT_ID}")
print(f"Using region: {LOCATION}")
print(f"Videos will be saved to: {LOCAL_OUTPUT_DIR}")
print(f"Note: Veo 3 API may only be available in us-central1. If you get errors, try changing GOOGLE_CLOUD_REGION to us-central1 in your .env file.")

# Check for credentials
print("\nChecking Google Cloud authentication...")
print("To authenticate, you need to run one of these commands:")
print("1. For user account: gcloud auth application-default login")
print("2. For service account: Set GOOGLE_APPLICATION_CREDENTIALS in .env to point to your service account key file")
print("")

# Create client
client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)

# Example 1: Generate video from text prompt
print("Generating video from text prompt...")
text_prompt = "a garden gnome singing a pop song in a whimsical outdoor garden"

result = generate_video_from_text(client, text_prompt)
if result:
    video = result.generated_videos[0].video
    if hasattr(video, 'uri') and video.uri:
        save_video_locally(video.uri, LOCAL_OUTPUT_DIR)
    elif hasattr(video, 'video_bytes') and video.video_bytes:
        save_video_locally(video.video_bytes, LOCAL_OUTPUT_DIR)
    else:
        print("No video URI or bytes found in result")

# # Example 2: Generate video from image
# print("\nGenerating video from image...")
# image_prompt = "zoom out of the flower field, play whimsical music"
# # You can use a URL or local path for the image
# image_url = "https://storage.googleapis.com/cloud-samples-data/generative-ai/image/flowers.png"
# 
# result2 = generate_video_from_image(client, image_prompt, image_url)
# if result2:
#     video_url = result2.generated_videos[0].video.uri
#     save_video_locally(video_url, LOCAL_OUTPUT_DIR)

if name == "main":
main()