/
/
/
1"""URI-addressable read-only player and queue resources."""
2# ruff: noqa: TID252 -- relative imports are the canonical MA-provider pattern.
3
4from __future__ import annotations
5
6from typing import TYPE_CHECKING, Any
7
8from ..tags import Tag
9from ..tools._common import (
10 safe_active_queue,
11 to_brief_player,
12 to_brief_queue,
13 to_resource_text,
14)
15
16if TYPE_CHECKING:
17 from music_assistant.mass import MusicAssistant
18
19
20def register_player_resources(mcp: Any, mass: MusicAssistant) -> None:
21 """Register ``player://`` and ``queue://`` resources on the given FastMCP root."""
22
23 @mcp.resource("player://{player_id}", tags={Tag.QUERY_PLAYERS}) # type: ignore[untyped-decorator, unused-ignore]
24 async def player_resource(player_id: str) -> str | None:
25 """Player snapshot by id."""
26 player = mass.players.get_player(player_id)
27 return to_resource_text(
28 to_brief_player(player, safe_active_queue(mass, player_id))
29 if player is not None
30 else None
31 )
32
33 @mcp.resource("queue://{queue_id}", tags={Tag.QUERY_QUEUE}) # type: ignore[untyped-decorator, unused-ignore]
34 async def queue_resource(queue_id: str) -> str | None:
35 """Queue snapshot by id (up to 500 items â MA's default page size)."""
36 queue = mass.player_queues.get(queue_id)
37 if queue is None:
38 return None
39 items = mass.player_queues.items(queue_id, limit=500)
40 return to_resource_text(to_brief_queue(queue, items=list(items)))
41