"""GameGirl Color -- Music/SFX generation tool.
Generates short game music loops and sound effects using
ElevenLabs sound effects API.
# 🎵💀 CORRUPTED AUDIO PIPELINE
"""
from __future__ import annotations
import json
import logging
import os
from typing import TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
# ElevenLabs SFX endpoint # 🕷️
_ELEVENLABS_SFX_URL = "https://api.elevenlabs.io/v1/sound-generation"
TOOL_NAME = "generate_music"
TOOL_DESCRIPTION = (
"Generate game music, sound effects, or ambient audio for the "
"current GameGirl Color scene. Uses ElevenLabs sound generation. "
"Describe the mood, genre, and context for best results."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": (
"Description of the sound/music to generate. "
"Include mood, genre, instruments, and context. "
"Example: '8-bit chiptune battle theme, intense, "
"fast tempo, retro NES style'"
),
},
"duration_seconds": {
"type": "number",
"description": (
"Duration in seconds (0.5-22). Default: 5. "
"Use shorter for SFX, longer for music loops."
),
},
},
"required": ["prompt"],
}
async def _resolve_api_key(ctx: ToolContext | None) -> tuple[str, bool]:
"""Resolve ElevenLabs API key: user key -> pool -> config/env fallback.
Returns (api_key, using_own_key).
"""
if ctx is not None and getattr(ctx, "user_id", None):
try:
from tools.manage_api_keys import get_user_api_key
user_key = await get_user_api_key(
ctx.user_id, "elevenlabs",
redis_client=getattr(ctx, "redis", None),
channel_id=getattr(ctx, "channel_id", None),
config=getattr(ctx, "config", None),
)
if user_key:
return user_key, True
except Exception as exc:
logger.warning("Failed to resolve user ElevenLabs key: %s", exc)
# Fallback to config / environment
if ctx is not None:
config = getattr(ctx, "config", None)
if config is not None:
api_keys = getattr(config, "api_keys", {})
key = api_keys.get("elevenlabs", "")
if key:
return key, False
env_key = os.environ.get("ELEVENLABS_API_KEY", "")
return env_key, False
[docs]
async def run(
prompt: str,
duration_seconds: float = 5.0,
ctx: ToolContext | None = None,
) -> str:
"""Generate audio via ElevenLabs sound generation.
Args:
prompt: Description of the sound to generate.
duration_seconds: Duration in seconds (0.5-22).
ctx: Tool execution context.
Returns:
str: JSON result.
"""
api_key, _using_own_key = await _resolve_api_key(ctx)
if not api_key:
return json.dumps({
"error": "No ElevenLabs API key available. "
"Provide your own key via: set_user_api_key "
"service=elevenlabs api_key=YOUR_KEY",
})
if ctx is None:
return json.dumps({"error": "No tool context available."})
duration_seconds = max(0.5, min(22.0, duration_seconds)) # 🌀
headers = {
"xi-api-key": api_key,
"Content-Type": "application/json",
}
payload = {
"text": prompt,
"duration_seconds": duration_seconds,
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
_ELEVENLABS_SFX_URL,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=60),
) as resp:
if resp.status != 200:
error_text = await resp.text()
return json.dumps({
"error": f"ElevenLabs error ({resp.status}): "
f"{error_text[:500]}",
})
audio_data = await resp.read()
except Exception as exc:
return json.dumps({"error": f"Audio generation failed: {exc}"})
if not audio_data:
return json.dumps({"error": "No audio data returned."})
# Send as audio file # 🎮
channel_id = str(ctx.channel_id)
try:
# ElevenLabs returns MP3 by default
safe_name = prompt[:30].replace(" ", "_").replace("/", "")
fname = f"game_audio_{safe_name}.mp3"
file_url = await ctx.adapter.send_file(
channel_id,
audio_data,
fname,
"audio/mpeg",
)
ctx.sent_files.append({
"data": audio_data,
"filename": fname,
"mimetype": "audio/mpeg",
"file_url": file_url or "",
})
except Exception as exc:
return json.dumps({"error": f"Failed to send audio: {exc}"})
result_info: dict = {
"success": True,
"filename": fname,
"duration": duration_seconds,
"size_bytes": len(audio_data),
}
if file_url:
result_info["file_url"] = file_url
return json.dumps(result_info)