music-assistant-server

1"""Playback: play, pause, seek, skip, play media."""
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 FastMCP
9from fastmcp.exceptions import ToolError
10from mcp.types import ToolAnnotations
11from music_assistant_models.errors import MusicAssistantError
12from music_assistant_models.media_items import BrowseFolder
13
14from music_assistant.providers.radio_playlist import radio_playlist_uri
15
16from ..tags import Tag
17from ._common import TIMEOUT_MUTATION, TIMEOUT_QUERY
18
19if TYPE_CHECKING:
20    from music_assistant.mass import MusicAssistant
21
22
23def _control_annotations(*, title: str, idempotent: bool = False) -> ToolAnnotations:
24    """Build default annotations for transport-control tools (mutate but non-destructive)."""
25    return ToolAnnotations(
26        title=title,
27        readOnlyHint=False,
28        destructiveHint=False,
29        idempotentHint=idempotent,
30        openWorldHint=False,
31    )
32
33
34def build_playback_server(mass: MusicAssistant) -> FastMCP:
35    """Construct the ``playback/*`` sub-server."""
36    sub: FastMCP = FastMCP(name="playback")
37
38    @sub.tool(
39        tags={Tag.CONTROL_PLAYBACK},
40        annotations=_control_annotations(title="Pause playback"),
41        timeout=TIMEOUT_MUTATION,
42    )  # type: ignore[untyped-decorator, unused-ignore]
43    async def pause(queue_id: str) -> None:
44        """
45        Pause playback on the given queue.
46
47        Always pauses — unlike ``playback_play_pause``, this does not toggle. Prefer this
48        when the user asks to pause. Returns nothing.
49
50        :param queue_id: Queue identifier — same as ``player_id`` from
51            ``players_list_players``.
52        """
53        await mass.player_queues.pause(queue_id)
54
55    @sub.tool(
56        tags={Tag.CONTROL_PLAYBACK},
57        annotations=_control_annotations(title="Resume playback"),
58        timeout=TIMEOUT_MUTATION,
59    )  # type: ignore[untyped-decorator, unused-ignore]
60    async def resume(queue_id: str) -> None:
61        """
62        Resume paused playback on the given queue.
63
64        Always resumes — unlike ``playback_play_pause``, this does not toggle. Prefer this
65        when the user asks to resume or unpause. Returns nothing.
66
67        :param queue_id: Queue identifier — same as ``player_id`` from
68            ``players_list_players``.
69        """
70        await mass.player_queues.resume(queue_id)
71
72    @sub.tool(
73        tags={Tag.CONTROL_PLAYBACK},
74        annotations=_control_annotations(title="Toggle play / pause"),
75        timeout=TIMEOUT_MUTATION,
76    )  # type: ignore[untyped-decorator, unused-ignore]
77    async def play_pause(queue_id: str) -> None:
78        """
79        Toggle play/pause on the given queue.
80
81        Playing → pauses, paused → resumes. Prefer ``playback_pause`` or
82        ``playback_resume`` when the user gives an explicit instruction —
83        toggling can flip the wrong way if state is unknown. Use
84        ``playback_stop`` to halt and reset position. Returns
85        nothing.
86
87        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
88        """
89        await mass.player_queues.play_pause(queue_id)
90
91    @sub.tool(
92        tags={Tag.CONTROL_PLAYBACK},
93        annotations=_control_annotations(title="Stop playback", idempotent=True),
94        timeout=TIMEOUT_MUTATION,
95    )  # type: ignore[untyped-decorator, unused-ignore]
96    async def stop(queue_id: str) -> None:
97        """
98        Stop playback and reset the playback position. The queue is preserved.
99
100        Use ``playback_play_pause`` to resume without losing position. Returns nothing.
101
102        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
103        """
104        await mass.player_queues.stop(queue_id)
105
106    @sub.tool(
107        tags={Tag.CONTROL_PLAYBACK},
108        annotations=_control_annotations(title="Next track"),
109        timeout=TIMEOUT_MUTATION,
110    )  # type: ignore[untyped-decorator, unused-ignore]
111    async def next_track(queue_id: str) -> None:
112        """
113        Skip to the next item in the queue.
114
115        At the end of the queue the behaviour depends on the current repeat
116        mode. Use ``playback_play_index`` to jump to a specific position. Returns
117        nothing.
118
119        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
120        """
121        await mass.player_queues.next(queue_id)
122
123    @sub.tool(
124        tags={Tag.CONTROL_PLAYBACK},
125        annotations=_control_annotations(title="Previous track"),
126        timeout=TIMEOUT_MUTATION,
127    )  # type: ignore[untyped-decorator, unused-ignore]
128    async def previous_track(queue_id: str) -> None:
129        """
130        Go back in the queue.
131
132        If the current track has been playing past Music Assistant's
133        rewind threshold the call restarts the current track instead of
134        moving to the previous one — invoke a second time to actually
135        step back. Returns nothing.
136
137        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
138        """
139        await mass.player_queues.previous(queue_id)
140
141    @sub.tool(
142        tags={Tag.CONTROL_PLAYBACK},
143        annotations=_control_annotations(title="Skip by seconds"),
144        timeout=TIMEOUT_MUTATION,
145    )  # type: ignore[untyped-decorator, unused-ignore]
146    async def skip(queue_id: str, seconds: int = 10) -> None:
147        """
148        Skip relative to the current playback position.
149
150        Use ``seek`` for an absolute position. Returns nothing.
151
152        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
153        :param seconds: Seconds to skip; negative values skip backward.
154            Defaults to ``10``.
155        """
156        await mass.player_queues.skip(queue_id, seconds)
157
158    @sub.tool(
159        tags={Tag.CONTROL_PLAYBACK},
160        annotations=_control_annotations(title="Seek to position"),
161        timeout=TIMEOUT_MUTATION,
162    )  # type: ignore[untyped-decorator, unused-ignore]
163    async def seek(queue_id: str, position: int) -> None:
164        """
165        Seek to an absolute position within the current track.
166
167        Use ``skip`` for relative offsets. Returns nothing.
168
169        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
170        :param position: Seconds from the start of the current track (``>= 0``).
171        """
172        await mass.player_queues.seek(queue_id, position)
173
174    @sub.tool(
175        tags={Tag.CONTROL_PLAYBACK},
176        annotations=ToolAnnotations(
177            title="Play media on a queue",
178            readOnlyHint=False,
179            destructiveHint=True,
180            idempotentHint=False,
181            openWorldHint=False,
182        ),
183        timeout=TIMEOUT_QUERY,
184    )  # type: ignore[untyped-decorator, unused-ignore]
185    async def play_media(
186        queue_id: str,
187        uri: str,
188        radio: bool = False,
189    ) -> None:
190        """
191        Load and start playing an artist, album, track, playlist, or radio station on a queue.
192
193        Replaces whatever the queue was playing. Use ``playback_play_index`` to start an
194        item that is already in the queue. Returns nothing.
195
196        :param queue_id: Queue identifier — for a player queue this is the same
197            ``player_id`` returned by ``players_list_players``.
198        :param uri: Music Assistant URI of the artist, album, track, playlist
199            or radio station to play, of the form
200            ``<provider>://<media_type>/<id>`` (e.g. as found on
201            ``TrackBrief.uri`` / ``AlbumBrief.uri`` / ...).
202        :param radio: When ``True``, play an endless "radio" seeded from ``uri``
203            instead of just the item itself. Music Assistant builds a dynamic
204            playlist from the seed (artist, album, track, playlist or genre),
205            mixing in its own tracks and continuously refilling the queue with
206            similar tracks.
207        """
208        if radio:
209            try:
210                seed = await mass.music.get_item_by_uri(uri)
211            except MusicAssistantError as err:
212                msg = f"Could not resolve URI for radio: {uri!r} ({err})"
213                raise ToolError(msg) from err
214            if isinstance(seed, BrowseFolder):
215                msg = f"Cannot start a radio from a browse folder: {uri!r}"
216                raise ToolError(msg)
217            uri = radio_playlist_uri(seed)
218        await mass.player_queues.play_media(queue_id, uri)
219
220    @sub.tool(
221        tags={Tag.CONTROL_PLAYBACK},
222        annotations=_control_annotations(title="Play queue item at index"),
223        timeout=TIMEOUT_MUTATION,
224    )  # type: ignore[untyped-decorator, unused-ignore]
225    async def play_index(queue_id: str, index: int) -> None:
226        """
227        Start playing the item at the given position in the existing queue.
228
229        Does not load new media — use ``playback_play_media`` for that. Returns nothing.
230
231        :param queue_id: Queue identifier from ``QueueBrief.queue_id``.
232        :param index: Zero-based position in the queue (``>= 0``).
233        """
234        await mass.player_queues.play_index(queue_id, index)
235
236    return sub
237