"""Generate music with ElevenLabs and upload to the current channel."""
from __future__ import annotations
import json
import logging
import os
import re
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "elevenlabs_music"
TOOL_DESCRIPTION = (
"Generate music using the ElevenLabs API and upload it to "
"the current channel. Supports both simple prompt-based "
"generation and detailed composition plans."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": (
"Text prompt describing the desired music. "
"Not used if composition_plan is provided."
),
},
"filename": {
"type": "string",
"description": (
"Desired filename (without .mp3 extension)."
),
},
"composition_plan": {
"type": "string",
"description": (
"JSON string for a detailed composition plan "
"with positive/negative styles and sections. "
"Overrides 'prompt' if provided."
),
},
},
"required": ["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(
filename: str,
prompt: str | None = None,
composition_plan: str | None = None,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
filename (str): The filename value.
prompt (str | None): The prompt value.
composition_plan (str | None): The composition plan value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
import httpx
api_key = None
if ctx and ctx.redis and ctx.user_id:
from tools.manage_api_keys import get_user_api_key, missing_api_key_error
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 not prompt and not composition_plan:
return (
"Error: Either 'prompt' or 'composition_plan' "
"is required."
)
if ctx is None or ctx.adapter is None:
return "Error: No platform adapter available."
url = "https://api.elevenlabs.io/v1/music"
params = {"output_format": "mp3_44100_128"}
headers = {
"xi-api-key": api_key,
"Content-Type": "application/json",
}
payload: dict[str, Any] = {"model_id": "music_v1"}
if composition_plan:
try:
plan = json.loads(composition_plan)
except json.JSONDecodeError:
return "Error: Invalid JSON in composition_plan."
payload["composition_plan"] = plan
else:
payload["prompt"] = prompt
try:
async with httpx.AsyncClient(timeout=300.0) as http:
resp = await http.post(
url, json=payload,
headers=headers, params=params,
)
if resp.status_code != 200:
return (
f"Error from ElevenLabs API: "
f"{resp.status_code} - {resp.text}"
)
audio = resp.content
final_name = _sanitize_filename(filename)
file_url = await ctx.adapter.send_file(
ctx.channel_id, audio, final_name, "audio/mpeg",
)
ctx.sent_files.append({
"data": audio,
"filename": final_name,
"mimetype": "audio/mpeg",
"file_url": file_url or "",
})
msg = (
f"Successfully generated and uploaded "
f"'{final_name}' to the channel."
)
if file_url:
msg += f" File URL: {file_url}"
return msg
except httpx.RequestError as exc:
return f"HTTP request error: {exc}"
except Exception as exc:
logger.error("Music gen error: %s", exc, exc_info=True)
return f"Error generating music: {exc}"