music-assistant-server

1"""Metadata: lyrics, recommendations, similar tracks, refresh."""
2# ruff: noqa: TID252  -- relative imports are the canonical MA-provider pattern.
3
4from __future__ import annotations
5
6from typing import TYPE_CHECKING
7
8from fastmcp import Context, FastMCP
9from mcp.types import ToolAnnotations
10from music_assistant_models.enums import MediaType
11
12from ..models import RecommendationFolderBrief, RecommendationItemBrief, TrackBrief
13from ..tags import Tag
14from ._common import TIMEOUT_QUERY, page_args, resolve_typed_uri, to_brief_track
15
16if TYPE_CHECKING:
17    from music_assistant.mass import MusicAssistant
18
19
20def _readonly(title: str) -> ToolAnnotations:
21    """Read-only metadata tool annotations with the supplied UI title."""
22    return ToolAnnotations(
23        title=title,
24        readOnlyHint=True,
25        destructiveHint=False,
26        idempotentHint=True,
27        openWorldHint=False,
28    )
29
30
31def build_metadata_server(mass: MusicAssistant) -> FastMCP:
32    """Construct the ``metadata/*`` sub-server."""
33    sub: FastMCP = FastMCP(name="metadata")
34
35    @sub.tool(
36        name="recommendations",
37        tags={Tag.QUERY_METADATA},
38        annotations=ToolAnnotations(
39            title="Recommendations",
40            readOnlyHint=True,
41            destructiveHint=False,
42            idempotentHint=True,
43            openWorldHint=False,
44        ),
45        timeout=TIMEOUT_QUERY,
46    )  # type: ignore[untyped-decorator, unused-ignore]
47    async def recommendation_rows(
48        ctx: Context | None = None,
49    ) -> list[RecommendationFolderBrief]:
50        """
51        Return Music Assistant's curated recommendation rows, without items.
52
53        Each ``RecommendationFolderBrief`` has a ``name`` plus the
54        ``provider`` / ``item_id`` pair identifying the row; pass both to
55        ``recommendation_items`` to fetch the row's items.
56        """
57        if ctx is not None:
58            await ctx.info("Fetching MA curated recommendations…")
59        folders = await mass.music.recommendations.get_recommendations()
60        return [
61            RecommendationFolderBrief(
62                name=str(getattr(folder, "name", "")),
63                provider=str(getattr(folder, "provider", "")),
64                item_id=str(getattr(folder, "item_id", "")),
65            )
66            for folder in folders
67        ]
68
69    @sub.tool(
70        tags={Tag.QUERY_METADATA},
71        annotations=_readonly("Recommendation items"),
72        timeout=TIMEOUT_QUERY,
73    )  # type: ignore[untyped-decorator, unused-ignore]
74    async def recommendation_items(provider: str, item_id: str) -> list[RecommendationItemBrief]:
75        """
76        Return the items of a single recommendation row.
77
78        Each ``RecommendationItemBrief`` has a ``uri`` that can be passed to
79        ``play_media`` or to the matching ``library_get_*_by_uri`` tool to
80        inspect further.
81
82        :param provider: The row's ``provider``, as returned by ``recommendations``.
83        :param item_id: The row's ``item_id``, as returned by ``recommendations``.
84        """
85        items = await mass.music.recommendations.get_recommendation_items(provider, item_id)
86        result: list[RecommendationItemBrief] = []
87        for it in items:
88            media_type = getattr(it, "media_type", None)
89            result.append(
90                RecommendationItemBrief(
91                    uri=str(getattr(it, "uri", "")),
92                    name=str(getattr(it, "name", "")),
93                    media_type=str(getattr(media_type, "value", media_type))
94                    if media_type
95                    else None,
96                )
97            )
98        return result
99
100    @sub.tool(
101        tags={Tag.QUERY_METADATA},
102        annotations=ToolAnnotations(
103            title="Recently played tracks",
104            readOnlyHint=True,
105            destructiveHint=False,
106            idempotentHint=True,
107            openWorldHint=False,
108        ),
109        timeout=TIMEOUT_QUERY,
110    )  # type: ignore[untyped-decorator, unused-ignore]
111    async def recently_played(limit: int = 10) -> list[TrackBrief]:
112        """
113        Return the user's most recently played tracks, newest first.
114
115        Returns ``TrackBrief`` items. Items without a resolved name are
116        filtered out.
117
118        :param limit: Max results to return (clamped to ``[1, 200]``).
119        """
120        _, limit = page_args(0, limit)
121        items = await mass.music.recently_played(limit=limit)
122        return [to_brief_track(it) for it in items if getattr(it, "name", None)]
123
124    @sub.tool(
125        tags={Tag.QUERY_METADATA},
126        annotations=_readonly("Get lyrics"),
127        timeout=TIMEOUT_QUERY,
128    )  # type: ignore[untyped-decorator, unused-ignore]
129    async def get_lyrics(track_uri: str) -> str | None:
130        """
131        Return lyrics for a track on a best-effort basis.
132
133        Returns ``None`` if lyrics are not available for the track. Raises
134        ``ToolError`` if the URI resolves to a non-track — without that
135        guard, querying ``metadata.lyrics`` on an album / playlist would
136        silently return ``None`` and the type confusion would never
137        surface to the caller.
138
139        :param track_uri: Music Assistant track URI (e.g. as found on
140            ``TrackBrief.uri``).
141        """
142        item = await resolve_typed_uri(
143            mass,
144            track_uri,
145            MediaType.TRACK,
146            type_label="track",
147            hint="lyrics only apply to tracks.",
148        )
149        metadata = getattr(item, "metadata", None)
150        lyrics = getattr(metadata, "lyrics", None) if metadata else None
151        return str(lyrics) if lyrics else None
152
153    return sub
154