Merge remote-tracking branch 'origin/main' into codex/repair-pr-442

This commit is contained in:
calesthio
2026-08-13 11:11:02 -07:00
20 changed files with 2349 additions and 34 deletions

View File

@@ -36,7 +36,9 @@ class ElevenLabsTTS(BaseTool):
install_instructions = (
"Set the ELEVENLABS_API_KEY environment variable:\n"
" export ELEVENLABS_API_KEY=your_key_here\n"
"Get a key at https://elevenlabs.io"
"Get a key at https://elevenlabs.io\n"
"If fal_elevenlabs_tts is available, use it instead to access ElevenLabs "
"speech through fal.ai without a separate ElevenLabs key."
)
fallback = "openai_tts"
fallback_tools = ["openai_tts", "piper_tts"]

View File

@@ -0,0 +1,231 @@
"""Generate music with ElevenLabs Music through fal.ai.
This provider uses OpenMontage's shared ``FAL_KEY`` credential and the fal.ai
queue API, then downloads the generated MP3 to a project-local path.
"""
from __future__ import annotations
import math
import os
import time
from pathlib import Path
from typing import Any
from tools.base_tool import (
BaseTool,
Determinism,
ExecutionMode,
ResourceProfile,
RetryPolicy,
ToolResult,
ToolRuntime,
ToolStability,
ToolStatus,
ToolTier,
)
class FalElevenLabsMusic(BaseTool):
"""Generate a precisely timed music track through fal.ai."""
name = "fal_elevenlabs_music"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "music_generation"
provider = "fal.ai"
stability = ToolStability.BETA
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = ["env:FAL_KEY"]
install_instructions = (
"Set FAL_KEY to a fal.ai API key. "
"Get one at https://fal.ai/dashboard/keys"
)
fallback_tools = ["music_gen", "google_music"]
agent_skills = ["music", "elevenlabs"]
capabilities = [
"generate_background_music",
"generate_instrumental",
]
supports = {
"instrumental": True,
"style_control": True,
"exact_duration": True,
}
best_for = [
"precisely timed instrumental background music",
"high-quality ElevenLabs Music through a shared fal.ai account",
"short-form social video soundtracks",
]
not_good_for = [
"offline generation",
"free music sourcing",
"sub-3-second sound effects",
]
input_schema = {
"type": "object",
"required": ["prompt", "duration_seconds"],
"properties": {
"prompt": {
"type": "string",
"description": "Music description including mood, style, and instruments",
},
"duration_seconds": {
"type": "number",
"minimum": 3,
"maximum": 600,
"description": "Exact target length in seconds",
},
"force_instrumental": {
"type": "boolean",
"default": True,
"description": "Generate music without vocals",
},
"output_path": {
"type": "string",
"default": "fal_music_output.mp3",
},
},
}
resource_profile = ResourceProfile(
cpu_cores=1,
ram_mb=256,
vram_mb=0,
disk_mb=50,
network_required=True,
)
retry_policy = RetryPolicy(
max_retries=0,
retryable_errors=["rate_limit", "timeout"],
)
idempotency_key_fields = ["prompt", "duration_seconds", "force_instrumental"]
side_effects = [
"writes an MP3 file to output_path",
"submits one paid fal.ai generation request",
]
user_visible_verification = [
"Listen to the generated music for mood, mix, and duration",
]
_MODEL = "fal-ai/elevenlabs/music"
_QUEUE_URL = f"https://queue.fal.run/{_MODEL}"
_POLL_INTERVAL_SECONDS = 5
_MAX_WAIT_SECONDS = 600
def _get_api_key(self) -> str | None:
return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY")
def get_status(self) -> ToolStatus:
return ToolStatus.AVAILABLE if self._get_api_key() else ToolStatus.UNAVAILABLE
def estimate_cost(self, inputs: dict[str, Any]) -> float:
"""fal bills $0.80 per output minute, rounded up to a full minute."""
duration = inputs.get("duration_seconds")
if duration is None:
raise ValueError("duration_seconds is required for cost estimation")
return round(math.ceil(float(duration) / 60.0) * 0.80, 2)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = self._get_api_key()
if not api_key:
return ToolResult(
success=False,
error="No fal.ai API key found. " + self.install_instructions,
)
duration = inputs.get("duration_seconds")
if duration is None:
return ToolResult(success=False, error="duration_seconds is required")
duration = float(duration)
if not 3 <= duration <= 600:
return ToolResult(
success=False,
error="duration_seconds must be between 3 and 600",
)
import requests
started = time.time()
headers = {
"Authorization": f"Key {api_key}",
"Content-Type": "application/json",
}
payload = {
"prompt": inputs["prompt"],
"music_length_ms": round(duration * 1000),
"force_instrumental": bool(inputs.get("force_instrumental", True)),
}
try:
submit_response = requests.post(
self._QUEUE_URL,
headers=headers,
json=payload,
timeout=30,
)
submit_response.raise_for_status()
queue_data = submit_response.json()
status_url = queue_data["status_url"]
response_url = queue_data["response_url"]
deadline = time.monotonic() + self._MAX_WAIT_SECONDS
while True:
if time.monotonic() >= deadline:
return ToolResult(
success=False,
error="fal.ai music generation timed out while waiting in the queue",
duration_seconds=round(time.time() - started, 2),
)
time.sleep(self._POLL_INTERVAL_SECONDS)
status_response = requests.get(status_url, headers=headers, timeout=20)
status_response.raise_for_status()
status = status_response.json().get("status", "UNKNOWN")
if status == "COMPLETED":
break
if status in {"FAILED", "CANCELLED"}:
return ToolResult(
success=False,
error=f"fal.ai music generation {status.lower()}",
duration_seconds=round(time.time() - started, 2),
)
result_response = requests.get(response_url, headers=headers, timeout=30)
result_response.raise_for_status()
result_data = result_response.json()
audio_url = result_data["audio"]["url"]
audio_response = requests.get(audio_url, timeout=120)
audio_response.raise_for_status()
output_path = Path(inputs.get("output_path", "fal_music_output.mp3"))
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(audio_response.content)
except Exception as exc:
safe_error = str(exc).replace(api_key, "[REDACTED]")
return ToolResult(
success=False,
error=f"fal.ai ElevenLabs Music generation failed: {safe_error}",
duration_seconds=round(time.time() - started, 2),
)
return ToolResult(
success=True,
data={
"provider": "fal.ai",
"model": self._MODEL,
"prompt": inputs["prompt"],
"duration_seconds": duration,
"force_instrumental": payload["force_instrumental"],
"output": str(output_path),
"format": "mp3",
},
artifacts=[str(output_path)],
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - started, 2),
model=self._MODEL,
)

View File

@@ -0,0 +1,359 @@
"""Generate ElevenLabs speech through fal.ai using the shared FAL credential."""
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,
)
class FalElevenLabsTTS(BaseTool):
"""Generate expressive narration with ElevenLabs models hosted by fal.ai."""
name = "fal_elevenlabs_tts"
version = "0.1.0"
tier = ToolTier.VOICE
capability = "tts"
provider = "fal.ai"
stability = ToolStability.BETA
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.STOCHASTIC
runtime = ToolRuntime.API
dependencies = ["env:FAL_KEY"]
install_instructions = (
"Set FAL_KEY to a fal.ai API key. No separate ElevenLabs key is needed. "
"Get a fal.ai key at https://fal.ai/dashboard/keys"
)
fallback_tools = ["google_tts", "piper_tts", "elevenlabs_tts"]
agent_skills = ["elevenlabs"]
capabilities = [
"text_to_speech",
"voice_selection",
"expressive_delivery",
"multilingual",
"word_timestamps",
]
supports = {
"voice_cloning": False,
"multilingual": True,
"offline": False,
"native_audio": True,
"inline_audio_tags": True,
"word_timestamps": True,
}
best_for = [
"expressive ElevenLabs narration through an existing fal.ai connection",
"emotionally directed voiceover with Eleven v3 audio tags",
"multilingual narration without a separate ElevenLabs credential",
]
not_good_for = [
"offline generation",
"voice cloning or private custom ElevenLabs voices",
]
_MODELS = {
"eleven-v3": "fal-ai/elevenlabs/tts/eleven-v3",
"multilingual-v2": "fal-ai/elevenlabs/tts/multilingual-v2",
"turbo-v2.5": "fal-ai/elevenlabs/tts/turbo-v2.5",
}
_MODEL_ALIASES = {
"eleven_v3": "eleven-v3",
"eleven_multilingual_v2": "multilingual-v2",
"multilingual_v2": "multilingual-v2",
"eleven_turbo_v2_5": "turbo-v2.5",
"turbo_v2_5": "turbo-v2.5",
**{value: key for key, value in _MODELS.items()},
}
_PRICE_PER_CHARACTER = {
"eleven-v3": 0.0001,
"multilingual-v2": 0.0001,
"turbo-v2.5": 0.00005,
}
_POLL_INTERVAL_SECONDS = 2
_MAX_WAIT_SECONDS = 300
input_schema = {
"type": "object",
"required": ["text"],
"properties": {
"text": {
"type": "string",
"description": "Text to speak. Eleven v3 supports inline tags such as [whispers].",
},
"voice": {
"type": "string",
"default": "Rachel",
"description": "fal.ai ElevenLabs voice name or ID",
},
"voice_id": {
"type": "string",
"description": "Alias for voice, for compatibility with tts_selector",
},
"model_id": {
"type": "string",
"default": "eleven-v3",
"description": "eleven-v3, multilingual-v2, or turbo-v2.5",
},
"stability": {
"type": "number",
"default": 0.5,
"minimum": 0,
"maximum": 1,
},
"similarity_boost": {
"type": "number",
"default": 0.75,
"minimum": 0,
"maximum": 1,
},
"style": {
"type": "number",
"minimum": 0,
"maximum": 1,
},
"speed": {
"type": "number",
"default": 1.0,
"minimum": 0.7,
"maximum": 1.2,
},
"language_code": {
"type": "string",
"description": "Optional ISO 639-1 language code",
},
"timestamps": {
"type": "boolean",
"default": False,
},
"apply_text_normalization": {
"type": "string",
"default": "auto",
"enum": ["auto", "on", "off"],
},
"output_format": {
"type": "string",
"default": "mp3_44100_128",
"enum": [
"mp3_22050_32",
"mp3_44100_64",
"mp3_44100_96",
"mp3_44100_128",
"mp3_44100_192",
"pcm_16000",
"pcm_24000",
"pcm_44100",
"pcm_48000",
"opus_48000_64",
"opus_48000_96",
"opus_48000_128",
"opus_48000_192",
],
},
"seed": {"type": "integer"},
"output_path": {"type": "string"},
},
}
resource_profile = ResourceProfile(
cpu_cores=1,
ram_mb=256,
vram_mb=0,
disk_mb=50,
network_required=True,
)
retry_policy = RetryPolicy(
max_retries=0,
retryable_errors=["rate_limit", "timeout"],
)
idempotency_key_fields = [
"text",
"voice",
"voice_id",
"model_id",
"stability",
"similarity_boost",
"style",
"speed",
"language_code",
"seed",
]
side_effects = [
"writes an audio file to output_path",
"submits one paid fal.ai ElevenLabs speech request",
]
user_visible_verification = [
"Listen to the generated voice sample before approving full narration",
]
def _get_api_key(self) -> str | None:
return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY")
def get_status(self) -> ToolStatus:
return ToolStatus.AVAILABLE if self._get_api_key() else ToolStatus.UNAVAILABLE
def _resolve_model(self, requested: str | None) -> tuple[str, str]:
model_name = requested or "eleven-v3"
model_name = self._MODEL_ALIASES.get(model_name, model_name)
if model_name not in self._MODELS:
choices = ", ".join(self._MODELS)
raise ValueError(f"model_id must be one of: {choices}")
return model_name, self._MODELS[model_name]
def estimate_cost(self, inputs: dict[str, Any]) -> float:
model_name, _ = self._resolve_model(inputs.get("model_id"))
return round(
len(inputs.get("text", "")) * self._PRICE_PER_CHARACTER[model_name],
4,
)
@staticmethod
def _output_extension(output_format: str) -> str:
return {
"mp3": "mp3",
"pcm": "pcm",
"opus": "opus",
}.get(output_format.split("_", 1)[0], "audio")
def execute(self, inputs: dict[str, Any]) -> ToolResult:
api_key = self._get_api_key()
if not api_key:
return ToolResult(
success=False,
error="No fal.ai API key found. " + self.install_instructions,
)
text = str(inputs.get("text", "")).strip()
if not text:
return ToolResult(success=False, error="text is required")
try:
model_name, model_id = self._resolve_model(inputs.get("model_id"))
except ValueError as exc:
return ToolResult(success=False, error=str(exc))
stability = float(inputs.get("stability", 0.5))
similarity_boost = float(inputs.get("similarity_boost", 0.75))
speed = float(inputs.get("speed", 1.0))
if not 0 <= stability <= 1 or not 0 <= similarity_boost <= 1:
return ToolResult(
success=False,
error="stability and similarity_boost must be between 0 and 1",
)
if not 0.7 <= speed <= 1.2:
return ToolResult(success=False, error="speed must be between 0.7 and 1.2")
output_format = inputs.get("output_format", "mp3_44100_128")
voice = inputs.get("voice") or inputs.get("voice_id") or "Rachel"
payload: dict[str, Any] = {
"text": text,
"voice": voice,
"stability": stability,
"similarity_boost": similarity_boost,
"speed": speed,
"timestamps": bool(inputs.get("timestamps", False)),
"apply_text_normalization": inputs.get("apply_text_normalization", "auto"),
"output_format": output_format,
}
for optional in ("language_code", "seed", "style"):
if inputs.get(optional) is not None:
payload[optional] = inputs[optional]
import requests
started = time.time()
headers = {
"Authorization": f"Key {api_key}",
"Content-Type": "application/json",
}
queue_url = f"https://queue.fal.run/{model_id}"
try:
submit_response = requests.post(
queue_url,
headers=headers,
json=payload,
timeout=30,
)
submit_response.raise_for_status()
queue_data = submit_response.json()
status_url = queue_data["status_url"]
response_url = queue_data["response_url"]
deadline = time.monotonic() + self._MAX_WAIT_SECONDS
while True:
if time.monotonic() >= deadline:
return ToolResult(
success=False,
error="fal.ai ElevenLabs speech timed out while waiting in the queue",
duration_seconds=round(time.time() - started, 2),
)
time.sleep(self._POLL_INTERVAL_SECONDS)
status_response = requests.get(status_url, headers=headers, timeout=20)
status_response.raise_for_status()
status = status_response.json().get("status", "UNKNOWN")
if status == "COMPLETED":
break
if status in {"FAILED", "CANCELLED"}:
return ToolResult(
success=False,
error=f"fal.ai ElevenLabs speech {status.lower()}",
duration_seconds=round(time.time() - started, 2),
)
result_response = requests.get(response_url, headers=headers, timeout=30)
result_response.raise_for_status()
result_data = result_response.json()
audio_url = result_data["audio"]["url"]
audio_response = requests.get(audio_url, timeout=120)
audio_response.raise_for_status()
default_output = f"fal_elevenlabs_tts.{self._output_extension(output_format)}"
output_path = Path(inputs.get("output_path", default_output))
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(audio_response.content)
except Exception as exc:
safe_error = str(exc).replace(api_key, "[REDACTED]")
return ToolResult(
success=False,
error=f"fal.ai ElevenLabs speech failed: {safe_error}",
duration_seconds=round(time.time() - started, 2),
)
data = {
"provider": self.provider,
"model": model_id,
"voice": voice,
"text_length": len(text),
"stability": stability,
"similarity_boost": similarity_boost,
"speed": speed,
"output": str(output_path),
"format": output_format,
}
if payload["timestamps"] and "timestamps" in result_data:
data["timestamps"] = result_data["timestamps"]
return ToolResult(
success=True,
data=data,
artifacts=[str(output_path)],
cost_usd=self.estimate_cost({**inputs, "model_id": model_name, "text": text}),
duration_seconds=round(time.time() - started, 2),
model=model_id,
)

View File

@@ -27,7 +27,6 @@ from tools.base_tool import (
from tools.google_credentials import (
get_access_token,
service_account_configured,
has_google_credentials,
)
@@ -44,7 +43,8 @@ class GoogleTTS(BaseTool):
dependencies = []
install_instructions = (
"Auth option A — API key: set GOOGLE_API_KEY (or GEMINI_API_KEY) to a\n"
"Auth option A — TTS-only API key: set GOOGLE_TTS_API_KEY.\n"
" GOOGLE_API_KEY or GEMINI_API_KEY remain supported for broader Google setups.\n"
" Google Cloud API key with Text-to-Speech enabled.\n"
" Enable the API at https://console.cloud.google.com/apis/library/texttospeech.googleapis.com\n"
"Auth option B — service account: set GOOGLE_APPLICATION_CREDENTIALS to the\n"
@@ -149,12 +149,16 @@ class GoogleTTS(BaseTool):
}
def _get_api_key(self) -> str | None:
return os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
return (
os.environ.get("GOOGLE_TTS_API_KEY")
or os.environ.get("GOOGLE_API_KEY")
or os.environ.get("GEMINI_API_KEY")
)
def get_status(self) -> ToolStatus:
# Available via either an API key or a service-account JSON. Both paths
# are honoured by execute() — so this no longer over-reports.
if has_google_credentials():
if self._get_api_key() or service_account_configured():
return ToolStatus.AVAILABLE
return ToolStatus.UNAVAILABLE
@@ -204,7 +208,11 @@ class GoogleTTS(BaseTool):
try:
result = self._generate(inputs, api_key=api_key, bearer_token=bearer_token)
except Exception as exc:
return ToolResult(success=False, error=f"Google TTS failed: {exc}")
safe_error = str(exc)
for credential in (api_key, bearer_token):
if credential:
safe_error = safe_error.replace(credential, "[REDACTED]")
return ToolResult(success=False, error=f"Google TTS failed: {safe_error}")
result.duration_seconds = round(time.time() - start, 2)
result.cost_usd = self.estimate_cost(inputs)
@@ -266,16 +274,14 @@ class GoogleTTS(BaseTool):
url = f"https://texttospeech.googleapis.com/{api_version}/text:synthesize"
headers = {"Content-Type": "application/json"}
params: dict[str, str] = {}
if bearer_token:
headers["Authorization"] = f"Bearer {bearer_token}"
elif api_key:
params["key"] = api_key
headers["x-goog-api-key"] = api_key
response = requests.post(
url,
headers=headers,
params=params,
json=payload,
timeout=120,
)

View File

@@ -45,6 +45,10 @@ class TTSSelector(BaseTool):
"type": "string",
"description": "Provider-specific voice ID. Passed through to the selected TTS provider.",
},
"voice": {
"type": "string",
"description": "Provider-specific voice name or ID. fal.ai ElevenLabs accepts names such as Rachel.",
},
"voice_language": {
"type": "string",
"enum": ["zh", "en"],
@@ -58,7 +62,7 @@ class TTSSelector(BaseTool):
},
"model_id": {
"type": "string",
"description": "TTS model to use (e.g. eleven_multilingual_v2). Passed through to provider.",
"description": "TTS model to use (e.g. eleven-v3 or eleven_multilingual_v2). Passed through to provider.",
},
"stability": {
"type": "number", "minimum": 0, "maximum": 1,
@@ -100,6 +104,24 @@ class TTSSelector(BaseTool):
"default": "text",
"description": "Use 'ssml' only when the selected provider supports tags such as <break>.",
},
"language_code": {
"type": "string",
"description": "Provider-specific language code, such as en-US for Google or en for fal.ai ElevenLabs.",
},
"timestamps": {
"type": "boolean",
"default": False,
"description": "Request word timestamps when the selected provider supports them.",
},
"apply_text_normalization": {
"type": "string",
"enum": ["auto", "on", "off"],
"description": "Text normalization mode for providers that support it.",
},
"seed": {
"type": "integer",
"description": "Optional generation seed for providers that support reproducible speech.",
},
"voice_performance": {
"type": "object",
"description": "Structured voice-performance plan or section delivery cues from the script artifact.",

View File

@@ -119,9 +119,12 @@ class GoogleImagen(BaseTool):
"imagen-4.0-generate-001",
"imagen-4.0-fast-generate-001",
"imagen-4.0-ultra-generate-001",
"gemini-2.5-flash-image",
],
"default": "imagen-4.0-generate-001",
"description": "Imagen model variant",
"description": "Imagen model variant, or a Gemini image model "
"(gemini-*) routed through generate_content. Use "
"gemini-2.5-flash-image when the project has no Imagen access.",
},
"number_of_images": {
"type": "integer",
@@ -177,13 +180,111 @@ class GoogleImagen(BaseTool):
def estimate_cost(self, inputs: dict[str, Any]) -> float:
model = inputs.get("model", "imagen-4.0-generate-001")
n = inputs.get("number_of_images", 1)
if model.startswith("gemini-"):
# ~1290 output tokens per image at $30/1M tokens
return 0.039 * n
if "ultra" in model:
return 0.06 * n
if "fast" in model:
return 0.02 * n
return 0.04 * n
def _resolve_aspect_ratio(self, inputs: dict[str, Any]) -> str:
"""Explicit aspect_ratio > derived from width/height > default 1:1."""
if "aspect_ratio" in inputs:
return inputs["aspect_ratio"]
if "width" in inputs and "height" in inputs:
import logging
aspect_ratio = _dims_to_aspect_ratio(inputs["width"], inputs["height"])
logging.getLogger(__name__).info(
"google_imagen: remapped %sx%s to nearest supported aspect ratio %s",
inputs["width"],
inputs["height"],
aspect_ratio,
)
return aspect_ratio
return "1:1"
def _execute_gemini(self, inputs: dict[str, Any], model: str) -> ToolResult:
"""Generate via a Gemini image model (e.g. gemini-2.5-flash-image).
These models use generate_content with an image_config instead of the
Imagen :predict endpoint, and work on both auth paths (API key and
Vertex service account) through the shared genai client.
"""
start = time.time()
try:
from google.genai import types
from tools.google_credentials import get_genai_client
client = get_genai_client()
except Exception as e:
return ToolResult(
success=False,
error=f"Failed to initialize Google GenAI client: {e}",
)
prompt = inputs["prompt"]
aspect_ratio = self._resolve_aspect_ratio(inputs)
number_of_images = inputs.get("number_of_images", 1)
config = types.GenerateContentConfig(
image_config=types.ImageConfig(aspect_ratio=aspect_ratio),
)
image_bytes: list[bytes] = []
try:
for _ in range(number_of_images):
response = client.models.generate_content(
model=model, contents=prompt, config=config
)
for part in response.candidates[0].content.parts or []:
inline = getattr(part, "inline_data", None)
if inline and inline.data:
image_bytes.append(inline.data)
break
except Exception as e:
return ToolResult(
success=False, error=f"Gemini image generation failed: {e}"
)
if not image_bytes:
return ToolResult(
success=False,
error=f"No image data returned by {model} (text-only response).",
)
output_paths = self._output_paths(inputs.get("output_path"), len(image_bytes))
outputs: list[str] = []
for data, out_path in zip(image_bytes, output_paths):
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_bytes(data)
outputs.append(str(out_path))
return ToolResult(
success=True,
data={
"provider": "google_imagen",
"model": model,
"prompt": prompt,
"aspect_ratio": aspect_ratio,
"output": outputs[0],
"outputs": outputs,
"images_generated": len(outputs),
},
artifacts=outputs,
cost_usd=self.estimate_cost(inputs),
duration_seconds=round(time.time() - start, 2),
model=model,
)
def execute(self, inputs: dict[str, Any]) -> ToolResult:
# Gemini image models go through generate_content via the shared genai
# client, which resolves auth (API key or Vertex service account) itself.
model = inputs.get("model", "imagen-4.0-generate-001")
if model.startswith("gemini-"):
return self._execute_gemini(inputs, model)
# Two auth paths: an AI Studio API key, or a service-account JSON that
# routes to Vertex AI (the AI Studio endpoint does not accept service
# accounts). API key wins when both are present.
@@ -213,27 +314,9 @@ class GoogleImagen(BaseTool):
import requests
start = time.time()
model = inputs.get("model", "imagen-4.0-generate-001")
prompt = inputs["prompt"]
import logging
logger = logging.getLogger(__name__)
# Resolve aspect ratio: explicit > derived from width/height > default
if "aspect_ratio" in inputs:
aspect_ratio = inputs["aspect_ratio"]
elif "width" in inputs and "height" in inputs:
requested_ratio = f"{inputs['width']}x{inputs['height']}"
aspect_ratio = _dims_to_aspect_ratio(inputs["width"], inputs["height"])
logger.info(
"google_imagen: remapped %s to nearest supported aspect ratio %s",
requested_ratio,
aspect_ratio,
)
else:
aspect_ratio = "1:1"
aspect_ratio = self._resolve_aspect_ratio(inputs)
number_of_images = inputs.get("number_of_images", 1)
parameters: dict[str, Any] = {

View File

@@ -0,0 +1,580 @@
"""Tencent Hunyuan (腾讯混元) cloud image generation (3.0) 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 image models — no TC3-HMAC-SHA256 signing required.
API flow: POST /v1/api/image/submit -> poll /v1/api/image/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/image/submit"
_QUERY_PATH = "/v1/api/image/query"
# TokenHub model identifier for 混元生图 3.0
_MODEL = "hy-image-v3.0"
class HunyuanImage(BaseTool):
"""Tencent Hunyuan cloud image generation (3.0) via TokenHub API."""
name = "hunyuan_image"
version = "0.1.0"
tier = ToolTier.GENERATE
capability = "image_generation"
provider = "hunyuan_cloud"
stability = ToolStability.EXPERIMENTAL
execution_mode = ExecutionMode.ASYNC
determinism = Determinism.SEEDED
runtime = ToolRuntime.API
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"
)
agent_skills = ["visual-style"]
capabilities = ["generate_image", "text_to_image"]
supports = {
"negative_prompt": False,
"seed": True,
"custom_size": True,
"reference_image": True,
"prompt_rewrite": True,
}
best_for = [
"Hunyuan text-to-image 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 = ["dashscope_image", "flux_image", "openai_image", "recraft_image"]
input_schema = {
"type": "object",
"required": ["prompt"],
"properties": {
"prompt": {
"type": "string",
"maxLength": 8192,
"description": (
"Image description. Max 8192 UTF-8 characters. "
"Supports Chinese and English. Be specific about subject, "
"composition, style, and mood."
),
},
"images": {
"type": "array",
"items": {"type": "string"},
"maxItems": 3,
"description": (
"Reference images per upstream Images.N param (max 3). "
"Each entry is a publicly accessible URL or a local file path "
"(auto-encoded to base64 data URI). "
"Single image: 50-5000px per side, base64 < 6MB. "
"Formats: jpg/png/jpeg/webp/bmp/tiff."
),
},
"resolution": {
"type": "string",
"default": "1024:1024",
"description": (
'Image resolution as "W:H" (colon separator, per upstream '
"Resolution param). W, H in [512, 2048], product (W*H) <= "
'1024x1024 pixels. Examples: "1024:1024", "768:1024", '
'"1024:576".'
),
},
"seed": {
"type": "integer",
"minimum": 1,
"maximum": 4294967295,
"description": (
"Random seed in [1, 4294967295]. "
"Note: seed is ignored when revise is enabled (default)."
),
},
"revise": {
"type": "integer",
"enum": [0, 1],
"default": 1,
"description": (
"Prompt auto-rewrite toggle per upstream Revise param. "
"1 = enabled (default, adds ~20s processing), 0 = disabled. "
"When disabled, caller should handle prompt rewriting."
),
},
"logo_add": {
"type": "integer",
"enum": [0, 1],
"default": 1,
"description": (
"Add 'AI-generated' watermark per upstream LogoAdd param. "
"1 = add watermark (default), 0 = no watermark. "
"Values other than 0 or 1 are treated as 1."
),
},
"logo_param": {
"type": "object",
"properties": {
"logo_url": {
"type": "string",
"description": "Custom watermark image URL.",
},
"logo_image": {
"type": "string",
"description": "Custom watermark image as base64-encoded string.",
},
},
"description": (
"Custom watermark settings per upstream LogoParam. "
"Default: \"图片由 AI 生成\" at bottom-right. "
"Note: custom watermarks may not be supported by the engine "
"(error: InvalidParameterValue.LogoParamErr)."
),
},
"output_path": {
"type": "string",
"description": "Output file path for the generated image (PNG).",
},
"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=100, network_required=True,
)
retry_policy = RetryPolicy(
max_retries=2,
backoff_seconds=2.0,
retryable_errors=["rate_limit", "timeout"],
)
idempotency_key_fields = [
"prompt",
"resolution",
"images",
"seed",
"revise",
"logo_add",
"logo_param",
]
side_effects = [
"writes image file to output_path",
"calls Tencent TokenHub API (Bearer-token submit + poll + download)",
]
user_visible_verification = [
"Inspect generated image for quality 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.
Tencent TokenHub credit-based pricing (1 credit = 1.2 RMB ≈ $0.167 USD):
- hy-image-v3.0: ~0.5 credits/image → ~$0.08
Source: https://cloud.tencent.com.cn/document/product/1823/130054
"""
_CREDIT_TO_USD = 1.2 / 7.2 # 1 credit = 1.2 RMB, ~7.2 RMB/USD
credits = 0.5
return round(credits * _CREDIT_TO_USD, 2)
def estimate_runtime(self, inputs: dict[str, Any]) -> float:
"""Estimate wall-clock time in seconds.
Per upstream docs, prompt rewrite (revise=1) adds ~20s. Including
queuing and download, 120s is a safe upper-bound.
"""
return 120.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,
)
start = time.time()
try:
result = self._generate(inputs, api_key=api_key)
except Exception as exc:
return ToolResult(
success=False,
error=f"Hunyuan TokenHub image 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
# Guard: refuse to make paid API calls without an explicit output_path.
# A CWD-relative default would leak files into the project root when
# called by selectors or other automated tooling.
if not inputs.get("output_path"):
return ToolResult(
success=False,
error="output_path is required for hunyuan_image generation.",
)
payload = self._build_payload(inputs)
task_id = self._submit_task(payload, model=_MODEL, api_key=api_key)
image_urls = 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)),
)
output_paths = self._resolve_output_paths(
inputs["output_path"],
count=len(image_urls),
)
for path, url in zip(output_paths, image_urls):
download = requests.get(url, timeout=120)
download.raise_for_status()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(download.content)
return ToolResult(
success=True,
data={
"provider": "hunyuan_cloud",
"route": "tokenhub",
"model": _MODEL,
"prompt": inputs["prompt"],
"resolution": payload.get("resolution", "1024:1024"),
"revise": payload.get("revise", 1),
"logo_add": payload.get("logo_add", 1),
"task_id": task_id,
"output": str(output_paths[0]),
"outputs": [str(p) for p in output_paths],
"images_generated": len(output_paths),
},
artifacts=[str(p) for p in output_paths],
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 image submit.
TokenHub async endpoints use snake_case versions of the upstream
SubmitTextToImageJob parameter names (e.g. Resolution -> resolution,
Images -> images).
"""
payload: dict[str, Any] = {
"prompt": inputs["prompt"],
}
# Optional parameters (snake_case of upstream SubmitTextToImageJob params)
if inputs.get("resolution"):
payload["resolution"] = inputs["resolution"]
if inputs.get("seed") is not None:
payload["seed"] = int(inputs["seed"])
if "revise" in inputs:
payload["revise"] = int(inputs["revise"])
if "logo_add" in inputs:
payload["logo_add"] = int(inputs["logo_add"])
if inputs.get("logo_param"):
logo_param: dict[str, str] = {}
lp = inputs["logo_param"]
if lp.get("logo_url"):
logo_param["logo_url"] = lp["logo_url"]
if lp.get("logo_image"):
logo_param["logo_image"] = lp["logo_image"]
if logo_param:
payload["logo_param"] = logo_param
# Reference images — maps to upstream Images.N
# TokenHub accepts URLs or base64 data URIs in the images array
image_refs = inputs.get("images")
if image_refs:
payload["images"] = self._resolve_images(image_refs)
return payload
@staticmethod
def _resolve_images(refs: list[str]) -> list[str]:
"""Resolve reference images to strings for the TokenHub API.
Each entry may be:
- An HTTP(S) URL → passed through unchanged
- A data URI (``data:...``) → passed through unchanged
- A local file path → base64-encoded as a data URI
Per upstream docs: single image 50-5000px per side, base64 < 6MB.
Formats: jpg/jpeg/png/bmp/tiff/webp.
"""
import base64
resolved: list[str] = []
for ref in refs:
if ref.startswith("data:") or ref.startswith("http://") or ref.startswith("https://"):
resolved.append(ref)
continue
image_path = Path(ref)
if not image_path.is_file():
raise FileNotFoundError(f"Reference image not found: {ref}")
raw = image_path.read_bytes()
max_raw = 6 * 1024 * 1024 # 6MB per upstream limit
if len(raw) > max_raw:
raise ValueError(
f"Image too large ({len(raw)} bytes). Max ~6MB raw."
)
suffix = image_path.suffix.lower()
mime_map = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".bmp": "image/bmp",
".tiff": "image/tiff",
".tif": "image/tiff",
".webp": "image/webp",
}
mime = mime_map.get(suffix, "image/png")
data = base64.b64encode(raw).decode("ascii")
resolved.append(f"data:{mime};base64,{data}")
return resolved
# ------------------------------------------------------------------
# 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 an image generation task and return the task ID.
POST /v1/api/image/submit
Request: {"model": "hy-image-v3.0", "prompt": "...", ...}
Response: {"id": "...", "status": "queued", ...}
"""
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,
) -> list[str]:
"""Poll /v1/api/image/query until completion, return image download URLs.
Response when completed:
{"status": "completed", "data": [{"url": "...", "revised_prompt": "..."}]}
The data array may contain multiple images. Each URL is valid for ~1 hour.
"""
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 []
urls = [item.get("url") for item in result_data if item.get("url")]
if not urls:
raise RuntimeError(
f"TokenHub task {task_id} completed but no data[].url: {data}"
)
return urls
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}"
)
# ------------------------------------------------------------------
# Output helpers
# ------------------------------------------------------------------
@staticmethod
def _resolve_output_paths(base: str, count: int) -> list[Path]:
"""Derive distinct paths for ``count`` images.
Single image keeps the base path unchanged; multiple images insert an
index before the extension (foo.png -> foo_1.png, foo_2.png, ...).
"""
base_path = Path(base)
if count <= 1:
return [base_path]
stem = base_path.stem
suffix = base_path.suffix
parent = base_path.parent
return [parent / f"{stem}_{i}{suffix}" for i in range(1, count + 1)]

View File

@@ -242,6 +242,25 @@ class ImageSelector(BaseTool):
props = tool.input_schema.get("properties", {})
if "query" in props and "query" not in adapted:
adapted["query"] = adapted.get("prompt", "")
# Normalize the selector's shared reference-image inputs for
# providers whose native contract accepts an ``images`` array.
if "images" in props and "images" not in adapted:
refs = (
adapted.get("image_paths")
or adapted.get("image_urls")
or ([adapted["image_path"]] if adapted.get("image_path") else None)
or ([adapted["image_url"]] if adapted.get("image_url") else None)
)
if refs:
adapted["images"] = refs
# The selector exposes a provider-neutral ``model_name`` field,
# while several providers call the same input ``model``.
if (
"model_name" in adapted
and "model" in props
and "model" not in adapted
):
adapted["model"] = adapted["model_name"]
if "n" in adapted and "num_images" in props and "num_images" not in adapted:
adapted["num_images"] = adapted["n"]