diff --git a/plugins/spotify/__init__.py b/plugins/spotify/__init__.py index 0f68bba1f7..d0acfc09b5 100644 --- a/plugins/spotify/__init__.py +++ b/plugins/spotify/__init__.py @@ -1,25 +1,13 @@ """Spotify integration plugin — bundled, auto-loaded. -Registers 7 tools (playback, devices, queue, search, playlists, albums, -library) into the ``spotify`` toolset. Each tool's handler is gated by -``_check_spotify_available()`` — when the user has not run ``hermes auth -spotify``, the tools remain registered (so they appear in ``hermes -tools``) but the runtime check prevents dispatch. +Registers 7 tools into the ``spotify`` toolset. Tools stay registered (so they +appear in ``hermes tools``) but ``_check_spotify_available()`` gates dispatch +until the user has run ``hermes auth spotify``. -Why a plugin instead of a top-level ``tools/`` file? - -- ``plugins/`` is where third-party service integrations live (see - ``plugins/image_gen/`` for the backend-provider pattern, ``plugins/ - disk-cleanup/`` for the standalone pattern). ``tools/`` is reserved - for foundational capabilities (terminal, read_file, web_search, etc.). -- Mirroring the image_gen plugin layout (``plugins///`` - for categories, flat ``plugins//`` for standalones) makes new - service integrations a pattern contributors can copy. -- Bundled + ``kind: backend`` auto-loads on startup just like image_gen - backends — no user opt-in needed, no ``plugins.enabled`` config. - -The Spotify auth flow (``hermes auth spotify``), CLI plumbing, and docs -are unchanged. This move is purely structural. +Why a plugin rather than a ``tools/`` module: ``tools/`` is reserved for +foundational capabilities; third-party service integrations live under +``plugins/`` (flat ``plugins//`` for standalones, like image_gen backends), +and ``kind: backend`` bundled plugins auto-load with no ``plugins.enabled`` opt-in. """ from __future__ import annotations @@ -56,11 +44,4 @@ _TOOLS = ( def register(ctx) -> None: """Register all Spotify tools. Called once by the plugin loader.""" for name, schema, handler, emoji in _TOOLS: - ctx.register_tool( - name=name, - toolset="spotify", - schema=schema, - handler=handler, - check_fn=_check_spotify_available, - emoji=emoji, - ) + ctx.register_tool(name=name, toolset="spotify", schema=schema, handler=handler, check_fn=_check_spotify_available, emoji=emoji) diff --git a/plugins/spotify/client.py b/plugins/spotify/client.py index 2195cc20a8..e87e1f8fda 100644 --- a/plugins/spotify/client.py +++ b/plugins/spotify/client.py @@ -1,17 +1,17 @@ -"""Thin Spotify Web API helper used by Hermes native tools.""" +"""Thin Spotify Web API helper used by Hermes native tools. + +Owns auth (token refresh/401 retry), error mapping and id/URI normalization; +endpoint paths live with their tool handlers in ``tools.py``. +""" from __future__ import annotations -import json from typing import Any, Dict, Iterable, Optional from urllib.parse import urlparse import httpx -from hermes_cli.auth import ( - AuthError, - resolve_spotify_runtime_credentials, -) +from hermes_cli.auth import AuthError, resolve_spotify_runtime_credentials class SpotifyError(RuntimeError): @@ -25,17 +25,16 @@ class SpotifyAuthRequiredError(SpotifyError): class SpotifyAPIError(SpotifyError): """Structured Spotify API failure.""" - def __init__( - self, - message: str, - *, - status_code: Optional[int] = None, - response_body: Optional[str] = None, - ) -> None: + def __init__(self, message: str, *, status_code: Optional[int] = None, response_body: Optional[str] = None) -> None: super().__init__(message) self.status_code = status_code self.response_body = response_body - self.path = None + self.path: Optional[str] = None + + +def _empty_204(message: str) -> Dict[str, Any]: + """Explanatory payload returned instead of a bare 204 for the player endpoints.""" + return {"status_code": 204, "empty": True, "message": message} class SpotifyClient: @@ -44,10 +43,7 @@ class SpotifyClient: def _resolve_runtime(self, *, force_refresh: bool = False, refresh_if_expiring: bool = True) -> Dict[str, Any]: try: - return resolve_spotify_runtime_credentials( - force_refresh=force_refresh, - refresh_if_expiring=refresh_if_expiring, - ) + return resolve_spotify_runtime_credentials(force_refresh=force_refresh, refresh_if_expiring=refresh_if_expiring) except AuthError as exc: raise SpotifyAuthRequiredError(str(exc)) from exc @@ -55,12 +51,6 @@ class SpotifyClient: def base_url(self) -> str: return str(self._runtime.get("base_url") or "").rstrip("/") - def _headers(self) -> Dict[str, str]: - return { - "Authorization": f"Bearer {self._runtime['access_token']}", - "Content-Type": "application/json", - } - def request( self, method: str, @@ -71,365 +61,130 @@ class SpotifyClient: allow_retry_on_401: bool = True, empty_response: Optional[Dict[str, Any]] = None, ) -> Any: - url = f"{self.base_url}{path}" response = httpx.request( method, - url, - headers=self._headers(), + f"{self.base_url}{path}", + headers={"Authorization": f"Bearer {self._runtime['access_token']}", "Content-Type": "application/json"}, params=_strip_none(params), json=_strip_none(json_body) if json_body is not None else None, timeout=30.0, ) if response.status_code == 401 and allow_retry_on_401: + # One forced token refresh, then retry exactly once. self._runtime = self._resolve_runtime(force_refresh=True, refresh_if_expiring=True) - return self.request( - method, - path, - params=params, - json_body=json_body, - allow_retry_on_401=False, - ) + return self.request(method, path, params=params, json_body=json_body, allow_retry_on_401=False) if response.status_code >= 400: - self._raise_api_error(response, method=method, path=path) + detail = response.text.strip() + message = _friendly_spotify_error_message( + status_code=response.status_code, + detail=_extract_spotify_error_detail(response, fallback=detail), + path=path, + retry_after=response.headers.get("Retry-After"), + ) + error = SpotifyAPIError(message, status_code=response.status_code, response_body=detail) + error.path = path + raise error if response.status_code == 204 or not response.content: return empty_response or {"success": True, "status_code": response.status_code, "empty": True} if "application/json" in response.headers.get("content-type", ""): return response.json() return {"success": True, "text": response.text} - def _raise_api_error(self, response: httpx.Response, *, method: str, path: str) -> None: - detail = response.text.strip() - message = _friendly_spotify_error_message( - status_code=response.status_code, - detail=_extract_spotify_error_detail(response, fallback=detail), - method=method, - path=path, - retry_after=response.headers.get("Retry-After"), - ) - error = SpotifyAPIError(message, status_code=response.status_code, response_body=detail) - error.path = path - raise error + # -- player ----------------------------------------------------------- - def get_devices(self) -> Any: - return self.request("GET", "/me/player/devices") - - def transfer_playback(self, *, device_id: str, play: bool = False) -> Any: - return self.request("PUT", "/me/player", json_body={ - "device_ids": [device_id], - "play": play, - }) + # Player reads return an explanatory payload instead of a bare 204. def get_playback_state(self, *, market: Optional[str] = None) -> Any: - return self.request( - "GET", - "/me/player", - params={"market": market}, - empty_response={ - "status_code": 204, - "empty": True, - "message": "No active Spotify playback session was found. Open Spotify on a device and start playback, or transfer playback to an available device.", - }, - ) + return self.request("GET", "/me/player", params={"market": market}, empty_response=_empty_204( + "No active Spotify playback session was found. Open Spotify on a device and start playback, or transfer playback to an available device." + )) def get_currently_playing(self, *, market: Optional[str] = None) -> Any: - return self.request( - "GET", - "/me/player/currently-playing", - params={"market": market}, - empty_response={ - "status_code": 204, - "empty": True, - "message": "Spotify is not currently playing anything. Start playback in Spotify and try again.", - }, - ) - - def start_playback( - self, - *, - device_id: Optional[str] = None, - context_uri: Optional[str] = None, - uris: Optional[list[str]] = None, - offset: Optional[Dict[str, Any]] = None, - position_ms: Optional[int] = None, - ) -> Any: - return self.request( - "PUT", - "/me/player/play", - params={"device_id": device_id}, - json_body={ - "context_uri": context_uri, - "uris": uris, - "offset": offset, - "position_ms": position_ms, - }, - ) - - def pause_playback(self, *, device_id: Optional[str] = None) -> Any: - return self.request("PUT", "/me/player/pause", params={"device_id": device_id}) - - def skip_next(self, *, device_id: Optional[str] = None) -> Any: - return self.request("POST", "/me/player/next", params={"device_id": device_id}) - - def skip_previous(self, *, device_id: Optional[str] = None) -> Any: - return self.request("POST", "/me/player/previous", params={"device_id": device_id}) - - def seek(self, *, position_ms: int, device_id: Optional[str] = None) -> Any: - return self.request("PUT", "/me/player/seek", params={ - "position_ms": position_ms, - "device_id": device_id, - }) - - def set_repeat(self, *, state: str, device_id: Optional[str] = None) -> Any: - return self.request("PUT", "/me/player/repeat", params={"state": state, "device_id": device_id}) - - def set_shuffle(self, *, state: bool, device_id: Optional[str] = None) -> Any: - return self.request("PUT", "/me/player/shuffle", params={"state": str(bool(state)).lower(), "device_id": device_id}) - - def set_volume(self, *, volume_percent: int, device_id: Optional[str] = None) -> Any: - return self.request("PUT", "/me/player/volume", params={ - "volume_percent": volume_percent, - "device_id": device_id, - }) - - def get_queue(self) -> Any: - return self.request("GET", "/me/player/queue") - - def add_to_queue(self, *, uri: str, device_id: Optional[str] = None) -> Any: - return self.request("POST", "/me/player/queue", params={"uri": uri, "device_id": device_id}) - - def search( - self, - *, - query: str, - search_types: list[str], - limit: int = 10, - offset: int = 0, - market: Optional[str] = None, - include_external: Optional[str] = None, - ) -> Any: - return self.request("GET", "/search", params={ - "q": query, - "type": ",".join(search_types), - "limit": limit, - "offset": offset, - "market": market, - "include_external": include_external, - }) - - def get_my_playlists(self, *, limit: int = 20, offset: int = 0) -> Any: - return self.request("GET", "/me/playlists", params={"limit": limit, "offset": offset}) - - def get_playlist(self, *, playlist_id: str, market: Optional[str] = None) -> Any: - return self.request("GET", f"/playlists/{playlist_id}", params={"market": market}) - - def create_playlist( - self, - *, - name: str, - public: bool = False, - collaborative: bool = False, - description: Optional[str] = None, - ) -> Any: - return self.request("POST", "/me/playlists", json_body={ - "name": name, - "public": public, - "collaborative": collaborative, - "description": description, - }) - - def add_playlist_items( - self, - *, - playlist_id: str, - uris: list[str], - position: Optional[int] = None, - ) -> Any: - return self.request("POST", f"/playlists/{playlist_id}/items", json_body={ - "uris": uris, - "position": position, - }) - - def remove_playlist_items( - self, - *, - playlist_id: str, - uris: list[str], - snapshot_id: Optional[str] = None, - ) -> Any: - return self.request("DELETE", f"/playlists/{playlist_id}/items", json_body={ - "items": [{"uri": uri} for uri in uris], - "snapshot_id": snapshot_id, - }) - - def update_playlist_details( - self, - *, - playlist_id: str, - name: Optional[str] = None, - public: Optional[bool] = None, - collaborative: Optional[bool] = None, - description: Optional[str] = None, - ) -> Any: - return self.request("PUT", f"/playlists/{playlist_id}", json_body={ - "name": name, - "public": public, - "collaborative": collaborative, - "description": description, - }) - - def get_album(self, *, album_id: str, market: Optional[str] = None) -> Any: - return self.request("GET", f"/albums/{album_id}", params={"market": market}) - - def get_album_tracks(self, *, album_id: str, limit: int = 20, offset: int = 0, market: Optional[str] = None) -> Any: - return self.request("GET", f"/albums/{album_id}/tracks", params={ - "limit": limit, - "offset": offset, - "market": market, - }) - - def get_saved_tracks(self, *, limit: int = 20, offset: int = 0, market: Optional[str] = None) -> Any: - return self.request("GET", "/me/tracks", params={"limit": limit, "offset": offset, "market": market}) - - def save_library_items(self, *, uris: list[str]) -> Any: - return self.request("PUT", "/me/library", params={"uris": ",".join(uris)}) - - def library_contains(self, *, uris: list[str]) -> Any: - return self.request("GET", "/me/library/contains", params={"uris": ",".join(uris)}) - - def get_saved_albums(self, *, limit: int = 20, offset: int = 0, market: Optional[str] = None) -> Any: - return self.request("GET", "/me/albums", params={"limit": limit, "offset": offset, "market": market}) - - def remove_saved_tracks(self, *, track_ids: list[str]) -> Any: - uris = [f"spotify:track:{track_id}" for track_id in track_ids] - return self.request("DELETE", "/me/library", params={"uris": ",".join(uris)}) - - def remove_saved_albums(self, *, album_ids: list[str]) -> Any: - uris = [f"spotify:album:{album_id}" for album_id in album_ids] - return self.request("DELETE", "/me/library", params={"uris": ",".join(uris)}) - - def get_recently_played( - self, - *, - limit: int = 20, - after: Optional[int] = None, - before: Optional[int] = None, - ) -> Any: - return self.request("GET", "/me/player/recently-played", params={ - "limit": limit, - "after": after, - "before": before, - }) + return self.request("GET", "/me/player/currently-playing", params={"market": market}, empty_response=_empty_204( + "Spotify is not currently playing anything. Start playback in Spotify and try again." + )) def _extract_spotify_error_detail(response: httpx.Response, *, fallback: str) -> str: + """Prefer Spotify's ``{"error": {"message": ...}}`` (or ``{"error": "..."}``) body over raw text.""" detail = fallback try: - payload = response.json() - if isinstance(payload, dict): - error_obj = payload.get("error") - if isinstance(error_obj, dict): - detail = str(error_obj.get("message") or detail) - elif isinstance(error_obj, str): - detail = error_obj + error_obj = response.json().get("error") + if isinstance(error_obj, dict): + detail = str(error_obj.get("message") or detail) + elif isinstance(error_obj, str): + detail = error_obj except Exception: pass return detail.strip() -def _friendly_spotify_error_message( - *, - status_code: int, - detail: str, - method: str, - path: str, - retry_after: Optional[str], -) -> str: - normalized_detail = detail.lower() +def _friendly_spotify_error_message(*, status_code: int, detail: str, path: str, retry_after: Optional[str]) -> str: is_playback_path = path.startswith("/me/player") - if status_code == 401: return "Spotify authentication failed or expired. Run `hermes auth spotify` again." - if status_code == 403: if is_playback_path: return ( "Spotify rejected this playback request. Playback control usually requires a Spotify Premium account " "and an active Spotify Connect device." ) - if "scope" in normalized_detail or "permission" in normalized_detail: + if "scope" in detail.lower() or "permission" in detail.lower(): return "Spotify rejected the request because the current auth scope is insufficient. Re-run `hermes auth spotify` to refresh permissions." return "Spotify rejected the request. The account may not have permission for this action." - if status_code == 404: if is_playback_path: return "Spotify could not find an active playback device or player session for this request." return "Spotify resource not found." - if status_code == 429: - message = "Spotify rate limit exceeded." - if retry_after: - message += f" Retry after {retry_after} seconds." - return message - - if detail: - return detail - return f"Spotify API request failed with status {status_code}." + return "Spotify rate limit exceeded." + (f" Retry after {retry_after} seconds." if retry_after else "") + return detail or f"Spotify API request failed with status {status_code}." def _strip_none(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]: - if not payload: - return {} - return {key: value for key, value in payload.items() if value is not None} + return {key: value for key, value in (payload or {}).items() if value is not None} + + +def _check_type(item_type: str, expected_type: Optional[str]) -> None: + if expected_type and item_type != expected_type: + raise SpotifyError(f"Expected a Spotify {expected_type}, got {item_type}.") def normalize_spotify_id(value: str, expected_type: Optional[str] = None) -> str: + """Accept a bare id, ``spotify::`` URI, or open.spotify.com URL; return the id.""" cleaned = (value or "").strip() if not cleaned: raise SpotifyError("Spotify id/uri/url is required.") if cleaned.startswith("spotify:"): parts = cleaned.split(":") if len(parts) >= 3: - item_type = parts[1] - if expected_type and item_type != expected_type: - raise SpotifyError(f"Expected a Spotify {expected_type}, got {item_type}.") + _check_type(parts[1], expected_type) return parts[2] if "open.spotify.com" in cleaned: - parsed = urlparse(cleaned) - path_parts = [part for part in parsed.path.split("/") if part] + path_parts = [part for part in urlparse(cleaned).path.split("/") if part] if len(path_parts) >= 2: - item_type, item_id = path_parts[0], path_parts[1] - if expected_type and item_type != expected_type: - raise SpotifyError(f"Expected a Spotify {expected_type}, got {item_type}.") - return item_id + _check_type(path_parts[0], expected_type) + return path_parts[1] return cleaned def normalize_spotify_uri(value: str, expected_type: Optional[str] = None) -> str: + """Like normalize_spotify_id but returns a URI; bare ids need *expected_type* to become one.""" cleaned = (value or "").strip() if not cleaned: raise SpotifyError("Spotify URI/url/id is required.") if cleaned.startswith("spotify:"): - if expected_type: - parts = cleaned.split(":") - if len(parts) >= 3 and parts[1] != expected_type: - raise SpotifyError(f"Expected a Spotify {expected_type}, got {parts[1]}.") + parts = cleaned.split(":") + if expected_type and len(parts) >= 3: + _check_type(parts[1], expected_type) return cleaned item_id = normalize_spotify_id(cleaned, expected_type) - if expected_type: - return f"spotify:{expected_type}:{item_id}" - return cleaned + return f"spotify:{expected_type}:{item_id}" if expected_type else cleaned def normalize_spotify_uris(values: Iterable[str], expected_type: Optional[str] = None) -> list[str]: - uris: list[str] = [] - for value in values: - uri = normalize_spotify_uri(str(value), expected_type) - if uri not in uris: - uris.append(uri) + """Normalize each value, dropping duplicates while keeping first-seen order.""" + uris = list(dict.fromkeys(normalize_spotify_uri(str(value), expected_type) for value in values)) if not uris: raise SpotifyError("At least one Spotify item is required.") return uris - - -def compact_json(data: Any) -> str: - return json.dumps(data, ensure_ascii=False) diff --git a/plugins/spotify/tools.py b/plugins/spotify/tools.py index 4bd18a02b6..8301002847 100644 --- a/plugins/spotify/tools.py +++ b/plugins/spotify/tools.py @@ -1,13 +1,16 @@ -"""Native Spotify tools for Hermes (registered via plugins/spotify).""" +"""Native Spotify tools for Hermes (registered via plugins/spotify). + +Each tool routes ``args["action"]`` through a dict dispatch table; every entry +has the signature ``(client, args, action) -> str`` and issues its Spotify Web +API call through ``client.request`` (auth refresh + error mapping live there). +""" from __future__ import annotations -from typing import Any, List +from typing import Any, Callable, Dict, List, Optional from hermes_cli.auth import get_auth_status from plugins.spotify.client import ( - SpotifyAPIError, - SpotifyAuthRequiredError, SpotifyClient, SpotifyError, normalize_spotify_id, @@ -16,6 +19,8 @@ from plugins.spotify.client import ( ) from tools.registry import tool_error, tool_result +_Handler = Callable[[SpotifyClient, dict, str], str] + def _check_spotify_available() -> bool: try: @@ -24,26 +29,12 @@ def _check_spotify_available() -> bool: return False -def _spotify_client() -> SpotifyClient: - return SpotifyClient() - - def _spotify_tool_error(exc: Exception) -> str: - if isinstance(exc, (SpotifyError, SpotifyAuthRequiredError)): + if isinstance(exc, SpotifyError): # includes SpotifyAPIError / SpotifyAuthRequiredError return tool_error(str(exc)) - if isinstance(exc, SpotifyAPIError): - return tool_error(str(exc), status_code=exc.status_code) return tool_error(f"Spotify tool failed: {type(exc).__name__}: {exc}") -def _coerce_limit(raw: Any, *, default: int = 20, minimum: int = 1, maximum: int = 50) -> int: - try: - value = int(raw) - except Exception: - value = default - return max(minimum, min(maximum, value)) - - def _coerce_bool(raw: Any, default: bool = False) -> bool: if isinstance(raw, bool): return raw @@ -59,396 +50,364 @@ def _coerce_bool(raw: Any, default: bool = False) -> bool: def _as_list(raw: Any) -> List[str]: if raw is None: return [] - if isinstance(raw, list): - return [str(item).strip() for item in raw if str(item).strip()] - return [str(raw).strip()] if str(raw).strip() else [] + items = raw if isinstance(raw, list) else [raw] + return [str(item).strip() for item in items if str(item).strip()] -def _describe_empty_playback(payload: Any, *, action: str) -> dict | None: - if not isinstance(payload, dict) or not payload.get("empty"): - return None - if action == "get_currently_playing": - return { +def _offset(args: dict) -> int: + return max(0, int(args.get("offset") or 0)) + + +def _limit(args: dict, default: int = 20) -> int: + """Clamp ``limit`` to Spotify's 1..50 window; non-numeric input falls back to *default*.""" + raw: Any = args.get("limit") + try: + value = int(raw) + except Exception: + value = default + return max(1, min(50, value)) + + +def _ok(action: str, result: Any, **extra: Any) -> str: + return tool_result({"success": True, "action": action, **extra, "result": result}) + + +def _dispatcher(tool_name: str, default: str, table: Dict[str, _Handler], prepare: Optional[Callable[[dict], dict]] = None): + """Build a tool handler that routes ``args['action']`` through *table*. + + The client is constructed outside the error boundary (auth failures propagate + as exceptions, as they always have). *prepare* runs inside the boundary before + the action lookup, so its validation errors surface even for unknown actions. + """ + def handle(args: dict, **kw) -> str: + action = str(args.get("action") or default).strip().lower() + client = SpotifyClient() + try: + if prepare is not None: + args = prepare(args) + handler = table.get(action) + if handler is None: + return tool_error(f"Unknown {tool_name} action: {action}") + return handler(client, args, action) + except Exception as exc: + return _spotify_tool_error(exc) + handle.__name__ = handle.__qualname__ = f"_handle_{tool_name}" + return handle + + +# -- spotify_playback --------------------------------------------------------- + +# action -> (flag key reported False, fallback message) when Spotify returns 204/empty. +_EMPTY_PLAYBACK = { + "get_currently_playing": ("is_playing", "Spotify is not currently playing anything."), + "get_state": ("has_active_device", "No active Spotify playback session was found."), +} + + +def _pb_read(fetch: Callable[..., Any], args: dict, action: str) -> str: + payload = fetch(market=args.get("market")) + if isinstance(payload, dict) and payload.get("empty"): + flag, fallback = _EMPTY_PLAYBACK[action] + payload = { "success": True, "action": action, - "is_playing": False, + flag: False, "status_code": payload.get("status_code", 204), - "message": payload.get("message") or "Spotify is not currently playing anything.", + "message": payload.get("message") or fallback, } - if action == "get_state": - return { - "success": True, - "action": action, - "has_active_device": False, - "status_code": payload.get("status_code", 204), - "message": payload.get("message") or "No active Spotify playback session was found.", - } - return None + return tool_result(payload) -def _handle_spotify_playback(args: dict, **kw) -> str: - action = str(args.get("action") or "get_state").strip().lower() - client = _spotify_client() - try: - if action == "get_state": - payload = client.get_playback_state(market=args.get("market")) - empty_result = _describe_empty_playback(payload, action=action) - return tool_result(empty_result or payload) - if action == "get_currently_playing": - payload = client.get_currently_playing(market=args.get("market")) - empty_result = _describe_empty_playback(payload, action=action) - return tool_result(empty_result or payload) - if action == "play": - offset = args.get("offset") - if isinstance(offset, dict): - payload_offset = {k: v for k, v in offset.items() if v is not None} - else: - payload_offset = None - uris = normalize_spotify_uris(_as_list(args.get("uris")), "track") if args.get("uris") else None - context_uri = None - if args.get("context_uri"): - raw_context = str(args.get("context_uri")) - context_type = None - if raw_context.startswith("spotify:album:") or "/album/" in raw_context: - context_type = "album" - elif raw_context.startswith("spotify:playlist:") or "/playlist/" in raw_context: - context_type = "playlist" - elif raw_context.startswith("spotify:artist:") or "/artist/" in raw_context: - context_type = "artist" - context_uri = normalize_spotify_uri(raw_context, context_type) - result = client.start_playback( - device_id=args.get("device_id"), - context_uri=context_uri, - uris=uris, - offset=payload_offset, - position_ms=args.get("position_ms"), - ) - return tool_result({"success": True, "action": action, "result": result}) - if action == "pause": - result = client.pause_playback(device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "next": - result = client.skip_next(device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "previous": - result = client.skip_previous(device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "seek": - if args.get("position_ms") is None: - return tool_error("position_ms is required for action='seek'") - result = client.seek(position_ms=int(args["position_ms"]), device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "set_repeat": - state = str(args.get("state") or "").strip().lower() - if state not in {"track", "context", "off"}: - return tool_error("state must be one of: track, context, off") - result = client.set_repeat(state=state, device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "set_shuffle": - result = client.set_shuffle(state=_coerce_bool(args.get("state")), device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "set_volume": - if args.get("volume_percent") is None: - return tool_error("volume_percent is required for action='set_volume'") - result = client.set_volume(volume_percent=max(0, min(100, int(args["volume_percent"]))), device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "result": result}) - if action == "recently_played": - after = args.get("after") - before = args.get("before") - if after and before: - return tool_error("Provide only one of 'after' or 'before'") - return tool_result(client.get_recently_played( - limit=_coerce_limit(args.get("limit"), default=20), - after=int(after) if after is not None else None, - before=int(before) if before is not None else None, - )) - return tool_error(f"Unknown spotify_playback action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) +_CONTEXT_TYPES = (("album", "spotify:album:", "/album/"), ("playlist", "spotify:playlist:", "/playlist/"), ("artist", "spotify:artist:", "/artist/")) -def _handle_spotify_devices(args: dict, **kw) -> str: - action = str(args.get("action") or "list").strip().lower() - client = _spotify_client() - try: - if action == "list": - return tool_result(client.get_devices()) - if action == "transfer": - device_id = str(args.get("device_id") or "").strip() - if not device_id: - return tool_error("device_id is required for action='transfer'") - result = client.transfer_playback(device_id=device_id, play=_coerce_bool(args.get("play"))) - return tool_result({"success": True, "action": action, "result": result}) - return tool_error(f"Unknown spotify_devices action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) +def _pb_play(client: SpotifyClient, args: dict, action: str) -> str: + offset = args.get("offset") + payload_offset = {k: v for k, v in offset.items() if v is not None} if isinstance(offset, dict) else None + uris = normalize_spotify_uris(_as_list(args.get("uris")), "track") if args.get("uris") else None + context_uri = None + if args.get("context_uri"): + raw = str(args.get("context_uri")) + # Infer the context type so mismatches raise; unknown kinds pass through unchecked. + context_type = next((t for t, prefix, frag in _CONTEXT_TYPES if raw.startswith(prefix) or frag in raw), None) + context_uri = normalize_spotify_uri(raw, context_type) + body = {"context_uri": context_uri, "uris": uris, "offset": payload_offset, "position_ms": args.get("position_ms")} + return _ok(action, client.request("PUT", "/me/player/play", params={"device_id": args.get("device_id")}, json_body=body)) -def _handle_spotify_queue(args: dict, **kw) -> str: - action = str(args.get("action") or "get").strip().lower() - client = _spotify_client() - try: - if action == "get": - return tool_result(client.get_queue()) - if action == "add": - uri = normalize_spotify_uri(str(args.get("uri") or ""), None) - result = client.add_to_queue(uri=uri, device_id=args.get("device_id")) - return tool_result({"success": True, "action": action, "uri": uri, "result": result}) - return tool_error(f"Unknown spotify_queue action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) +def _pb_device_cmd(method: str, path: str, **extra: Any) -> _Handler: + """Handler for a player command whose only free argument is ``device_id`` (*extra* = fixed params).""" + return lambda c, a, act: _ok(act, c.request(method, path, params={**extra, "device_id": a.get("device_id")})) + + +def _pb_seek(client: SpotifyClient, args: dict, action: str) -> str: + if args.get("position_ms") is None: + return tool_error("position_ms is required for action='seek'") + return _pb_device_cmd("PUT", "/me/player/seek", position_ms=int(args["position_ms"]))(client, args, action) + + +def _pb_set_repeat(client: SpotifyClient, args: dict, action: str) -> str: + state = str(args.get("state") or "").strip().lower() + if state not in {"track", "context", "off"}: + return tool_error("state must be one of: track, context, off") + return _pb_device_cmd("PUT", "/me/player/repeat", state=state)(client, args, action) + + +def _pb_set_shuffle(client: SpotifyClient, args: dict, action: str) -> str: + state = str(_coerce_bool(args.get("state"))).lower() + return _pb_device_cmd("PUT", "/me/player/shuffle", state=state)(client, args, action) + + +def _pb_set_volume(client: SpotifyClient, args: dict, action: str) -> str: + if args.get("volume_percent") is None: + return tool_error("volume_percent is required for action='set_volume'") + volume = max(0, min(100, int(args["volume_percent"]))) + return _pb_device_cmd("PUT", "/me/player/volume", volume_percent=volume)(client, args, action) + + +def _pb_recently_played(client: SpotifyClient, args: dict, action: str) -> str: + after, before = args.get("after"), args.get("before") + if after and before: + return tool_error("Provide only one of 'after' or 'before'") + params = { + "limit": _limit(args), + "after": int(after) if after is not None else None, + "before": int(before) if before is not None else None, + } + return tool_result(client.request("GET", "/me/player/recently-played", params=params)) + + +_handle_spotify_playback = _dispatcher("spotify_playback", "get_state", { + "get_state": lambda c, a, act: _pb_read(c.get_playback_state, a, act), + "get_currently_playing": lambda c, a, act: _pb_read(c.get_currently_playing, a, act), + "play": _pb_play, + "pause": _pb_device_cmd("PUT", "/me/player/pause"), + "next": _pb_device_cmd("POST", "/me/player/next"), + "previous": _pb_device_cmd("POST", "/me/player/previous"), + "seek": _pb_seek, + "set_repeat": _pb_set_repeat, + "set_shuffle": _pb_set_shuffle, + "set_volume": _pb_set_volume, + "recently_played": _pb_recently_played, +}) + + +# -- spotify_devices / spotify_queue / spotify_search --------------------------- + +def _dev_transfer(client: SpotifyClient, args: dict, action: str) -> str: + device_id = str(args.get("device_id") or "").strip() + if not device_id: + return tool_error("device_id is required for action='transfer'") + body = {"device_ids": [device_id], "play": _coerce_bool(args.get("play"))} + return _ok(action, client.request("PUT", "/me/player", json_body=body)) + + +_handle_spotify_devices = _dispatcher("spotify_devices", "list", { + "list": lambda c, a, act: tool_result(c.request("GET", "/me/player/devices")), + "transfer": _dev_transfer, +}) + + +def _queue_add(client: SpotifyClient, args: dict, action: str) -> str: + uri = normalize_spotify_uri(str(args.get("uri") or ""), None) + result = client.request("POST", "/me/player/queue", params={"uri": uri, "device_id": args.get("device_id")}) + return _ok(action, result, uri=uri) + + +_handle_spotify_queue = _dispatcher("spotify_queue", "get", { + "get": lambda c, a, act: tool_result(c.request("GET", "/me/player/queue")), + "add": _queue_add, +}) + +_SEARCH_TYPES = {"album", "artist", "playlist", "track", "show", "episode", "audiobook"} def _handle_spotify_search(args: dict, **kw) -> str: - client = _spotify_client() + client = SpotifyClient() query = str(args.get("query") or "").strip() if not query: return tool_error("query is required") raw_types = _as_list(args.get("types") or args.get("type") or ["track"]) - search_types = [value.lower() for value in raw_types if value.lower() in {"album", "artist", "playlist", "track", "show", "episode", "audiobook"}] + search_types = [value.lower() for value in raw_types if value.lower() in _SEARCH_TYPES] if not search_types: return tool_error("types must contain one or more of: album, artist, playlist, track, show, episode, audiobook") + params = { + "q": query, "type": ",".join(search_types), "limit": _limit(args, 10), "offset": _offset(args), + "market": args.get("market"), "include_external": args.get("include_external"), + } try: - return tool_result(client.search( - query=query, - search_types=search_types, - limit=_coerce_limit(args.get("limit"), default=10), - offset=max(0, int(args.get("offset") or 0)), - market=args.get("market"), - include_external=args.get("include_external"), - )) + return tool_result(client.request("GET", "/search", params=params)) except Exception as exc: return _spotify_tool_error(exc) -def _handle_spotify_playlists(args: dict, **kw) -> str: - action = str(args.get("action") or "list").strip().lower() - client = _spotify_client() - try: - if action == "list": - return tool_result(client.get_my_playlists( - limit=_coerce_limit(args.get("limit"), default=20), - offset=max(0, int(args.get("offset") or 0)), - )) - if action == "get": - playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist") - return tool_result(client.get_playlist(playlist_id=playlist_id, market=args.get("market"))) - if action == "create": - name = str(args.get("name") or "").strip() - if not name: - return tool_error("name is required for action='create'") - return tool_result(client.create_playlist( - name=name, - public=_coerce_bool(args.get("public")), - collaborative=_coerce_bool(args.get("collaborative")), - description=args.get("description"), - )) - if action == "add_items": - playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist") - uris = normalize_spotify_uris(_as_list(args.get("uris"))) - return tool_result(client.add_playlist_items( - playlist_id=playlist_id, - uris=uris, - position=args.get("position"), - )) - if action == "remove_items": - playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist") - uris = normalize_spotify_uris(_as_list(args.get("uris"))) - return tool_result(client.remove_playlist_items( - playlist_id=playlist_id, - uris=uris, - snapshot_id=args.get("snapshot_id"), - )) - if action == "update_details": - playlist_id = normalize_spotify_id(str(args.get("playlist_id") or ""), "playlist") - return tool_result(client.update_playlist_details( - playlist_id=playlist_id, - name=args.get("name"), - public=args.get("public"), - collaborative=args.get("collaborative"), - description=args.get("description"), - )) - return tool_error(f"Unknown spotify_playlists action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) +# -- spotify_playlists --------------------------------------------------------- + +def _playlist_path(args: dict, suffix: str = "") -> str: + return f"/playlists/{normalize_spotify_id(str(args.get('playlist_id') or ''), 'playlist')}{suffix}" -def _handle_spotify_albums(args: dict, **kw) -> str: - action = str(args.get("action") or "get").strip().lower() - client = _spotify_client() - try: - album_id = normalize_spotify_id(str(args.get("album_id") or args.get("id") or ""), "album") - if action == "get": - return tool_result(client.get_album(album_id=album_id, market=args.get("market"))) - if action == "tracks": - return tool_result(client.get_album_tracks( - album_id=album_id, - limit=_coerce_limit(args.get("limit"), default=20), - offset=max(0, int(args.get("offset") or 0)), - market=args.get("market"), - )) - return tool_error(f"Unknown spotify_albums action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) +def _pl_create(client: SpotifyClient, args: dict, action: str) -> str: + name = str(args.get("name") or "").strip() + if not name: + return tool_error("name is required for action='create'") + body = { + "name": name, "public": _coerce_bool(args.get("public")), + "collaborative": _coerce_bool(args.get("collaborative")), "description": args.get("description"), + } + return tool_result(client.request("POST", "/me/playlists", json_body=body)) + + +def _pl_remove_items(client: SpotifyClient, args: dict, action: str) -> str: + path = _playlist_path(args, "/items") + body = {"items": [{"uri": u} for u in normalize_spotify_uris(_as_list(args.get("uris")))], "snapshot_id": args.get("snapshot_id")} + return tool_result(client.request("DELETE", path, json_body=body)) + + +_handle_spotify_playlists = _dispatcher("spotify_playlists", "list", { + "list": lambda c, a, act: tool_result(c.request("GET", "/me/playlists", params={"limit": _limit(a), "offset": _offset(a)})), + "get": lambda c, a, act: tool_result(c.request("GET", _playlist_path(a), params={"market": a.get("market")})), + "create": _pl_create, + "add_items": lambda c, a, act: tool_result(c.request("POST", _playlist_path(a, "/items"), json_body={ + "uris": normalize_spotify_uris(_as_list(a.get("uris"))), "position": a.get("position"), + })), + "remove_items": _pl_remove_items, + "update_details": lambda c, a, act: tool_result(c.request("PUT", _playlist_path(a), json_body={ + "name": a.get("name"), "public": a.get("public"), "collaborative": a.get("collaborative"), "description": a.get("description"), + })), +}) + + +# -- spotify_albums ------------------------------------------------------------ + +def _page_params(args: dict) -> dict: + return {"limit": _limit(args), "offset": _offset(args), "market": args.get("market")} + + +def _prepare_album(args: dict) -> dict: + return {**args, "_path": f"/albums/{normalize_spotify_id(str(args.get('album_id') or args.get('id') or ''), 'album')}"} + + +_handle_spotify_albums = _dispatcher("spotify_albums", "get", { + "get": lambda c, a, act: tool_result(c.request("GET", a["_path"], params={"market": a.get("market")})), + "tracks": lambda c, a, act: tool_result(c.request("GET", a["_path"] + "/tracks", params=_page_params(a))), +}, prepare=_prepare_album) + + +# -- spotify_library — saved tracks + saved albums, selected by `kind` --------- + +def _lib_list(client: SpotifyClient, args: dict, action: str) -> str: + return tool_result(client.request("GET", f"/me/{args['kind']}", params=_page_params(args))) + + +def _lib_save(client: SpotifyClient, args: dict, action: str) -> str: + uris = normalize_spotify_uris(_as_list(args.get("uris") or args.get("items")), args["_item_type"]) + return tool_result(client.request("PUT", "/me/library", params={"uris": ",".join(uris)})) + + +def _lib_remove(client: SpotifyClient, args: dict, action: str) -> str: + item_type = args["_item_type"] + ids = [normalize_spotify_id(item, item_type) for item in _as_list(args.get("ids") or args.get("items"))] + if not ids: + return tool_error("ids/items is required for action='remove'") + uris = ",".join(f"spotify:{item_type}:{i}" for i in ids) + return tool_result(client.request("DELETE", "/me/library", params={"uris": uris})) + + +_dispatch_library = _dispatcher("spotify_library", "list", {"list": _lib_list, "save": _lib_save, "remove": _lib_remove}) def _handle_spotify_library(args: dict, **kw) -> str: - """Unified handler for saved tracks + saved albums (formerly two tools).""" kind = str(args.get("kind") or "").strip().lower() if kind not in {"tracks", "albums"}: return tool_error("kind must be one of: tracks, albums") - action = str(args.get("action") or "list").strip().lower() - item_type = "track" if kind == "tracks" else "album" - client = _spotify_client() - try: - if action == "list": - limit = _coerce_limit(args.get("limit"), default=20) - offset = max(0, int(args.get("offset") or 0)) - market = args.get("market") - if kind == "tracks": - return tool_result(client.get_saved_tracks(limit=limit, offset=offset, market=market)) - return tool_result(client.get_saved_albums(limit=limit, offset=offset, market=market)) - if action == "save": - uris = normalize_spotify_uris(_as_list(args.get("uris") or args.get("items")), item_type) - return tool_result(client.save_library_items(uris=uris)) - if action == "remove": - ids = [normalize_spotify_id(item, item_type) for item in _as_list(args.get("ids") or args.get("items"))] - if not ids: - return tool_error("ids/items is required for action='remove'") - if kind == "tracks": - return tool_result(client.remove_saved_tracks(track_ids=ids)) - return tool_result(client.remove_saved_albums(album_ids=ids)) - return tool_error(f"Unknown spotify_library action: {action}") - except Exception as exc: - return _spotify_tool_error(exc) + return _dispatch_library({**args, "kind": kind, "_item_type": kind[:-1]}) # tracks->track, albums->album +# -- Schemas (sent to the model — byte-stable; property order matters) ---------- + COMMON_STRING = {"type": "string"} +_INT = {"type": "integer"} +_BOOL = {"type": "boolean"} +_STR_ARRAY = {"type": "array", "items": COMMON_STRING} -SPOTIFY_PLAYBACK_SCHEMA = { - "name": "spotify_playback", - "description": "Control Spotify playback, inspect the active playback state, or fetch recently played tracks.", - "parameters": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"]}, - "device_id": COMMON_STRING, - "market": COMMON_STRING, - "context_uri": COMMON_STRING, - "uris": {"type": "array", "items": COMMON_STRING}, - "offset": {"type": "object"}, - "position_ms": {"type": "integer"}, - "state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]}, - "volume_percent": {"type": "integer"}, - "limit": {"type": "integer", "description": "For recently_played: number of tracks (max 50)"}, - "after": {"type": "integer", "description": "For recently_played: Unix ms cursor (after this timestamp)"}, - "before": {"type": "integer", "description": "For recently_played: Unix ms cursor (before this timestamp)"}, - }, - "required": ["action"], - }, -} -SPOTIFY_DEVICES_SCHEMA = { - "name": "spotify_devices", - "description": "List Spotify Connect devices or transfer playback to a different device.", - "parameters": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["list", "transfer"]}, - "device_id": COMMON_STRING, - "play": {"type": "boolean"}, - }, - "required": ["action"], - }, -} +def _strs(*names: str) -> dict: + return dict.fromkeys(names, COMMON_STRING) -SPOTIFY_QUEUE_SCHEMA = { - "name": "spotify_queue", - "description": "Inspect the user's Spotify queue or add an item to it.", - "parameters": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["get", "add"]}, - "uri": COMMON_STRING, - "device_id": COMMON_STRING, - }, - "required": ["action"], - }, -} -SPOTIFY_SEARCH_SCHEMA = { - "name": "spotify_search", - "description": "Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.", - "parameters": { - "type": "object", - "properties": { - "query": COMMON_STRING, - "types": {"type": "array", "items": COMMON_STRING}, - "type": COMMON_STRING, - "limit": {"type": "integer"}, - "offset": {"type": "integer"}, - "market": COMMON_STRING, - "include_external": COMMON_STRING, - }, - "required": ["query"], - }, -} +def _enum(*values: str) -> dict: + return {"type": "string", "enum": list(values)} -SPOTIFY_PLAYLISTS_SCHEMA = { - "name": "spotify_playlists", - "description": "List, inspect, create, update, and modify Spotify playlists.", - "parameters": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["list", "get", "create", "add_items", "remove_items", "update_details"]}, - "playlist_id": COMMON_STRING, - "market": COMMON_STRING, - "limit": {"type": "integer"}, - "offset": {"type": "integer"}, - "name": COMMON_STRING, - "description": COMMON_STRING, - "public": {"type": "boolean"}, - "collaborative": {"type": "boolean"}, - "uris": {"type": "array", "items": COMMON_STRING}, - "position": {"type": "integer"}, - "snapshot_id": COMMON_STRING, - }, - "required": ["action"], - }, -} -SPOTIFY_ALBUMS_SCHEMA = { - "name": "spotify_albums", - "description": "Fetch Spotify album metadata or album tracks.", - "parameters": { - "type": "object", - "properties": { - "action": {"type": "string", "enum": ["get", "tracks"]}, - "album_id": COMMON_STRING, - "id": COMMON_STRING, - "market": COMMON_STRING, - "limit": {"type": "integer"}, - "offset": {"type": "integer"}, - }, - "required": ["action"], - }, -} +def _schema(name: str, description: str, properties: dict, required: list) -> dict: + return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}} -SPOTIFY_LIBRARY_SCHEMA = { - "name": "spotify_library", - "description": "List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.", - "parameters": { - "type": "object", - "properties": { - "kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"}, - "action": {"type": "string", "enum": ["list", "save", "remove"]}, - "limit": {"type": "integer"}, - "offset": {"type": "integer"}, - "market": COMMON_STRING, - "uris": {"type": "array", "items": COMMON_STRING}, - "ids": {"type": "array", "items": COMMON_STRING}, - "items": {"type": "array", "items": COMMON_STRING}, - }, - "required": ["kind", "action"], + +SPOTIFY_PLAYBACK_SCHEMA = _schema( + "spotify_playback", + "Control Spotify playback, inspect the active playback state, or fetch recently played tracks.", + { + "action": _enum("get_state", "get_currently_playing", "play", "pause", "next", "previous", "seek", "set_repeat", "set_shuffle", "set_volume", "recently_played"), + **_strs("device_id", "market", "context_uri"), + "uris": _STR_ARRAY, + "offset": {"type": "object"}, + "position_ms": _INT, + "state": {"description": "For set_repeat use track/context/off. For set_shuffle use boolean-like true/false.", "oneOf": [{"type": "string"}, {"type": "boolean"}]}, + "volume_percent": _INT, + "limit": {"type": "integer", "description": "For recently_played: number of tracks (max 50)"}, + "after": {"type": "integer", "description": "For recently_played: Unix ms cursor (after this timestamp)"}, + "before": {"type": "integer", "description": "For recently_played: Unix ms cursor (before this timestamp)"}, }, -} + ["action"], +) + +SPOTIFY_DEVICES_SCHEMA = _schema( + "spotify_devices", + "List Spotify Connect devices or transfer playback to a different device.", + {"action": _enum("list", "transfer"), "device_id": COMMON_STRING, "play": _BOOL}, + ["action"], +) + +SPOTIFY_QUEUE_SCHEMA = _schema( + "spotify_queue", + "Inspect the user's Spotify queue or add an item to it.", + {"action": _enum("get", "add"), **_strs("uri", "device_id")}, + ["action"], +) + +SPOTIFY_SEARCH_SCHEMA = _schema( + "spotify_search", + "Search the Spotify catalog for tracks, albums, artists, playlists, shows, or episodes.", + {"query": COMMON_STRING, "types": _STR_ARRAY, "type": COMMON_STRING, "limit": _INT, "offset": _INT, **_strs("market", "include_external")}, + ["query"], +) + +SPOTIFY_PLAYLISTS_SCHEMA = _schema( + "spotify_playlists", + "List, inspect, create, update, and modify Spotify playlists.", + { + "action": _enum("list", "get", "create", "add_items", "remove_items", "update_details"), + **_strs("playlist_id", "market"), "limit": _INT, "offset": _INT, **_strs("name", "description"), + "public": _BOOL, "collaborative": _BOOL, "uris": _STR_ARRAY, "position": _INT, "snapshot_id": COMMON_STRING, + }, + ["action"], +) + +SPOTIFY_ALBUMS_SCHEMA = _schema( + "spotify_albums", + "Fetch Spotify album metadata or album tracks.", + {"action": _enum("get", "tracks"), **_strs("album_id", "id", "market"), "limit": _INT, "offset": _INT}, + ["action"], +) + +SPOTIFY_LIBRARY_SCHEMA = _schema( + "spotify_library", + "List, save, or remove the user's saved Spotify tracks or albums. Use `kind` to select which.", + { + "kind": {"type": "string", "enum": ["tracks", "albums"], "description": "Which library to operate on"}, + "action": _enum("list", "save", "remove"), + "limit": _INT, "offset": _INT, "market": COMMON_STRING, "uris": _STR_ARRAY, "ids": _STR_ARRAY, "items": _STR_ARRAY, + }, + ["kind", "action"], +) diff --git a/tests/tools/test_spotify_client.py b/tests/tools/test_spotify_client.py index 3271b474a2..763438f6a3 100644 --- a/tests/tools/test_spotify_client.py +++ b/tests/tools/test_spotify_client.py @@ -59,7 +59,7 @@ def test_spotify_client_retries_once_after_401(monkeypatch: pytest.MonkeyPatch) monkeypatch.setattr(spotify_mod.httpx, "request", fake_request) client = spotify_mod.SpotifyClient() - payload = client.get_devices() + payload = client.request("GET", "/me/player/devices") assert payload["devices"][0]["id"] == "dev-1" assert calls == ["Bearer token-1", "Bearer token-2"]