/
/
/
1"""Library: search, list, and get tools (read-only)."""
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 (
13 AlbumBrief,
14 AlbumTracksResult,
15 ArtistAlbumsResult,
16 ArtistBrief,
17 PlaylistBrief,
18 RadioBrief,
19 TrackBrief,
20)
21from ..tags import Tag
22from ._common import (
23 TIMEOUT_QUERY,
24 album_tracks_from_uri,
25 artist_albums_from_uri,
26 brief_from_uri,
27 page_args,
28 to_brief_album,
29 to_brief_artist,
30 to_brief_playlist,
31 to_brief_radio,
32 to_brief_track,
33)
34
35if TYPE_CHECKING:
36 from music_assistant.mass import MusicAssistant
37
38
39def _readonly(title: str) -> ToolAnnotations:
40 """Read-only library tool annotations with the supplied UI title."""
41 return ToolAnnotations(
42 title=title,
43 readOnlyHint=True,
44 destructiveHint=False,
45 idempotentHint=True,
46 openWorldHint=False,
47 )
48
49
50def _register_uri_tools(sub: FastMCP, mass: MusicAssistant) -> None:
51 """Register URI resolver and drill-down library tools."""
52
53 @sub.tool(
54 tags={Tag.QUERY_LIBRARY},
55 annotations=_readonly("Get track by URI"),
56 timeout=TIMEOUT_QUERY,
57 ) # type: ignore[untyped-decorator, unused-ignore]
58 async def get_track_by_uri(uri: str) -> TrackBrief:
59 """
60 Resolve a track by its Music Assistant URI to a brief summary.
61
62 Returns the same ``TrackBrief`` shape that search and list tools emit.
63 Raises ``ToolError`` if the URI does not resolve, or if it resolves
64 to a non-track (album, playlist, â¦) â otherwise the brief would
65 silently carry the wrong shape and downstream tools would
66 misinterpret it. Use ``search_tracks`` first if you only have a
67 name or partial identifier.
68
69 :param uri: A Music Assistant track URI of the form
70 ``<provider>://track/<id>`` (e.g. as found on
71 ``TrackBrief.uri``).
72 """
73 return await brief_from_uri(
74 mass,
75 uri,
76 MediaType.TRACK,
77 to_brief=to_brief_track,
78 type_label="track",
79 )
80
81 @sub.tool(
82 tags={Tag.QUERY_LIBRARY},
83 annotations=_readonly("Get album by URI"),
84 timeout=TIMEOUT_QUERY,
85 ) # type: ignore[untyped-decorator, unused-ignore]
86 async def get_album_by_uri(uri: str) -> AlbumBrief:
87 """
88 Resolve an album by its Music Assistant URI to a brief summary.
89
90 Returns the same ``AlbumBrief`` shape that search and list tools emit.
91 Raises ``ToolError`` if the URI does not resolve, or if it resolves
92 to a non-album. Use ``search_albums`` first if you only have a
93 name or partial identifier. For the track listing, use
94 ``get_album_tracks`` with the same URI.
95
96 :param uri: A Music Assistant album URI of the form
97 ``<provider>://album/<id>`` (e.g. as found on ``AlbumBrief.uri``).
98 """
99 return await brief_from_uri(
100 mass,
101 uri,
102 MediaType.ALBUM,
103 to_brief=to_brief_album,
104 type_label="album",
105 )
106
107 @sub.tool(
108 tags={Tag.QUERY_LIBRARY},
109 annotations=_readonly("Get artist by URI"),
110 timeout=TIMEOUT_QUERY,
111 ) # type: ignore[untyped-decorator, unused-ignore]
112 async def get_artist_by_uri(uri: str) -> ArtistBrief:
113 """
114 Resolve an artist by its Music Assistant URI to a brief summary.
115
116 Returns the same ``ArtistBrief`` shape that search and list tools emit.
117 Raises ``ToolError`` if the URI does not resolve, or if it resolves
118 to a non-artist. Use ``search_artists`` first if you only have a
119 name or partial identifier. For the artist's albums, use
120 ``get_artist_albums`` with the same URI.
121
122 :param uri: A Music Assistant artist URI of the form
123 ``<provider>://artist/<id>`` (e.g. as found on ``ArtistBrief.uri``).
124 """
125 return await brief_from_uri(
126 mass,
127 uri,
128 MediaType.ARTIST,
129 to_brief=to_brief_artist,
130 type_label="artist",
131 )
132
133 @sub.tool(
134 tags={Tag.QUERY_LIBRARY},
135 annotations=_readonly("Get artist albums"),
136 timeout=TIMEOUT_QUERY,
137 ) # type: ignore[untyped-decorator, unused-ignore]
138 async def get_artist_albums(uri: str, ctx: Context | None = None) -> ArtistAlbumsResult:
139 """
140 List albums by an artist, newest first.
141
142 Returns an ``ArtistAlbumsResult`` with a brief artist header and one
143 ``AlbumBrief`` per album (including ``uri`` for drill-down or playback).
144 This is a summary â not full metadata (genres, artwork URLs, etc.).
145
146 Typical flow: ``search_artists`` â ``get_artist_albums`` â
147 ``get_album_tracks`` or ``playback_play_media``.
148
149 Raises ``ToolError`` if the URI does not resolve or is not an artist.
150
151 :param uri: A Music Assistant artist URI of the form
152 ``<provider>://artist/<id>`` (e.g. as found on ``ArtistBrief.uri``).
153 """
154 if ctx is not None:
155 await ctx.info(f"Fetching albums for artist {uri!r}")
156 return await artist_albums_from_uri(mass, uri)
157
158 @sub.tool(
159 tags={Tag.QUERY_LIBRARY},
160 annotations=_readonly("Get playlist by URI"),
161 timeout=TIMEOUT_QUERY,
162 ) # type: ignore[untyped-decorator, unused-ignore]
163 async def get_playlist_by_uri(uri: str) -> PlaylistBrief:
164 """
165 Resolve a playlist by its Music Assistant URI to a brief summary.
166
167 Returns the same ``PlaylistBrief`` shape that list tools emit.
168 Raises ``ToolError`` if the URI does not resolve, or if it resolves
169 to a non-playlist.
170
171 :param uri: A Music Assistant playlist URI of the form
172 ``<provider>://playlist/<id>`` (e.g. as found on
173 ``PlaylistBrief.uri``).
174 """
175 return await brief_from_uri(
176 mass,
177 uri,
178 MediaType.PLAYLIST,
179 to_brief=to_brief_playlist,
180 type_label="playlist",
181 )
182
183 @sub.tool(
184 tags={Tag.QUERY_LIBRARY},
185 annotations=_readonly("Get radio by URI"),
186 timeout=TIMEOUT_QUERY,
187 ) # type: ignore[untyped-decorator, unused-ignore]
188 async def get_radio_by_uri(uri: str) -> RadioBrief:
189 """
190 Resolve a radio station by its Music Assistant URI to a brief summary.
191
192 Returns the same ``RadioBrief`` shape that list tools emit.
193 Raises ``ToolError`` if the URI does not resolve, or if it resolves
194 to a non-radio station.
195
196 :param uri: A Music Assistant radio URI of the form
197 ``<provider>://radio/<id>`` (e.g. as found on ``RadioBrief.uri``).
198 """
199 return await brief_from_uri(
200 mass,
201 uri,
202 MediaType.RADIO,
203 to_brief=to_brief_radio,
204 type_label="radio",
205 )
206
207 @sub.tool(
208 tags={Tag.QUERY_LIBRARY},
209 annotations=_readonly("Get album tracks"),
210 timeout=TIMEOUT_QUERY,
211 ) # type: ignore[untyped-decorator, unused-ignore]
212 async def get_album_tracks(uri: str, ctx: Context | None = None) -> AlbumTracksResult:
213 """
214 List the tracks on an album, in disc and track order.
215
216 Returns an ``AlbumTracksResult`` with a brief album header and one
217 ``TrackBrief`` per track (including ``uri`` for playback). This is a
218 summary â not full album metadata (genres, artwork URLs, etc.).
219
220 Typical flow: ``search_albums`` â ``get_album_tracks`` â pick a track
221 URI or pass the album URI to ``playback_play_media`` for the full album.
222
223 Raises ``ToolError`` if the URI does not resolve or is not an album.
224
225 :param uri: A Music Assistant album URI of the form
226 ``<provider>://album/<id>`` (e.g. as found on ``AlbumBrief.uri``).
227 """
228 if ctx is not None:
229 await ctx.info(f"Fetching tracks for album {uri!r}")
230 return await album_tracks_from_uri(mass, uri)
231
232
233def build_library_server(mass: MusicAssistant) -> FastMCP:
234 """Construct the ``library/*`` sub-server."""
235 sub: FastMCP = FastMCP(name="library")
236
237 @sub.tool(
238 tags={Tag.QUERY_LIBRARY},
239 annotations=ToolAnnotations(
240 title="Search tracks",
241 readOnlyHint=True,
242 destructiveHint=False,
243 idempotentHint=True,
244 openWorldHint=False,
245 ),
246 timeout=TIMEOUT_QUERY,
247 ) # type: ignore[untyped-decorator, unused-ignore]
248 async def search_tracks(
249 query: str, limit: int = 25, ctx: Context | None = None
250 ) -> list[TrackBrief]:
251 """
252 Search for tracks by free-text query across all enabled music providers.
253
254 Returns ``TrackBrief`` items with ``uri``, ``name``, ``artists``, ``album``,
255 and ``duration``. Use ``list_library_tracks`` instead to enumerate only
256 tracks already saved to the user's library.
257
258 :param query: Free-text search string.
259 :param limit: Max results to return (clamped to ``[1, 200]``).
260 """
261 _, limit = page_args(0, limit)
262 if ctx is not None:
263 await ctx.info(f"Searching MA for tracks matching {query!r} (limit={limit})")
264 results = await mass.music.search(query, [MediaType.TRACK], limit=limit)
265 return [to_brief_track(t) for t in (results.tracks or [])]
266
267 @sub.tool(
268 tags={Tag.QUERY_LIBRARY},
269 annotations=ToolAnnotations(
270 title="Search albums",
271 readOnlyHint=True,
272 destructiveHint=False,
273 idempotentHint=True,
274 openWorldHint=False,
275 ),
276 timeout=TIMEOUT_QUERY,
277 ) # type: ignore[untyped-decorator, unused-ignore]
278 async def search_albums(
279 query: str, limit: int = 25, ctx: Context | None = None
280 ) -> list[AlbumBrief]:
281 """
282 Search for albums by free-text query across all enabled music providers.
283
284 Returns ``AlbumBrief`` items with ``uri``, ``name``, ``artists`` and
285 ``year``. Use ``list_library_albums`` to enumerate only albums already
286 saved to the user's library.
287
288 :param query: Free-text search string.
289 :param limit: Max results to return (clamped to ``[1, 200]``).
290 """
291 _, limit = page_args(0, limit)
292 if ctx is not None:
293 await ctx.info(f"Searching MA for albums matching {query!r} (limit={limit})")
294 results = await mass.music.search(query, [MediaType.ALBUM], limit=limit)
295 return [to_brief_album(a) for a in (results.albums or [])]
296
297 @sub.tool(
298 tags={Tag.QUERY_LIBRARY},
299 annotations=ToolAnnotations(
300 title="Search artists",
301 readOnlyHint=True,
302 destructiveHint=False,
303 idempotentHint=True,
304 openWorldHint=False,
305 ),
306 timeout=TIMEOUT_QUERY,
307 ) # type: ignore[untyped-decorator, unused-ignore]
308 async def search_artists(
309 query: str, limit: int = 25, ctx: Context | None = None
310 ) -> list[ArtistBrief]:
311 """
312 Search for artists by free-text query across all enabled music providers.
313
314 Returns ``ArtistBrief`` items with ``uri`` and ``name``. Use
315 ``list_library_artists`` to enumerate only artists already saved to the
316 user's library.
317
318 :param query: Free-text search string.
319 :param limit: Max results to return (clamped to ``[1, 200]``).
320 """
321 _, limit = page_args(0, limit)
322 if ctx is not None:
323 await ctx.info(f"Searching MA for artists matching {query!r} (limit={limit})")
324 results = await mass.music.search(query, [MediaType.ARTIST], limit=limit)
325 return [to_brief_artist(a) for a in (results.artists or [])]
326
327 @sub.tool(
328 tags={Tag.QUERY_LIBRARY},
329 annotations=_readonly("List library tracks"),
330 timeout=TIMEOUT_QUERY,
331 ) # type: ignore[untyped-decorator, unused-ignore]
332 async def list_library_tracks(offset: int = 0, limit: int = 50) -> list[TrackBrief]:
333 """
334 List tracks already saved to the user's library, paginated.
335
336 Returns ``TrackBrief`` items in library order. Does not query external
337 providers â use ``search_tracks`` for that.
338
339 :param offset: Zero-based start position (clamped to ``>= 0``).
340 :param limit: Page size (clamped to ``[1, 200]``).
341 """
342 offset, limit = page_args(offset, limit)
343 items = await mass.music.tracks.library_items(limit=limit, offset=offset, summary=False)
344 return [to_brief_track(t) for t in items]
345
346 @sub.tool(
347 tags={Tag.QUERY_LIBRARY},
348 annotations=_readonly("List library albums"),
349 timeout=TIMEOUT_QUERY,
350 ) # type: ignore[untyped-decorator, unused-ignore]
351 async def list_library_albums(offset: int = 0, limit: int = 50) -> list[AlbumBrief]:
352 """
353 List albums already saved to the user's library, paginated.
354
355 Returns ``AlbumBrief`` items in library order. Does not query external
356 providers â use ``search_albums`` for that.
357
358 :param offset: Zero-based start position (clamped to ``>= 0``).
359 :param limit: Page size (clamped to ``[1, 200]``).
360 """
361 offset, limit = page_args(offset, limit)
362 items = await mass.music.albums.library_items(limit=limit, offset=offset, summary=False)
363 return [to_brief_album(a) for a in items]
364
365 @sub.tool(
366 tags={Tag.QUERY_LIBRARY},
367 annotations=_readonly("List library artists"),
368 timeout=TIMEOUT_QUERY,
369 ) # type: ignore[untyped-decorator, unused-ignore]
370 async def list_library_artists(offset: int = 0, limit: int = 50) -> list[ArtistBrief]:
371 """
372 List artists already saved to the user's library, paginated.
373
374 Returns ``ArtistBrief`` items in library order. Does not query external
375 providers â use ``search_artists`` for that.
376
377 :param offset: Zero-based start position (clamped to ``>= 0``).
378 :param limit: Page size (clamped to ``[1, 200]``).
379 """
380 offset, limit = page_args(offset, limit)
381 items = await mass.music.artists.library_items(limit=limit, offset=offset, summary=False)
382 return [to_brief_artist(a) for a in items]
383
384 @sub.tool(
385 tags={Tag.QUERY_LIBRARY},
386 annotations=_readonly("List library playlists"),
387 timeout=TIMEOUT_QUERY,
388 ) # type: ignore[untyped-decorator, unused-ignore]
389 async def list_library_playlists(offset: int = 0, limit: int = 50) -> list[PlaylistBrief]:
390 """
391 List playlists already saved to the user's library, paginated.
392
393 Returns ``PlaylistBrief`` items in library order.
394
395 :param offset: Zero-based start position (clamped to ``>= 0``).
396 :param limit: Page size (clamped to ``[1, 200]``).
397 """
398 offset, limit = page_args(offset, limit)
399 items = await mass.music.playlists.library_items(limit=limit, offset=offset, summary=False)
400 return [to_brief_playlist(p) for p in items]
401
402 @sub.tool(
403 tags={Tag.QUERY_LIBRARY},
404 annotations=_readonly("List library radio"),
405 timeout=TIMEOUT_QUERY,
406 ) # type: ignore[untyped-decorator, unused-ignore]
407 async def list_library_radio(offset: int = 0, limit: int = 50) -> list[RadioBrief]:
408 """
409 List radio stations already saved to the user's library, paginated.
410
411 Returns ``RadioBrief`` items in library order.
412
413 :param offset: Zero-based start position (clamped to ``>= 0``).
414 :param limit: Page size (clamped to ``[1, 200]``).
415 """
416 offset, limit = page_args(offset, limit)
417 items = await mass.music.radio.library_items(limit=limit, offset=offset, summary=False)
418 return [to_brief_radio(r) for r in items]
419
420 @sub.tool(
421 tags={Tag.QUERY_LIBRARY},
422 annotations=_readonly("Recently added tracks"),
423 timeout=TIMEOUT_QUERY,
424 ) # type: ignore[untyped-decorator, unused-ignore]
425 async def recently_added_tracks(limit: int = 10) -> list[TrackBrief]:
426 """
427 Return tracks most recently added to the user's library, newest first.
428
429 Returns ``TrackBrief`` items. Does not paginate further than the first
430 page â pick a higher ``limit`` if more history is needed.
431
432 :param limit: Max results to return (clamped to ``[1, 200]``).
433 """
434 _, limit = page_args(0, limit)
435 items = await mass.music.recently_added_tracks(limit=limit)
436 return [to_brief_track(t) for t in items]
437
438 _register_uri_tools(sub, mass)
439
440 return sub
441