"""Generate sound effects with ElevenLabs and upload to the current channel."""
from __future__ import annotations
import logging
import os
import re
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "elevenlabs_sfx"
TOOL_DESCRIPTION = (
"Generate sound effects from a text description using the "
"ElevenLabs API and upload to the current channel. Good for "
"foley, ambient sounds, sci-fi effects, etc."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": (
"Descriptive prompt for the sound effect. "
"E.g. 'footsteps on gravel', 'thunder rolling', "
"'sci-fi laser blast'."
),
},
"filename": {
"type": "string",
"description": (
"Desired filename (without .mp3 extension)."
),
},
"duration_seconds": {
"type": "number",
"description": (
"Duration in seconds (0.5 - 22). "
"Omit to let the API decide."
),
},
"prompt_influence": {
"type": "number",
"description": (
"How closely to follow the prompt (0.0 - 1.0). "
"Lower = more creative."
),
},
},
"required": ["prompt", "filename"],
}
def _sanitize_filename(name: str) -> str:
"""Internal helper: sanitize filename.
Args:
name (str): Human-readable name.
Returns:
str: Result string.
"""
if name.endswith(".mp3"):
name = name[:-4]
s = re.sub(r'[^\w\s-]', '', name).strip().lower()
return re.sub(r'[-\s]+', '-', s)[:80] + ".mp3"
[docs]
async def run(
prompt: str,
filename: str,
duration_seconds: float | None = None,
prompt_influence: float | None = None,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
prompt (str): The prompt value.
filename (str): The filename value.
duration_seconds (float | None): The duration seconds value.
prompt_influence (float | None): The prompt influence value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
from elevenlabs.client import AsyncElevenLabs
api_key = None
if ctx and ctx.redis and ctx.user_id:
from tools.manage_api_keys import get_user_api_key
api_key = await get_user_api_key(
ctx.user_id, "elevenlabs",
redis_client=ctx.redis, channel_id=ctx.channel_id,
config=getattr(ctx, "config", None),
)
api_key = api_key or os.getenv("ELEVENLABS_API_KEY")
if not api_key:
from tools.manage_api_keys import missing_api_key_error
return missing_api_key_error("elevenlabs")
if duration_seconds is not None:
try:
duration_seconds = float(duration_seconds)
except (ValueError, TypeError):
return (
f"Error: duration_seconds must be a number, "
f"got '{duration_seconds}'."
)
if not (0.5 <= duration_seconds <= 22):
return (
"Error: duration_seconds must be between "
"0.5 and 22."
)
if prompt_influence is not None:
try:
prompt_influence = float(prompt_influence)
except (ValueError, TypeError):
return (
f"Error: prompt_influence must be a number, "
f"got '{prompt_influence}'."
)
if not (0.0 <= prompt_influence <= 1.0):
return (
"Error: prompt_influence must be between "
"0.0 and 1.0."
)
if ctx is None or ctx.adapter is None:
return "Error: No platform adapter available."
try:
el = AsyncElevenLabs(api_key=api_key)
api_params: dict = {"text": prompt}
if duration_seconds is not None:
api_params["duration_seconds"] = duration_seconds
if prompt_influence is not None:
api_params["prompt_influence"] = prompt_influence
log_prompt = (prompt or "")[:120]
if len(prompt or "") > 120:
log_prompt += "…"
logger.info("Generating SFX for prompt: %s", log_prompt)
audio_gen = el.text_to_sound_effects.convert(**api_params)
audio_bytes = b""
async for chunk in audio_gen:
audio_bytes += chunk
if not audio_bytes:
return "Error: No audio data received from API."
final_name = _sanitize_filename(filename)
file_url = await ctx.adapter.send_file(
ctx.channel_id, audio_bytes, final_name, "audio/mpeg",
)
ctx.sent_files.append({
"data": audio_bytes,
"filename": final_name,
"mimetype": "audio/mpeg",
"file_url": file_url or "",
})
dur_info = (
f" ({duration_seconds}s)" if duration_seconds else ""
)
msg = (
f"Generated and uploaded SFX '{final_name}'"
f"{dur_info} to the channel."
)
if file_url:
msg += f" File URL: {file_url}"
return msg
except Exception as exc:
logger.error("SFX gen error: %s", exc, exc_info=True)
return f"Error generating sound effect: {exc}"