diff --git a/intentkit/tools/music/__init__.py b/intentkit/tools/music/__init__.py new file mode 100644 index 00000000..3504ae5d --- /dev/null +++ b/intentkit/tools/music/__init__.py @@ -0,0 +1,44 @@ +"""Music generation tools.""" + +from collections.abc import Callable + +from intentkit.config.config import config as system_config +from intentkit.tools.meta import ToolsetMeta +from intentkit.tools.music.base import MusicBaseTool +from intentkit.tools.music.minimax import MiniMaxMusicGeneration + +toolset = ToolsetMeta( + title="Music Generation", + description="Generate songs from prompts and lyrics.", + tags=["AI", "Audio"], + icon="/tools/music/music.svg", +) + +_cache: dict[str, MusicBaseTool] = {} + +_TOOL_NAME_TO_CLASS: dict[str, Callable[[], MusicBaseTool]] = { + "music_minimax_generate": MiniMaxMusicGeneration, +} + + +async def get_tools(tool_names: list[str], **_) -> list[MusicBaseTool]: + """Return requested music generation tools and skip unknown names.""" + return [tool for name in tool_names if (tool := get_music_tool(name))] + + +def get_music_tool(tool_name: str) -> MusicBaseTool | None: + """Get a cached music generation tool by name.""" + if tool_name in _cache: + return _cache[tool_name] + + tool_class = _TOOL_NAME_TO_CLASS.get(tool_name) + if tool_class is None: + return None + + _cache[tool_name] = tool_class() + return _cache[tool_name] + + +def available() -> bool: + """Check whether the music generation API is configured.""" + return bool(system_config.minimax_plan_api_key) diff --git a/intentkit/tools/music/base.py b/intentkit/tools/music/base.py new file mode 100644 index 00000000..60d7337e --- /dev/null +++ b/intentkit/tools/music/base.py @@ -0,0 +1,18 @@ +"""Base class for music generation tools.""" + +from langchain_core.tools.base import ToolException + +from intentkit.config.config import config +from intentkit.tools.base import IntentKitTool + + +class MusicBaseTool(IntentKitTool): + """Shared configuration for music generation tools.""" + + category: str = "music" + + def get_api_key(self) -> str: + """Return the configured API key or raise a tool error.""" + if not config.minimax_plan_api_key: + raise ToolException("Music generation API key is not configured") + return config.minimax_plan_api_key diff --git a/intentkit/tools/music/minimax.py b/intentkit/tools/music/minimax.py new file mode 100644 index 00000000..c960bceb --- /dev/null +++ b/intentkit/tools/music/minimax.py @@ -0,0 +1,187 @@ +"""MiniMax music generation tool.""" + +import json +from typing import Any, Literal + +import httpx +from langchain_core.tools import ArgsSchema +from langchain_core.tools.base import ToolException +from pydantic import BaseModel, Field, model_validator + +from intentkit.tools.music.base import MusicBaseTool + +_ENDPOINTS = { + "global_en": "https://api.minimax.io/v1/music_generation", + "cn_zh": "https://api.minimaxi.com/v1/music_generation", +} + +MusicModel = Literal[ + "music-3.0", + "music-2.6", + "music-3.0-free", + "music-2.6-free", +] +MusicRegion = Literal["global_en", "cn_zh"] +MusicOutputFormat = Literal["url", "hex"] +MusicAudioFormat = Literal["mp3", "wav", "pcm"] + + +class MusicAudioSetting(BaseModel): + """Audio encoding options for generated music.""" + + sample_rate: Literal[16000, 24000, 32000, 44100] = 44100 + bitrate: Literal[32000, 64000, 128000, 256000] = 256000 + format: MusicAudioFormat = "mp3" + + +class MiniMaxMusicGenerationInput(BaseModel): + """Input schema for text-to-music generation.""" + + model: MusicModel = Field(default="music-3.0", description="Generation model") + prompt: str | None = Field( + default=None, + max_length=2000, + description="Music style, mood, and scenario", + ) + lyrics: str | None = Field( + default=None, + min_length=1, + max_length=3500, + description="Song lyrics with newline-separated sections", + ) + stream: bool = Field(default=False, description="Stream hex audio chunks") + output_format: MusicOutputFormat = Field( + default="url", + description="Return a 24-hour URL or hex-encoded audio", + ) + audio_setting: MusicAudioSetting = Field(default_factory=MusicAudioSetting) + lyrics_optimizer: bool = Field( + default=False, + description="Generate lyrics from the prompt when lyrics are omitted", + ) + is_instrumental: bool = Field( + default=False, + description="Generate music without vocals", + ) + region: MusicRegion = Field( + default="global_en", + description="Use the global or China API endpoint", + ) + aigc_watermark: bool | None = Field( + default=None, + description="Append the China-region audio watermark", + ) + + @model_validator(mode="after") + def validate_generation_options(self): + """Validate conditional API requirements.""" + if self.stream and self.output_format != "hex": + raise ValueError("streaming requires hex output") + if self.region != "cn_zh" and self.aigc_watermark is not None: + raise ValueError("aigc_watermark is only available in the China region") + if self.is_instrumental and not self.prompt: + raise ValueError("instrumental generation requires a prompt") + if not self.is_instrumental and not self.lyrics and not self.lyrics_optimizer: + raise ValueError("vocal generation requires lyrics or lyrics optimization") + if self.lyrics_optimizer and not self.prompt: + raise ValueError("lyrics optimization requires a prompt") + return self + + +def _parse_payload(payload: dict[str, Any]) -> tuple[str, int]: + """Validate a response payload and return its audio and status.""" + base_response = payload.get("base_resp") or {} + if base_response.get("status_code") != 0: + message = base_response.get("status_msg") or "unknown API error" + raise ToolException(f"Music generation failed: {message}") + + data = payload.get("data") or {} + status = data.get("status") + audio = data.get("audio") + if status not in (1, 2) or not isinstance(audio, str) or not audio: + raise ToolException("Music generation returned no audio") + return audio, status + + +class MiniMaxMusicGeneration(MusicBaseTool): + """Generate music using MiniMax Music models.""" + + name: str = "music_minimax_generate" + title: str = "MiniMax Music Generation" + description: str = ( + "Generate instrumental or vocal music from a prompt and optional lyrics. " + "Supports global and China endpoints, URL or hex output, MP3, WAV, and PCM." + ) + args_schema: ArgsSchema | None = MiniMaxMusicGenerationInput + + async def _arun( + self, + model: MusicModel = "music-3.0", + prompt: str | None = None, + lyrics: str | None = None, + stream: bool = False, + output_format: MusicOutputFormat = "url", + audio_setting: MusicAudioSetting | dict[str, Any] | None = None, + lyrics_optimizer: bool = False, + is_instrumental: bool = False, + region: MusicRegion = "global_en", + aigc_watermark: bool | None = None, + **_, + ) -> dict[str, Any]: + """Generate music and parse the completed audio response.""" + settings = MusicAudioSetting.model_validate(audio_setting or {}) + body: dict[str, Any] = { + "model": model, + "stream": stream, + "output_format": output_format, + "audio_setting": settings.model_dump(), + "lyrics_optimizer": lyrics_optimizer, + "is_instrumental": is_instrumental, + } + if prompt is not None: + body["prompt"] = prompt + if lyrics is not None: + body["lyrics"] = lyrics + if region == "cn_zh" and aigc_watermark is not None: + body["aigc_watermark"] = aigc_watermark + + headers = { + "Authorization": f"Bearer {self.get_api_key()}", + "Content-Type": "application/json", + } + + try: + async with httpx.AsyncClient(timeout=300) as client: + if not stream: + response = await client.post( + _ENDPOINTS[region], json=body, headers=headers + ) + response.raise_for_status() + audio, status = _parse_payload(response.json()) + else: + chunks: list[str] = [] + status = 1 + async with client.stream( + "POST", _ENDPOINTS[region], json=body, headers=headers + ) as response: + response.raise_for_status() + async for line in response.aiter_lines(): + line = line.removeprefix("data:").strip() + if not line or line == "[DONE]": + continue + chunk, status = _parse_payload(json.loads(line)) + chunks.append(chunk) + audio = "".join(chunks) + if not audio: + raise ToolException("Music generation returned no audio") + + return { + "status": status, + "output_format": output_format, + "audio_url" if output_format == "url" else "audio_hex": audio, + "url_ttl_hours": 24 if output_format == "url" else None, + } + except ToolException: + raise + except (httpx.HTTPError, json.JSONDecodeError) as error: + raise ToolException(f"Music generation API error: {error}") from error diff --git a/intentkit/tools/music/music.svg b/intentkit/tools/music/music.svg new file mode 100644 index 00000000..bce20fb2 --- /dev/null +++ b/intentkit/tools/music/music.svg @@ -0,0 +1,4 @@ + + + + diff --git a/tests/tools/test_music.py b/tests/tools/test_music.py new file mode 100644 index 00000000..4abcd72d --- /dev/null +++ b/tests/tools/test_music.py @@ -0,0 +1,135 @@ +"""Tests for music generation tools.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from langchain_core.tools.base import ToolException +from pydantic import ValidationError + +from intentkit.tools.music import available, get_tools +from intentkit.tools.music.minimax import ( + MiniMaxMusicGeneration, + MiniMaxMusicGenerationInput, +) + + +def test_input_schema_validates_conditional_fields(): + instrumental = MiniMaxMusicGenerationInput( + prompt="ambient piano", is_instrumental=True + ) + assert instrumental.model == "music-3.0" + assert instrumental.output_format == "url" + + optimized = MiniMaxMusicGenerationInput(prompt="upbeat pop", lyrics_optimizer=True) + assert optimized.lyrics is None + + with pytest.raises(ValidationError, match="requires lyrics"): + MiniMaxMusicGenerationInput() + with pytest.raises(ValidationError, match="China region"): + MiniMaxMusicGenerationInput( + prompt="ambient piano", + is_instrumental=True, + aigc_watermark=True, + ) + with pytest.raises(ValidationError, match="hex output"): + MiniMaxMusicGenerationInput( + prompt="ambient piano", is_instrumental=True, stream=True + ) + + +@pytest.mark.asyncio +async def test_get_tools_selects_by_name(): + tools = await get_tools(["music_minimax_generate", "music_unknown"]) + assert [tool.name for tool in tools] == ["music_minimax_generate"] + assert await get_tools([]) == [] + + +def test_available_uses_configured_key(): + with patch("intentkit.tools.music.system_config") as config: + config.minimax_plan_api_key = "test-key" + assert available() is True + config.minimax_plan_api_key = None + assert available() is False + + +@pytest.mark.asyncio +async def test_generation_uses_global_endpoint_and_parses_url(): + response = MagicMock() + response.json.return_value = { + "base_resp": {"status_code": 0}, + "data": {"status": 2, "audio": "https://example.test/song.mp3"}, + } + client = AsyncMock() + client.post.return_value = response + context = AsyncMock() + context.__aenter__.return_value = client + + with ( + patch("intentkit.tools.music.base.config") as config, + patch("intentkit.tools.music.minimax.httpx.AsyncClient", return_value=context), + ): + config.minimax_plan_api_key = "test-key" + result = await MiniMaxMusicGeneration()._arun( + prompt="ambient piano", is_instrumental=True + ) + + assert result == { + "status": 2, + "output_format": "url", + "audio_url": "https://example.test/song.mp3", + "url_ttl_hours": 24, + } + call = client.post.call_args + assert call.args[0] == "https://api.minimax.io/v1/music_generation" + assert call.kwargs["headers"]["Authorization"] == "Bearer test-key" + assert call.kwargs["json"]["model"] == "music-3.0" + assert call.kwargs["json"]["audio_setting"] == { + "sample_rate": 44100, + "bitrate": 256000, + "format": "mp3", + } + + +@pytest.mark.asyncio +async def test_generation_uses_china_endpoint_and_watermark(): + response = MagicMock() + response.json.return_value = { + "base_resp": {"status_code": 0}, + "data": {"status": 2, "audio": "494433"}, + } + client = AsyncMock() + client.post.return_value = response + context = AsyncMock() + context.__aenter__.return_value = client + + with ( + patch("intentkit.tools.music.base.config") as config, + patch("intentkit.tools.music.minimax.httpx.AsyncClient", return_value=context), + ): + config.minimax_plan_api_key = "test-key" + result = await MiniMaxMusicGeneration()._arun( + prompt="ambient piano", + is_instrumental=True, + region="cn_zh", + aigc_watermark=True, + output_format="hex", + ) + + assert result["audio_hex"] == "494433" + call = client.post.call_args + assert call.args[0] == "https://api.minimaxi.com/v1/music_generation" + assert call.kwargs["json"]["aigc_watermark"] is True + + +def test_api_errors_raise_tool_exception(): + from intentkit.tools.music.minimax import _parse_payload + + with pytest.raises(ToolException, match="insufficient balance"): + _parse_payload( + { + "base_resp": { + "status_code": 1008, + "status_msg": "insufficient balance", + } + } + )