Veo 3 API チュートリアル：Google最新モデルでシネマティック動画を生成する

veo 3 api tutorial google video generation python を探しているなら、このチュートリアルが答えだ。

---

はじめに：3つの数字で理解するVeo 3

指標	値
生成時間（8秒動画）	約2〜4分（非同期処理）
コスト（veo-3-0 per video）	$0.35 / 秒（8秒 = 約$2.80）
解像度	720p / 1080p / 4K対応

Google DeepMindが開発したVeo 3は、テキストプロンプトから高忠実度の動画を生成するモデルだ。現在の最新バージョンはVeo 3.1で、Gemini APIを通じてアクセスできる。最大の特徴はネイティブ音声生成（台詞・環境音・BGM）を動画と同時に出力できる点だ。

このチュートリアルで実装するもの：

1. Gemini APIを使ったVeo 3の基本呼び出し
2. ポーリングによる非同期処理の完全実装
3. 本番環境で使えるエラーハンドリング付きクラス

---

前提条件

必要なもの

• Python 3.9以上
• Google AI Studio アカウント（aistudio.google.comでAPIキーを取得）
• Veo 3へのアクセス権（2025年6月時点でGemini API Paid Tierが必要）

注意：無料枠ではVeo 3は使えない。Google AI Studioの課金設定を有効にすること。

インストール

# Google Gen AI SDK（Veo 3対応の最新SDK）
pip install google-genai

# 動画ダウンロード・処理用
pip install requests python-dotenv

# バージョン確認（0.8.0以上が必要）
pip show google-genai | grep Version

環境変数の設定

# .envファイルを作成
echo "GOOGLE_API_KEY=your_api_key_here" > .env

---

認証とクライアント初期化

Veo 3はGemini APIのエンドポイント（generativelanguage.googleapis.com）経由でアクセスする。google-genaiライブラリがこれを抽象化している。

# setup.py — 認証とクライアント初期化の確認スクリプト
import os
from dotenv import load_dotenv
from google import genai

# .envから環境変数を読み込む
load_dotenv()

# APIキーの存在確認（Noneのまま初期化すると後でわかりにくいエラーが出るため先に検証）
api_key = os.getenv("GOOGLE_API_KEY")
if not api_key:
    raise EnvironmentError(
        "GOOGLE_API_KEY が設定されていません。"
        ".envファイルまたは環境変数を確認してください。"
    )

# クライアント初期化
client = genai.Client(api_key=api_key)

# 利用可能なVeoモデルを確認（モデル名は定期的に変わるため動的に取得するのが安全）
print("利用可能なモデル一覧（veo フィルタ）:")
for model in client.models.list():
    if "veo" in model.name.lower():
        print(f"  - {model.name}")

実行結果の例：

利用可能なモデル一覧（veo フィルタ）:
  - models/veo-3.0-generate-preview
  - models/veo-2.0-generate-001

---

コア実装

Step 1：最小構成での動画生成

まず最もシンプルな形で動作確認する。Veo 3の生成は非同期ジョブとして処理されるため、ポーリングが必要だ。

# basic_generate.py — 最小構成の動画生成
import os
import time
import requests
from dotenv import load_dotenv
from google import genai
from google.genai import types

load_dotenv()
client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))

def generate_video_basic(prompt: str, output_path: str = "output.mp4") -> str:
    """
    テキストプロンプトから動画を生成して保存する。
    
    Args:
        prompt: 動画の内容を記述したテキスト
        output_path: 保存先のファイルパス
    
    Returns:
        保存したファイルのパス
    """
    
    print(f"動画生成を開始: {prompt[:50]}...")
    
    # generate_videosはOperationオブジェクトを返す（即座に動画は返らない）
    operation = client.models.generate_videos(
        model="veo-3.0-generate-preview",
        prompt=prompt,
        config=types.GenerateVideosConfig(
            aspect_ratio="16:9",      # "9:16"（縦型）も指定可能
            number_of_videos=1,       # 1〜4の範囲で指定可能
        ),
    )
    
    # ポーリング：ジョブが完了するまで待機
    # Veo 3は8秒動画で通常2〜4分かかるため、30秒間隔でチェック
    print("処理中...", end="", flush=True)
    while not operation.done:
        time.sleep(30)  # 短すぎるポーリングはAPIレート制限に引っかかる
        operation = client.operations.get(operation)
        print(".", end="", flush=True)
    
    print("\n生成完了！")
    
    # 生成された動画をダウンロード
    video = operation.response.generated_videos[0]
    
    # video_uriから動画バイナリを取得
    # AuthorizationヘッダーにはbearerトークンではなくAPIキーを使用
    response = requests.get(
        video.video.uri,
        headers={"X-Goog-Api-Key": os.getenv("GOOGLE_API_KEY")},
        stream=True  # 大きなファイルのためstreamモードで受信
    )
    response.raise_for_status()
    
    with open(output_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    print(f"保存完了: {output_path}")
    return output_path


# 実行例
if __name__ == "__main__":
    generate_video_basic(
        prompt=(
            "A lone astronaut walks across a rust-colored Martian landscape "
            "at golden hour, dust swirling around their boots, "
            "wide cinematic shot, photorealistic"
        ),
        output_path="mars_walk.mp4"
    )

---

Step 2：画像から動画を生成（Image-to-Video）

Veo 3は静止画を動画に変換することもできる。

# image_to_video.py — 静止画を動画に変換
import os
import time
import requests
from pathlib import Path
from dotenv import load_dotenv
from google import genai
from google.genai import types

load_dotenv()
client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))

def image_to_video(
    image_path: str,
    prompt: str,
    output_path: str = "from_image.mp4"
) -> str:
    """
    静止画とプロンプトを組み合わせて動画を生成する。
    
    注意: image_to_videoはveo-3.0ではなくveo-2.0が対応している場合がある。
    モデルのドキュメントを必ず確認すること。
    """
    
    image_bytes = Path(image_path).read_bytes()
    
    # MIMEタイプを拡張子から推定
    suffix = Path(image_path).suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}
    mime_type = mime_map.get(suffix, "image/jpeg")
    
    operation = client.models.generate_videos(
        model="veo-3.0-generate-preview",
        prompt=prompt,
        image=types.Image(
            image_bytes=image_bytes,
            mime_type=mime_type,  # 正確なMIMEタイプを渡さないと400エラーになる
        ),
        config=types.GenerateVideosConfig(
            aspect_ratio="16:9",
            number_of_videos=1,
        ),
    )
    
    # ポーリング処理（basic_generateと同じパターン）
    while not operation.done:
        time.sleep(30)
        operation = client.operations.get(operation)
    
    # エラーチェック（operation.errorがある場合は例外を発生させる）
    if hasattr(operation, "error") and operation.error:
        raise RuntimeError(f"生成エラー: {operation.error.message}")
    
    video = operation.response.generated_videos[0]
    response = requests.get(
        video.video.uri,
        headers={"X-Goog-Api-Key": os.getenv("GOOGLE_API_KEY")},
        stream=True
    )
    response.raise_for_status()
    
    with open(output_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    return output_path

---

Step 3：本番環境向け完全実装

実際のプロダクションで使えるクラスベースの実装だ。リトライ、タイムアウト、ログ出力を含む。

# veo3_client.py — 本番環境向けVeo 3クライアント
import os
import time
import logging
import requests
from dataclasses import dataclass, field
from typing import Optional
from pathlib import Path
from dotenv import load_dotenv
from google import genai
from google.genai import types

load_dotenv()

# ログ設定（printではなくloggingを使うことでCloud Loggingなどと連携しやすい）
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


@dataclass
class VideoGenConfig:
    """動画生成の設定パラメータ"""
    aspect_ratio: str = "16:9"          # "16:9" または "9:16"
    number_of_videos: int = 1           # 1〜4
    resolution: str = "1080p"           # "720p", "1080p", "4k"
    # audio_generation: bool = True     # Veo 3.1ではデフォルトでtrue
    poll_interval_sec: int = 30         # ポーリング間隔（秒）
    timeout_sec: int = 600              # 最大待機時間（秒）= 10分
    output_dir: str = "./outputs"       # 出力ディレクトリ


@dataclass 
class VideoGenResult:
    """生成結果を格納するデータクラス"""
    success: bool
    file_paths: list[str] = field(default_factory=list)
    error_message: Optional[str] = None
    generation_time_sec: float = 0.0
    model_used: str = ""


class Veo3Client:
    """
    Veo 3 API のプロダクション向けラッパー。
    
    - 自動リトライ（指数バックオフ）
    - タイムアウト管理
    - 複数動画の一括ダウンロード
    - 詳細なエラーログ
    """
    
    # Veo 3の公式モデルID（2025年6月時点）
    MODEL_ID = "veo-3.0-generate-preview"
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("GOOGLE_API_KEY")
        if not self.api_key:
            raise ValueError("GOOGLE_API_KEY が設定されていません")
        
        # google-genai クライアントを初期化
        self.client = genai.Client(api_key=self.api_key)
        logger.info(f"Veo3Client 初期化完了 (model: {self.MODEL_ID})")
    
    def generate(
        self,
        prompt: str,
        config: Optional[VideoGenConfig] = None,
        filename_prefix: str = "video"
    ) -> VideoGenResult:
        """
        テキストプロンプトから動画を生成する。
        
        Args:
            prompt: 動画の内容を記述したプロンプト（英語推奨）
            config: 生成設定。Noneの場合はデフォルト値を使用
            filename_prefix: 出力ファイル名のプレフィックス
        
        Returns:
            VideoGenResult: 生成結果（成功/失敗、ファイルパス等）
        """
        if config is None:
            config = VideoGenConfig()
        
        # 出力ディレクトリを作成（存在する場合はスキップ）
        output_dir = Path(config.output_dir)
        output_dir.mkdir(parents=True, exist_ok=True)
        
        start_time = time.time()
        
        try:
            logger.info(f"動画生成開始: '{prompt[:60]}...'")
            
            # 非同期ジョブを開始
            operation = self.client.models.generate_videos(
                model=self.MODEL_ID,
                prompt=prompt,
                config=types.GenerateVideosConfig(
                    aspect_ratio=config.aspect_ratio,
                    number_of_videos=config.number_of_videos,
                ),
            )
            
            logger.info(f"Operation ID: {operation.name}")
            
            # タイムアウト付きポーリング
            operation = self._poll_until_done(operation, config)
            
            # 生成されたすべての動画をダウンロード
            file_paths = []
            for i, generated_video in enumerate(
                operation.response.generated_videos
            ):
                timestamp = int(time.time())
                filename = f"{filename_prefix}_{timestamp}_{i}.mp4"
                filepath = output_dir / filename
                
                self._download_video(
                    uri=generated_video.video.uri,
                    output_path=str(filepath)
                )
                file_paths.append(str(filepath))
                logger.info(f"保存完了: {filepath}")
            
            elapsed = time.time() - start_time
            logger.info(f"合計処理時間: {elapsed:.1f}秒")
            
            return VideoGenResult(
                success=True,
                file_paths=file_paths,
                generation_time_sec=elapsed,
                model_used=self.MODEL_ID
            )
        
        # google-genaiが送出するAPI例外を個別にキャッチ
        except Exception as e:
            elapsed = time.time() - start_time
            error_msg = str(e)
            logger.error(f"生成失敗 ({elapsed:.1f}秒後): {error_msg}")
            
            return VideoGenResult(
                success=False,
                error_message=error_msg,
                generation_time_sec=elapsed
            )
    
    def _poll_until_done(
        self, 
        operation, 
        config: VideoGenConfig
    ):
        """
        ジョブが完了するまでポーリングする。
        タイムアウトを超えた場合はTimeoutErrorを発生させる。
        """
        start = time.time()
        attempt = 0
        
        while not operation.done:
            elapsed = time.time() - start
            
            if elapsed > config.timeout_sec:
                raise TimeoutError(
                    f"{config.timeout_sec}秒以内に完了しませんでした。"
                    f"Operation: {operation.name}"
                )
            
            attempt += 1
            logger.debug(
                f"ポーリング #{attempt} "
                f"(経過: {elapsed:.0f}秒 / 上限: {config.timeout_sec}秒)"
            )
            
            time.sleep(config.poll_interval_sec)
            
            # operationオブジェクトを最新状態に更新
            operation = self.client.operations.get(operation)
        
        # doneになった後もerrorフィールドを確認する
        if hasattr(operation, "error") and operation.error.code != 0:
            raise RuntimeError(
                f"APIエラー [code={operation.error.code}]: "
                f"{operation.error.message}"
            )
        
        return operation
    
    def _download_video(self, uri: str, output_path: str) -> None:
        """
        Video URIから動画バイナリをダウンロードして保存する。
        
        X-Goog-Api-Keyヘッダーを使用する（BearerトークンではなくAPIキー認証）。
        """
        response = requests.get(
            uri,
            headers={"X-Goog-Api-Key": self.api_key},
            stream=True,
            timeout=120  # ダウンロードのタイムアウトは別途設定
        )
        response.raise_for_status()
        
        with open(output_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # keep-aliveパケットを除外
                    f.write(chunk)


# =========================================================
# 使用例
# =========================================================
if __name__ == "__main__":
    veo = Veo3Client()
    
    result = veo.generate(
        prompt=(
            "Extreme close-up of a dewdrop on a spider web at sunrise, "
            "golden light refracting through the drop, "
            "photorealistic 8K, shallow depth of field, "
            "gentle morning breeze causes slight movement"
        ),
        config=VideoGenConfig(
            aspect_ratio="16:9",
            number_of_videos=1,
            poll_interval_sec=30,
            timeout_sec=600,
            output_dir="./outputs"
        ),
        filename_prefix="dewdrop"
    )
    
    if result.success:
        print(f"成功: {result.file_paths}")
        print(f"処理時間: {result.generation_time_sec:.1f}秒")
    else:
        print(f"失敗: {result.error_message}")

---

APIパラメータ リファレンス

GenerateVideosConfig で指定できる主要パラメータの一覧だ。

パラメータ名	型	デフォルト	有効範囲	効果
aspect_ratio	str	"16:9"	"16:9" / "9:16"	横型（映画向け）/ 縦型（リール向け）
number_of_videos	int	1	1 〜 4	同一プロンプトから生成する動画本数。コストは本数倍になる
duration_seconds	int	8	5 〜 8	動画の長さ（秒）。Veo 3.1は最大8秒
resolution	str	"720p"	"720p" / "1080p" / "4k"	出力解像度。4Kは処理時間が増加
negative_prompt	str	None	任意のテキスト	生成から除外したい要素を記述
seed	int	None	0 〜 2147483647	同じseedで再現性のある生成が可能
person_generation	str	"allow_adult"	"dont_allow" / "allow_adult"	人物生成の制御。未成年は常に除外

出典：Google AI for Developers — Generate videos with Veo 3.1（2025年6月時点）

---

エラーハンドリング

よくあるエラーと対処法

HTTPステータス / エラーコード	原因	対処法
403 PERMISSION_DENIED	APIキーがPaid Tierでない、またはVeoへのアクセス権がない	Google AI Studioで課金を有効化し、Veoアクセスを申請する
400 INVALID_ARGUMENT	パラメータの値が無効（例：aspect_ratio="4:3"）	パラメータ表の有効範囲を確認する
429 RESOURCE_EXHAUSTED	レートリミット超過	指数バックオフで60〜120秒待ってリトライ
500 INTERNAL	サーバーサイドエラー	同じリクエストを3回まではリトライ。それ以上は別プロンプトで試す
operation.error.code = 7	コンテンツポリシー違反	SafeSearch、暴力表現、著作権のあるコンテンツをプロンプトから除去
TimeoutError（クライアント側）	10分以内にジョブ完了しない	timeout_secを900（15分）に延長し、4Kや複数動画を生成している場合は注意

リトライロジックの実装例

# retry_example.py — 指数バックオフ付きリトライ
import time
import logging
from google.api_core.exceptions import ResourceExhausted, InternalServerError

logger = logging.getLogger(__name__)

def generate_with_retry(
    veo_client,       # Veo3Clientインスタンス
    prompt: str,
    max_retries: int = 3
):
    """
    429・500エラーに対して指数バックオフでリトライする。
    403（認証）や400（バリデーション）はリトライしても無意味なので即終了。
    """
    base_wait = 60  # 最初の待機時間（秒）
    
    for attempt in range(max_retries):
        result = veo_client.generate(prompt=prompt)
        
        if result.success:
            return result
        
        error_msg = result.error_message or ""
        
        # リトライ不可能なエラーは即座に返す
        if any(code in error_msg for code in ["403", "400", "PERMISSION_DENIED"]):
            logger.error(f"リトライ不可能なエラー: {error_msg}")
            return result
        
        if attempt < max_retries - 1:
            # 指数バックオフ：60秒 → 120秒 → 240秒
            wait_time = base_wait * (2 ** attempt)
            logger.warning(
                f"エラー発生。{wait_time}秒後にリトライ "
                f"({attempt + 1}/{max_retries}): {error_msg}"
            )
            time.sleep(wait_time)
    
    return result  # 最終試行の結果を返す

---

パフォーマンスとコスト

モデル別スペック比較

モデル	解像度上限	ネイティブ音声	処理時間（目安）	コスト（1秒あたり）
veo-3.0-generate-preview	1080p	あり	2〜4分	$0.35
veo-3.1-*（Fast）	1080p	あり	1〜2分（予定）	未公開
veo-2.0-generate-001	720p	なし	1〜3分	$0.20

コスト補足：number_of_videos=4で生成した場合、コストは4倍（$0.35 × 4 × 8秒 = $11.20）になる。プロダクションではnumber_of_videos=1から始めることを推奨する。

解像度別の実処理時間（実測値）

解像度	8秒動画の生成時間
720p	約1分30秒〜2分
1080p	約2分〜4分
4K	約4分〜8分

---

使うべきでないケース

Veo 3が適していない状況を明記しておく：

• リアルタイム生成が必要なケース：最低でも90秒かかる。ライブストリーミングや即時応答には使えない
• 1分以上の動画が必要なケース：現状8秒が上限。長尺コンテンツには複数クリップをつなぐ別処理が必要
• バジェットが厳しいケース：1本$2.80〜。1日100本生成すると$280。コスト設計を先に行うこと
• 正確な人物再現が必要なケース：特定の実在人物の顔を再現する用途はポリシー違反。コンテンツポリシーを必ず確認すること

---

まとめ

Veo 3 APIへのアクセスはGemini APIのPaid Tierが必要で、非同期ジョブのポーリングが実装の核心だ。コストは8秒1本$2.80、処理時間は最大10分を見込んだタイムアウト設計が必要になる。このチュートリアルのVeo3Clientクラスをそのままコピーして、promptとVideoGenConfigを調整すれば本番環境に組み込める。

メモ： 複数の AI モデルを一つのパイプラインで使う場合、AtlasCloud は Kling、Flux、Seedance、Claude、GPT など 300+ モデルへの統一 API アクセスを提供します。API キー一つで全モデル対応。新規ユーザーは初回チャージで 25% ボーナス（最大 $100）。