"""GameGirl Color -- Suno AI Music Generation (direct, no proxy).
Calls Suno's internal API directly using session cookies.
Star IS the API proxy. No separate deployment needed.
Auth flow: cookie -> Clerk session refresh -> studio-api.suno.ai
# 🎵💀 DIRECT NEURAL AUDIO PIPELINE
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
from typing import Any, TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
# Suno internal endpoints # 🔥
_STUDIO_API = "https://studio-api.suno.ai"
_CLERK_BASE = "https://clerk.suno.com"
_CLERK_JS_VERSION = "5.56.0-snapshot.v20250312181236"
# Polling config
_POLL_INTERVAL = 5.0
_POLL_TIMEOUT = 300.0 # 5 min max
TOOL_NAME = "generate_suno_music"
TOOL_DESCRIPTION = (
"Generate a music track using Suno AI. Creates full songs "
"with vocals, instrumentals, or ambient soundscapes. "
"DEFAULT STYLE (always use unless user overrides): "
"gangster rap, chiptune, dubstep, heavy bass, distortion, "
"reverb, beatmasher, bitcrusher, glitchcore vocal effects, "
"8-bit, pokemon, undertale, sonic, gameboy, nerdcore rap. "
"DEFAULT VOCALS: female. "
"For YouTube URLs, use the auto_cover tool instead."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": (
"Text description of the music to generate. "
"E.g. '8-bit chiptune battle theme, intense'"
),
},
"style": {
"type": "string",
"description": (
"Music style tags. DEFAULT (use this if user does "
"not specify): 'gangster rap, chiptune, dubstep, "
"heavy bass, distortion, reverb, beatmasher, "
"bitcrusher, glitchcore vocal effects, 8-bit, "
"pokemon, undertale, sonic, gameboy, nerdcore rap'"
),
},
"title": {
"type": "string",
"description": "Optional title for the generated track.",
},
"lyrics": {
"type": "string",
"description": (
"Optional lyrics. Use [Verse], [Chorus], [Bridge] "
"tags. Leave empty for instrumental."
),
},
"instrumental": {
"type": "boolean",
"description": "If true, instrumental only (no vocals).",
},
"save_as": {
"type": "string",
"description": "Optional name to save as a game asset.",
},
"source_url": {
"type": "string",
"description": (
"URL of an existing audio track to cover/remix. "
"When provided, Suno will restyle the track while "
"preserving the melody. Use with 'style' param."
),
},
"weirdness": {
"type": "integer",
"description": "Weirdness 0-100. Default 64.",
},
"style_influence": {
"type": "integer",
"description": "Style influence 0-100. Default 64.",
},
"audio_influence": {
"type": "integer",
"description": "Audio influence 0-100. Default 28.",
},
},
"required": ["prompt"],
}
# ------------------------------------------------------------------
# Cookie + Clerk auth # 🕷️
# ------------------------------------------------------------------
def _get_cookie(ctx: ToolContext | None) -> str:
"""Get the Suno session cookie.
Checks in order:
1. config.api_keys.suno_cookie
2. SUNO_COOKIE env var
3. .suno_cookie file in project root
"""
# 1. From config
if ctx is not None:
config = getattr(ctx, "config", None)
if config is not None:
api_keys = getattr(config, "api_keys", {})
cookie = api_keys.get("suno_cookie", "")
if cookie:
return cookie
# 2. From env var
env_cookie = os.environ.get("SUNO_COOKIE", "")
if env_cookie:
return env_cookie
# 3. From file (easiest for shell tool) # 💀
cookie_paths = [
os.path.join(os.path.dirname(__file__), "..", ".suno_cookie"),
os.path.expanduser("~/.suno_cookie"),
]
for path in cookie_paths:
try:
with open(path, "r", encoding="utf-8") as f:
cookie = f.read().strip()
if cookie:
return cookie
except FileNotFoundError:
continue
return ""
def _extract_session_id(cookie: str) -> str:
"""Extract the active Clerk session ID from the cookie string.
Cookies may contain multiple __client_uat_<sid> entries.
We pick the one with the highest (most recent) value,
which is the active session.
"""
# Find all __client_uat_<sid>=<timestamp> pairs
matches = re.findall(
r"__client_uat_([A-Za-z0-9_-]+)=(\d+)", cookie,
)
if matches:
# Pick the session with the highest timestamp (active)
best = max(matches, key=lambda m: int(m[1]))
if int(best[1]) > 0:
return best[0]
# Fallback: look for __client_<sid>=eyJ pattern
match = re.search(r"__client_([A-Za-z0-9_-]+)=eyJ", cookie)
if match:
return match.group(1)
return ""
async def _refresh_token(cookie: str) -> str | None:
"""Refresh the Suno session token via Clerk.
Returns a fresh JWT bearer token for studio-api requests.
"""
session_id = _extract_session_id(cookie)
if not session_id:
logger.error("Could not extract Clerk session ID from cookie")
return None
token_url = (
f"{_CLERK_BASE}/v1/client/sessions/{session_id}/tokens"
f"?_clerk_js_version={_CLERK_JS_VERSION}"
)
headers = {
"Cookie": cookie,
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36"
),
"Origin": "https://suno.com",
"Referer": "https://suno.com/",
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
token_url,
headers=headers,
data="", # Empty POST body
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
if resp.status != 200:
err = await resp.text()
logger.error(
"Clerk token refresh failed (%s): %s",
resp.status, err[:300],
)
# Surface the actual error so Star can report it
return f"CLERK_ERROR:{resp.status}:{err[:300]}"
data = await resp.json()
return data.get("jwt", None)
except Exception as exc:
logger.error("Clerk token refresh error: %s", exc)
return None
# ------------------------------------------------------------------
# Main tool # 🎵
# ------------------------------------------------------------------
[docs]
async def run(
prompt: str,
style: str = "",
title: str = "",
lyrics: str = "",
source_url: str = "",
instrumental: bool = False,
save_as: str = "",
weirdness: int = 64,
style_influence: int = 64,
audio_influence: int = 28,
ctx: ToolContext | None = None,
) -> str:
"""Generate music via Suno's internal API."""
cookie = await asyncio.to_thread(_get_cookie, ctx)
if not cookie:
return json.dumps({
"error": "No Suno cookie configured. "
"Set SUNO_COOKIE env var or config.api_keys.suno_cookie",
})
if ctx is None:
return json.dumps({"error": "No tool context available."})
channel_id = str(ctx.channel_id)
# Step 0: Get a fresh JWT via Clerk # 🔥
jwt = await _refresh_token(cookie)
if not jwt:
return json.dumps({
"error": "Failed to refresh Suno session token. "
"Cookie may be expired -- re-grab from browser.",
})
if isinstance(jwt, str) and jwt.startswith("CLERK_ERROR:"):
return json.dumps({
"error": f"Clerk auth failed: {jwt}",
"hint": "Cookie may be expired or session suspended.",
})
api_headers = {
"Authorization": f"Bearer {jwt}",
"Content-Type": "application/json",
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36"
),
"Origin": "https://suno.com",
"Referer": "https://suno.com/",
}
# Suno slider values (convert 0-100 int to 0.0-1.0 float) # 🎵
suno_weirdness = max(0.0, min(1.0, weirdness / 100.0))
suno_style_inf = max(0.0, min(1.0, style_influence / 100.0))
suno_audio_inf = max(0.0, min(1.0, audio_influence / 100.0))
# Step 1: Create the music # 💀
if source_url:
# Cover/Remix mode -- upload source, restyle it
endpoint = f"{_STUDIO_API}/api/generate/v2/"
payload: dict[str, Any] = {
"prompt": lyrics if lyrics else "",
"tags": style or prompt,
"title": title or f"Cover - {prompt[:50]}",
"make_instrumental": instrumental,
"mv": "chirp-v4",
"generation_type": "TEXT",
"continue_clip_id": None,
"upload_url": source_url,
"weirdness": suno_weirdness,
"cfg_coef": suno_style_inf,
}
elif lyrics or style or title:
# Custom mode
endpoint = f"{_STUDIO_API}/api/generate/v2/"
payload = {
"prompt": lyrics if lyrics else "",
"tags": style or prompt,
"title": title or prompt[:60],
"make_instrumental": instrumental,
"mv": "chirp-v4",
"generation_type": "TEXT",
"weirdness": suno_weirdness,
"cfg_coef": suno_style_inf,
}
if not lyrics:
payload["prompt"] = ""
payload["make_instrumental"] = True
payload["gpt_description_prompt"] = prompt
else:
# Simple mode
endpoint = f"{_STUDIO_API}/api/generate/v2/"
payload = {
"gpt_description_prompt": prompt,
"make_instrumental": instrumental,
"mv": "chirp-v4",
"weirdness": suno_weirdness,
"cfg_coef": suno_style_inf,
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
endpoint,
headers=api_headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30),
) as resp:
if resp.status != 200:
error_text = await resp.text()
return json.dumps({
"error": f"Suno API error ({resp.status}): "
f"{error_text[:500]}",
})
create_data = await resp.json()
except Exception as exc:
return json.dumps({"error": f"Music creation failed: {exc}"})
# Extract clip IDs # 🕷️
clip_ids = []
clips_data = create_data.get("clips", [])
if isinstance(clips_data, list):
for clip in clips_data:
if isinstance(clip, dict) and "id" in clip:
clip_ids.append(clip["id"])
if not clip_ids:
return json.dumps({
"error": "No clip IDs returned.",
"raw": str(create_data)[:300],
})
# Step 2: Poll /api/feed/ for completion # 🌀
audio_url = None
clip_title = ""
image_url = ""
elapsed = 0.0
while elapsed < _POLL_TIMEOUT:
await asyncio.sleep(_POLL_INTERVAL)
elapsed += _POLL_INTERVAL
# Refresh token periodically (every 60s)
if int(elapsed) % 60 == 0 and elapsed > 0:
new_jwt = await _refresh_token(cookie)
if new_jwt:
jwt = new_jwt
api_headers["Authorization"] = f"Bearer {jwt}"
feed_url = (
f"{_STUDIO_API}/api/feed/"
f"?ids={','.join(clip_ids)}"
)
try:
async with aiohttp.ClientSession() as session:
async with session.get(
feed_url,
headers=api_headers,
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
if resp.status != 200:
continue
feed_data = await resp.json()
if not isinstance(feed_data, list):
continue
for clip in feed_data:
if not isinstance(clip, dict):
continue
status = clip.get("status", "")
if status == "complete":
audio_url = clip.get("audio_url", "")
clip_title = clip.get("title", "")
image_url = clip.get("image_url", "")
break
elif status in ("error", "failed"):
meta = clip.get("metadata", {})
return json.dumps({
"error": f"Generation failed: {meta}",
})
if audio_url:
break
except Exception:
continue
if not audio_url:
return json.dumps({
"error": "Music generation timed out (5 min). "
"Try again or check Suno status.",
})
# Step 3: Download and send # 🎮
try:
async with aiohttp.ClientSession() as session:
async with session.get(
audio_url,
timeout=aiohttp.ClientTimeout(total=60),
) as resp:
if resp.status != 200:
return json.dumps({
"error": f"Download failed ({resp.status}).",
})
audio_data = await resp.read()
safe_name = (
save_as or clip_title or "suno_track"
)[:30].replace(" ", "_")
fname = f"{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: {exc}"})
# Save as game asset if requested # 💾
if save_as:
try:
from game_session import get_session
from game_assets import upload_asset
session_obj = get_session(channel_id)
if session_obj and session_obj.active:
redis = getattr(ctx, "redis", None)
await upload_asset(
game_id=session_obj.game_id,
name=save_as,
category="special",
url=audio_url,
user_id=str(getattr(ctx, "user_id", "")),
turn=session_obj.turn_number,
redis=redis,
)
except Exception as exc:
logger.warning("Failed to save music asset: %s", exc)
result_info: dict = {
"success": True,
"title": clip_title or title or prompt[:50],
"style": style or "auto",
"audio_url": audio_url,
"image_url": image_url,
"filename": fname,
"instrumental": instrumental,
"saved_as": save_as if save_as else None,
}
if file_url:
result_info["file_url"] = file_url
return json.dumps(result_info)