Add MiniMax music generation tool

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)

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

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

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",
+                }
+            }
+        )