gh-christianlouis-quizzicalbeats/musicround/mcp_server.py

"""MCP server for agentic Quizzical Beats workflows.

Run with:
    python -m musicround.mcp_server
"""

from __future__ import annotations

from functools import lru_cache
from typing import Any

from mcp.server.fastmcp import FastMCP

from musicround import create_app
from musicround.services import automation


mcp = FastMCP("Quizzical Beats")


@lru_cache(maxsize=1)
def _app():
    """Create the Flask app once for the MCP server process."""
    return create_app()


def _with_app_context(func, *args, **kwargs) -> dict[str, Any]:
    """Run a service function inside the Quizzical Beats app context."""
    app = _app()
    with app.app_context():
        return func(*args, **kwargs)


@mcp.tool()
def find_songs(
    query: str | None = None,
    title: str | None = None,
    artist: str | None = None,
    spotify_id: str | None = None,
    deezer_id: str | None = None,
    isrc: str | None = None,
    limit: int = 20,
) -> dict[str, Any]:
    """Search the local Quizzical Beats catalog before adding a song."""
    return _with_app_context(
        automation.find_songs,
        query=query,
        title=title,
        artist=artist,
        spotify_id=spotify_id,
        deezer_id=deezer_id,
        isrc=isrc,
        limit=limit,
    )


@mcp.tool()
def add_song(
    title: str,
    artist: str,
    album_name: str | None = None,
    genre: str | None = None,
    year: int | None = None,
    preview_url: str | None = None,
    cover_url: str | None = None,
    spotify_id: str | None = None,
    deezer_id: str | None = None,
    isrc: str | None = None,
    tags: list[str] | None = None,
    source: str = "manual",
) -> dict[str, Any]:
    """Add a song to Quizzical Beats if it is not already present."""
    return _with_app_context(
        automation.add_song,
        title=title,
        artist=artist,
        album_name=album_name,
        genre=genre,
        year=year,
        preview_url=preview_url,
        cover_url=cover_url,
        spotify_id=spotify_id,
        deezer_id=deezer_id,
        isrc=isrc,
        tags=tags,
        source=source,
    )


@mcp.tool()
def import_catalog_item(
    service_name: str,
    item_type: str,
    item_id_or_url: str,
    user_id: int | None = None,
) -> dict[str, Any]:
    """Import a Spotify or Deezer track, album, or playlist into the catalog."""
    return _with_app_context(
        automation.import_catalog_item,
        service_name=service_name,
        item_type=item_type,
        item_id_or_url=item_id_or_url,
        user_id=user_id,
    )


@mcp.tool()
def compile_round(
    name: str | None = None,
    round_type: str = "random",
    count: int = 8,
    criteria: str | None = None,
    song_ids: list[int] | None = None,
) -> dict[str, Any]:
    """Compile songs into a named quiz round."""
    return _with_app_context(
        automation.create_round,
        name=name,
        round_type=round_type,
        count=count,
        criteria=criteria,
        song_ids=song_ids,
    )


@mcp.tool()
def rename_round(round_id: int, name: str | None) -> dict[str, Any]:
    """Set or clear the display name for a round."""
    return _with_app_context(automation.rename_round, round_id=round_id, name=name)


@mcp.tool()
def create_round_from_playlist(
    service_name: str,
    playlist_id_or_url: str,
    name: str | None = None,
    count: int = 8,
    user_id: int | None = None,
) -> dict[str, Any]:
    """Import a Spotify or Deezer playlist and turn it into a quiz round."""
    return _with_app_context(
        automation.create_round_from_playlist,
        service_name=service_name,
        playlist_id_or_url=playlist_id_or_url,
        name=name,
        count=count,
        user_id=user_id,
    )


@mcp.tool()
def generate_round_assets(
    round_id: int,
    user_id: int | None = None,
    include_pdf: bool = True,
    include_mp3: bool = True,
) -> dict[str, Any]:
    """Generate the PDF and/or MP3 files for a round."""
    return _with_app_context(
        automation.generate_round_assets,
        round_id=round_id,
        user_id=user_id,
        include_pdf=include_pdf,
        include_mp3=include_mp3,
    )


@mcp.tool()
def inspect_round_mp3(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
    """Check a round MP3 for duration, loudness, clipping, and silence issues."""
    return _with_app_context(automation.inspect_mp3_quality, path=path, round_id=round_id)


@mcp.tool()
def inspect_round_pdf(path: str | None = None, round_id: int | None = None) -> dict[str, Any]:
    """Check that a round PDF exists and has a valid basic PDF structure."""
    return _with_app_context(automation.inspect_pdf_quality, path=path, round_id=round_id)


@mcp.tool()
def send_round_email(
    round_id: int,
    recipient: str | None = None,
    user_id: int | None = None,
    subject: str | None = None,
    body_text: str | None = None,
) -> dict[str, Any]:
    """Generate assets and email the completed round bundle."""
    return _with_app_context(
        automation.email_round,
        round_id=round_id,
        recipient=recipient,
        user_id=user_id,
        subject=subject,
        body_text=body_text,
    )


@mcp.tool()
def generate_tts_snippet(
    user_id: int,
    mp3_type: str,
    text: str,
    service: str = "openai",
    voice: str | None = None,
    model: str | None = None,
    stability: float | None = None,
    similarity: float | None = None,
) -> dict[str, Any]:
    """Generate and assign a custom intro, replay, or outro TTS MP3."""
    return _with_app_context(
        automation.generate_tts_snippet,
        user_id=user_id,
        mp3_type=mp3_type,
        text=text,
        service=service,
        voice=voice,
        model=model,
        stability=stability,
        similarity=similarity,
    )


if __name__ == "__main__":
    mcp.run()