"""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 jsonutil as 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:
"""Resolve the Suno browser session cookie from config, env, or disk.
Supplies the raw Clerk session cookie that authenticates every Suno request,
since this tool talks to Suno's internal API directly (Star is the proxy)
rather than through an official key. It is a blocking, filesystem-touching
accessor, which is why :func:`run` calls it via ``asyncio.to_thread``.
It checks three sources in order and returns the first non-empty match: the
ToolContext config at ``ctx.config.api_keys.suno_cookie``, the ``SUNO_COOKIE``
environment variable, then a ``.suno_cookie`` file in the project root or the
user's home directory. No Redis, network, or LLM work.
Called only by :func:`run`.
Args:
ctx (ToolContext | None): Execution context; only
``ctx.config.api_keys`` is consulted.
Returns:
str: The cookie string, or an empty string when none is configured.
"""
# 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:
"""Parse the active Clerk session ID out of the raw cookie string.
A Suno cookie can hold several ``__client_uat_<sid>`` entries (one per past
session), so this picks the right one before a token refresh by selecting the
session whose Unix-timestamp value is highest, i.e. the most recently active.
If no positive-timestamp entry is found it falls back to the
``__client_<sid>=eyJ`` JWT-prefixed pattern. Pure regex parsing with no Redis,
network, or LLM work.
Called only by :func:`_refresh_token`.
Args:
cookie (str): The full Suno session cookie string.
Returns:
str: The resolved Clerk session ID, or an empty string when none can be
extracted.
"""
# 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:
"""Exchange the session cookie for a fresh Clerk JWT bearer token.
Mints the short-lived JWT that authorizes ``studio-api.suno.ai`` requests,
so the caller can mint a token at the start of generation and re-mint it
periodically during long polls before it expires. It first resolves the
session ID via :func:`_extract_session_id`, then makes an HTTP POST to the
Clerk tokens endpoint with browser-like headers via an ``aiohttp`` session
(15s timeout). No Redis, filesystem, or LLM work.
Called by :func:`run` — once before generation and again roughly every 60
seconds while polling for completion.
Args:
cookie (str): The full Suno session cookie string.
Returns:
str | None: The fresh JWT on success; ``None`` when the session ID
cannot be extracted or the request raises; or a sentinel string of
the form ``CLERK_ERROR:{status}:{body}`` when Clerk returns a
non-200 response, so :func:`run` can surface the real error.
"""
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 a music track end-to-end via Suno's internal studio API.
The entry point for the ``generate_suno_music`` tool. It authenticates,
submits a generation request, polls until the track is rendered, then
downloads the audio and delivers it to the originating channel. Three request
shapes are chosen automatically from the arguments: cover/remix mode when a
``source_url`` is given, custom mode when lyrics/style/title are supplied, and
simple description mode otherwise; the 0-100 ``weirdness`` /
``style_influence`` / ``audio_influence`` sliders are normalized to Suno's
0.0-1.0 ``cfg_coef`` and ``weirdness`` parameters.
Side effects span auth, HTTP, the platform, and optionally the game session.
It pulls the cookie via :func:`_get_cookie` (off-thread) and a JWT via
:func:`_refresh_token` (re-refreshing roughly every 60s mid-poll), then drives
``aiohttp`` calls against ``studio-api.suno.ai`` — POST ``/api/generate/v2/``
to create clips and GET ``/api/feed/`` to poll (up to a 5-minute timeout) for
a ``complete`` status and ``audio_url``. On completion it downloads the MP3
and sends it through ``ctx.adapter.send_file``, appending the bytes to
``ctx.sent_files``. When ``save_as`` is set and a game session is active it
registers the track as a game asset via :func:`game_assets.upload_asset`
(passing ``ctx.redis``). It does no LLM work.
Invoked through the tool dispatch registry (this module uses the single-tool
``TOOL_NAME`` / ``run`` format loaded by ``tool_loader.py``); the ``auto_cover``
tool also instructs the model to call ``generate_suno_music`` as a follow-up
step.
Args:
prompt (str): Text description of the music to generate.
style (str): Optional Suno style tags.
title (str): Optional track title.
lyrics (str): Optional lyrics with ``[Verse]`` / ``[Chorus]`` tags; empty
implies instrumental.
source_url (str): Optional existing-track URL for cover/remix mode.
instrumental (bool): When ``True``, suppress vocals.
save_as (str): Optional asset name; when set and a game is active, the
track is saved as a game asset.
weirdness (int): 0-100 weirdness slider (default 64).
style_influence (int): 0-100 style-influence slider (default 64).
audio_influence (int): 0-100 audio-influence slider (default 28).
ctx (ToolContext | None): Execution context; ``channel_id``, ``adapter``,
``sent_files``, ``config``, ``user_id``, and ``redis`` are used.
Returns:
str: A JSON string with the track metadata and delivered file info on
success, or an ``{"error": ...}`` payload on any failure (missing
cookie/context, auth failure, Suno API error, no clips, timeout, or a
download/send error).
"""
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)