diff --git a/model_tools.py b/model_tools.py index c651d93ed73d..ceffb7fa1016 100644 --- a/model_tools.py +++ b/model_tools.py @@ -142,6 +142,7 @@ def _discover_tools(): "tools.vision_tools", "tools.mixture_of_agents_tool", "tools.image_generation_tool", + "tools.video_generation_tool", "tools.skills_tool", "tools.skill_manager_tool", "tools.browser_tool", diff --git a/tests/tools/test_video_generation_tool.py b/tests/tools/test_video_generation_tool.py new file mode 100644 index 000000000000..e82eca216cb0 --- /dev/null +++ b/tests/tools/test_video_generation_tool.py @@ -0,0 +1,612 @@ +""" +Tests for video_generation_tool.py + +Covers: +- Parameter validation (prompt, model, duration, aspect_ratio) +- _extract_video_url for different FAL.ai response shapes +- _build_arguments per model family +- check_video_generation_requirements +- video_generate_tool success path (mocked fal_client) +- video_generate_tool failure paths (missing key, bad response, download error) +- Registry registration +""" + +import json +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers to isolate fal_client and urllib from the real network +# --------------------------------------------------------------------------- + +def _make_fal_mock(video_url: str): + """Return a mock fal_client whose submit().get() yields a video URL.""" + mock_handler = MagicMock() + mock_handler.get.return_value = {"video": {"url": video_url, "width": 1280, "height": 720}} + + mock_fal = MagicMock() + mock_fal.submit.return_value = mock_handler + return mock_fal + + +# --------------------------------------------------------------------------- +# Unit tests — internal helpers (no network) +# --------------------------------------------------------------------------- + +class TestExtractVideoUrl: + def setup_method(self): + from tools.video_generation_tool import _extract_video_url + self._extract = _extract_video_url + + def test_single_video_object(self): + result = {"video": {"url": "https://cdn.fal.ai/video.mp4"}} + assert self._extract(result) == "https://cdn.fal.ai/video.mp4" + + def test_videos_list(self): + result = {"videos": [{"url": "https://cdn.fal.ai/v1.mp4"}, {"url": "https://cdn.fal.ai/v2.mp4"}]} + assert self._extract(result) == "https://cdn.fal.ai/v1.mp4" + + def test_none_result(self): + assert self._extract(None) is None + + def test_empty_dict(self): + assert self._extract({}) is None + + def test_empty_videos_list(self): + assert self._extract({"videos": []}) is None + + def test_video_without_url_key(self): + result = {"video": {"width": 1280, "height": 720}} + assert self._extract(result) is None + + +class TestBuildArguments: + def setup_method(self): + from tools.video_generation_tool import _build_arguments + self._build = _build_arguments + + def test_kling_includes_duration_and_aspect(self): + args = self._build("kling", "a dog running", 5, "landscape", None, None) + assert args["prompt"] == "a dog running" + assert args["duration"] == "5" + assert args["aspect_ratio"] == "16:9" + assert "negative_prompt" not in args + + def test_kling_10s_duration(self): + args = self._build("kling", "sunset timelapse", 10, "portrait", None, None) + assert args["duration"] == "10" + assert args["aspect_ratio"] == "9:16" + + def test_kling_negative_prompt_included(self): + args = self._build("kling", "a scene", 5, "square", "blurry, shaky", None) + assert args["negative_prompt"] == "blurry, shaky" + assert args["aspect_ratio"] == "1:1" + + def test_luma_has_aspect_and_loop(self): + args = self._build("luma", "a galaxy", 5, "landscape", None, None) + assert args["aspect_ratio"] == "16:9" + assert args["loop"] is False + assert "duration" not in args + + def test_minimax_prompt_only(self): + args = self._build("minimax", "waves crashing", 5, "landscape", None, None) + assert args["prompt"] == "waves crashing" + assert "aspect_ratio" not in args + assert "duration" not in args + + def test_prompt_stripped(self): + args = self._build("kling", " spaced prompt ", 5, "landscape", None, None) + assert args["prompt"] == "spaced prompt" + + def test_kling_image_url_included(self): + args = self._build("kling", "make it move", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert args["image_url"] == "https://cdn.fal.ai/img.png" + + def test_luma_image_url_included(self): + args = self._build("luma", "ocean waves", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert args["image_url"] == "https://cdn.fal.ai/img.png" + + def test_minimax_image_url_included(self): + args = self._build("minimax", "gentle motion", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert args["image_url"] == "https://cdn.fal.ai/img.png" + + def test_hunyuan_no_image_url(self): + args = self._build("hunyuan", "a galaxy", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert "image_url" not in args + assert args["aspect_ratio"] == "16:9" + + def test_veo2_no_image_url(self): + args = self._build("veo2", "a volcano", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert "image_url" not in args + assert args["aspect_ratio"] == "16:9" + + def test_ltx_image_url_included(self): + args = self._build("ltx", "slow zoom", 5, "landscape", None, "https://cdn.fal.ai/img.png") + assert args["image_url"] == "https://cdn.fal.ai/img.png" + + +# --------------------------------------------------------------------------- +# Requirements check +# --------------------------------------------------------------------------- + +class TestCheckRequirements: + def test_returns_false_without_fal_key(self): + from tools.video_generation_tool import check_video_generation_requirements + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("FAL_KEY", None) + assert check_video_generation_requirements() is False + + def test_returns_true_with_fal_key(self): + from tools.video_generation_tool import check_video_generation_requirements + with patch.dict(os.environ, {"FAL_KEY": "test-key"}): + assert check_video_generation_requirements() is True + + def test_returns_false_when_fal_client_missing(self): + from tools.video_generation_tool import check_video_generation_requirements + with patch.dict(os.environ, {"FAL_KEY": "test-key"}): + with patch.dict(sys.modules, {"fal_client": None}): + assert check_video_generation_requirements() is False + + +# --------------------------------------------------------------------------- +# video_generate_tool — validation paths (no network) +# --------------------------------------------------------------------------- + +class TestVideoGenerateToolValidation: + def _call(self, **kwargs): + from tools.video_generation_tool import video_generate_tool + return json.loads(video_generate_tool(**kwargs)) + + def test_empty_prompt_fails(self): + with patch.dict(os.environ, {"FAL_KEY": "key"}): + result = self._call(prompt="") + assert result["success"] is False + assert result["video_path"] is None + + def test_whitespace_prompt_fails(self): + with patch.dict(os.environ, {"FAL_KEY": "key"}): + result = self._call(prompt=" ") + assert result["success"] is False + + def test_missing_fal_key_fails(self): + env = {k: v for k, v in os.environ.items() if k != "FAL_KEY"} + with patch.dict(os.environ, env, clear=True): + result = self._call(prompt="a scene") + assert result["success"] is False + + def test_invalid_model_fails(self): + with patch.dict(os.environ, {"FAL_KEY": "key"}): + result = self._call(prompt="a scene", model="unknown_model") + assert result["success"] is False + + def test_invalid_duration_defaults_silently(self): + """Invalid duration should fall back to 5 (no crash).""" + fake_fal = _make_fal_mock("https://cdn.fal.ai/video.mp4") + fake_content = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve") as mock_dl: + # Simulate download writing bytes to the temp file + def fake_download(url, path): + with open(path, "wb") as f: + f.write(fake_content) + mock_dl.side_effect = fake_download + + result = self._call(prompt="a scene", duration=999) + assert result["success"] is True + + def test_invalid_aspect_ratio_defaults_silently(self): + fake_fal = _make_fal_mock("https://cdn.fal.ai/video.mp4") + fake_content = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve") as mock_dl: + def fake_download(url, path): + with open(path, "wb") as f: + f.write(fake_content) + mock_dl.side_effect = fake_download + + result = self._call(prompt="a scene", aspect_ratio="widescreen_banana") + assert result["success"] is True + + +# --------------------------------------------------------------------------- +# video_generate_tool — success path (mocked fal_client + download) +# --------------------------------------------------------------------------- + +class TestVideoGenerateToolSuccess: + VIDEO_URL = "https://cdn.fal.ai/generated/video.mp4" + FAKE_BYTES = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + + def _run(self, prompt="a cat walking", model="kling", duration=5, aspect_ratio="landscape"): + from tools.video_generation_tool import video_generate_tool + + fake_fal = _make_fal_mock(self.VIDEO_URL) + + def fake_download(url, path): + with open(path, "wb") as f: + f.write(self.FAKE_BYTES) + + with patch.dict(os.environ, {"FAL_KEY": "test-key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve", side_effect=fake_download): + raw = video_generate_tool( + prompt=prompt, model=model, + duration=duration, aspect_ratio=aspect_ratio, + ) + return json.loads(raw), fake_fal + + def test_returns_success_true(self): + result, _ = self._run() + assert result["success"] is True + + def test_returns_video_path(self): + result, _ = self._run() + assert result["video_path"] is not None + assert result["video_path"].endswith(".mp4") + + def test_returns_video_url(self): + result, _ = self._run() + assert result["video_url"] == self.VIDEO_URL + + def test_returns_media_tag(self): + result, _ = self._run() + assert "media_tag" in result + assert result["media_tag"].startswith("MEDIA:") + assert result["video_path"] in result["media_tag"] + + def test_returns_model_name(self): + result, _ = self._run(model="luma") + assert result["model"] == "luma" + + def test_returns_duration(self): + result, _ = self._run(duration=10) + assert result["duration_seconds"] == 10 + + def test_fal_submit_called_with_kling_model_id(self): + _, fake_fal = self._run(model="kling") + call_args = fake_fal.submit.call_args + assert call_args[0][0] == "fal-ai/kling-video/v1/standard/text-to-video" + + def test_fal_submit_called_with_luma_model_id(self): + _, fake_fal = self._run(model="luma") + call_args = fake_fal.submit.call_args + assert call_args[0][0] == "fal-ai/luma-dream-machine" + + def test_fal_submit_called_with_minimax_model_id(self): + _, fake_fal = self._run(model="minimax") + call_args = fake_fal.submit.call_args + assert call_args[0][0] == "fal-ai/minimax/video-01-live" + + def test_video_path_is_local_file(self): + result, _ = self._run() + # File was created during the mocked download + assert os.path.exists(result["video_path"]) + # Clean up + os.unlink(result["video_path"]) + + def test_model_case_insensitive(self): + from tools.video_generation_tool import video_generate_tool + + fake_fal = _make_fal_mock(self.VIDEO_URL) + + def fake_download(url, path): + with open(path, "wb") as f: + f.write(self.FAKE_BYTES) + + with patch.dict(os.environ, {"FAL_KEY": "test-key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve", side_effect=fake_download): + raw = video_generate_tool(prompt="test", model="KLING") + result = json.loads(raw) + assert result["success"] is True + if result["video_path"]: + os.unlink(result["video_path"]) + + +# --------------------------------------------------------------------------- +# video_generate_tool — failure paths +# --------------------------------------------------------------------------- + +class TestVideoGenerateToolFailures: + def test_fal_api_raises_exception(self): + from tools.video_generation_tool import video_generate_tool + + mock_fal = MagicMock() + mock_fal.submit.side_effect = RuntimeError("API unavailable") + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", mock_fal): + result = json.loads(video_generate_tool(prompt="a scene")) + + assert result["success"] is False + assert result["video_path"] is None + + def test_no_video_url_in_response(self): + from tools.video_generation_tool import video_generate_tool + + mock_handler = MagicMock() + mock_handler.get.return_value = {"images": []} # wrong key + mock_fal = MagicMock() + mock_fal.submit.return_value = mock_handler + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", mock_fal): + result = json.loads(video_generate_tool(prompt="a scene")) + + assert result["success"] is False + + def test_download_failure_handled(self): + from tools.video_generation_tool import video_generate_tool + + mock_handler = MagicMock() + mock_handler.get.return_value = {"video": {"url": "https://cdn.fal.ai/v.mp4"}} + mock_fal = MagicMock() + mock_fal.submit.return_value = mock_handler + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", mock_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve", + side_effect=OSError("network error")): + result = json.loads(video_generate_tool(prompt="a scene")) + + assert result["success"] is False + + +# --------------------------------------------------------------------------- +# Registry registration +# --------------------------------------------------------------------------- + +class TestRegistryRegistration: + def test_video_generate_registered(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 — ensure module loaded + + entry = registry._tools.get("video_generate") + assert entry is not None + + def test_toolset_is_video_gen(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + assert registry._tools["video_generate"].toolset == "video_gen" + + def test_requires_fal_key(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + assert "FAL_KEY" in registry._tools["video_generate"].requires_env + + def test_is_sync(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + assert registry._tools["video_generate"].is_async is False + + def test_schema_has_required_prompt(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + schema = registry._tools["video_generate"].schema + assert "prompt" in schema["parameters"]["required"] + + def test_schema_model_enum(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + schema = registry._tools["video_generate"].schema + model_enum = schema["parameters"]["properties"]["model"]["enum"] + assert set(model_enum) == {"kling", "luma", "minimax", "hunyuan", "veo2", "ltx"} + + def test_schema_has_image_url_param(self): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 + schema = registry._tools["video_generate"].schema + assert "image_url" in schema["parameters"]["properties"] + + +# --------------------------------------------------------------------------- +# _handle_video_generate — registry handler argument mapping +# --------------------------------------------------------------------------- + +class TestHandleVideoGenerate: + """_handle_video_generate maps args dict → video_generate_tool kwargs.""" + + FAKE_BYTES = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + VIDEO_URL = "https://cdn.fal.ai/v.mp4" + + def _dispatch(self, args: dict): + from tools.video_generation_tool import _handle_video_generate + + fake_fal = _make_fal_mock(self.VIDEO_URL) + + def fake_download(url, path): + with open(path, "wb") as f: + f.write(self.FAKE_BYTES) + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve", + side_effect=fake_download): + return json.loads(_handle_video_generate(args)), fake_fal + + def test_missing_prompt_returns_error(self): + from tools.video_generation_tool import _handle_video_generate + result = json.loads(_handle_video_generate({})) + assert "error" in result + + def test_empty_prompt_returns_error(self): + from tools.video_generation_tool import _handle_video_generate + result = json.loads(_handle_video_generate({"prompt": ""})) + assert "error" in result + + def test_prompt_forwarded(self): + result, fake_fal = self._dispatch({"prompt": "a volcano erupting"}) + assert result["success"] is True + call_kwargs = fake_fal.submit.call_args[1]["arguments"] + assert call_kwargs["prompt"] == "a volcano erupting" + + def test_model_forwarded(self): + result, fake_fal = self._dispatch({"prompt": "ocean", "model": "luma"}) + assert result["model"] == "luma" + + def test_duration_forwarded_as_int(self): + """Handler must cast duration to int (model may pass it as string).""" + result, fake_fal = self._dispatch({"prompt": "sky", "duration": "10"}) + assert result["success"] is True + # Kling receives duration as string "10" in API args + call_kwargs = fake_fal.submit.call_args[1]["arguments"] + assert call_kwargs["duration"] == "10" + + def test_aspect_ratio_forwarded(self): + result, fake_fal = self._dispatch({"prompt": "forest", "aspect_ratio": "portrait"}) + assert result["success"] is True + call_kwargs = fake_fal.submit.call_args[1]["arguments"] + assert call_kwargs["aspect_ratio"] == "9:16" + + def test_negative_prompt_forwarded(self): + result, fake_fal = self._dispatch({ + "prompt": "a city", "negative_prompt": "blurry, shaky" + }) + assert result["success"] is True + call_kwargs = fake_fal.submit.call_args[1]["arguments"] + assert call_kwargs["negative_prompt"] == "blurry, shaky" + + def test_defaults_applied_when_optional_args_absent(self): + result, fake_fal = self._dispatch({"prompt": "clouds"}) + assert result["success"] is True + assert result["model"] == "kling" + assert result["duration_seconds"] == 5 + + def test_image_url_forwarded(self): + img_url = "https://cdn.fal.ai/image.png" + result, fake_fal = self._dispatch({"prompt": "make it move", "image_url": img_url}) + assert result["success"] is True + call_kwargs = fake_fal.submit.call_args[1]["arguments"] + assert call_kwargs["image_url"] == img_url + + def cleanup_paths(self, result): + path = result.get("video_path") + if path and os.path.exists(path): + os.unlink(path) + + +# --------------------------------------------------------------------------- +# _download_video — extension / suffix logic +# --------------------------------------------------------------------------- + +class TestDownloadVideoExtension: + """_download_video picks the correct temp file suffix from the URL.""" + + FAKE_BYTES = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + + def _download(self, url: str) -> str: + from tools.video_generation_tool import _download_video + with patch("tools.video_generation_tool.urllib.request.urlretrieve") as mock_dl: + def fake_retrieve(u, path): + with open(path, "wb") as f: + f.write(self.FAKE_BYTES) + mock_dl.side_effect = fake_retrieve + return _download_video(url) + + def test_mp4_url_gives_mp4_suffix(self): + path = self._download("https://cdn.fal.ai/output/video.mp4") + assert path.endswith(".mp4") + os.unlink(path) + + def test_webm_url_gives_webm_suffix(self): + path = self._download("https://cdn.fal.ai/output/video.webm") + assert path.endswith(".webm") + os.unlink(path) + + def test_mov_url_gives_mov_suffix(self): + path = self._download("https://cdn.fal.ai/output/clip.mov") + assert path.endswith(".mov") + os.unlink(path) + + def test_avi_url_gives_avi_suffix(self): + path = self._download("https://cdn.fal.ai/output/clip.avi") + assert path.endswith(".avi") + os.unlink(path) + + def test_unknown_extension_defaults_to_mp4(self): + path = self._download("https://cdn.fal.ai/output/video.ts") + assert path.endswith(".mp4") + os.unlink(path) + + def test_no_extension_defaults_to_mp4(self): + path = self._download("https://cdn.fal.ai/output/abc123") + assert path.endswith(".mp4") + os.unlink(path) + + def test_query_params_ignored(self): + """URL with ?token=... should still parse the extension correctly.""" + path = self._download("https://cdn.fal.ai/output/video.mp4?token=xyz&expires=99") + assert path.endswith(".mp4") + os.unlink(path) + + def test_file_has_hermes_prefix(self): + path = self._download("https://cdn.fal.ai/output/video.mp4") + assert os.path.basename(path).startswith("hermes_video_") + os.unlink(path) + + +# --------------------------------------------------------------------------- +# Integration — registry.dispatch end-to-end +# --------------------------------------------------------------------------- + +class TestRegistryDispatchIntegration: + """Full pipeline: registry.dispatch('video_generate', args) → JSON result.""" + + FAKE_BYTES = b"\x00\x00\x00\x18ftypmp42" + b"\x00" * 11_000 # > 10 KB size guard + VIDEO_URL = "https://cdn.fal.ai/integration/video.mp4" + + def _dispatch(self, args: dict): + from tools.registry import registry + import tools.video_generation_tool # noqa: F401 — ensure registered + + fake_fal = _make_fal_mock(self.VIDEO_URL) + + def fake_download(url, path): + with open(path, "wb") as f: + f.write(self.FAKE_BYTES) + + with patch.dict(os.environ, {"FAL_KEY": "key"}): + with patch("tools.video_generation_tool.fal_client", fake_fal): + with patch("tools.video_generation_tool.urllib.request.urlretrieve", + side_effect=fake_download): + raw = registry.dispatch("video_generate", args) + return json.loads(raw) + + def test_dispatch_returns_success(self): + result = self._dispatch({"prompt": "a waterfall at dawn"}) + assert result["success"] is True + + def test_dispatch_returns_video_path(self): + result = self._dispatch({"prompt": "a waterfall at dawn"}) + assert result["video_path"] is not None + assert os.path.exists(result["video_path"]) + os.unlink(result["video_path"]) + + def test_dispatch_unknown_tool_returns_error(self): + from tools.registry import registry + result = json.loads(registry.dispatch("nonexistent_tool", {})) + assert "error" in result + + def test_dispatch_missing_prompt_returns_error(self): + result = self._dispatch({}) + assert "error" in result + + def test_dispatch_all_params_forwarded(self): + result = self._dispatch({ + "prompt": "a volcano", + "model": "kling", + "duration": 10, + "aspect_ratio": "portrait", + }) + assert result["success"] is True + assert result["model"] == "kling" + assert result["duration_seconds"] == 10 + if result.get("video_path"): + os.unlink(result["video_path"]) diff --git a/tools/__init__.py b/tools/__init__.py index 9b2542296913..3214b979e514 100644 --- a/tools/__init__.py +++ b/tools/__init__.py @@ -1,262 +1,25 @@ #!/usr/bin/env python3 -""" -Tools Package - -This package contains all the specific tool implementations for the Hermes Agent. -Each module provides specialized functionality for different capabilities: - -- web_tools: Web search, content extraction, and crawling -- terminal_tool: Command execution (local/docker/modal/daytona/ssh/singularity backends) -- vision_tools: Image analysis and understanding -- mixture_of_agents_tool: Multi-model collaborative reasoning -- image_generation_tool: Text-to-image generation with upscaling - -The tools are imported into model_tools.py which provides a unified interface -for the AI agent to access all capabilities. -""" - -# Export all tools for easy importing -from .web_tools import ( - web_search_tool, - web_extract_tool, - web_crawl_tool, - check_firecrawl_api_key -) - -# Primary terminal tool (local/docker/singularity/modal/daytona/ssh) -from .terminal_tool import ( - terminal_tool, - check_terminal_requirements, - cleanup_vm, - cleanup_all_environments, - get_active_environments_info, - register_task_env_overrides, - clear_task_env_overrides, - TERMINAL_TOOL_DESCRIPTION -) - -from .vision_tools import ( - vision_analyze_tool, - check_vision_requirements -) - -from .mixture_of_agents_tool import ( - mixture_of_agents_tool, - check_moa_requirements -) +"""Tools package namespace. -from .image_generation_tool import ( - image_generate_tool, - check_image_generation_requirements -) +Keep package import side effects minimal. Importing ``tools`` should not +eagerly import the full tool stack, because several subsystems load tools while +``hermes_cli.config`` is still initializing. -from .skills_tool import ( - skills_list, - skill_view, - check_skills_requirements, - SKILLS_TOOL_DESCRIPTION -) +Callers should import concrete submodules directly, for example: -from .skill_manager_tool import ( - skill_manage, - check_skill_manage_requirements, - SKILL_MANAGE_SCHEMA -) + import tools.web_tools + from tools import browser_tool -# Browser automation tools (agent-browser + Browserbase) -from .browser_tool import ( - browser_navigate, - browser_snapshot, - browser_click, - browser_type, - browser_scroll, - browser_back, - browser_press, - browser_close, - browser_get_images, - browser_vision, - cleanup_browser, - cleanup_all_browsers, - get_active_browser_sessions, - check_browser_requirements, - BROWSER_TOOL_SCHEMAS -) - -# Cronjob management tools (CLI-only, hermes-cli toolset) -from .cronjob_tools import ( - cronjob, - schedule_cronjob, - list_cronjobs, - remove_cronjob, - check_cronjob_requirements, - get_cronjob_tool_definitions, - CRONJOB_SCHEMA, -) - -# RL Training tools (Tinker-Atropos) -from .rl_training_tool import ( - rl_list_environments, - rl_select_environment, - rl_get_current_config, - rl_edit_config, - rl_start_training, - rl_check_status, - rl_stop_training, - rl_get_results, - rl_list_runs, - rl_test_inference, - check_rl_api_keys, - get_missing_keys, -) - -# File manipulation tools (read, write, patch, search) -from .file_tools import ( - read_file_tool, - write_file_tool, - patch_tool, - search_tool, - get_file_tools, - clear_file_ops_cache, -) - -# Text-to-speech tools (Edge TTS / ElevenLabs / OpenAI) -from .tts_tool import ( - text_to_speech_tool, - check_tts_requirements, -) - -# Planning & task management tool -from .todo_tool import ( - todo_tool, - check_todo_requirements, - TODO_SCHEMA, - TodoStore, -) - -# Clarifying questions tool (interactive Q&A with the user) -from .clarify_tool import ( - clarify_tool, - check_clarify_requirements, - CLARIFY_SCHEMA, -) - -# Code execution sandbox (programmatic tool calling) -from .code_execution_tool import ( - execute_code, - check_sandbox_requirements, - EXECUTE_CODE_SCHEMA, -) +Python will resolve those submodules via the package path without needing them +to be re-exported here. +""" -# Subagent delegation (spawn child agents with isolated context) -from .delegate_tool import ( - delegate_task, - check_delegate_requirements, - DELEGATE_TASK_SCHEMA, -) -# File tools have no external requirements - they use the terminal backend def check_file_requirements(): - """File tools only require terminal backend to be available.""" + """File tools only require terminal backend availability.""" from .terminal_tool import check_terminal_requirements + return check_terminal_requirements() -__all__ = [ - # Web tools - 'web_search_tool', - 'web_extract_tool', - 'web_crawl_tool', - 'check_firecrawl_api_key', - # Terminal tools - 'terminal_tool', - 'check_terminal_requirements', - 'cleanup_vm', - 'cleanup_all_environments', - 'get_active_environments_info', - 'register_task_env_overrides', - 'clear_task_env_overrides', - 'TERMINAL_TOOL_DESCRIPTION', - # Vision tools - 'vision_analyze_tool', - 'check_vision_requirements', - # MoA tools - 'mixture_of_agents_tool', - 'check_moa_requirements', - # Image generation tools - 'image_generate_tool', - 'check_image_generation_requirements', - # Skills tools - 'skills_list', - 'skill_view', - 'check_skills_requirements', - 'SKILLS_TOOL_DESCRIPTION', - # Skill management - 'skill_manage', - 'check_skill_manage_requirements', - 'SKILL_MANAGE_SCHEMA', - # Browser automation tools - 'browser_navigate', - 'browser_snapshot', - 'browser_click', - 'browser_type', - 'browser_scroll', - 'browser_back', - 'browser_press', - 'browser_close', - 'browser_get_images', - 'browser_vision', - 'cleanup_browser', - 'cleanup_all_browsers', - 'get_active_browser_sessions', - 'check_browser_requirements', - 'BROWSER_TOOL_SCHEMAS', - # Cronjob management tools (CLI-only) - 'cronjob', - 'schedule_cronjob', - 'list_cronjobs', - 'remove_cronjob', - 'check_cronjob_requirements', - 'get_cronjob_tool_definitions', - 'CRONJOB_SCHEMA', - # RL Training tools - 'rl_list_environments', - 'rl_select_environment', - 'rl_get_current_config', - 'rl_edit_config', - 'rl_start_training', - 'rl_check_status', - 'rl_stop_training', - 'rl_get_results', - 'rl_list_runs', - 'rl_test_inference', - 'check_rl_api_keys', - 'get_missing_keys', - # File manipulation tools - 'read_file_tool', - 'write_file_tool', - 'patch_tool', - 'search_tool', - 'get_file_tools', - 'clear_file_ops_cache', - 'check_file_requirements', - # Text-to-speech tools - 'text_to_speech_tool', - 'check_tts_requirements', - # Planning & task management tool - 'todo_tool', - 'check_todo_requirements', - 'TODO_SCHEMA', - 'TodoStore', - # Clarifying questions tool - 'clarify_tool', - 'check_clarify_requirements', - 'CLARIFY_SCHEMA', - # Code execution sandbox - 'execute_code', - 'check_sandbox_requirements', - 'EXECUTE_CODE_SCHEMA', - # Subagent delegation - 'delegate_task', - 'check_delegate_requirements', - 'DELEGATE_TASK_SCHEMA', -] +__all__ = ["check_file_requirements"] diff --git a/tools/video_generation_tool.py b/tools/video_generation_tool.py new file mode 100644 index 000000000000..2a7137c86129 --- /dev/null +++ b/tools/video_generation_tool.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +""" +Video Generation Tools Module + +Generates short MP4 videos from text prompts (or images) using FAL.ai video models. +Follows the same pattern as image_generation_tool.py. + +Available tools: +- video_generate_tool: Generate video from a text prompt or image URL + +Supported models: +- kling → fal-ai/kling-video/v1.5/pro (default, text+image-to-video) +- luma → fal-ai/luma-dream-machine (text+image-to-video) +- minimax → fal-ai/minimax/video-01-live (text+image-to-video) +- hunyuan → fal-ai/hunyuan-video (text-to-video, high quality) +- veo2 → fal-ai/veo2 (text-to-video, up to 4K) +- ltx → fal-ai/ltx-video-v095/multiconditioning (text+image-to-video) + +Output: local MP4 file path (downloaded from FAL.ai CDN). + +Usage: + from tools.video_generation_tool import video_generate_tool + + # Text-to-video + result = video_generate_tool( + prompt="A cat walking through a neon-lit Tokyo street at night", + model="kling", + duration=5, + aspect_ratio="landscape", + ) + + # Image-to-video + result = video_generate_tool( + prompt="Make it move, gentle waves", + model="kling", + image_url="https://cdn.fal.ai/some-image.png", + ) +""" + +import json +import logging +import os +import tempfile +import datetime +import urllib.request +from typing import Optional + +import fal_client +from tools.debug_helpers import DebugSession + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Model registry +# --------------------------------------------------------------------------- + +MODEL_MAP = { + "kling": "fal-ai/kling-video/v1/standard/text-to-video", + "luma": "fal-ai/luma-dream-machine", + "minimax": "fal-ai/minimax/video-01-live", + "hunyuan": "fal-ai/hunyuan-video", + "veo2": "fal-ai/veo2", + "ltx": "fal-ai/ltx-video-v095/multiconditioning", +} + +# Models that support image_url as input +IMAGE_TO_VIDEO_MODELS = {"kling", "luma", "minimax", "ltx"} + +DEFAULT_MODEL_KEY = "kling" + +# --------------------------------------------------------------------------- +# Parameter constants +# --------------------------------------------------------------------------- + +# Aspect ratio display → API value mappings per model family +_ASPECT_MAP = { + "landscape": "16:9", + "portrait": "9:16", + "square": "1:1", +} + +VALID_DURATIONS = [5, 10] # seconds; Kling supports both, others default to 5 +VALID_ASPECT_RATIOS = ["landscape", "portrait", "square"] +DEFAULT_ASPECT_RATIO = "landscape" +DEFAULT_DURATION = 5 + +# --------------------------------------------------------------------------- +# Debug session +# --------------------------------------------------------------------------- + +_debug = DebugSession("video_tools", env_var="VIDEO_TOOLS_DEBUG") + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _build_arguments(model_key: str, prompt: str, duration: int, aspect_ratio: str, + negative_prompt: Optional[str], image_url: Optional[str]) -> dict: + """Build the FAL.ai arguments dict for the given model.""" + args: dict = {"prompt": prompt.strip()} + + if model_key == "kling": + args["duration"] = str(duration) # Kling expects string "5" / "10" + args["aspect_ratio"] = _ASPECT_MAP[aspect_ratio] + if negative_prompt: + args["negative_prompt"] = negative_prompt + if image_url: + args["image_url"] = image_url + + elif model_key == "luma": + args["aspect_ratio"] = _ASPECT_MAP[aspect_ratio] + args["loop"] = False + # Luma Dream Machine root endpoint does not accept a duration parameter + if image_url: + args["image_url"] = image_url + + elif model_key == "minimax": + if image_url: + args["image_url"] = image_url + + elif model_key == "hunyuan": + # hunyuan only supports 16:9 and 9:16 (no 1:1), text-to-video only + ar = _ASPECT_MAP[aspect_ratio] if aspect_ratio != "square" else "16:9" + args["aspect_ratio"] = ar + + elif model_key == "veo2": + args["aspect_ratio"] = _ASPECT_MAP[aspect_ratio] + veo2_dur = min(duration, 8) # veo2 max is 8s, valid: "5s"–"8s" + args["duration"] = f"{veo2_dur}s" + # veo2 is text-to-video only + + elif model_key == "ltx": + args["aspect_ratio"] = _ASPECT_MAP[aspect_ratio] + if image_url: + args["image_url"] = image_url + + return args + + +def _extract_video_url(result: dict) -> Optional[str]: + """ + Extract video URL from FAL.ai response. + Models return either {"video": {"url": ...}} or {"videos": [{"url": ...}]}. + """ + if not result: + return None + + # Single video object + if "video" in result: + v = result["video"] + return v.get("url") if isinstance(v, dict) else None + + # List of videos + if "videos" in result: + videos = result["videos"] + if videos and isinstance(videos[0], dict): + return videos[0].get("url") + + return None + + +def _download_video(url: str) -> str: + """Download video from URL to a temp file. Returns local file path.""" + suffix = ".mp4" + # Preserve extension if present in URL + url_path = url.split("?")[0] + if "." in url_path.split("/")[-1]: + ext = "." + url_path.split("/")[-1].rsplit(".", 1)[-1] + if ext.lower() in (".mp4", ".mov", ".webm", ".avi"): + suffix = ext + + tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False, prefix="hermes_video_") + tmp_path = tmp.name + tmp.close() + + logger.info("Downloading video from FAL.ai CDN → %s", tmp_path) + urllib.request.urlretrieve(url, tmp_path) + size_bytes = os.path.getsize(tmp_path) + size_mb = size_bytes / (1024 * 1024) + logger.info("Downloaded %.1f MB (%d bytes)", size_mb, size_bytes) + if size_bytes < 10_000: # < 10 KB is clearly not a real video + os.unlink(tmp_path) + raise ValueError( + f"Downloaded file is too small ({size_bytes} bytes) — " + "CDN placeholder or generation failed. URL: " + url + ) + return tmp_path + + +# --------------------------------------------------------------------------- +# Public tool function +# --------------------------------------------------------------------------- + +def video_generate_tool( + prompt: str, + model: str = DEFAULT_MODEL_KEY, + duration: int = DEFAULT_DURATION, + aspect_ratio: str = DEFAULT_ASPECT_RATIO, + negative_prompt: Optional[str] = None, + image_url: Optional[str] = None, +) -> str: + """ + Generate a short video from a text prompt (or image) using FAL.ai. + + Uses fal_client.submit() (sync) to avoid event-loop lifecycle issues + in the gateway thread-pool pattern — same reason as image_generation_tool. + + Args: + prompt: Text description / motion description for the video. + model: "kling" (default), "luma", "minimax", "hunyuan", "veo2", "ltx". + duration: Video length in seconds — 5 (default) or 10. + aspect_ratio: "landscape" (16:9, default), "portrait" (9:16), "square" (1:1). + negative_prompt: Things to avoid in the video (kling only). + image_url: Image URL for image-to-video. Supported by: kling, luma, minimax, ltx. + Pass the URL returned by image_generate to animate a still image. + + Returns: + JSON string: + { + "success": bool, + "video_path": str | null, # local temp file path + "video_url": str | null, # original CDN URL + "media_tag": str | null, # MEDIA: — include in response for delivery + "model": str, + "duration_seconds": int + } + """ + start = datetime.datetime.now() + debug_data: dict = { + "prompt": prompt, + "model": model, + "duration": duration, + "aspect_ratio": aspect_ratio, + "image_url": image_url, + "error": None, + "success": False, + } + + def _fail(msg: str) -> str: + logger.error(msg) + debug_data["error"] = msg + _debug.log_call("video_generate_tool", debug_data) + _debug.save() + return json.dumps({"success": False, "video_path": None, "video_url": None, + "model": model, "duration_seconds": duration}) + + # --- Validation --- + if not prompt or not isinstance(prompt, str) or not prompt.strip(): + return _fail("prompt is required and must be a non-empty string") + + if not os.getenv("FAL_KEY"): + return _fail("FAL_KEY environment variable not set") + + model_key = model.lower().strip() + if model_key not in MODEL_MAP: + return _fail(f"Unknown model '{model}'. Valid choices: {list(MODEL_MAP.keys())}") + + if duration not in VALID_DURATIONS: + logger.warning("Invalid duration %s, defaulting to %s", duration, DEFAULT_DURATION) + duration = DEFAULT_DURATION + + aspect_key = aspect_ratio.lower().strip() + if aspect_key not in VALID_ASPECT_RATIOS: + logger.warning("Invalid aspect_ratio '%s', defaulting to '%s'", aspect_ratio, DEFAULT_ASPECT_RATIO) + aspect_key = DEFAULT_ASPECT_RATIO + + fal_model_id = MODEL_MAP[model_key] + arguments = _build_arguments(model_key, prompt, duration, aspect_key, negative_prompt, image_url) + + logger.info("Generating video | model=%s | duration=%ss | aspect=%s", model_key, duration, aspect_key) + logger.info("Prompt: %s", prompt[:120]) + logger.info("FAL model ID: %s", fal_model_id) + + try: + # Sync fal_client — same pattern as image_generation_tool.py. + # submit_async() caches a global httpx.AsyncClient via @cached_property + # which breaks when asyncio.run() destroys the event loop between calls + # (gateway thread-pool pattern). submit() uses httpx.Client (no loop). + handler = fal_client.submit(fal_model_id, arguments=arguments) + result = handler.get() + + elapsed = (datetime.datetime.now() - start).total_seconds() + logger.info("FAL.ai responded in %.1fs", elapsed) + + video_url = _extract_video_url(result) + if not video_url: + return _fail(f"FAL.ai returned no video URL. Raw response keys: {list(result.keys()) if result else 'empty'}") + + # Download to local temp file so send_message_tool can attach it directly + video_path = _download_video(video_url) + + total_elapsed = (datetime.datetime.now() - start).total_seconds() + logger.info("Video ready in %.1fs total | path=%s", total_elapsed, video_path) + + debug_data.update({"success": True, "video_url": video_url, "video_path": video_path}) + _debug.log_call("video_generate_tool", debug_data) + _debug.save() + + return json.dumps({ + "success": True, + "video_path": video_path, + "video_url": video_url, + "model": model_key, + "duration_seconds": duration, + "media_tag": f"MEDIA:{video_path}", + }, indent=2) + + except Exception as exc: + return _fail(f"Error generating video: {exc}") + + +# --------------------------------------------------------------------------- +# Requirements check +# --------------------------------------------------------------------------- + +def check_video_generation_requirements() -> bool: + """Return True if FAL_KEY is set and fal_client is importable.""" + try: + if not os.getenv("FAL_KEY"): + return False + import fal_client # noqa: F401 + return True + except ImportError: + return False + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +from tools.registry import registry # noqa: E402 + +VIDEO_GENERATE_SCHEMA = { + "name": "video_generate", + "description": ( + "Generate a short MP4 video from a text prompt or image using FAL.ai video models. " + "Pass image_url (from image_generate) to animate a still image. " + "Returns a media_tag (MEDIA:) — include it in your response to deliver the video " + "as a native video message (Telegram inline playback, Discord/Slack attachment). " + "Generation takes 30–120 seconds — warn the user before starting. " + "Models: 'kling' (default, image+text), 'luma' (cinematic, image+text), " + "'minimax' (image+text), 'hunyuan' (high quality, text only), " + "'veo2' (4K, text only), 'ltx' (fast, image+text)." + ), + "parameters": { + "type": "object", + "properties": { + "prompt": { + "type": "string", + "description": "Detailed text description of the video to generate.", + }, + "model": { + "type": "string", + "enum": ["kling", "luma", "minimax", "hunyuan", "veo2", "ltx"], + "description": "Video generation model. Default: 'kling'. For image-to-video use kling/luma/minimax/ltx.", + "default": "kling", + }, + "duration": { + "type": "integer", + "enum": [5, 10], + "description": "Video length in seconds. 10s only supported by kling. Default: 5.", + "default": 5, + }, + "aspect_ratio": { + "type": "string", + "enum": ["landscape", "portrait", "square"], + "description": "Video aspect ratio — landscape (16:9), portrait (9:16), square (1:1). Default: landscape.", + "default": "landscape", + }, + "negative_prompt": { + "type": "string", + "description": "Things to avoid in the video (supported by kling only). Optional.", + }, + "image_url": { + "type": "string", + "description": "Image URL to animate (image-to-video). Supported by kling, luma, minimax, ltx. Pass the URL returned by image_generate. Optional.", + }, + }, + "required": ["prompt"], + }, +} + + +def _handle_video_generate(args, **kw): + prompt = args.get("prompt", "") + if not prompt: + return json.dumps({"error": "prompt is required for video generation"}) + return video_generate_tool( + prompt=prompt, + model=args.get("model", DEFAULT_MODEL_KEY), + duration=int(args.get("duration", DEFAULT_DURATION)), + aspect_ratio=args.get("aspect_ratio", DEFAULT_ASPECT_RATIO), + negative_prompt=args.get("negative_prompt"), + image_url=args.get("image_url"), + ) + + +registry.register( + name="video_generate", + toolset="video_gen", + schema=VIDEO_GENERATE_SCHEMA, + handler=_handle_video_generate, + check_fn=check_video_generation_requirements, + requires_env=["FAL_KEY"], + is_async=False, + emoji="🎬", +) diff --git a/toolsets.py b/toolsets.py index a314f277b738..ea073c3f9923 100644 --- a/toolsets.py +++ b/toolsets.py @@ -35,8 +35,8 @@ "terminal", "process", # File manipulation "read_file", "write_file", "patch", "search_files", - # Vision + image generation - "vision_analyze", "image_generate", + # Vision + image + video generation + "vision_analyze", "image_generate", "video_generate", # MoA "mixture_of_agents", # Skills @@ -94,7 +94,13 @@ "tools": ["image_generate"], "includes": [] }, - + + "video_gen": { + "description": "Video generation tools (text-to-video and image-to-video via FAL.ai)", + "tools": ["video_generate"], + "includes": [] + }, + "terminal": { "description": "Terminal/command execution and process management tools", "tools": ["terminal", "process"],