music-assistant-server

1"""Players: list, inspect, power, group."""
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 mcp.types import ToolAnnotations
10
11from ..models import PlayerBrief
12from ..tags import Tag
13from ._common import TIMEOUT_FAST, TIMEOUT_MUTATION, safe_active_queue, to_brief_player
14
15if TYPE_CHECKING:
16    from music_assistant.mass import MusicAssistant
17
18
19def build_players_server(mass: MusicAssistant) -> FastMCP:
20    """Construct the ``players/*`` sub-server."""
21    sub: FastMCP = FastMCP(name="players")
22
23    @sub.tool(
24        tags={Tag.QUERY_PLAYERS},
25        annotations=ToolAnnotations(
26            title="List players",
27            readOnlyHint=True,
28            destructiveHint=False,
29            idempotentHint=True,
30            openWorldHint=False,
31        ),
32        timeout=TIMEOUT_FAST,
33    )  # type: ignore[untyped-decorator, unused-ignore]
34    async def list_players(
35        include_unavailable: bool = False,
36        include_disabled: bool = False,
37    ) -> list[PlayerBrief]:
38        """
39        List players known to Music Assistant.
40
41        Returns ``PlayerBrief`` items with ``player_id``, ``name``, ``state``,
42        ``powered``, ``volume_level``, ``available``, ``enabled``,
43        ``needs_setup``, ``active_group``, ``synced_to``, ``external_source``
44        and the currently playing item (if any). ``state`` summarises
45        usability — values are ``unavailable`` (offline), ``disabled``
46        (admin-disabled), ``needs_setup`` (first-run config pending),
47        ``synced`` (member of an active sync group; its queue belongs to the
48        group leader), or the normal playback states
49        (``idle`` / ``playing`` / ``paused`` / ...). When a player is being
50        driven by an external "Connect"-style source (e.g. Spotify Connect,
51        AirPlay, Yandex Ynison), ``state`` reflects the active queue
52        (``playing`` / ``paused``) rather than ``idle``, ``external_source``
53        holds the controlling provider instance id, and ``current_item``
54        shows the real track title rather than the source wrapper name.
55        Offline and admin-disabled players are hidden by default — flip the
56        corresponding ``include_*`` flag to get them back. Does not include
57        queue contents — use the ``queue`` tools for that.
58
59        :param include_unavailable: When ``True``, include players whose
60            ``available`` flag is ``False`` (offline / unreachable
61            devices). Defaults to ``False``.
62        :param include_disabled: When ``True``, include players that the
63            admin has disabled in Music Assistant settings. Defaults to
64            ``False`` (matching MA's own ``return_disabled`` default).
65        """
66        # Delegate filtering to MA's native ``return_unavailable`` /
67        # ``return_disabled`` knobs rather than re-implementing them in
68        # Python — MA short-circuits the build at the controller level and
69        # applies the same user-role visibility filters as every other
70        # consumer.
71        players = mass.players.all_players(
72            return_unavailable=include_unavailable,
73            return_disabled=include_disabled,
74        )
75        return [to_brief_player(p, safe_active_queue(mass, p.player_id)) for p in players]
76
77    @sub.tool(
78        tags={Tag.QUERY_PLAYERS},
79        annotations=ToolAnnotations(
80            title="Get player by id",
81            readOnlyHint=True,
82            destructiveHint=False,
83            idempotentHint=True,
84            openWorldHint=False,
85        ),
86        timeout=TIMEOUT_FAST,
87    )  # type: ignore[untyped-decorator, unused-ignore]
88    async def get_player(player_id: str) -> PlayerBrief | None:
89        """
90        Return a single player by id, or ``None`` if it doesn't exist.
91
92        Same ``PlayerBrief`` shape as ``list_players``. Prefer this over
93        ``list_players`` when the id is already known.
94
95        :param player_id: Player identifier (from ``PlayerBrief.player_id``).
96        """
97        player = mass.players.get_player(player_id)
98        if player is None:
99            return None
100        return to_brief_player(player, safe_active_queue(mass, player_id))
101
102    @sub.tool(
103        tags={Tag.CONTROL_PLAYERS},
104        annotations=ToolAnnotations(
105            title="Power player on / off",
106            readOnlyHint=False,
107            destructiveHint=False,
108            idempotentHint=True,
109            openWorldHint=False,
110        ),
111        timeout=TIMEOUT_MUTATION,
112    )  # type: ignore[untyped-decorator, unused-ignore]
113    async def set_power(player_id: str, powered: bool) -> None:
114        """
115        Power a player on or off.
116
117        Does not affect sync-group membership — use ``group_player`` to add a
118        player to a sync group, or ``ungroup_player`` to remove one. Setting
119        the current power state again is a no-op. Returns nothing.
120
121        :param player_id: Player identifier from ``PlayerBrief.player_id``.
122        :param powered: ``True`` to power on, ``False`` to power off.
123        """
124        await mass.players.cmd_power(player_id, powered)
125
126    @sub.tool(
127        tags={Tag.CONTROL_PLAYERS},
128        annotations=ToolAnnotations(
129            title="Group player into sync group",
130            readOnlyHint=False,
131            destructiveHint=False,
132            idempotentHint=False,
133            openWorldHint=False,
134        ),
135        timeout=TIMEOUT_MUTATION,
136    )  # type: ignore[untyped-decorator, unused-ignore]
137    async def group_player(player_id: str, target_player_id: str) -> None:
138        """
139        Add a player to another player's sync group so both play in lockstep.
140
141        Does not change volume — use ``set_group_volume`` on the volume
142        sub-server for that. Use ``ungroup_player`` to remove a player from
143        a sync group. Returns nothing.
144
145        :param player_id: Player to add to the group.
146        :param target_player_id: Player whose sync group ``player_id`` joins
147            (typically the group leader).
148        """
149        await mass.players.cmd_group(player_id, target_player_id)
150
151    @sub.tool(
152        tags={Tag.CONTROL_PLAYERS},
153        annotations=ToolAnnotations(
154            title="Ungroup player from sync group",
155            readOnlyHint=False,
156            destructiveHint=False,
157            idempotentHint=True,
158            openWorldHint=False,
159        ),
160        timeout=TIMEOUT_MUTATION,
161    )  # type: ignore[untyped-decorator, unused-ignore]
162    async def ungroup_player(player_id: str) -> None:
163        """
164        Remove a player from its sync group so it plays independently again.
165
166        If the player is not currently grouped, this is a no-op. Use
167        ``group_player`` to add a player to another player's sync group.
168        Returns nothing.
169
170        :param player_id: Player to remove from any active sync group.
171        """
172        await mass.players.cmd_ungroup(player_id)
173
174    return sub
175