diff --git a/tests/contracts/test_phase3_contracts.py b/tests/contracts/test_phase3_contracts.py index 6f225f14..fb1a8884 100644 --- a/tests/contracts/test_phase3_contracts.py +++ b/tests/contracts/test_phase3_contracts.py @@ -6,8 +6,11 @@ stage director skills, meta skills, and the animated-explainer pipeline. import sys import builtins +import base64 +import os import shutil from pathlib import Path +from unittest.mock import MagicMock, patch import pytest @@ -32,10 +35,51 @@ from tools.audio.elevenlabs_tts import ElevenLabsTTS from tools.audio.openai_tts import OpenAITTS from tools.audio.piper_tts import PiperTTS from tools.audio.tts_selector import TTSSelector +from tools.audio.google_tts import GoogleTTS +from tools.graphics.google_imagen import GoogleImagen +from tools.audio.google_music import GoogleMusic +from tools.video.veo_video import VeoVideo + + +# ---- Google Credentials ---- + + +class TestGoogleCredentials: + def test_get_genai_client_with_google_api_key(self): + from tools.google_credentials import get_genai_client + from google.genai import types + + mock_client = MagicMock() + with ( + patch.dict( + os.environ, + { + "GOOGLE_API_KEY": "my_google_key", + "GEMINI_API_KEY": "", + "GOOGLE_GENAI_USE_VERTEXAI": "false", + }, + ), + patch("google.genai.Client", return_value=mock_client) as mock_genai_client, + ): + # 1. Call with default options (None) + client = get_genai_client() + assert client is not None + kwargs = mock_genai_client.call_args[1] + assert kwargs["api_key"] == "my_google_key" + assert kwargs["http_options"] is None + + # 2. Call with explicit options + my_opts = types.HttpOptions(timeout=12345) + client_custom = get_genai_client(http_options=my_opts) + assert client_custom is not None + kwargs_custom = mock_genai_client.call_args[1] + assert kwargs_custom["api_key"] == "my_google_key" + assert kwargs_custom["http_options"].timeout == 12345 # ---- TTS Provider Tools ---- + class TestElevenLabsTTS: def test_identity(self): tool = ElevenLabsTTS() @@ -92,6 +136,30 @@ class TestPiperTTS: assert PiperTTS().get_status() == ToolStatus.UNAVAILABLE +class TestGoogleTTS: + def test_identity(self): + tool = GoogleTTS() + info = tool.get_info() + assert info["name"] == "google_tts" + assert info["tier"] == "voice" + assert info["capability"] == "tts" + assert info["provider"] == "google_tts" + + def test_cost_estimate(self): + tool = GoogleTTS() + cost = tool.estimate_cost({"text": "Hello world, this is a test."}) + assert cost > 0 + assert cost < 0.01 # short text should be cheap + + def test_capabilities(self): + tool = GoogleTTS() + assert "text_to_speech" in tool.capabilities + assert "voice_selection" in tool.capabilities + + +# ---- Music Generation Tools ---- + + class TestMusicGen: def test_identity(self): tool = MusicGen() @@ -110,6 +178,451 @@ class TestMusicGen: assert "generate_background_music" in tool.capabilities +class TestGoogleMusic: + def test_identity(self): + tool = GoogleMusic() + info = tool.get_info() + assert info["name"] == "google_music" + assert info["tier"] == "generate" + assert info["capability"] == "music_generation" + assert info["provider"] == "google" + + def test_duration_validation(self): + tool = GoogleMusic() + mock_client = MagicMock() + + mock_interaction = MagicMock() + mock_interaction.status = "completed" + mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode( + "utf-8" + ) + mock_client.interactions.create.return_value = mock_interaction + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + patch("pathlib.Path.write_bytes"), + ): + # 1. duration > 184 and auto_fix is True -> coerced to 184 + inputs = { + "prompt": "melodic pop", + "duration_seconds": 200, + "auto_fix": True, + "output_path": "test_out.mp3", + } + res = tool.execute(inputs) + assert res.success is True + assert res.data["duration_seconds"] == 184.0 + + # 2. duration > 184 and auto_fix is False -> raises error + inputs = { + "prompt": "melodic pop", + "duration_seconds": 200, + "auto_fix": False, + } + res = tool.execute(inputs) + assert res.success is False + assert res.error is not None + assert "maximum duration is 184" in res.error + + def test_execute_success_convenience_extraction(self, tmp_path): + tool = GoogleMusic() + mock_client = MagicMock() + + mock_interaction = MagicMock() + mock_interaction.status = "completed" + mock_interaction.output_audio.data = base64.b64encode( + b"my_fake_google_lyria_audio" + ).decode("utf-8") + mock_client.interactions.create.return_value = mock_interaction + + output_file = tmp_path / "test_music.mp3" + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + inputs = { + "prompt": "atmospheric electronic ambient beat", + "duration_seconds": 30, + "output_path": str(output_file), + } + res = tool.execute(inputs) + assert res.success is True + assert res.data["provider"] == "google" + assert res.data["model"] == "lyria-3-pro-preview" + assert res.data["output"] == str(output_file) + assert output_file.read_bytes() == b"my_fake_google_lyria_audio" + + def test_execute_success_fallback_extraction(self, tmp_path): + tool = GoogleMusic() + mock_client = MagicMock() + + mock_interaction = MagicMock() + mock_interaction.status = "completed" + del mock_interaction.output_audio + + mock_part = MagicMock() + mock_part.type = "audio" + mock_part.data = base64.b64encode(b"raw_step_audio").decode("utf-8") + + mock_step = MagicMock() + mock_step.type = "model_output" + mock_step.content = [mock_part] + mock_interaction.steps = [mock_step] + + mock_client.interactions.create.return_value = mock_interaction + + output_file = tmp_path / "test_music_step.mp3" + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + inputs = { + "prompt": "jazz piano solo", + "duration_seconds": 30, + "output_path": str(output_file), + } + res = tool.execute(inputs) + assert res.success is True + assert res.data["output"] == str(output_file) + assert output_file.read_bytes() == b"raw_step_audio" + + @patch("os.path.exists") + @patch("requests.get") + def test_multimodal_image(self, mock_get, mock_exists, tmp_path): + tool = GoogleMusic() + mock_client = MagicMock() + + mock_interaction = MagicMock() + mock_interaction.status = "completed" + mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode( + "utf-8" + ) + mock_client.interactions.create.return_value = mock_interaction + + mock_exists.return_value = True + + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "image/jpeg"} + mock_resp.content = b"url_image_bytes" + mock_get.return_value = mock_resp + + local_image = tmp_path / "ref.png" + with open(local_image, "wb") as f: + f.write(b"local_image_bytes") + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + # 1. Local image path + inputs = { + "prompt": "music inspired by image", + "image_path": str(local_image), + "output_path": str(tmp_path / "out1.mp3"), + } + res = tool.execute(inputs) + assert res.success is True + + called_input = mock_client.interactions.create.call_args[1]["input"] + assert len(called_input) == 2 + assert called_input[0] == { + "type": "text", + "text": "music inspired by image\n\n[Target Duration: 30 seconds]", + } + assert called_input[1]["type"] == "image" + assert called_input[1]["mime_type"] == "image/png" + assert called_input[1]["data"] == base64.b64encode( + b"local_image_bytes" + ).decode("utf-8") + + # 2. Remote image URL + inputs = { + "prompt": "music inspired by url", + "image_url": "https://example.com/art.jpg", + "output_path": str(tmp_path / "out2.mp3"), + } + res = tool.execute(inputs) + assert res.success is True + + called_input = mock_client.interactions.create.call_args[1]["input"] + assert len(called_input) == 2 + assert called_input[0] == { + "type": "text", + "text": "music inspired by url\n\n[Target Duration: 30 seconds]", + } + assert called_input[1]["type"] == "image" + assert called_input[1]["mime_type"] == "image/jpeg" + assert called_input[1]["data"] == base64.b64encode( + b"url_image_bytes" + ).decode("utf-8") + + def test_minimum_duration_validation(self, caplog): + import logging + + tool = GoogleMusic() + mock_client = MagicMock() + + mock_interaction = MagicMock() + mock_interaction.status = "completed" + mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode( + "utf-8" + ) + mock_client.interactions.create.return_value = mock_interaction + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + patch("pathlib.Path.write_bytes"), + caplog.at_level(logging.WARNING), + ): + # 1. duration < 5 and auto_fix is True -> coerced to 5 with warning logged + inputs = { + "prompt": "melodic pop", + "duration_seconds": 3, + "auto_fix": True, + "output_path": "test_out.mp3", + } + res = tool.execute(inputs) + assert res.success is True + assert res.data["duration_seconds"] == 5.0 + + warnings = [ + rec.message + for rec in caplog.records + if "minimum duration" in rec.message + ] + assert len(warnings) == 1 + assert "minimum duration of 5 seconds" in warnings[0] + + # Clear records for next check + caplog.clear() + + # 2. duration < 5 and auto_fix is False -> raises error + inputs = { + "prompt": "melodic pop", + "duration_seconds": 3, + "auto_fix": False, + } + res = tool.execute(inputs) + assert res.success is False + assert res.error is not None + assert "minimum duration is 5" in res.error + + def test_missing_image_path_error(self, tmp_path): + tool = GoogleMusic() + with patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}): + inputs = { + "prompt": "music with missing image", + "image_path": str(tmp_path / "does_not_exist.png"), + "output_path": str(tmp_path / "out.mp3"), + } + res = tool.execute(inputs) + assert res.success is False + assert res.error is not None + assert "Failed to load visual conditioning image" in res.error + assert "Local reference image not found" in res.error + + +# ---- Image Generation Tools ---- + + +class TestGoogleImagen: + def test_identity(self): + tool = GoogleImagen() + info = tool.get_info() + assert info["name"] == "google_imagen" + assert info["tier"] == "generate" + assert info["capability"] == "image_generation" + assert info["provider"] == "google_imagen" + + def test_capabilities(self): + tool = GoogleImagen() + assert "text_to_image" in tool.capabilities + + +# ---- Video Generation Tools ---- + + +class TestVeoVideo: + def test_identity(self): + tool = VeoVideo() + info = tool.get_info() + assert info["name"] == "veo_video" + assert info["tier"] == "generate" + assert info["capability"] == "video_generation" + assert info["provider"] == "veo" + + def test_backend_auto_detect(self): + tool = VeoVideo() + + with patch.dict(os.environ, {"GEMINI_API_KEY": "test_key", "FAL_KEY": ""}): + if "FAL_KEY" in os.environ: + del os.environ["FAL_KEY"] + if "FAL_AI_API_KEY" in os.environ: + del os.environ["FAL_AI_API_KEY"] + assert tool._get_google_credentials_status() is True + assert not tool._get_fal_api_key() + assert tool.get_status() == ToolStatus.AVAILABLE + + with patch.dict( + os.environ, + {"GEMINI_API_KEY": "", "GOOGLE_API_KEY": "", "FAL_KEY": "test_fal_key"}, + ): + assert tool._get_google_credentials_status() is False + assert tool._get_fal_api_key() == "test_fal_key" + assert tool.get_status() == ToolStatus.AVAILABLE + + @patch("tools.video._shared.probe_output") + def test_duration_coercion(self, mock_probe): + tool = VeoVideo() + mock_probe.return_value = {"width": 1920, "height": 1080, "duration": 8.0} + + mock_client = MagicMock() + mock_client._api_client.vertexai = False + mock_operation = MagicMock() + mock_operation.done = True + mock_operation.error = None + + mock_video_result = MagicMock() + mock_video_result.video = MagicMock() + mock_operation.response.generated_videos = [mock_video_result] + mock_client.models.generate_videos.return_value = mock_operation + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + # auto_fix = True -> coerced to 8s + inputs = { + "prompt": "Test prompt", + "backend": "google", + "resolution": "1080p", + "duration": "4s", + "auto_fix": True, + "output_path": "test_out.mp4", + } + res = tool.execute(inputs) + assert res.success is True + + called_config = mock_client.models.generate_videos.call_args[1]["config"] + assert called_config.duration_seconds == 8 + + @patch("tools.video._shared.probe_output") + @patch("PIL.Image.open") + @patch("os.path.exists") + @patch("requests.get") + def test_operations_mapping( + self, mock_req_get, mock_exists, mock_img_open, mock_probe + ): + tool = VeoVideo() + mock_probe.return_value = {"width": 1920, "height": 1080, "duration": 8.0} + mock_exists.return_value = True + + mock_img = MagicMock() + mock_img.format = "PNG" + mock_img_open.return_value = mock_img + + mock_resp = MagicMock() + mock_resp.content = b"fake_image_bytes" + mock_req_get.return_value = mock_resp + + mock_client = MagicMock() + mock_client._api_client.vertexai = False + mock_operation = MagicMock() + mock_operation.done = True + mock_operation.error = None + mock_video_result = MagicMock() + mock_video_result.video = MagicMock() + mock_operation.response.generated_videos = [mock_video_result] + mock_client.models.generate_videos.return_value = mock_operation + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + # text_to_video + inputs = { + "prompt": "Test text to video", + "backend": "google", + "operation": "text_to_video", + "duration": "8s", + } + res = tool.execute(inputs) + assert res.success is True + + called_kwargs = mock_client.models.generate_videos.call_args[1] + assert called_kwargs["image"] is None + + # image_to_video + inputs = { + "prompt": "Test image to video", + "backend": "google", + "operation": "image_to_video", + "image_path": "local_img.png", + "duration": "8s", + } + res = tool.execute(inputs) + assert res.success is True + + called_kwargs = mock_client.models.generate_videos.call_args[1] + assert called_kwargs["image"] is not None + + def test_vertex_ai_mode_rejection(self): + tool = VeoVideo() + mock_client = MagicMock() + mock_client.vertexai = True + if hasattr(mock_client, "_api_client"): + delattr(mock_client, "_api_client") + + with ( + patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}), + patch("google.genai.Client", return_value=mock_client), + ): + inputs = { + "prompt": "cinematic shot", + "backend": "google", + } + res = tool.execute(inputs) + assert res.success is False + assert res.error is not None + assert "only supported using the Gemini Developer API" in res.error + + def test_missing_local_image_paths(self): + tool = VeoVideo() + with patch.dict( + os.environ, + {"GEMINI_API_KEY": "test_key", "GOOGLE_GENAI_USE_VERTEXAI": "false"}, + ): + inputs = { + "prompt": "cinematic shot", + "backend": "google", + "operation": "image_to_video", + "image_path": "non_existent_file_path_12345.png", + } + res = tool.execute(inputs) + assert res.success is False + assert "Local input image not found" in res.error + + def test_missing_reference_image_paths(self): + tool = VeoVideo() + with patch.dict( + os.environ, + {"GEMINI_API_KEY": "test_key", "GOOGLE_GENAI_USE_VERTEXAI": "false"}, + ): + inputs = { + "prompt": "cinematic shot", + "backend": "google", + "operation": "reference_to_video", + "reference_image_paths": ["non_existent_reference_12345.png"], + } + res = tool.execute(inputs) + assert res.success is False + assert "Local reference image not found" in res.error + + class TestNewToolsRegistry: def test_all_register(self): reg = ToolRegistry() @@ -135,7 +648,9 @@ class TestCapabilityMetadata: info = tool.get_info() assert info["capability"] == "tts" assert info["provider"] == "elevenlabs" - assert info["usage_location"].endswith("tools\\audio\\elevenlabs_tts.py") or info["usage_location"].endswith("tools/audio/elevenlabs_tts.py") + assert info["usage_location"].endswith( + "tools\\audio\\elevenlabs_tts.py" + ) or info["usage_location"].endswith("tools/audio/elevenlabs_tts.py") assert "related_skills" in info assert "fallback_tools" in info @@ -151,7 +666,9 @@ class TestCapabilityMetadata: "piper_tts", "tts_selector", } - assert {tool.name for tool in reg.get_by_provider("elevenlabs")} == {"elevenlabs_tts"} + assert {tool.name for tool in reg.get_by_provider("elevenlabs")} == { + "elevenlabs_tts" + } def test_registry_catalog_views(self): reg = ToolRegistry() @@ -173,6 +690,7 @@ class TestCapabilityMetadata: # ---- Animated Explainer Pipeline ---- + class TestAnimatedExplainerManifest: def test_loads(self): manifest = load_pipeline("animated-explainer") @@ -182,7 +700,16 @@ class TestAnimatedExplainerManifest: def test_all_stages_present(self): manifest = load_pipeline("animated-explainer") stage_names = get_stage_order(manifest) - expected = ["research", "proposal", "script", "scene_plan", "assets", "edit", "compose", "publish"] + expected = [ + "research", + "proposal", + "script", + "scene_plan", + "assets", + "edit", + "compose", + "publish", + ] assert stage_names == expected def test_every_stage_has_skill(self): @@ -197,7 +724,9 @@ class TestAnimatedExplainerManifest: manifest = load_pipeline("animated-explainer") for stage in manifest["stages"]: focus = get_stage_review_focus(manifest, stage["name"]) - assert len(focus) >= 3, f"Stage {stage['name']} needs more review focus items" + assert len(focus) >= 3, ( + f"Stage {stage['name']} needs more review focus items" + ) def test_required_tools_complete(self): manifest = load_pipeline("animated-explainer") @@ -221,6 +750,7 @@ class TestAnimatedExplainerManifest: # ---- Style Playbooks ---- + class TestStylePlaybooks: def test_all_listed(self): playbooks = list_playbooks() @@ -228,13 +758,17 @@ class TestStylePlaybooks: assert "flat-motion-graphics" in playbooks assert "minimalist-diagram" in playbooks - @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]) + @pytest.mark.parametrize( + "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"] + ) def test_loads_and_validates(self, name): pb = load_playbook(name) assert pb["identity"]["name"] assert pb["identity"]["category"] - @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]) + @pytest.mark.parametrize( + "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"] + ) def test_has_required_sections(self, name): pb = load_playbook(name) assert "visual_language" in pb @@ -245,7 +779,9 @@ class TestStylePlaybooks: assert "quality_rules" in pb assert len(pb["quality_rules"]) >= 3 - @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]) + @pytest.mark.parametrize( + "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"] + ) def test_color_palette_complete(self, name): pb = load_playbook(name) palette = pb["visual_language"]["color_palette"] @@ -254,7 +790,9 @@ class TestStylePlaybooks: assert "background" in palette assert "text" in palette - @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]) + @pytest.mark.parametrize( + "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"] + ) def test_pacing_rules_present(self, name): pb = load_playbook(name) pacing = pb["motion"]["pacing_rules"] @@ -268,60 +806,75 @@ class TestStylePlaybooks: # compatible_playbooks is a dict with recommended/also_works lists playbook_names = compat.get("recommended", []) + compat.get("also_works", []) for name in playbook_names: - assert name in available, f"Manifest references unavailable playbook: {name}" + assert name in available, ( + f"Manifest references unavailable playbook: {name}" + ) # ---- Skills Existence ---- + class TestSkillsExist: SKILLS_DIR = PROJECT_ROOT / "skills" - @pytest.mark.parametrize("skill_path", [ - "pipelines/explainer/idea-director.md", - "pipelines/explainer/script-director.md", - "pipelines/explainer/scene-director.md", - "pipelines/explainer/asset-director.md", - "pipelines/explainer/edit-director.md", - "pipelines/explainer/compose-director.md", - "pipelines/explainer/publish-director.md", - ]) + @pytest.mark.parametrize( + "skill_path", + [ + "pipelines/explainer/idea-director.md", + "pipelines/explainer/script-director.md", + "pipelines/explainer/scene-director.md", + "pipelines/explainer/asset-director.md", + "pipelines/explainer/edit-director.md", + "pipelines/explainer/compose-director.md", + "pipelines/explainer/publish-director.md", + ], + ) def test_director_skills_exist(self, skill_path): full_path = self.SKILLS_DIR / skill_path assert full_path.exists(), f"Missing director skill: {skill_path}" content = full_path.read_text(encoding="utf-8") assert len(content) > 500, f"Skill too short to be useful: {skill_path}" - @pytest.mark.parametrize("skill_path", [ - "meta/reviewer.md", - "meta/checkpoint-protocol.md", - "meta/skill-creator.md", - ]) + @pytest.mark.parametrize( + "skill_path", + [ + "meta/reviewer.md", + "meta/checkpoint-protocol.md", + "meta/skill-creator.md", + ], + ) def test_meta_skills_exist(self, skill_path): full_path = self.SKILLS_DIR / skill_path assert full_path.exists(), f"Missing meta skill: {skill_path}" content = full_path.read_text(encoding="utf-8") assert len(content) > 500, f"Skill too short to be useful: {skill_path}" - @pytest.mark.parametrize("skill_path", [ - "pipelines/explainer/idea-director.md", - "pipelines/explainer/script-director.md", - "pipelines/explainer/scene-director.md", - "pipelines/explainer/asset-director.md", - "pipelines/explainer/edit-director.md", - "pipelines/explainer/compose-director.md", - "pipelines/explainer/publish-director.md", - ]) + @pytest.mark.parametrize( + "skill_path", + [ + "pipelines/explainer/idea-director.md", + "pipelines/explainer/script-director.md", + "pipelines/explainer/scene-director.md", + "pipelines/explainer/asset-director.md", + "pipelines/explainer/edit-director.md", + "pipelines/explainer/compose-director.md", + "pipelines/explainer/publish-director.md", + ], + ) def test_director_skills_have_required_sections(self, skill_path): content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8") assert "## When to Use" in content assert "## Process" in content or "## Protocol" in content assert "Self-Evaluate" in content or "self-evaluate" in content.lower() - @pytest.mark.parametrize("skill_path", [ - "meta/reviewer.md", - "meta/checkpoint-protocol.md", - "meta/skill-creator.md", - ]) + @pytest.mark.parametrize( + "skill_path", + [ + "meta/reviewer.md", + "meta/checkpoint-protocol.md", + "meta/skill-creator.md", + ], + ) def test_meta_skills_have_required_sections(self, skill_path): content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8") assert "## When to Use" in content @@ -330,6 +883,7 @@ class TestSkillsExist: # ---- Remotion Scaffold ---- + class TestRemotionScaffold: REMOTION_DIR = PROJECT_ROOT / "remotion-composer" @@ -354,17 +908,23 @@ class TestRemotionScaffold: # ---- Video Compose Operations ---- + class TestVideoComposeOperations: def test_render_operation_exists(self): + from typing import Any from tools.video.video_compose import VideoCompose + tool = VideoCompose() - ops = tool.input_schema["properties"]["operation"]["enum"] + schema: Any = tool.input_schema + ops = schema["properties"]["operation"]["enum"] assert "render" in ops assert "remotion_render" in ops def test_render_rejects_missing_inputs(self): from tools.video.video_compose import VideoCompose + tool = VideoCompose() result = tool.execute({"operation": "render"}) assert not result.success + assert result.error is not None assert "edit_decisions" in result.error diff --git a/tools/audio/google_music.py b/tools/audio/google_music.py new file mode 100644 index 00000000..defb56fd --- /dev/null +++ b/tools/audio/google_music.py @@ -0,0 +1,333 @@ +"""Generate music using Google Lyria via Google GenAI SDK. + +Generate background music and audio tracks for video production using lyria-3-pro-preview. +""" + +from __future__ import annotations + +import base64 +import mimetypes +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class GoogleMusic(BaseTool): + name = "google_music" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "music_generation" + provider = "google" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Configure Google credentials:\n" + " - Set GEMINI_API_KEY (or GOOGLE_API_KEY) in environment.\n" + " - Or set GOOGLE_APPLICATION_CREDENTIALS for Vertex AI service account." + ) + fallback_tools = ["music_gen"] + agent_skills = ["music"] + + capabilities = [ + "generate_background_music", + ] + supports = { + "instrumental": True, + "vocals": True, + "custom_lyrics": True, + "style_control": True, + "long_form": True, + } + best_for = [ + "high-quality instrumental background music", + "genre-specific music guided by rich text prompts", + "Google ecosystem integration", + ] + not_good_for = [ + "offline generation", + "sub-5-second sound effects", + ] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": { + "type": "string", + "description": "Music description (mood, genre, instruments, tempo)", + }, + "duration_seconds": { + "type": "number", + "minimum": 5, + "maximum": 184, + "default": 30, + "description": "Target duration in seconds (model hard limit is 184s)", + }, + "image_url": { + "type": "string", + "description": "Reference image URL for visual music conditioning", + }, + "image_path": { + "type": "string", + "description": "Local reference image path for visual music conditioning", + }, + "auto_fix": {"type": "boolean", "default": True}, + "output_path": { + "type": "string", + "default": "music_output.mp3", + "description": "Path where the generated MP3 file should be written", + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True + ) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) + idempotency_key_fields = ["prompt", "duration_seconds", "image_url", "image_path"] + side_effects = [ + "writes audio file to output_path", + "calls Google Gemini/Vertex API", + ] + user_visible_verification = [ + "Listen to generated music for style and quality", + ] + + def _get_google_credentials_status(self) -> bool: + """Check whether Google API keys or Vertex AI service account credentials are set.""" + from tools.google_credentials import has_google_credentials + + return has_google_credentials() + + def get_status(self) -> ToolStatus: + """Determine whether the tool is available based on configured credentials.""" + if self._get_google_credentials_status(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + """Estimate the generation cost in USD.""" + # Lyria 3 Pro is a flat $0.08 per generation request + return 0.08 + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + """Execute the music generation tool using the Google GenAI SDK.""" + if not self._get_google_credentials_status(): + return ToolResult( + success=False, + error="No Google credentials configured. " + self.install_instructions, + ) + + start = time.time() + + try: + import requests + from google.genai import types + from tools.google_credentials import get_genai_client, GOOGLE_API_TIMEOUT_MS + + http_options = types.HttpOptions(timeout=GOOGLE_API_TIMEOUT_MS) + client = get_genai_client(http_options=http_options) + except ImportError as e: + return ToolResult( + success=False, + error=f"Failed to import required Google libraries: {e}. Run 'uv pip install google-genai requests'", + ) + except Exception as e: + return ToolResult( + success=False, + error=f"Failed to initialize Google GenAI Client: {e}", + ) + + prompt = inputs["prompt"] + duration = float(inputs.get("duration_seconds", 30)) + auto_fix = inputs.get("auto_fix", True) + output_path = inputs.get("output_path", "music_output.mp3") + + # Ensure minimum duration of 5 seconds + if duration < 5: + if auto_fix: + import logging + + logging.getLogger(__name__).warning( + "Lyria 3 Pro requires a minimum duration of 5 seconds. Coercing duration_seconds to 5.0." + ) + duration = 5.0 + else: + return ToolResult( + success=False, + error="lyria-3-pro-preview minimum duration is 5 seconds.", + ) + + # Cap at 184 seconds + if duration > 184: + if auto_fix: + import logging + + logging.getLogger(__name__).warning( + "Lyria 3 Pro supports up to 184 seconds of audio. Coercing duration_seconds to 184." + ) + duration = 184.0 + else: + return ToolResult( + success=False, + error="lyria-3-pro-preview maximum duration is 184 seconds.", + ) + + # Helper to load reference image bytes + mime type + def _get_image_data( + url: str | None, path: str | None + ) -> tuple[str, str] | None: + if path: + if not os.path.exists(path): + raise FileNotFoundError(f"Local reference image not found: {path}") + img_bytes = Path(path).read_bytes() + mime, _ = mimetypes.guess_type(path) + if not mime: + mime = "image/png" + b64 = base64.b64encode(img_bytes).decode("utf-8") + return b64, mime + if url: + resp = requests.get(url, timeout=30) + resp.raise_for_status() + mime = resp.headers.get("Content-Type") + if not mime or "image" not in mime: + mime = "image/png" + b64 = base64.b64encode(resp.content).decode("utf-8") + return b64, mime + return None + + # Build payload input incorporating target duration instructions + timed_prompt = f"{prompt}\n\n[Target Duration: {int(duration)} seconds]" + input_list: list[dict[str, Any]] = [{"type": "text", "text": timed_prompt}] + try: + image_data = _get_image_data( + inputs.get("image_url"), inputs.get("image_path") + ) + if image_data: + b64, mime = image_data + input_list.append({"type": "image", "mime_type": mime, "data": b64}) + except Exception as e: + return ToolResult( + success=False, + error=f"Failed to load visual conditioning image: {e}", + ) + + model_name = "lyria-3-pro-preview" + + try: + # Create parent dirs if needed + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + + interaction = client.interactions.create(model=model_name, input=input_list) + + if hasattr(interaction, "status") and interaction.status in ( + "failed", + "cancelled", + ): + return ToolResult( + success=False, + error=f"Google Lyria music generation failed. Status: {interaction.status}", + ) + + audio_data = None + if hasattr(interaction, "output_audio") and interaction.output_audio: + audio_data = getattr(interaction.output_audio, "data", None) + + # Fall back to outputs list + outputs = getattr(interaction, "outputs", None) + if not audio_data and isinstance(outputs, list): + for output in outputs: + if hasattr(output, "inline_data") and output.inline_data: + audio_data = getattr(output.inline_data, "data", None) + if audio_data: + break + + # Fall back to step traversal + steps = getattr(interaction, "steps", None) + if not audio_data and isinstance(steps, list): + for step in steps: + if ( + hasattr(step, "type") + and step.type == "model_output" + and hasattr(step, "content") + and step.content + ): + for content_part in step.content: + if ( + hasattr(content_part, "type") + and content_part.type == "audio" + ): + audio_data = getattr(content_part, "data", None) + break + if audio_data: + break + + if not audio_data: + return ToolResult( + success=False, + error=f"No audio data returned by model {model_name}.", + ) + + # Decode and save output file + if isinstance(audio_data, str): + audio_bytes = base64.b64decode(audio_data) + else: + # If it's already bytes, it could be raw audio or base64 bytes + if audio_data.startswith(b"ID3") or ( + len(audio_data) > 2 + and audio_data[0] == 0xFF + and (audio_data[1] & 0xE0) == 0xE0 + ): + audio_bytes = audio_data + else: + try: + audio_bytes = base64.b64decode(audio_data) + except Exception: + audio_bytes = audio_data + + Path(output_path).write_bytes(audio_bytes) + + except Exception as e: + return ToolResult( + success=False, error=f"Google Lyria music generation failed: {e}" + ) + + duration_seconds = round(time.time() - start, 2) + cost_usd = self.estimate_cost(inputs) + + return ToolResult( + success=True, + data={ + "provider": "google", + "model": model_name, + "prompt": prompt, + "duration_seconds": duration, + "output": str(output_path), + "output_path": str(output_path), + "format": "mp3", + }, + artifacts=[str(output_path)], + cost_usd=cost_usd, + duration_seconds=duration_seconds, + model=model_name, + ) diff --git a/tools/audio/google_tts.py b/tools/audio/google_tts.py index b6218a1c..424e4b30 100644 --- a/tools/audio/google_tts.py +++ b/tools/audio/google_tts.py @@ -24,7 +24,11 @@ from tools.base_tool import ( ToolStatus, ToolTier, ) -from tools.google_credentials import get_access_token, service_account_configured +from tools.google_credentials import ( + get_access_token, + service_account_configured, + has_google_credentials, +) class GoogleTTS(BaseTool): @@ -121,8 +125,17 @@ class GoogleTTS(BaseTool): resource_profile = ResourceProfile( cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True ) - retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) - idempotency_key_fields = ["text", "input_type", "voice", "language_code", "speaking_rate", "pitch"] + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) + idempotency_key_fields = [ + "text", + "input_type", + "voice", + "language_code", + "speaking_rate", + "pitch", + ] side_effects = ["writes audio file to output_path", "calls Google Cloud TTS API"] user_visible_verification = ["Listen to generated audio for natural speech quality"] @@ -141,7 +154,7 @@ class GoogleTTS(BaseTool): def get_status(self) -> ToolStatus: # Available via either an API key or a service-account JSON. Both paths # are honoured by execute() — so this no longer over-reports. - if self._get_api_key() or service_account_configured(): + if has_google_credentials(): return ToolStatus.AVAILABLE return ToolStatus.UNAVAILABLE @@ -226,7 +239,11 @@ class GoogleTTS(BaseTool): if input_type == "ssml": stripped = text.strip() - ssml = stripped if stripped.startswith("{stripped}" + ssml = ( + stripped + if stripped.startswith("{stripped}" + ) synthesis_input = {"ssml": ssml} else: synthesis_input = {"text": text} @@ -252,7 +269,7 @@ class GoogleTTS(BaseTool): params: dict[str, str] = {} if bearer_token: headers["Authorization"] = f"Bearer {bearer_token}" - else: + elif api_key: params["key"] = api_key response = requests.post( diff --git a/tools/google_credentials.py b/tools/google_credentials.py index 6336bc24..71158c73 100644 --- a/tools/google_credentials.py +++ b/tools/google_credentials.py @@ -13,10 +13,15 @@ surfaces as an actionable runtime error rather than a hard import failure. from __future__ import annotations import os +from typing import Any # Broad scope that covers Cloud Text-to-Speech and Vertex AI prediction. CLOUD_PLATFORM_SCOPE = "https://www.googleapis.com/auth/cloud-platform" +# Shared constants for long-running Google/Vertex AI generation calls (e.g. music, video) +GOOGLE_API_TIMEOUT_SECONDS = 600 +GOOGLE_API_TIMEOUT_MS = GOOGLE_API_TIMEOUT_SECONDS * 1000 + def service_account_configured() -> bool: """True when GOOGLE_APPLICATION_CREDENTIALS points to an existing file.""" @@ -24,6 +29,41 @@ def service_account_configured() -> bool: return bool(path and os.path.exists(path)) +def has_google_credentials() -> bool: + """True when GOOGLE_API_KEY, GEMINI_API_KEY, or service account is configured.""" + return bool( + os.environ.get("GOOGLE_API_KEY") + or os.environ.get("GEMINI_API_KEY") + or service_account_configured() + ) + + +def get_genai_client(http_options: Any | None = None) -> Any: + """Lazily import and initialize the Google GenAI Client based on configured credentials.""" + from google import genai + + api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY") + use_vertex = os.environ.get("GOOGLE_GENAI_USE_VERTEXAI", "").lower() in ( + "true", + "1", + ) or os.environ.get("GOOGLE_GENAI_USE_ENTERPRISE", "").lower() in ("true", "1") + + if use_vertex or (not api_key and service_account_configured()): + kwargs = { + "vertexai": True, + "location": os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"), + "http_options": http_options, + } + project_id = resolve_project_id() + if project_id: + kwargs["project"] = project_id + return genai.Client(**kwargs) + else: + if api_key: + return genai.Client(api_key=api_key, http_options=http_options) + return genai.Client(http_options=http_options) + + def resolve_project_id(creds_project_id: str | None = None) -> str | None: """Resolve the GCP project id from env vars, falling back to the key file's. @@ -77,4 +117,12 @@ def get_access_token(scopes: list[str] | None = None) -> tuple[str, str | None]: f"Failed to load/refresh service-account credentials from {path}: {exc}" ) from exc - return creds.token, getattr(creds, "project_id", None) + token = creds.token + if not token or not isinstance(token, str): + raise RuntimeError( + "Service-account credentials did not yield a valid access token." + ) + + project_id = getattr(creds, "project_id", None) + ret_project_id = str(project_id) if project_id is not None else None + return token, ret_project_id diff --git a/tools/graphics/google_imagen.py b/tools/graphics/google_imagen.py index 72eebba0..a222374c 100644 --- a/tools/graphics/google_imagen.py +++ b/tools/graphics/google_imagen.py @@ -24,6 +24,7 @@ from tools.google_credentials import ( get_access_token, resolve_project_id, service_account_configured, + has_google_credentials, ) # Aspect ratio to approximate pixel dimensions (for cost/reporting only) @@ -93,7 +94,10 @@ class GoogleImagen(BaseTool): "type": "object", "required": ["prompt"], "properties": { - "prompt": {"type": "string", "description": "Image description (max 480 tokens)"}, + "prompt": { + "type": "string", + "description": "Image description (max 480 tokens)", + }, "aspect_ratio": { "type": "string", "enum": ["1:1", "3:4", "4:3", "9:16", "16:9"], @@ -131,9 +135,14 @@ class GoogleImagen(BaseTool): resource_profile = ResourceProfile( cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True ) - retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) idempotency_key_fields = ["prompt", "aspect_ratio", "model"] - side_effects = ["writes image file to output_path", "calls Google Generative AI API"] + side_effects = [ + "writes image file to output_path", + "calls Google Generative AI API", + ] user_visible_verification = ["Inspect generated image for relevance and quality"] def _get_api_key(self) -> str | None: @@ -141,7 +150,7 @@ class GoogleImagen(BaseTool): def get_status(self) -> ToolStatus: # API key -> AI Studio endpoint; service-account JSON -> Vertex AI. - if self._get_api_key() or service_account_configured(): + if has_google_credentials(): return ToolStatus.AVAILABLE return ToolStatus.UNAVAILABLE @@ -188,6 +197,7 @@ class GoogleImagen(BaseTool): prompt = inputs["prompt"] import logging + logger = logging.getLogger(__name__) # Resolve aspect ratio: explicit > derived from width/height > default @@ -198,7 +208,8 @@ class GoogleImagen(BaseTool): aspect_ratio = _dims_to_aspect_ratio(inputs["width"], inputs["height"]) logger.info( "google_imagen: remapped %s to nearest supported aspect ratio %s", - requested_ratio, aspect_ratio, + requested_ratio, + aspect_ratio, ) else: aspect_ratio = "1:1" @@ -228,7 +239,7 @@ class GoogleImagen(BaseTool): ) headers = { "Content-Type": "application/json", - "x-goog-api-key": api_key, + "x-goog-api-key": api_key or "", } try: @@ -246,11 +257,11 @@ class GoogleImagen(BaseTool): predictions = data.get("predictions", []) if not predictions: - return ToolResult(success=False, error="No images returned from Imagen API") + return ToolResult( + success=False, error="No images returned from Imagen API" + ) - image_bytes = base64.b64decode( - predictions[0]["bytesBase64Encoded"] - ) + image_bytes = base64.b64decode(predictions[0]["bytesBase64Encoded"]) output_path = Path(inputs.get("output_path", "generated_image.png")) output_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tools/video/veo_video.py b/tools/video/veo_video.py index 33b177a6..de4efea2 100644 --- a/tools/video/veo_video.py +++ b/tools/video/veo_video.py @@ -1,6 +1,6 @@ -"""Google Veo 3.1 video generation via fal.ai API. +"""Generate video using Google Veo 3.1 via fal.ai or Google GenAI API. -Supports text-to-video, image-to-video, reference-to-video, and first/last-frame +Support text-to-video, image-to-video, reference-to-video, and first/last-frame interpolation so agents can preserve visual consistency instead of relying only on raw text prompts. """ @@ -14,6 +14,8 @@ import time from pathlib import Path from typing import Any +from tools.google_credentials import GOOGLE_API_TIMEOUT_SECONDS + from tools.base_tool import ( BaseTool, Determinism, @@ -30,7 +32,7 @@ from tools.base_tool import ( class VeoVideo(BaseTool): name = "veo_video" - version = "0.1.0" + version = "0.2.0" tier = ToolTier.GENERATE capability = "video_generation" provider = "veo" @@ -41,12 +43,21 @@ class VeoVideo(BaseTool): dependencies = [] install_instructions = ( - "Set FAL_KEY or FAL_AI_API_KEY to your fal.ai API key.\n" - " Get one at https://fal.ai/dashboard/keys" + "Configure at least one backend API key:\n" + " - Direct Google GenAI backend: Set GEMINI_API_KEY (or GOOGLE_API_KEY).\n" + " Get a key at https://aistudio.google.com/\n" + " Or set GOOGLE_APPLICATION_CREDENTIALS for Vertex AI service account.\n" + " - FAL.ai backend: Set FAL_KEY (or FAL_AI_API_KEY).\n" + " Get one at https://fal.ai/dashboard/keys" ) agent_skills = ["ai-video-gen"] - capabilities = ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"] + capabilities = [ + "text_to_video", + "image_to_video", + "reference_to_video", + "first_last_frame_to_video", + ] supports = { "text_to_video": True, "image_to_video": True, @@ -69,21 +80,31 @@ class VeoVideo(BaseTool): "required": ["prompt"], "properties": { "prompt": {"type": "string"}, + "backend": { + "type": "string", + "enum": ["auto", "google", "fal"], + "default": "auto", + "description": "API backend provider to use for generation", + }, "operation": { "type": "string", - "enum": ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"], + "enum": [ + "text_to_video", + "image_to_video", + "reference_to_video", + "first_last_frame_to_video", + ], "default": "text_to_video", }, "model_variant": { "type": "string", - "enum": ["veo3", "veo3/fast", "veo3.1", "veo3.1/fast"], "default": "veo3.1", + "description": "Model variant for FAL (e.g. veo3.1) or custom model for Google", }, "duration": { "type": "string", - "enum": ["4s", "6s", "8s"], "default": "8s", - "description": "Duration in seconds", + "description": "Duration (e.g., '4s', '6s', '8s')", }, "aspect_ratio": { "type": "string", @@ -108,8 +129,14 @@ class VeoVideo(BaseTool): "enum": ["1", "2", "3", "4", "5", "6"], "default": "4", }, - "image_url": {"type": "string", "description": "Reference image URL for image_to_video"}, - "image_path": {"type": "string", "description": "Local reference image path for image_to_video"}, + "image_url": { + "type": "string", + "description": "Reference image URL for image_to_video", + }, + "image_path": { + "type": "string", + "description": "Local reference image path for image_to_video", + }, "reference_image_urls": { "type": "array", "items": {"type": "string"}, @@ -131,26 +158,56 @@ class VeoVideo(BaseTool): resource_profile = ResourceProfile( cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True ) - retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) + retry_policy = RetryPolicy( + max_retries=2, retryable_errors=["rate_limit", "timeout"] + ) idempotency_key_fields = ["prompt", "model_variant", "operation", "duration"] - side_effects = ["writes video file to output_path", "calls fal.ai API"] + side_effects = ["writes video file to output_path", "calls fal.ai or Google APIs"] user_visible_verification = [ "Watch generated clip for visual quality and motion", "Listen for audio synchronization and quality", ] - def _get_api_key(self) -> str | None: + def _get_google_credentials_status(self) -> bool: + """Check whether Google API keys or Vertex AI service account credentials are set.""" + from tools.google_credentials import has_google_credentials + + return has_google_credentials() + + def _get_fal_api_key(self) -> str | None: + """Retrieve the FAL API key from environment variables.""" return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY") def get_status(self) -> ToolStatus: - if self._get_api_key(): + """Determine whether the tool is available based on configured credentials.""" + if self._get_google_credentials_status() or self._get_fal_api_key(): return ToolStatus.AVAILABLE return ToolStatus.UNAVAILABLE def estimate_cost(self, inputs: dict[str, Any]) -> float: + """Estimate the generation cost in USD based on input parameters.""" + # Determine active backend using inputs and environment + backend = inputs.get("backend", "auto") + if backend == "auto": + if self._get_google_credentials_status(): + backend = "google" + elif self._get_fal_api_key(): + backend = "fal" + else: + backend = "google" + + duration_text = str(inputs.get("duration", "8s")).lower().replace("s", "") + try: + duration = int(duration_text) + except ValueError: + duration = 8 + + if backend == "google": + # Standard Google Veo is $0.40 per second + return round(duration * 0.40, 4) + + # FAL cost estimation variant = inputs.get("model_variant", "veo3.1") - duration_text = str(inputs.get("duration", "8s")).replace("s", "") - duration = int(duration_text) resolution = inputs.get("resolution", "1080p") generate_audio = bool(inputs.get("generate_audio", True)) @@ -168,6 +225,14 @@ class VeoVideo(BaseTool): return (audio_per_second if generate_audio else base_per_second) * duration def estimate_runtime(self, inputs: dict[str, Any]) -> float: + """Estimate the expected runtime in seconds.""" + backend = inputs.get("backend", "auto") + if backend == "auto": + backend = "google" if self._get_google_credentials_status() else "fal" + + if backend == "google": + return 90.0 + variant = inputs.get("model_variant", "veo3.1") if "fast" in variant: return 45.0 @@ -175,6 +240,7 @@ class VeoVideo(BaseTool): @staticmethod def _file_to_data_uri(path_str: str) -> str: + """Convert a local file into a base64-encoded Data URI.""" path = Path(path_str) if not path.exists(): raise FileNotFoundError(f"Input file not found: {path}") @@ -184,7 +250,10 @@ class VeoVideo(BaseTool): encoded = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:{mime_type};base64,{encoded}" - def _normalize_file_input(self, url_value: str | None, path_value: str | None) -> str | None: + def _normalize_file_input( + self, url_value: str | None, path_value: str | None + ) -> str | None: + """Normalize file input by converting local file paths to Data URIs or returning URLs.""" if url_value: return url_value if path_value: @@ -192,7 +261,285 @@ class VeoVideo(BaseTool): return None def execute(self, inputs: dict[str, Any]) -> ToolResult: - api_key = self._get_api_key() + """Execute the video generation tool using the selected backend.""" + backend = inputs.get("backend", "auto") + if backend == "auto": + if self._get_google_credentials_status(): + backend = "google" + elif self._get_fal_api_key(): + backend = "fal" + else: + return ToolResult( + success=False, + error="No backend credentials configured. " + + self.install_instructions, + ) + + if backend == "google": + return self._execute_google(inputs) + return self._execute_fal(inputs) + + def _execute_google(self, inputs: dict[str, Any]) -> ToolResult: + """Execute the generation request using the Google GenAI SDK backend.""" + start = time.time() + try: + from google.genai import types + from PIL import Image + from io import BytesIO + import requests + from tools.google_credentials import get_genai_client, GOOGLE_API_TIMEOUT_MS + + http_options = types.HttpOptions(timeout=GOOGLE_API_TIMEOUT_MS) + client = get_genai_client(http_options=http_options) + except ImportError as e: + return ToolResult( + success=False, + error=f"Failed to import required Google libraries: {e}. Run 'uv pip install google-genai pillow requests'", + ) + except Exception as e: + return ToolResult( + success=False, + error=f"Failed to initialize Google GenAI Client: {e}", + ) + + is_vertex = getattr(client, "vertexai", None) + if is_vertex is None or not isinstance(is_vertex, bool): + is_vertex = getattr(client, "_api_client", None) and getattr( + client._api_client, "vertexai", False + ) + + if is_vertex: + return ToolResult( + success=False, + error="Google Veo video generation via google-genai is only supported using the Gemini Developer API (API key) backend. " + "Please configure GEMINI_API_KEY/GOOGLE_API_KEY or use the FAL.ai backend.", + ) + + prompt = inputs["prompt"] + operation = inputs.get("operation", "text_to_video") + model_variant = inputs.get("model_variant", "veo3.1") + auto_fix = inputs.get("auto_fix", True) + + # Map to the official preview model unless a custom model name is provided + if model_variant in {"veo3", "veo3/fast", "veo3.1", "veo3.1/fast"}: + if is_vertex: + model_name = "veo-3.1-generate-001" + else: + model_name = "veo-3.1-generate-preview" + else: + model_name = model_variant + + duration_text = str(inputs.get("duration", "8s")).lower().replace("s", "") + try: + duration_seconds = int(duration_text) + except ValueError: + duration_seconds = 8 + + aspect_ratio = inputs.get("aspect_ratio", "16:9") + resolution = inputs.get("resolution", "1080p") + + # Validate/Auto-Fix duration based on 1080p/4K or reference-to-video rules + needs_8s = (resolution in {"1080p", "4k"}) or ( + operation == "reference_to_video" + ) + if needs_8s and duration_seconds != 8: + if auto_fix: + import logging + + logging.getLogger(__name__).warning( + f"Google Veo 3.1 requires 8 seconds duration when using " + f"resolution={resolution} or operation={operation}. Coercing duration to 8s." + ) + duration_seconds = 8 + else: + return ToolResult( + success=False, + error=f"Google Veo 3.1 requires duration to be 8 seconds when resolution is {resolution} or operation is {operation}.", + ) + + # Construct generation configuration + config = types.GenerateVideosConfig( + aspect_ratio=aspect_ratio, + duration_seconds=duration_seconds, + resolution=resolution, + number_of_videos=1, + ) + + if inputs.get("generate_audio") is not None: + config.generate_audio = inputs["generate_audio"] + if inputs.get("negative_prompt"): + config.negative_prompt = inputs["negative_prompt"] + if inputs.get("seed") is not None: + config.seed = inputs["seed"] + + def _get_image(url: str | None, path: str | None) -> Image.Image | None: + if path: + if not os.path.exists(path): + raise FileNotFoundError(f"Local input image not found: {path}") + return Image.open(path) + if url: + resp = requests.get(url, timeout=30) + resp.raise_for_status() + return Image.open(BytesIO(resp.content)) + return None + + def _to_sdk_image(pil_img: Image.Image) -> types.Image: + buf = BytesIO() + fmt = pil_img.format or "PNG" + try: + pil_img.save(buf, format=fmt) + except KeyError: + pil_img.save(buf, format="PNG") + fmt = "PNG" + return types.Image( + image_bytes=buf.getvalue(), + mime_type=f"image/{fmt.lower()}", + ) + + # Build execution input args + sdk_image = None + try: + if operation == "image_to_video": + image_obj = _get_image( + inputs.get("image_url"), inputs.get("image_path") + ) + if not image_obj: + return ToolResult( + success=False, + error="image_to_video requires image_url or image_path", + ) + sdk_image = _to_sdk_image(image_obj) + + elif operation == "first_last_frame_to_video": + image_obj = _get_image( + inputs.get("first_frame_url"), inputs.get("first_frame_path") + ) + last_image = _get_image( + inputs.get("last_frame_url"), inputs.get("last_frame_path") + ) + if not image_obj or not last_image: + return ToolResult( + success=False, + error="first_last_frame_to_video requires first_frame_url/path and last_frame_url/path", + ) + config.last_frame = _to_sdk_image(last_image) + sdk_image = _to_sdk_image(image_obj) + + elif operation == "reference_to_video": + ref_images = [] + image_urls = list(inputs.get("reference_image_urls") or []) + image_paths = list(inputs.get("reference_image_paths") or []) + + for path in image_paths: + if not os.path.exists(path): + raise FileNotFoundError( + f"Local reference image not found: {path}" + ) + ref_images.append( + types.VideoGenerationReferenceImage( + image=_to_sdk_image(Image.open(path)), + reference_type=types.VideoGenerationReferenceType.ASSET, + ) + ) + for url in image_urls: + resp = requests.get(url, timeout=30) + resp.raise_for_status() + ref_images.append( + types.VideoGenerationReferenceImage( + image=_to_sdk_image(Image.open(BytesIO(resp.content))), + reference_type=types.VideoGenerationReferenceType.ASSET, + ) + ) + if not ref_images: + return ToolResult( + success=False, + error="reference_to_video requires reference_image_urls or reference_image_paths", + ) + config.reference_images = ref_images + + except Exception as e: + return ToolResult( + success=False, + error=f"Failed to load inputs for operation {operation}: {e}", + ) + + try: + # Submit generation request + operation_handle = client.models.generate_videos( + model=model_name, prompt=prompt, image=sdk_image, config=config + ) + + # Poll for completion with safety timeout + poll_interval = 5 + deadline = time.time() + GOOGLE_API_TIMEOUT_SECONDS + while not operation_handle.done: + if time.time() >= deadline: + return ToolResult( + success=False, + error=f"Veo video generation timed out after {GOOGLE_API_TIMEOUT_SECONDS} seconds.", + ) + time.sleep(poll_interval) + operation_handle = client.operations.get(operation_handle) + + if operation_handle.error: + return ToolResult( + success=False, + error=f"Veo direct API error: {operation_handle.error}", + ) + + # Download and save final file + response = operation_handle.response + if not response or not response.generated_videos: + return ToolResult( + success=False, + error="No video generation response received.", + ) + video_result = response.generated_videos[0] + video_asset = video_result.video + if not video_asset: + return ToolResult( + success=False, + error="No video asset returned in the response.", + ) + client.files.download(file=video_asset) + + output_path = Path(inputs.get("output_path", "veo_output.mp4")) + output_path.parent.mkdir(parents=True, exist_ok=True) + video_asset.save(str(output_path)) + + except Exception as e: + return ToolResult( + success=False, + error=f"Veo direct API generation failed: {e}", + ) + + from tools.video._shared import probe_output + + probed = probe_output(output_path) + + return ToolResult( + success=True, + data={ + "provider": "veo", + "gateway": "google", + "model": model_name, + "prompt": prompt, + "operation": operation, + "aspect_ratio": aspect_ratio, + "output": str(output_path), + "output_path": str(output_path), + "format": "mp4", + **probed, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=model_name, + ) + + def _execute_fal(self, inputs: dict[str, Any]) -> ToolResult: + """Execute the generation request using the fal.ai API backend.""" + api_key = self._get_fal_api_key() if not api_key: return ToolResult( success=False, @@ -207,7 +554,11 @@ class VeoVideo(BaseTool): duration = inputs.get("duration", "8s") # Current fal Veo 3.1 image-guided endpoints only accept 8-second clips. - if variant == "veo3.1" and operation in {"reference_to_video", "first_last_frame_to_video"} and duration != "8s": + if ( + variant == "veo3.1" + and operation in {"reference_to_video", "first_last_frame_to_video"} + and duration != "8s" + ): return ToolResult( success=False, error=( @@ -244,7 +595,9 @@ class VeoVideo(BaseTool): payload["safety_tolerance"] = inputs["safety_tolerance"] if operation == "image_to_video": - image_value = self._normalize_file_input(inputs.get("image_url"), inputs.get("image_path")) + image_value = self._normalize_file_input( + inputs.get("image_url"), inputs.get("image_path") + ) if not image_value: return ToolResult( success=False, @@ -297,9 +650,16 @@ class VeoVideo(BaseTool): status_url = queue_data["status_url"] response_url = queue_data["response_url"] - # Poll until complete + # Poll until complete with safety timeout + poll_interval = 5 + deadline = time.time() + GOOGLE_API_TIMEOUT_SECONDS while True: - time.sleep(5) + if time.time() >= deadline: + return ToolResult( + success=False, + error=f"Veo video generation timed out on FAL.ai after {GOOGLE_API_TIMEOUT_SECONDS} seconds.", + ) + time.sleep(poll_interval) status_resp = requests.get(status_url, headers=headers, timeout=15) status_resp.raise_for_status() status = status_resp.json().get("status", "UNKNOWN") @@ -332,15 +692,23 @@ class VeoVideo(BaseTool): except Exception as e: return ToolResult(success=False, error=f"Veo video generation failed: {e}") + from tools.video._shared import probe_output + + probed = probe_output(output_path) + return ToolResult( success=True, data={ "provider": "veo", + "gateway": "fal", "model": f"fal-ai/{model_path}", "prompt": inputs["prompt"], - "output": str(output_path), - "has_audio": inputs.get("generate_audio", True), "operation": operation, + "aspect_ratio": inputs.get("aspect_ratio", "16:9"), + "output": str(output_path), + "output_path": str(output_path), + "format": "mp4", + **probed, }, artifacts=[str(output_path)], cost_usd=self.estimate_cost(inputs),