From 287c77fa641416537b0db895f143ef3d0da19faa Mon Sep 17 00:00:00 2001 From: clarkh Date: Tue, 28 Jul 2026 16:48:06 +0800 Subject: [PATCH 1/4] feat: add Tencent Hunyuan cloud video provider via TokenHub API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a new video generation provider backed by the Tencent TokenHub API (tokenhub.tencentmaas.com), an OpenAI-compatible gateway for Tencent Hunyuan video models with simple Bearer-token auth. - Add hunyuan_cloud_video tool (submit → poll → download) supporting both text-to-video (hy-video-1.5) and image-to-video (yt-video-2.0) - Add env vars: TENCENT_TOKENHUB_API_KEY, TENCENT_TOKENHUB_MODEL - Add contract tests for the new tool - Document setup, API flow, model pricing, and schema constraints in PROVIDERS.md - Update provider tables and capability matrix throughout docs --- .env.example | 5 +- docs/PROVIDERS.md | 89 ++- tests/contracts/test_hunyuan_cloud_video.py | 820 ++++++++++++++++++++ tools/video/hunyuan_cloud_video.py | 523 +++++++++++++ 4 files changed, 1431 insertions(+), 6 deletions(-) create mode 100644 tests/contracts/test_hunyuan_cloud_video.py create mode 100644 tools/video/hunyuan_cloud_video.py diff --git a/.env.example b/.env.example index d60f8987..70266e18 100644 --- a/.env.example +++ b/.env.example @@ -51,7 +51,10 @@ SUNO_API_KEY= # Suno AI music generation (full songs, instrumenta HEYGEN_API_KEY= # HeyGen API (VEO, Sora, Runway, Kling, Seedance via single key) RUNWAY_API_KEY= # Runway Gen-4 (direct API, alternative to fal.ai routing) VOLC_ACCESSKEY= # Volcengine Jimeng (即梦 AI) video generation via official API (HMAC-SHA256 V4 signing) -VOLC_SECRETKEY= # Secret Access Key paired with VOLC_ACCESSKEY. Get both at https://console.volcengine.com/iam/keymanage +VOLC_SECRETKEY= # Secret Access Key paired with VOLC_ACCESSKEY. Get both at https://console.volcengine.com/iam/keymanage +TENCENT_TOKENHUB_API_KEY= # Tencent Hunyuan (腾讯混元) video generation via TokenHub API (Bearer token) + # Get it at https://console.cloud.tencent.com/tokenhub +# TENCENT_TOKENHUB_MODEL= # Optional: override default TokenHub model (hy-video-1.5 / yt-video-2.0) VIDEO_GEN_LOCAL_ENABLED= # Set to "true" for local video gen (needs GPU + diffusers) VIDEO_GEN_LOCAL_MODEL= # Local model: wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b MODAL_LTX2_ENDPOINT_URL= # Modal self-hosted LTX-2 endpoint (optional) diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index 7ad51bad..a9b86874 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -19,10 +19,11 @@ Everything you need to know about every provider in OpenMontage — setup instru | 7 | **~$0.04/image** | Google Imagen | Imagen 4 images (shares the Google API key) | | 8 | **pay-as-you-go** | Kling Official | Official direct Kling video, image, TTS, avatar, and lip-sync API, separate from fal.ai Kling | | 9 | **$12/month** | Runway | Gen-4 video — highest quality AI video | -| 10 | **pay-as-you-go** | HeyGen | Avatar videos, multi-model video gateway | -| 11 | **pay-as-you-go** | Suno | Full song generation with vocals and lyrics | -| 12 | **$0 + GPU** | Local video gen | WAN 2.1, Hunyuan, CogVideo, LTX — free, offline | -| 13 | **$0 + GPU** | Local Diffusion | Stable Diffusion images — free, offline | +| 10 | **pay-as-you-go** | Hunyuan cloud video | Chinese-friendly T2V + I2V | +| 11 | **pay-as-you-go** | HeyGen | Avatar videos, multi-model video gateway | +| 12 | **pay-as-you-go** | Suno | Full song generation with vocals and lyrics | +| 13 | **$0 + GPU** | Local video gen | WAN 2.1, Hunyuan, CogVideo, LTX — free, offline | +| 14 | **$0 + GPU** | Local Diffusion | Stable Diffusion images — free, offline | ### Environment Variable Summary @@ -60,6 +61,9 @@ HEYGEN_API_KEY= # HeyGen avatar video gateway RUNWAY_API_KEY= # Runway Gen-4 video (direct) SUNO_API_KEY= # Suno music generation +# TENCLOUD HUNYUAN VIDEO +TENCENT_TOKENHUB_API_KEY= # Tencent Hunyuan cloud video via TokenHub API + # LOCAL (no keys needed — just GPU + install) VIDEO_GEN_LOCAL_ENABLED= # Set to "true" for local video gen VIDEO_GEN_LOCAL_MODEL= # wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b @@ -343,6 +347,80 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon --- +### Tencent Hunyuan Cloud — Video Generation + +> **Tencent Hunyuan (腾讯混元) cloud video generation via TokenHub API.** Generates +> videos from text or images using Tencent's Hunyuan models through the Tencent +> TokenHub API — an OpenAI-compatible gateway (tokenhub.tencentmaas.com) with +> simple Bearer-token authentication. No TC3-HMAC-SHA256 signing required. + +**Tools unlocked:** `hunyuan_cloud_video` +**Env vars:** `TENCENT_TOKENHUB_API_KEY` (required), `TENCENT_TOKENHUB_MODEL` (optional, overrides the default model) + +#### Setup + +1. Go to the [Tencent Cloud TokenHub console](https://console.cloud.tencent.com/tokenhub). +2. Create an application or navigate to the **API Key** section. +3. Generate an API key and copy its value. +4. Add to `.env`: + ```bash + TENCENT_TOKENHUB_API_KEY=your-tokenhub-api-key + # TENCENT_TOKENHUB_MODEL=hy-video-1.5 # optional model override + ``` + +#### What It's Best For + +- **Chinese-friendly prompt understanding** — Hunyuan models natively understand Chinese prompts better than most Western APIs +- **Simple auth** — Bearer token, no complex signing (just an HTTP Authorization header) +- **Direct Tencent Cloud quota** — uses your own Tencent Cloud credits, not a third-party gateway mark-up +- **Both T2V and I2V** — one API key unlocks text-to-video and image-to-video + +#### API Notes + +TokenHub uses a **submit-then-poll** pattern: + +```text +# Submit a generation task +POST https://tokenhub.tencentmaas.com/v1/api/video/submit +Authorization: Bearer ${TENCENT_TOKENHUB_API_KEY} + +# Poll for results +POST https://tokenhub.tencentmaas.com/v1/api/video/query +Authorization: Bearer ${TENCENT_TOKENHUB_API_KEY} +``` + +| Model | Type | Pricing | +|-------|------|---------| +| `hy-video-1.5` | Text-to-video | 1.5 credits (~$0.25) | +| `yt-video-2.0` | Image-to-video | 2–5 credits (~$0.33–0.83) | + +Resolution options: **720p** (default) or **1080p**. + +A watermark (`logo_add`) is added by default. Set `logo_add: 0` to disable it (requires console approval from Tencent). + +**Schema constraints:** +- **Prompt:** max 200 UTF-8 characters +- **Image:** max 10MB, 50–5000 px per side, aspect ratio 1:4 to 4:1 +- **Formats:** jpg, png, jpeg, webp, bmp, tiff + +#### Fallback Tools + +If `hunyuan_cloud_video` returns an error, the agent may retry with: `jimeng_video`, `kling_official_video`, `minimax_video` + +#### Pricing + +Tencent TokenHub uses a credit-based pricing system (1 credit = 1.2 RMB ≈ $0.167 USD): + +| Model | Resolution | Credits | Estimated USD | +|-------|-----------|---------|---------------| +| HY-Video-1.5 | any | 1.5 | ~$0.25 | +| YT-Video-2.0 | 480p | 2 | ~$0.33 | +| YT-Video-2.0 | 720p / 1080p | 5 | ~$0.83 | + +> **Free tier:** Tencent occasionally offers new-user credits for TokenHub. Check the [TokenHub console](https://console.cloud.tencent.com/tokenhub) for current promotions. + +--- + ### Azure AI Speech — Speech-to-Text > **Cloud transcription.** Azure AI Speech Fast Transcription turns local audio into text with word-level timestamps, speaker diarization, and multi-language identification — no GPU required. Optional: the local faster-whisper `transcriber` remains the default offline STT path. When `AZURE_SPEECH_KEY` is set, the agent prefers `azure_stt` for cloud transcription. @@ -936,6 +1014,7 @@ These tools require only FFmpeg or Python packages — no GPU, no API key. | **Higgsfield** | `HIGGSFIELD_API_KEY` + `HIGGSFIELD_API_SECRET` | `higgsfield_video` | Subscription ($15-84/mo) | | **HeyGen** | `HEYGEN_API_KEY` | `heygen_video` | Pay-as-you-go | | **Suno** | `SUNO_API_KEY` | `suno_music` | Pay-as-you-go | +| **Tencent Hunyuan** | `TENCENT_TOKENHUB_API_KEY` | `hunyuan_cloud_video` | Pay-as-you-go (~$0.25–0.83/gen) | | **Local GPU** | `VIDEO_GEN_LOCAL_ENABLED` | `wan_video`, `hunyuan_video`, `cogvideo_video`, `ltx_video_local` | Free (GPU required) | | **Local Diffusion** | — (install only) | `local_diffusion` | Free (GPU required) | | **Modal** | `MODAL_LTX2_ENDPOINT_URL` | `ltx_video_modal` | Self-hosted cloud | @@ -949,7 +1028,7 @@ How many providers cover each capability: | Capability | Cloud Providers | Local Providers | Free Options | |-----------|----------------|-----------------|--------------| | **Image Generation** | FLUX, Kling Official, Grok, Google Imagen, GPT Image 2, Recraft | Local Diffusion | Pexels, Pixabay (stock) | -| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Runway, Veo, Gemini Omni, Higgsfield, MiniMax, HeyGen | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) | +| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Runway, Veo, Gemini Omni, Higgsfield, MiniMax, HeyGen, Tencent Hunyuan | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) | | **Text-to-Speech** | ElevenLabs, Google TTS, Kling Official, OpenAI | Piper | Piper, Google free tier, ElevenLabs free tier | | **Music Generation** | ElevenLabs, Suno, Google Lyria | — | ElevenLabs free tier | | **Post-Production** | — | FFmpeg (compose, stitch, trim, mix, enhance, grade) | All free | diff --git a/tests/contracts/test_hunyuan_cloud_video.py b/tests/contracts/test_hunyuan_cloud_video.py new file mode 100644 index 00000000..b445d5dd --- /dev/null +++ b/tests/contracts/test_hunyuan_cloud_video.py @@ -0,0 +1,820 @@ +"""Contract tests for the Tencent Hunyuan cloud video provider tool (TokenHub API). + +These tests verify that the tool satisfies the BaseTool contract without +requiring real Tencent Cloud credentials or making any API calls. + +Run: pytest tests/contracts/test_hunyuan_cloud_video.py -v +""" + +from __future__ import annotations + +import json +import sys +import types +from pathlib import Path + +import pytest + +from tools.base_tool import ( + BaseTool, + ExecutionMode, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) +from tools.video.hunyuan_cloud_video import HunyuanCloudVideo + + +# ------------------------------------------------------------------ +# Fake HTTP infrastructure (used by execute-path tests) +# ------------------------------------------------------------------ + +class FakeResponse: + def __init__(self, json_data=None, content=b"", ok=True, status_code=200, headers=None, text=""): + self._json = json_data + self.content = content + self.ok = ok + self.status_code = status_code + self.headers = headers or {} + self.text = text or (json.dumps(json_data) if json_data is not None else "") + + def json(self): + return self._json + + def raise_for_status(self): + if not self.ok: + raise RuntimeError(f"HTTP {self.status_code}") + + +def _install_fake_requests(monkeypatch, post_responses, get_responses): + """Inject a fake requests module; returns the recorded calls.""" + calls = {"post": [], "get": []} + fake = types.ModuleType("requests") + + def fake_post(url, headers=None, json=None, data=None, timeout=None, params=None): + calls["post"].append({"url": url, "headers": headers, "json": json, "data": data}) + return post_responses.pop(0) + + def fake_get(url, headers=None, timeout=None, params=None): + calls["get"].append({"url": url, "headers": headers, "params": params}) + return get_responses.pop(0) + + fake.post = fake_post + fake.get = fake_get + monkeypatch.setitem(sys.modules, "requests", fake) + return calls + + +# ------------------------------------------------------------------ +# Fixtures +# ------------------------------------------------------------------ + +@pytest.fixture() +def hunyuan_env(monkeypatch): + """Set fake TokenHub API credentials.""" + monkeypatch.setenv("TENCENT_TOKENHUB_API_KEY", "thub-fake-test-key") + + +@pytest.fixture() +def no_hunyuan_env(monkeypatch): + """Ensure no TokenHub credentials are set.""" + monkeypatch.delenv("TENCENT_TOKENHUB_API_KEY", raising=False) + + +# ------------------------------------------------------------------ +# Contract compliance +# ------------------------------------------------------------------ + +class TestContract: + + def test_inherits_base_tool(self): + assert issubclass(HunyuanCloudVideo, BaseTool) + + def test_has_required_identity(self): + tool = HunyuanCloudVideo() + assert tool.name == "hunyuan_cloud_video" + assert tool.version == "0.1.0" + assert tool.provider == "hunyuan_cloud" + assert tool.capability == "video_generation" + assert tool.tier == ToolTier.GENERATE + assert tool.stability == ToolStability.EXPERIMENTAL + assert tool.runtime == ToolRuntime.API + + def test_execution_mode_is_async(self): + assert HunyuanCloudVideo().execution_mode == ExecutionMode.ASYNC + + def test_has_input_schema(self): + schema = HunyuanCloudVideo().input_schema + assert schema.get("type") == "object" + props = schema.get("properties", {}) + required = schema.get("required", []) + assert required == ["prompt"] + for field in required: + assert field in props + + def test_has_capabilities(self): + tool = HunyuanCloudVideo() + assert "text_to_video" in tool.capabilities + assert "image_to_video" in tool.capabilities + + def test_has_agent_skills(self): + assert "ai-video-gen" in HunyuanCloudVideo().agent_skills + + def test_has_fallbacks(self): + tool = HunyuanCloudVideo() + assert "jimeng_video" in tool.fallback_tools + assert "kling_official_video" in tool.fallback_tools + assert "minimax_video" in tool.fallback_tools + + def test_has_install_instructions(self): + tool = HunyuanCloudVideo() + assert "TENCENT_TOKENHUB_API_KEY" in tool.install_instructions + + def test_get_info_returns_dict(self): + info = HunyuanCloudVideo().get_info() + assert isinstance(info, dict) + assert info["name"] == "hunyuan_cloud_video" + assert info["provider"] == "hunyuan_cloud" + assert info["runtime"] == "api" + assert info["capability"] == "video_generation" + + def test_status_unavailable_without_keys(self, no_hunyuan_env): + assert HunyuanCloudVideo().get_status() == ToolStatus.UNAVAILABLE + + def test_status_available_with_keys(self, hunyuan_env): + assert HunyuanCloudVideo().get_status() == ToolStatus.AVAILABLE + + def test_has_resource_profile(self): + rp = HunyuanCloudVideo().resource_profile + assert rp.network_required is True + assert rp.vram_mb == 0 + + def test_has_retry_policy(self): + assert HunyuanCloudVideo().retry_policy.max_retries >= 0 + + def test_has_side_effects(self): + side = HunyuanCloudVideo().side_effects + assert len(side) > 0 + assert any("API" in s for s in side) or any("TokenHub" in s for s in side) + + def test_has_user_visible_verification(self): + assert len(HunyuanCloudVideo().user_visible_verification) > 0 + + def test_estimate_cost_returns_float(self): + cost = HunyuanCloudVideo().estimate_cost({"prompt": "x"}) + assert isinstance(cost, float) + assert cost > 0.0 + + def test_dry_run_returns_dict(self): + result = HunyuanCloudVideo().dry_run({"prompt": "test"}) + assert isinstance(result, dict) + assert result["tool"] == "hunyuan_cloud_video" + + +# ------------------------------------------------------------------ +# Supports flags +# ------------------------------------------------------------------ + +class TestSupports: + + def test_text_to_video_supported(self): + assert HunyuanCloudVideo().supports["text_to_video"] is True + + def test_image_to_video_supported(self): + assert HunyuanCloudVideo().supports["image_to_video"] is True + + def test_native_audio_not_supported(self): + assert HunyuanCloudVideo().supports["native_audio"] is False + + def test_seed_not_supported(self): + assert HunyuanCloudVideo().supports["seed"] is False + + +# ------------------------------------------------------------------ +# Idempotency keys +# ------------------------------------------------------------------ + +class TestIdempotencyKeys: + + def test_includes_all_output_affecting_fields(self): + fields = HunyuanCloudVideo().idempotency_key_fields + for field in ("prompt", "operation", "model", "image_url", "resolution"): + assert field in fields, f"missing idempotency field: {field}" + + def test_excludes_execution_only_fields(self): + fields = HunyuanCloudVideo().idempotency_key_fields + for field in ("output_path", "poll_interval_seconds", "timeout_seconds"): + assert field not in fields + + def test_differs_on_operation(self): + tool = HunyuanCloudVideo() + base = {"prompt": "x"} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "operation": "image_to_video"} + ) + + def test_differs_on_model(self): + tool = HunyuanCloudVideo() + base = {"prompt": "x"} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "model": "yt-video-2.0"} + ) + + def test_differs_on_image_url(self): + tool = HunyuanCloudVideo() + base = {"prompt": "x", "operation": "image_to_video"} + assert tool.idempotency_key(base) != tool.idempotency_key( + {**base, "image_url": "https://example.com/img.png"} + ) + + def test_ignores_execution_params(self): + tool = HunyuanCloudVideo() + base = {"prompt": "x"} + assert tool.idempotency_key(base) == tool.idempotency_key( + {**base, "output_path": "/tmp/out.mp4", "timeout_seconds": 999} + ) + + +# ------------------------------------------------------------------ +# Tool-specific behavior +# ------------------------------------------------------------------ + +class TestToolSpecific: + + def test_default_operation_is_text_to_video(self): + tool = HunyuanCloudVideo() + assert tool.input_schema["properties"]["operation"]["default"] == "text_to_video" + + def test_default_resolution_is_720p(self): + tool = HunyuanCloudVideo() + assert tool.input_schema["properties"]["resolution"]["default"] == "720p" + + def test_default_logo_add_is_1(self): + tool = HunyuanCloudVideo() + assert tool.input_schema["properties"]["logo_add"]["default"] == 1 + + def test_default_poll_interval_is_5(self): + tool = HunyuanCloudVideo() + assert tool.input_schema["properties"]["poll_interval_seconds"]["default"] == 5.0 + + def test_default_timeout_is_600(self): + tool = HunyuanCloudVideo() + assert tool.input_schema["properties"]["timeout_seconds"]["default"] == 600 + + # -- _resolve_model -- + + def test_resolve_model_defaults_t2v(self): + model = HunyuanCloudVideo._resolve_model({"prompt": "test"}) + assert model == "hy-video-1.5" + + def test_resolve_model_defaults_i2v(self): + model = HunyuanCloudVideo._resolve_model({ + "prompt": "test", "operation": "image_to_video", + }) + assert model == "yt-video-2.0" + + def test_resolve_model_explicit_input(self): + model = HunyuanCloudVideo._resolve_model({ + "prompt": "test", "model": "yt-video-2.0", + }) + assert model == "yt-video-2.0" + + def test_resolve_model_from_env(self, monkeypatch): + monkeypatch.setenv("TENCENT_TOKENHUB_MODEL", "hy-video-1.5") + model = HunyuanCloudVideo._resolve_model({ + "prompt": "test", "operation": "image_to_video", + }) + assert model == "hy-video-1.5" + + def test_resolve_model_input_overrides_env(self, monkeypatch): + monkeypatch.setenv("TENCENT_TOKENHUB_MODEL", "hy-video-1.5") + model = HunyuanCloudVideo._resolve_model({ + "prompt": "test", "model": "yt-video-2.0", + }) + assert model == "yt-video-2.0" + + # -- _build_payload -- + + def test_build_payload_t2v_minimal(self): + """Minimal T2V payload — only prompt is required by TokenHub.""" + tool = HunyuanCloudVideo() + payload = tool._build_payload({"prompt": "一只猫"}) + assert payload["prompt"] == "一只猫" + # resolution and logo_add are optional in TokenHub; only included when + # explicitly passed in inputs + assert "resolution" not in payload + assert "logo_add" not in payload + + def test_build_payload_t2v_with_resolution_and_logo(self): + tool = HunyuanCloudVideo() + payload = tool._build_payload({ + "prompt": "test", "resolution": "720p", "logo_add": 0, + }) + assert payload["prompt"] == "test" + assert payload["resolution"] == "720p" + assert payload["logo_add"] == 0 + + def test_build_payload_t2v_custom_logo_add(self): + tool = HunyuanCloudVideo() + payload = tool._build_payload({"prompt": "test", "logo_add": 0}) + assert payload["logo_add"] == 0 + + def test_build_payload_i2v_includes_image_url(self): + tool = HunyuanCloudVideo() + payload = tool._build_payload({ + "prompt": "motion", + "operation": "image_to_video", + "image_url": "https://example.com/frame.png", + }) + assert payload["image_url"] == "https://example.com/frame.png" + + def test_build_payload_t2v_omits_image(self): + tool = HunyuanCloudVideo() + payload = tool._build_payload({"prompt": "a cat", "operation": "text_to_video"}) + assert "image_url" not in payload + assert "image" not in payload + + def test_build_payload_i2v_base64_from_path(self, tmp_path): + """image_path should be base64-encoded into the image field.""" + img = tmp_path / "frame.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0test-jpeg-data") + tool = HunyuanCloudVideo() + payload = tool._build_payload({ + "prompt": "test", + "operation": "image_to_video", + "image_path": str(img), + }) + assert payload["image"].startswith("/9j/") + + def test_encode_image_returns_base64(self, tmp_path): + img = tmp_path / "ref.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0\x00\x10JFIF") + encoded = HunyuanCloudVideo._encode_image(str(img)) + assert isinstance(encoded, str) + assert encoded.startswith("/9j/") + + def test_encode_image_raises_on_missing(self): + with pytest.raises(FileNotFoundError): + HunyuanCloudVideo._encode_image("/nonexistent/file.jpg") + + # -- Error paths -- + + def test_no_keys_returns_error(self, no_hunyuan_env): + result = HunyuanCloudVideo().execute({"prompt": "test"}) + assert result.success is False + assert "TENCENT_TOKENHUB_API_KEY" in result.error + + def test_i2v_without_image_fails(self, hunyuan_env): + result = HunyuanCloudVideo().execute( + {"prompt": "test", "operation": "image_to_video"} + ) + assert result.success is False + assert "image_url" in result.error or "image_path" in result.error + + def test_i2v_both_url_and_path_fails(self, hunyuan_env, tmp_path): + img = tmp_path / "ref.jpg" + img.write_bytes(b"fake-jpeg") + result = HunyuanCloudVideo().execute({ + "prompt": "test", + "operation": "image_to_video", + "image_url": "https://example.com/img.jpg", + "image_path": str(img), + }) + assert result.success is False + assert "not both" in result.error.lower() + + def test_safe_error_redacts_keys(self, monkeypatch): + monkeypatch.setenv("TENCENT_TOKENHUB_API_KEY", "thub-secret-key") + redacted = HunyuanCloudVideo._safe_error( + Exception("failed with thub-secret-key in message") + ) + assert "thub-secret-key" not in redacted + assert "[redacted]" in redacted + + def test_safe_error_no_empty_string_bug(self, no_hunyuan_env): + msg = HunyuanCloudVideo._safe_error(Exception("abc")) + assert msg == "abc" + + +# ------------------------------------------------------------------ +# TokenHub auth headers +# ------------------------------------------------------------------ + +class TestAuthHeaders: + + def test_auth_headers_includes_bearer(self): + headers = HunyuanCloudVideo._auth_headers("test-api-key") + assert headers["Authorization"] == "Bearer test-api-key" + assert headers["Content-Type"] == "application/json" + + +# ------------------------------------------------------------------ +# Error handling helpers +# ------------------------------------------------------------------ + +class TestErrorHelpers: + + def test_json_or_raise_returns_dict(self): + class FakeResp: + status_code = 200 + def json(self): + return {"id": "task-123", "status": "queued"} + result = HunyuanCloudVideo._json_or_raise(FakeResp()) + assert result == {"id": "task-123", "status": "queued"} + + def test_json_or_raise_raises_on_non_json(self): + class FakeResp: + status_code = 500 + def json(self): + raise ValueError("not JSON") + with pytest.raises(RuntimeError, match="Non-JSON"): + HunyuanCloudVideo._json_or_raise(FakeResp()) + + def test_check_response_passes_on_success(self): + HunyuanCloudVideo._check_response( + {"id": "task-123", "status": "completed"} + ) + + def test_check_response_passes_without_error_field(self): + HunyuanCloudVideo._check_response( + {"id": "task-456", "status": "running", "progress": 50} + ) + + def test_check_response_raises_on_api_error(self): + with pytest.raises(RuntimeError, match="Prompt too long"): + HunyuanCloudVideo._check_response({ + "error": {"code": "invalid_parameter", "message": "Prompt too long"}, + }) + + def test_check_response_raises_on_auth_failure(self): + with pytest.raises(RuntimeError, match="Invalid API key"): + HunyuanCloudVideo._check_response({ + "error": { + "type": "authentication_error", + "message": "Invalid API key", + }, + }) + + +# ------------------------------------------------------------------ +# Execute with mocked HTTP +# ------------------------------------------------------------------ + +class TestExecuteWithMocks: + + def test_text_to_video_success(self, hunyuan_env, tmp_path, monkeypatch): + """Full T2V flow: submit -> poll -> download -> write output.""" + task_id = "143-test-task-12345" + calls = _install_fake_requests( + monkeypatch, + post_responses=[ + # Submit response + FakeResponse({ + "id": task_id, + "request_id": "req-sub-001", + "object": "video", + "created_at": 1700000000, + "status": "queued", + }), + # Poll response (completed) + FakeResponse({ + "request_id": "req-poll-001", + "object": "video", + "created_at": 1700000000, + "completed_at": 1700000120, + "status": "completed", + "progress": 100, + "data": {"url": "https://example.com/output.mp4"}, + }), + ], + get_responses=[ + # Download response + FakeResponse(content=b"fake-hunyuan-mp4-data"), + ], + ) + + output_path = tmp_path / "hunyuan_out.mp4" + result = HunyuanCloudVideo().execute({ + "prompt": "一只猫在草原上奔跑", + "output_path": str(output_path), + }) + + assert result.success, result.error + assert output_path.read_bytes() == b"fake-hunyuan-mp4-data" + assert result.data["provider"] == "hunyuan_cloud" + assert result.data["route"] == "tokenhub" + assert result.data["model"] == "hy-video-1.5" + assert result.data["task_id"] == task_id + assert result.data["operation"] == "text_to_video" + assert result.cost_usd == pytest.approx(0.25) + assert len(result.artifacts) == 1 + assert result.artifacts[0] == str(output_path) + + # Verify submit was called to the correct TokenHub endpoint + submit_call = calls["post"][0] + assert "tokenhub.tencentmaas.com" in submit_call["url"] + assert submit_call["url"].endswith("/v1/api/video/submit") + assert submit_call["headers"]["Authorization"] == "Bearer thub-fake-test-key" + assert submit_call["json"]["model"] == "hy-video-1.5" + assert submit_call["json"]["prompt"] == "一只猫在草原上奔跑" + + # Verify poll was called + poll_call = calls["post"][1] + assert poll_call["url"].endswith("/v1/api/video/query") + assert poll_call["json"]["model"] == "hy-video-1.5" + assert poll_call["json"]["id"] == task_id + + # Verify download was called + assert len(calls["get"]) == 1 + assert calls["get"][0]["url"] == "https://example.com/output.mp4" + + def test_image_to_video_with_url_success(self, hunyuan_env, tmp_path, monkeypatch): + """Full I2V flow with an image URL.""" + task_id = "i2v-task-999" + _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "id": task_id, "request_id": "req-sub", "object": "video", + "created_at": 1700000000, "status": "queued", + }), + FakeResponse({ + "request_id": "req-poll", "object": "video", + "created_at": 1700000000, "completed_at": 1700000120, + "status": "completed", "progress": 100, + "data": {"url": "https://example.com/i2v_out.mp4"}, + }), + ], + get_responses=[ + FakeResponse(content=b"fake-i2v-video"), + ], + ) + + output_path = tmp_path / "i2v_out.mp4" + result = HunyuanCloudVideo().execute({ + "prompt": "让画面动起来", + "operation": "image_to_video", + "image_url": "https://example.com/frame.jpg", + "output_path": str(output_path), + }) + + assert result.success, result.error + assert output_path.read_bytes() == b"fake-i2v-video" + assert result.data["operation"] == "image_to_video" + assert result.data["model"] == "yt-video-2.0" + assert result.data["task_id"] == task_id + + def test_i2v_with_local_image_path(self, hunyuan_env, tmp_path, monkeypatch): + """Full I2V flow with a local image path -> base64 encoding.""" + img = tmp_path / "frame.jpg" + img.write_bytes(b"\xff\xd8\xff\xe0test-jpeg-image-data") + + task_id = "i2v-local-task" + _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "id": task_id, "request_id": "req-sub", "object": "video", + "created_at": 1700000000, "status": "queued", + }), + FakeResponse({ + "request_id": "req-poll", "object": "video", + "created_at": 1700000000, "completed_at": 1700000120, + "status": "completed", "progress": 100, + "data": {"url": "https://example.com/i2v_local.mp4"}, + }), + ], + get_responses=[ + FakeResponse(content=b"fake-i2v-local-video"), + ], + ) + + output_path = tmp_path / "i2v_local.mp4" + result = HunyuanCloudVideo().execute({ + "prompt": "animate this frame", + "operation": "image_to_video", + "image_path": str(img), + "output_path": str(output_path), + }) + + assert result.success, result.error + assert output_path.read_bytes() == b"fake-i2v-local-video" + + def test_explicit_model_takes_priority(self, hunyuan_env, tmp_path, monkeypatch): + """When model is explicitly set, it should be used for both T2V and I2V.""" + task_id = "model-override-1" + calls = _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "id": task_id, "request_id": "req-sub", "object": "video", + "created_at": 1700000000, "status": "queued", + }), + FakeResponse({ + "request_id": "req-poll", "object": "video", + "status": "completed", "progress": 100, + "data": {"url": "https://example.com/out.mp4"}, + }), + ], + get_responses=[FakeResponse(content=b"data")], + ) + + result = HunyuanCloudVideo().execute({ + "prompt": "test", + "operation": "image_to_video", + "model": "hy-video-1.5", + "image_url": "https://example.com/frame.jpg", + "poll_interval_seconds": 0.1, + "output_path": str(tmp_path / "out.mp4"), + }) + + assert result.success, result.error + assert result.data["model"] == "hy-video-1.5" + assert calls["post"][0]["json"]["model"] == "hy-video-1.5" + + def test_polling_retries_until_success(self, hunyuan_env, tmp_path, monkeypatch): + """Polling should retry when status is queued/running, then succeed.""" + task_id = "poll-retry-task" + _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "id": task_id, "request_id": "req-sub", "object": "video", + "created_at": 1700000000, "status": "queued", + }), + FakeResponse({ + "request_id": "r1", "object": "video", + "status": "queued", "progress": 0, + }), + FakeResponse({ + "request_id": "r2", "object": "video", + "status": "running", "progress": 45, + }), + FakeResponse({ + "request_id": "r3", "object": "video", + "status": "completed", "progress": 100, + "data": {"url": "https://example.com/final.mp4"}, + }), + ], + get_responses=[ + FakeResponse(content=b"final-video-data"), + ], + ) + + result = HunyuanCloudVideo().execute({ + "prompt": "test polling", + "poll_interval_seconds": 0.1, + "output_path": str(tmp_path / "polled.mp4"), + }) + + assert result.success, result.error + assert result.data["task_id"] == task_id + + def test_polling_fails_on_task_failed(self, hunyuan_env, tmp_path, monkeypatch): + """When the API returns status=failed, execute should return error.""" + task_id = "failed-task" + _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "id": task_id, "request_id": "req-sub", "object": "video", + "created_at": 1700000000, "status": "queued", + }), + FakeResponse({ + "request_id": "req-fail", "object": "video", + "status": "failed", + "error": {"code": "internal_error", "message": "Service unavailable"}, + }), + ], + get_responses=[], + ) + + result = HunyuanCloudVideo().execute({ + "prompt": "this will fail", + "poll_interval_seconds": 0.1, + "output_path": str(tmp_path / "failed.mp4"), + }) + + assert result.success is False + assert "failed" in result.error.lower() + assert "Service unavailable" in result.error + + def test_submit_error_returns_failure(self, hunyuan_env, tmp_path, monkeypatch): + """API-level error on submit should be returned as ToolResult error.""" + _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({ + "error": {"code": "invalid_parameter", "message": "Prompt exceeds limit"}, + }), + ], + get_responses=[], + ) + + result = HunyuanCloudVideo().execute({ + "prompt": "test", + "output_path": str(tmp_path / "err.mp4"), + }) + + assert result.success is False + assert "Prompt exceeds limit" in result.error + + +# ------------------------------------------------------------------ +# Registry discovery +# ------------------------------------------------------------------ + +class TestRegistryDiscovery: + + def test_discoverable(self, isolated_tool_registry): + isolated_tool_registry.discover() + tool = isolated_tool_registry.get("hunyuan_cloud_video") + assert tool is not None + assert tool.provider == "hunyuan_cloud" + assert tool.capability == "video_generation" + + def test_distinct_from_local_hunyuan_tool(self, isolated_tool_registry): + isolated_tool_registry.discover() + cloud = isolated_tool_registry.get("hunyuan_cloud_video") + local = isolated_tool_registry.get("hunyuan_video") + assert cloud is not None + assert local is not None + assert cloud.provider == "hunyuan_cloud" + assert local.provider == "hunyuan" + assert cloud.runtime == ToolRuntime.API + assert local.runtime == ToolRuntime.LOCAL_GPU + + +# ------------------------------------------------------------------ +# Schema validation +# ------------------------------------------------------------------ + +class TestSchemaValidation: + + def test_prompt_max_length_200(self): + schema = HunyuanCloudVideo().input_schema + assert schema["properties"]["prompt"]["maxLength"] == 200 + + def test_prompt_rejects_over_200_chars(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + instance = {"prompt": "x" * 201} + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate(instance, schema) + + def test_prompt_accepts_200_chars(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + instance = {"prompt": "x" * 200} + jsonschema.validate(instance, schema) + + def test_operation_accepts_valid_values(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + for op in ["text_to_video", "image_to_video"]: + jsonschema.validate({"prompt": "test", "operation": op}, schema) + + def test_operation_rejects_invalid_values(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + for invalid in ["video_to_video", "", "TEXT_TO_VIDEO"]: + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate({"prompt": "test", "operation": invalid}, schema) + + def test_model_accepts_valid_values(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + for m in ["hy-video-1.5", "yt-video-2.0"]: + jsonschema.validate({"prompt": "test", "model": m}, schema) + + def test_model_rejects_invalid_values(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate({"prompt": "test", "model": "invalid-model"}, schema) + + def test_logo_add_accepts_0_and_1(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + for val in [0, 1]: + jsonschema.validate({"prompt": "test", "logo_add": val}, schema) + + def test_logo_add_rejects_other_values(self): + import jsonschema + schema = HunyuanCloudVideo().input_schema + for invalid in [2, -1, 99]: + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate({"prompt": "test", "logo_add": invalid}, schema) + + def test_poll_interval_minimum_2(self): + schema = HunyuanCloudVideo().input_schema + assert schema["properties"]["poll_interval_seconds"]["minimum"] == 2 + + def test_timeout_minimum_60(self): + schema = HunyuanCloudVideo().input_schema + assert schema["properties"]["timeout_seconds"]["minimum"] == 60 + + def test_resolution_only_720p(self): + schema = HunyuanCloudVideo().input_schema + assert schema["properties"]["resolution"]["enum"] == ["720p", "1080p"] diff --git a/tools/video/hunyuan_cloud_video.py b/tools/video/hunyuan_cloud_video.py new file mode 100644 index 00000000..2ab430b2 --- /dev/null +++ b/tools/video/hunyuan_cloud_video.py @@ -0,0 +1,523 @@ +"""Tencent Hunyuan (腾讯混元) cloud video generation via TokenHub API. + +Calls the Tencent TokenHub API (tokenhub.tencentmaas.com) using simple Bearer +token authentication. This is the OpenAI-compatible API gateway for Tencent +Hunyuan video models — no TC3-HMAC-SHA256 signing required. + +API flow: POST /v1/api/video/submit -> poll /v1/api/video/query -> +download data.url. + +Authentication uses a TokenHub API key obtained from the Tencent Cloud +TokenHub console (https://console.cloud.tencent.com/tokenhub). +""" + +from __future__ import annotations + +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, +) + +_HOST = "tokenhub.tencentmaas.com" +_SUBMIT_PATH = "/v1/api/video/submit" +_QUERY_PATH = "/v1/api/video/query" + +# TokenHub model identifiers +_MODEL_T2V = "hy-video-1.5" # text-to-video +_MODEL_I2V = "yt-video-2.0" # image-to-video + + +class HunyuanCloudVideo(BaseTool): + """Tencent Hunyuan cloud video generation via TokenHub API.""" + + name = "hunyuan_cloud_video" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "video_generation" + provider = "hunyuan_cloud" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.ASYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set TENCENT_TOKENHUB_API_KEY to your Tencent Cloud TokenHub API key.\n" + " Get it at https://console.cloud.tencent.com/tokenhub\n" + " Optionally set TENCENT_TOKENHUB_MODEL to override the default model." + ) + agent_skills = ["ai-video-gen"] + + capabilities = ["text_to_video", "image_to_video"] + supports = { + "text_to_video": True, + "image_to_video": True, + "native_audio": False, + "seed": False, + } + best_for = [ + "Hunyuan text-to-video and image-to-video via Tencent TokenHub API", + "simple Bearer-token auth (no TC3 signing required)", + "direct Tencent Cloud quota usage (not through a third-party gateway)", + "Chinese-language prompt understanding" + ] + not_good_for = [ + "offline generation or air-gapped environments", + "users without Tencent Cloud account and real-name verification" + ] + fallback_tools = ["jimeng_video", "kling_official_video", "minimax_video"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": { + "type": "string", + "maxLength": 200, + "description": ( + "Video description. Max 200 UTF-8 characters. " + "Supports Chinese and English. Be specific about subject, action, " + "setting, and style." + ), + }, + "operation": { + "type": "string", + "enum": ["text_to_video", "image_to_video"], + "default": "text_to_video", + "description": "Generation mode.", + }, + "model": { + "type": "string", + "enum": ["hy-video-1.5", "yt-video-2.0"], + "description": ( + "TokenHub model ID. hy-video-1.5 for text-to-video, " + "yt-video-2.0 for image-to-video. Defaults to the recommended " + "model for the chosen operation." + ), + }, + "image_url": { + "type": "string", + "description": ( + "Reference image URL for image-to-video. " + "Must be publicly accessible. Max 10MB. " + "Formats: jpg/png/jpeg/webp/bmp/tiff. " + "Resolution: 50-5000 pixels per side, aspect ratio 1:4 to 4:1." + ), + }, + "image_path": { + "type": "string", + "description": ( + "Local path to a reference image for image-to-video. " + "Will be base64-encoded and sent inline. " + "Mutually exclusive with image_url." + ), + }, + "resolution": { + "type": "string", + "enum": ["720p", "1080p"], + "default": "720p", + "description": "Output resolution.", + }, + "logo_add": { + "type": "integer", + "enum": [0, 1], + "default": 1, + "description": ( + "Add 'AI-generated' watermark. 1 = add watermark (default), " + "0 = no watermark (requires console approval from Tencent)." + ), + }, + "output_path": { + "type": "string", + "description": "Output file path for the generated video (MP4).", + }, + "poll_interval_seconds": { + "type": "number", + "minimum": 2, + "default": 5.0, + "description": "Seconds between status polls.", + }, + "timeout_seconds": { + "type": "integer", + "minimum": 60, + "default": 600, + "description": "Maximum seconds to wait for generation.", + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True, + ) + retry_policy = RetryPolicy( + max_retries=2, + backoff_seconds=2.0, + retryable_errors=["rate_limit", "timeout"], + ) + idempotency_key_fields = [ + "prompt", + "operation", + "model", + "image_url", + "resolution", + ] + side_effects = [ + "writes video file to output_path", + "calls Tencent TokenHub API (Bearer-token submit + poll + download)", + ] + user_visible_verification = [ + "Watch generated clip for motion coherence and prompt adherence", + "Check for watermark if logo_add=0 was requested", + ] + + # ------------------------------------------------------------------ + # Credential helpers + # ------------------------------------------------------------------ + + @staticmethod + def _api_key() -> str | None: + val = os.environ.get("TENCENT_TOKENHUB_API_KEY", "") + if val and not val.strip().startswith("#"): + return val.strip() + return None + + # ------------------------------------------------------------------ + # Tool contract methods + # ------------------------------------------------------------------ + + def get_status(self) -> ToolStatus: + if self._api_key(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + """Estimate cost in USD based on model and resolution. + + Tencent TokenHub credit-based pricing (1 credit = 1.2 RMB ≈ $0.167 USD): + - HY-Video-1.5: 1.5 credits/generation → $0.25 + - YT-Video-2.0 480p: 2 credits/generation → $0.33 + - YT-Video-2.0 720p/1080p: 5 credits/generation → $0.83 + + Source: https://cloud.tencent.com.cn/document/product/1823/130054 + """ + model = self._resolve_model(inputs) + resolution = inputs.get("resolution", "720p") + + _CREDIT_TO_USD = 1.2 / 7.2 # 1 credit = 1.2 RMB, ~7.2 RMB/USD + + if model == _MODEL_I2V: + # YT-Video-2.0 has resolution-tiered pricing + if resolution in ("720p", "1080p"): + credits = 5.0 + else: + credits = 2.0 # 480p and below + else: + # HY-Video-1.5 (and fallback for unknown models) + credits = 1.5 + + return round(credits * _CREDIT_TO_USD, 2) + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + """Estimate wall-clock time in seconds. + + Empirical estimate: most cloud video generation APIs (Kling, Pika, etc.) + queue + generate in 60-180s. No official latency published by Tencent. + 180s is a safe upper-bound for timeout planning. + """ + return 180.0 + + # ------------------------------------------------------------------ + # Main execution + # ------------------------------------------------------------------ + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = self._api_key() + if not api_key: + return ToolResult( + success=False, + error="TENCENT_TOKENHUB_API_KEY not set. " + self.install_instructions, + ) + + operation = inputs.get("operation", "text_to_video") + if operation == "image_to_video" and not inputs.get("image_url") and not inputs.get("image_path"): + return ToolResult( + success=False, + error="image_to_video requires image_url (public URL) or image_path (local file).", + ) + + if inputs.get("image_url") and inputs.get("image_path"): + return ToolResult( + success=False, + error="Provide only one of image_url or image_path, not both.", + ) + + start = time.time() + try: + result = self._generate(inputs, api_key=api_key) + except Exception as exc: + return ToolResult( + success=False, + error=f"Hunyuan TokenHub video generation failed: {self._safe_error(exc)}", + ) + + result.duration_seconds = round(time.time() - start, 2) + return result + + # ------------------------------------------------------------------ + # Generation pipeline + # ------------------------------------------------------------------ + + def _generate( + self, inputs: dict[str, Any], *, api_key: str, + ) -> ToolResult: + import requests + + from tools.video._shared import probe_output + + payload = self._build_payload(inputs) + model = self._resolve_model(inputs) + task_id = self._submit_task(payload, model=model, api_key=api_key) + video_url = self._poll_task( + task_id, + model=model, + api_key=api_key, + poll_interval=float(inputs.get("poll_interval_seconds", 5.0)), + timeout_seconds=int(inputs.get("timeout_seconds", 600)), + ) + + download = requests.get(video_url, timeout=120) + download.raise_for_status() + + output_path = Path( + inputs.get("output_path", f"hunyuan_cloud_{task_id}.mp4") + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(download.content) + + probed = probe_output(output_path) + return ToolResult( + success=True, + data={ + "provider": "hunyuan_cloud", + "route": "tokenhub", + "model": model, + "prompt": inputs["prompt"], + "operation": inputs.get("operation", "text_to_video"), + "resolution": inputs.get("resolution", "720p"), + "logo_add": payload.get("logo_add", 1), + "task_id": task_id, + "video_url": video_url, + "output": str(output_path), + "format": "mp4", + **probed, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + model=model, + ) + + # ------------------------------------------------------------------ + # Payload construction + # ------------------------------------------------------------------ + + def _build_payload(self, inputs: dict[str, Any]) -> dict[str, Any]: + """Build the request body for TokenHub video submit.""" + operation = inputs.get("operation", "text_to_video") + payload: dict[str, Any] = { + "prompt": inputs["prompt"], + } + + # Optional parameters (TokenHub uses lowercase_with_underscores) + if inputs.get("resolution"): + payload["resolution"] = inputs["resolution"] + if "logo_add" in inputs: + payload["logo_add"] = int(inputs["logo_add"]) + + # Image for image-to-video + if operation == "image_to_video": + if inputs.get("image_url"): + payload["image_url"] = inputs["image_url"] + elif inputs.get("image_path"): + payload["image"] = self._encode_image(inputs["image_path"]) + + return payload + + @staticmethod + def _resolve_model(inputs: dict[str, Any]) -> str: + """Resolve the TokenHub model ID. + + Order of precedence: + 1. Explicit ``model`` input + 2. TENCENT_TOKENHUB_MODEL env var + 3. Default based on operation (hy-video-1.5 for T2V, yt-video-2.0 for I2V) + """ + if inputs.get("model"): + return inputs["model"] + env_model = os.environ.get("TENCENT_TOKENHUB_MODEL", "").strip() + if env_model: + return env_model + operation = inputs.get("operation", "text_to_video") + return _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V + + @staticmethod + def _encode_image(path: str) -> str: + """Read a local image file and return a base64-encoded string.""" + import base64 + + image_path = Path(path) + if not image_path.is_file(): + raise FileNotFoundError(f"Image not found: {path}") + + raw = image_path.read_bytes() + max_raw = 6 * 1024 * 1024 # 6MB raw ≈ 8MB base64 + if len(raw) > max_raw: + raise ValueError( + f"Image too large ({len(raw)} bytes). Max ~6MB raw (8MB base64-encoded)." + ) + + return base64.b64encode(raw).decode("ascii") + + # ------------------------------------------------------------------ + # API communication (TokenHub OpenAI-compatible) + # ------------------------------------------------------------------ + + @staticmethod + def _auth_headers(api_key: str) -> dict[str, str]: + """Build common request headers for TokenHub API calls.""" + return { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + def _submit_task( + self, payload: dict[str, Any], *, model: str, api_key: str, + ) -> str: + """Submit a video generation task and return the task ID.""" + import requests + + body = { + "model": model, + **payload, + } + url = f"https://{_HOST}{_SUBMIT_PATH}" + resp = requests.post( + url, + json=body, + headers=self._auth_headers(api_key), + timeout=30, + ) + data = self._json_or_raise(resp) + self._check_response(data) + + task_id = data.get("id") + if not task_id: + raise RuntimeError( + f"TokenHub submit returned no task id: {data}" + ) + return task_id + + def _poll_task( + self, + task_id: str, + *, + model: str, + api_key: str, + poll_interval: float, + timeout_seconds: int, + ) -> str: + """Poll /v1/api/video/query until completion, return video download URL.""" + import requests + + url = f"https://{_HOST}{_QUERY_PATH}" + + deadline = time.time() + timeout_seconds + while time.time() < deadline: + time.sleep(poll_interval) + + resp = requests.post( + url, + json={"model": model, "id": task_id}, + headers=self._auth_headers(api_key), + timeout=30, + ) + data = self._json_or_raise(resp) + self._check_response(data) + + status = data.get("status", "") + + if status == "completed": + result_data = data.get("data") or {} + video_url = result_data.get("url") + if not video_url: + raise RuntimeError( + f"TokenHub task {task_id} completed but no data.url: {data}" + ) + return video_url + + if status == "failed": + error_info = data.get("error") or {} + error_msg = error_info.get("message", "unknown error") + raise RuntimeError( + f"TokenHub task {task_id} failed: {error_msg}" + ) + + # queued / running / in_progress — continue polling + if status not in ("queued", "running", "in_progress"): + raise RuntimeError( + f"TokenHub task {task_id} returned unknown status: {status}" + ) + + raise TimeoutError( + f"TokenHub task {task_id} did not finish within {timeout_seconds}s" + ) + + # ------------------------------------------------------------------ + # Error handling helpers + # ------------------------------------------------------------------ + + @staticmethod + def _safe_error(exc: Exception) -> str: + """Redact secret values from exception messages.""" + msg = str(exc) + for var in ("TENCENT_TOKENHUB_API_KEY",): + val = os.environ.get(var, "") + if val: + msg = msg.replace(val, "[redacted]") + return msg + + @staticmethod + def _json_or_raise(response: Any) -> dict[str, Any]: + """Parse JSON response body or raise with HTTP status.""" + try: + return response.json() + except ValueError as exc: + raise RuntimeError( + f"Non-JSON response from TokenHub API: HTTP {response.status_code}" + ) from exc + + @staticmethod + def _check_response(payload: dict[str, Any]) -> None: + """Check the TokenHub API response for errors. + + TokenHub returns errors at the top level with an ``error`` field. + """ + error = payload.get("error") + if error: + message = error.get("message", "unknown error") + code = error.get("code", error.get("type", "unknown")) + raise RuntimeError( + f"TokenHub API error: code={code}, message={message}" + ) From 2237a9bcf9240bdbe417f3c6d9cd746970dcb6bc Mon Sep 17 00:00:00 2001 From: clarkh Date: Thu, 30 Jul 2026 10:53:33 +0800 Subject: [PATCH 2/4] refactor: remove TENCENT_TOKENHUB_MODEL env var, relocate Hunyuan entry in .env.example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The env var added complexity without meaningful benefit — explicit model selection via the "model" input parameter is sufficient for both image and video TokenHub tools. - tools/video/hunyuan_cloud_video.py: drop env var fallback from _resolve_model(), remove mention from install_instructions - tests/contracts/test_hunyuan_cloud_video.py: remove test_resolve_model_from_env and test_resolve_model_input_overrides_env - .env.example: move TENCENT_TOKENHUB_API_KEY to a dedicated "Tencent Hunyuan TokenHub API" section, drop TENCENT_TOKENHUB_MODEL comment, broaden description from "video generation" to generic "Tencent Hunyuan via TokenHub API" --- .env.example | 7 ++++--- tests/contracts/test_hunyuan_cloud_video.py | 14 -------------- tools/video/hunyuan_cloud_video.py | 9 ++------- 3 files changed, 6 insertions(+), 24 deletions(-) diff --git a/.env.example b/.env.example index 70266e18..165e364d 100644 --- a/.env.example +++ b/.env.example @@ -44,6 +44,10 @@ DOUBAO_SPEECH_VOICE_TYPE= # Default Doubao speaker/voice type, e.g. zh_female DASHSCOPE_API_KEY= # Qwen image gen (qwen-image-2.0-pro), TTS (qwen3-tts-flash), ASR with word timestamps (qwen3-asr-flash-filetrans) # Get one at https://dashscope.aliyun.com/ +# --- Tencent Hunyuan TokenHub API --- +TENCENT_TOKENHUB_API_KEY= # Tencent Hunyuan (腾讯混元) via TokenHub API (Bearer token) + # Get it at https://console.cloud.tencent.com/tokenhub + # --- Music --- SUNO_API_KEY= # Suno AI music generation (full songs, instrumentals, any genre) @@ -52,9 +56,6 @@ HEYGEN_API_KEY= # HeyGen API (VEO, Sora, Runway, Kling, Seedance vi RUNWAY_API_KEY= # Runway Gen-4 (direct API, alternative to fal.ai routing) VOLC_ACCESSKEY= # Volcengine Jimeng (即梦 AI) video generation via official API (HMAC-SHA256 V4 signing) VOLC_SECRETKEY= # Secret Access Key paired with VOLC_ACCESSKEY. Get both at https://console.volcengine.com/iam/keymanage -TENCENT_TOKENHUB_API_KEY= # Tencent Hunyuan (腾讯混元) video generation via TokenHub API (Bearer token) - # Get it at https://console.cloud.tencent.com/tokenhub -# TENCENT_TOKENHUB_MODEL= # Optional: override default TokenHub model (hy-video-1.5 / yt-video-2.0) VIDEO_GEN_LOCAL_ENABLED= # Set to "true" for local video gen (needs GPU + diffusers) VIDEO_GEN_LOCAL_MODEL= # Local model: wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b MODAL_LTX2_ENDPOINT_URL= # Modal self-hosted LTX-2 endpoint (optional) diff --git a/tests/contracts/test_hunyuan_cloud_video.py b/tests/contracts/test_hunyuan_cloud_video.py index b445d5dd..e1915fe1 100644 --- a/tests/contracts/test_hunyuan_cloud_video.py +++ b/tests/contracts/test_hunyuan_cloud_video.py @@ -280,20 +280,6 @@ class TestToolSpecific: }) assert model == "yt-video-2.0" - def test_resolve_model_from_env(self, monkeypatch): - monkeypatch.setenv("TENCENT_TOKENHUB_MODEL", "hy-video-1.5") - model = HunyuanCloudVideo._resolve_model({ - "prompt": "test", "operation": "image_to_video", - }) - assert model == "hy-video-1.5" - - def test_resolve_model_input_overrides_env(self, monkeypatch): - monkeypatch.setenv("TENCENT_TOKENHUB_MODEL", "hy-video-1.5") - model = HunyuanCloudVideo._resolve_model({ - "prompt": "test", "model": "yt-video-2.0", - }) - assert model == "yt-video-2.0" - # -- _build_payload -- def test_build_payload_t2v_minimal(self): diff --git a/tools/video/hunyuan_cloud_video.py b/tools/video/hunyuan_cloud_video.py index 2ab430b2..5a55dd22 100644 --- a/tools/video/hunyuan_cloud_video.py +++ b/tools/video/hunyuan_cloud_video.py @@ -56,8 +56,7 @@ class HunyuanCloudVideo(BaseTool): dependencies = [] install_instructions = ( "Set TENCENT_TOKENHUB_API_KEY to your Tencent Cloud TokenHub API key.\n" - " Get it at https://console.cloud.tencent.com/tokenhub\n" - " Optionally set TENCENT_TOKENHUB_MODEL to override the default model." + " Get it at https://console.cloud.tencent.com/tokenhub" ) agent_skills = ["ai-video-gen"] @@ -361,14 +360,10 @@ class HunyuanCloudVideo(BaseTool): Order of precedence: 1. Explicit ``model`` input - 2. TENCENT_TOKENHUB_MODEL env var - 3. Default based on operation (hy-video-1.5 for T2V, yt-video-2.0 for I2V) + 2. Default based on operation (hy-video-1.5 for T2V, yt-video-2.0 for I2V) """ if inputs.get("model"): return inputs["model"] - env_model = os.environ.get("TENCENT_TOKENHUB_MODEL", "").strip() - if env_model: - return env_model operation = inputs.get("operation", "text_to_video") return _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V From 190d165d8bd5675c06ee949f93fd1911f2892c11 Mon Sep 17 00:00:00 2001 From: calesthio Date: Thu, 13 Aug 2026 09:16:22 -0700 Subject: [PATCH 3/4] fix: enforce Hunyuan provider contracts --- docs/PROVIDERS.md | 3 +- tests/contracts/test_hunyuan_cloud_video.py | 87 +++++++++++++++------ tools/video/hunyuan_cloud_video.py | 14 +++- 3 files changed, 76 insertions(+), 28 deletions(-) diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index a9b86874..4d285c92 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -355,7 +355,7 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon > simple Bearer-token authentication. No TC3-HMAC-SHA256 signing required. **Tools unlocked:** `hunyuan_cloud_video` -**Env vars:** `TENCENT_TOKENHUB_API_KEY` (required), `TENCENT_TOKENHUB_MODEL` (optional, overrides the default model) +**Env var:** `TENCENT_TOKENHUB_API_KEY` #### Setup @@ -365,7 +365,6 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon 4. Add to `.env`: ```bash TENCENT_TOKENHUB_API_KEY=your-tokenhub-api-key - # TENCENT_TOKENHUB_MODEL=hy-video-1.5 # optional model override ``` #### What It's Best For diff --git a/tests/contracts/test_hunyuan_cloud_video.py b/tests/contracts/test_hunyuan_cloud_video.py index e1915fe1..dc9f6d8c 100644 --- a/tests/contracts/test_hunyuan_cloud_video.py +++ b/tests/contracts/test_hunyuan_cloud_video.py @@ -199,7 +199,10 @@ class TestIdempotencyKeys: def test_includes_all_output_affecting_fields(self): fields = HunyuanCloudVideo().idempotency_key_fields - for field in ("prompt", "operation", "model", "image_url", "resolution"): + for field in ( + "prompt", "operation", "model", "image_url", "image_path", + "resolution", "logo_add", + ): assert field in fields, f"missing idempotency field: {field}" def test_excludes_execution_only_fields(self): @@ -358,6 +361,28 @@ class TestToolSpecific: assert result.success is False assert "image_url" in result.error or "image_path" in result.error + @pytest.mark.parametrize( + ("operation", "model", "expected"), + [ + ("text_to_video", "yt-video-2.0", "hy-video-1.5"), + ("image_to_video", "hy-video-1.5", "yt-video-2.0"), + ], + ) + def test_incompatible_model_operation_fails_before_submit( + self, hunyuan_env, monkeypatch, operation, model, expected + ): + monkeypatch.setattr( + HunyuanCloudVideo, + "_generate", + lambda *args, **kwargs: pytest.fail("must not submit a paid task"), + ) + inputs = {"prompt": "test", "operation": operation, "model": model} + if operation == "image_to_video": + inputs["image_url"] = "https://example.com/frame.png" + result = HunyuanCloudVideo().execute(inputs) + assert not result.success + assert expected in (result.error or "") + def test_i2v_both_url_and_path_fails(self, hunyuan_env, tmp_path): img = tmp_path / "ref.jpg" img.write_bytes(b"fake-jpeg") @@ -587,37 +612,16 @@ class TestExecuteWithMocks: assert result.success, result.error assert output_path.read_bytes() == b"fake-i2v-local-video" - def test_explicit_model_takes_priority(self, hunyuan_env, tmp_path, monkeypatch): - """When model is explicitly set, it should be used for both T2V and I2V.""" - task_id = "model-override-1" - calls = _install_fake_requests( - monkeypatch, - post_responses=[ - FakeResponse({ - "id": task_id, "request_id": "req-sub", "object": "video", - "created_at": 1700000000, "status": "queued", - }), - FakeResponse({ - "request_id": "req-poll", "object": "video", - "status": "completed", "progress": 100, - "data": {"url": "https://example.com/out.mp4"}, - }), - ], - get_responses=[FakeResponse(content=b"data")], - ) - + def test_explicit_incompatible_model_is_rejected(self, hunyuan_env): result = HunyuanCloudVideo().execute({ "prompt": "test", "operation": "image_to_video", "model": "hy-video-1.5", "image_url": "https://example.com/frame.jpg", - "poll_interval_seconds": 0.1, - "output_path": str(tmp_path / "out.mp4"), }) - assert result.success, result.error - assert result.data["model"] == "hy-video-1.5" - assert calls["post"][0]["json"]["model"] == "hy-video-1.5" + assert not result.success + assert "yt-video-2.0" in (result.error or "") def test_polling_retries_until_success(self, hunyuan_env, tmp_path, monkeypatch): """Polling should retry when status is queued/running, then succeed.""" @@ -731,6 +735,39 @@ class TestRegistryDiscovery: assert cloud.runtime == ToolRuntime.API assert local.runtime == ToolRuntime.LOCAL_GPU + def test_video_selector_routes_to_hunyuan_cloud(self, hunyuan_env, monkeypatch): + from tools.base_tool import ToolResult + from tools.video.video_selector import VideoSelector + + tool = HunyuanCloudVideo() + selector = VideoSelector() + monkeypatch.setattr(selector, "_providers", lambda: [tool]) + monkeypatch.setattr( + selector, + "_select_best_tool", + lambda _inputs, _candidates, _context: (tool, None), + ) + monkeypatch.setattr( + tool, + "execute", + lambda inputs: ToolResult( + success=True, + data={"received": inputs}, + artifacts=[inputs["output_path"]], + ), + ) + + result = selector.execute( + { + "prompt": "test", + "preferred_provider": "hunyuan_cloud", + "output_path": "out.mp4", + } + ) + assert result.success + assert result.data["selected_tool"] == "hunyuan_cloud_video" + assert result.data["selected_provider"] == "hunyuan_cloud" + # ------------------------------------------------------------------ # Schema validation diff --git a/tools/video/hunyuan_cloud_video.py b/tools/video/hunyuan_cloud_video.py index 5a55dd22..ff443b88 100644 --- a/tools/video/hunyuan_cloud_video.py +++ b/tools/video/hunyuan_cloud_video.py @@ -53,7 +53,7 @@ class HunyuanCloudVideo(BaseTool): determinism = Determinism.STOCHASTIC runtime = ToolRuntime.API - dependencies = [] + dependencies = ["env:TENCENT_TOKENHUB_API_KEY"] install_instructions = ( "Set TENCENT_TOKENHUB_API_KEY to your Tencent Cloud TokenHub API key.\n" " Get it at https://console.cloud.tencent.com/tokenhub" @@ -171,7 +171,9 @@ class HunyuanCloudVideo(BaseTool): "operation", "model", "image_url", + "image_path", "resolution", + "logo_add", ] side_effects = [ "writes video file to output_path", @@ -251,6 +253,16 @@ class HunyuanCloudVideo(BaseTool): ) operation = inputs.get("operation", "text_to_video") + model = self._resolve_model(inputs) + expected_model = _MODEL_I2V if operation == "image_to_video" else _MODEL_T2V + if model != expected_model: + return ToolResult( + success=False, + error=( + f"Model '{model}' is not compatible with operation '{operation}'. " + f"Use '{expected_model}'." + ), + ) if operation == "image_to_video" and not inputs.get("image_url") and not inputs.get("image_path"): return ToolResult( success=False, From b42d360e3df619e2e2ba411b99a3ce1f3f2bc392 Mon Sep 17 00:00:00 2001 From: calesthio Date: Thu, 13 Aug 2026 09:50:12 -0700 Subject: [PATCH 4/4] fix: keep TokenHub example credential empty --- .env.example | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index a1b8f604..ffeac908 100644 --- a/.env.example +++ b/.env.example @@ -62,8 +62,9 @@ DOUBAO_SPEECH_VOICE_TYPE= DASHSCOPE_API_KEY= # --- Tencent Hunyuan TokenHub API --- -TENCENT_TOKENHUB_API_KEY= # Tencent Hunyuan (腾讯混元) via TokenHub API (Bearer token) - # Get it at https://console.cloud.tencent.com/tokenhub +# Tencent Hunyuan (腾讯混元) via TokenHub API (Bearer token). +# Get it at https://console.cloud.tencent.com/tokenhub. +TENCENT_TOKENHUB_API_KEY= # --- Music --- # Suno AI music generation (full songs, instrumentals, any genre).