refactor: move Spotify out of core into the spotify catalog plugin
The spotify toolset (plugins/spotify) and its loopback-PKCE login (hermes_cli/auth_spotify.py)
now live in NousResearch/hermes-spotify, installed from the plugin catalog. Same pattern as the
Home Assistant move (fb9fc8a3dd): a LEFT_CORE row installs the plugin for every home that was
using it (a Spotify login in auth.json, or the spotify toolset in platform_toolsets) on
`hermes update` / first start; the toolset name, tool names, schemas, env vars and stored login
(providers.spotify) are unchanged, so nobody re-authenticates.
The one user-visible change is the login command: plugins cannot extend `hermes auth`, so the
plugin registers `hermes spotify login|status|logout`. LeftCoreFeature gains a `cli` field and
moved_command_hint() turns `hermes auth [status|logout] spotify`, and `hermes spotify` while the
plugin is not loaded, into a line naming the new command and the install/enable step for this
home. The dashboard's "needs login" hint names the new command too.
Core drops the Spotify auth constants/dispatch, the CONFIGURABLE_TOOLSETS row and post-setup
step, the core `spotify` entries in the shared-metrics toolset enums, its tests, and the docs
(en + zh-Hans) now point at `hermes plugins install spotify` / `hermes spotify login`.
This commit is contained in:
Teknium
@@ -26,7 +26,8 @@ command. A hook with no concrete consumer is speculative infrastructure and is r
|
||||
backends ship as standalone repos implementing the same `MemoryProvider` ABC, discovered through
|
||||
the same path, integrated via `hermes memory setup` / `post_setup()`.
|
||||
- **Other features that leave core** (Home Assistant platform + toolset, Oct 2026 →
|
||||
`plugin-catalog/homeassistant.yaml`) keep their names/config/env and get a row in
|
||||
`plugin-catalog/homeassistant.yaml`; Spotify toolset → `plugin-catalog/spotify.yaml`) keep their
|
||||
names/config/env and get a row in
|
||||
`hermes_cli/left_core_migration.py::LEFT_CORE` (catalog plugin + read-only "this home uses it"
|
||||
predicate); `hermes update` and agent/gateway start install the plugin for homes that used it.
|
||||
Installing and converting core-era state are separate steps: the toolset-scope conversion runs
|
||||
@@ -34,7 +35,8 @@ command. A hook with no concrete consumer is speculative infrastructure and is r
|
||||
The automatic install also happens once per home and row (`_left_core_installed`, written only
|
||||
once the plugin dir exists): a later `hermes plugins remove` sticks, a failed install retries.
|
||||
Core special cases become `PlatformEntry` seams the plugin sets (`trusted_inbound`,
|
||||
`display_tier`, `shared_env_prefixes`), never a name check.
|
||||
`display_tier`, `shared_env_prefixes`), never a name check. A core `hermes auth <name>` login
|
||||
becomes the plugin's own `hermes <name>` command (row field `cli`; the old spelling prints the new one).
|
||||
- **No new third-party-product plugins (June 2026).** Observability/metrics backends, vendor SaaS
|
||||
connectors, analytics dashboards, paid-service tie-ins ship as standalone plugin repos
|
||||
(`~/.hermes/plugins/` or pip entry point) promoted in Discord `#plugins-skills-and-skins`. Reason:
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
"""Spotify integration plugin — bundled, auto-loaded.
|
||||
|
||||
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 rather than a ``tools/`` module: ``tools/`` is reserved for
|
||||
foundational capabilities; third-party service integrations live under
|
||||
``plugins/`` (flat ``plugins/<name>/`` for standalones, like image_gen backends),
|
||||
and ``kind: backend`` bundled plugins auto-load with no ``plugins.enabled`` opt-in.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from plugins.spotify import tools as _t
|
||||
|
||||
_TOOLS = (
|
||||
("spotify_playback", _t.SPOTIFY_PLAYBACK_SCHEMA, _t._handle_spotify_playback, "🎵"),
|
||||
("spotify_devices", _t.SPOTIFY_DEVICES_SCHEMA, _t._handle_spotify_devices, "🔈"),
|
||||
("spotify_queue", _t.SPOTIFY_QUEUE_SCHEMA, _t._handle_spotify_queue, "📻"),
|
||||
("spotify_search", _t.SPOTIFY_SEARCH_SCHEMA, _t._handle_spotify_search, "🔎"),
|
||||
("spotify_playlists", _t.SPOTIFY_PLAYLISTS_SCHEMA, _t._handle_spotify_playlists, "📚"),
|
||||
("spotify_albums", _t.SPOTIFY_ALBUMS_SCHEMA, _t._handle_spotify_albums, "💿"),
|
||||
("spotify_library", _t.SPOTIFY_LIBRARY_SCHEMA, _t._handle_spotify_library, "❤️"),
|
||||
)
|
||||
|
||||
|
||||
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=_t._check_spotify_available, emoji=emoji)
|
||||
@@ -1,161 +0,0 @@
|
||||
"""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
|
||||
|
||||
from typing import Any, Dict, Iterable, Optional
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
|
||||
from hermes_cli.auth import AuthError, resolve_spotify_runtime_credentials
|
||||
|
||||
|
||||
class SpotifyError(RuntimeError): """Base Spotify tool error."""
|
||||
class SpotifyAuthRequiredError(SpotifyError): """Raised when the user needs to authenticate with Spotify first."""
|
||||
|
||||
|
||||
class SpotifyAPIError(SpotifyError):
|
||||
"""Structured Spotify API failure."""
|
||||
|
||||
def __init__(self, message: str, *, status_code: Optional[int] = None, response_body: Optional[str] = None) -> None:
|
||||
super().__init__(message)
|
||||
self.status_code, self.response_body, self.path = status_code, response_body, None
|
||||
|
||||
|
||||
_empty_204 = lambda message: {"status_code": 204, "empty": True, "message": message}
|
||||
|
||||
|
||||
class SpotifyClient:
|
||||
def __init__(self) -> None:
|
||||
self._runtime = self._resolve_runtime(refresh_if_expiring=True)
|
||||
|
||||
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)
|
||||
except AuthError as exc:
|
||||
raise SpotifyAuthRequiredError(str(exc)) from exc
|
||||
|
||||
@property
|
||||
def base_url(self) -> str:
|
||||
return str(self._runtime.get("base_url") or "").rstrip("/")
|
||||
|
||||
def request(
|
||||
self, method: str, path: str, *, params: Optional[dict[str, Any]] = None, json_body: Optional[dict[str, Any]] = None,
|
||||
allow_retry_on_401: bool = True, empty_response: Optional[dict[str, Any]] = None,
|
||||
) -> Any:
|
||||
response = httpx.request(
|
||||
method, 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)
|
||||
if response.status_code >= 400:
|
||||
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}
|
||||
|
||||
# -- 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=_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=_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:
|
||||
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, 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 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:
|
||||
return "Spotify could not find an active playback device or player session for this request." if is_playback_path else "Spotify resource not found."
|
||||
if status_code == 429:
|
||||
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}."
|
||||
|
||||
|
||||
_strip_none = lambda payload: {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:<type>:<id>`` URI, or open.spotify.com URL; return the id."""
|
||||
cleaned = (value or "").strip()
|
||||
if not cleaned:
|
||||
raise SpotifyError("Spotify id/uri/url is required.")
|
||||
# (type, id) segments of a URI or URL; a bare id (or malformed ref) has no segments and passes through.
|
||||
parts = cleaned.split(":")[1:] if cleaned.startswith("spotify:") else []
|
||||
if len(parts) < 2 and "open.spotify.com" in cleaned:
|
||||
parts = [part for part in urlparse(cleaned).path.split("/") if part]
|
||||
if len(parts) >= 2:
|
||||
_check_type(parts[0], expected_type)
|
||||
return 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:"):
|
||||
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)
|
||||
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]:
|
||||
"""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
|
||||
@@ -1,13 +0,0 @@
|
||||
name: spotify
|
||||
version: 1.0.0
|
||||
description: "Native Spotify integration — 7 tools (playback, devices, queue, search, playlists, albums, library) using Spotify Web API + PKCE OAuth. Auth via `hermes auth spotify`. Tools gate on `providers.spotify` in ~/.hermes/auth.json."
|
||||
author: NousResearch
|
||||
kind: backend
|
||||
provides_tools:
|
||||
- spotify_playback
|
||||
- spotify_devices
|
||||
- spotify_queue
|
||||
- spotify_search
|
||||
- spotify_playlists
|
||||
- spotify_albums
|
||||
- spotify_library
|
||||
@@ -1,316 +0,0 @@
|
||||
"""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, Callable, Dict, List, Optional
|
||||
|
||||
from hermes_cli.auth import get_auth_status
|
||||
from plugins.spotify.client import (
|
||||
SpotifyClient, SpotifyError, normalize_spotify_id, normalize_spotify_uri, normalize_spotify_uris)
|
||||
from tools.registry import tool_error, tool_result
|
||||
|
||||
_Handler = Callable[[SpotifyClient, dict, str], str]
|
||||
|
||||
|
||||
def _check_spotify_available() -> bool:
|
||||
try:
|
||||
return bool(get_auth_status("spotify").get("logged_in"))
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _spotify_tool_error(exc: Exception) -> str:
|
||||
if isinstance(exc, SpotifyError): # includes SpotifyAPIError / SpotifyAuthRequiredError
|
||||
return tool_error(str(exc))
|
||||
return tool_error(f"Spotify tool failed: {type(exc).__name__}: {exc}")
|
||||
|
||||
|
||||
# Inside a ``_dispatcher`` boundary ``raise SpotifyError(msg)`` renders exactly like ``return tool_error(msg)``.
|
||||
def _required(value: Any, message: str) -> Any:
|
||||
if value is None:
|
||||
raise SpotifyError(message)
|
||||
return value
|
||||
|
||||
|
||||
def _nonblank(raw: Any, message: str) -> str:
|
||||
value = str(raw or "").strip()
|
||||
if not value:
|
||||
raise SpotifyError(message)
|
||||
return value
|
||||
|
||||
|
||||
def _coerce_bool(raw: Any, default: bool = False) -> bool:
|
||||
if isinstance(raw, bool):
|
||||
return raw
|
||||
if isinstance(raw, str):
|
||||
cleaned = raw.strip().lower()
|
||||
if cleaned in {"1", "true", "yes", "on"}:
|
||||
return True
|
||||
if cleaned in {"0", "false", "no", "off"}:
|
||||
return False
|
||||
return default
|
||||
|
||||
|
||||
def _as_list(raw: Any) -> list[str]:
|
||||
if raw is None:
|
||||
return []
|
||||
items = raw if isinstance(raw, list) else [raw]
|
||||
return [str(item).strip() for item in items if str(item).strip()]
|
||||
|
||||
|
||||
_offset = lambda args: 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))
|
||||
|
||||
|
||||
_ok = lambda action, result, **extra: 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, flag: False, "status_code": payload.get("status_code", 204),
|
||||
"message": payload.get("message") or fallback}
|
||||
return tool_result(payload)
|
||||
|
||||
|
||||
_CONTEXT_TYPES = (("album", "spotify:album:", "/album/"), ("playlist", "spotify:playlist:", "/playlist/"), ("artist", "spotify:artist:", "/artist/"))
|
||||
|
||||
|
||||
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 _pb_cmd(client: SpotifyClient, args: dict, action: str, method: str, path: str, **extra: Any) -> str:
|
||||
"""Player command whose only free argument is ``device_id`` (*extra* = fixed params)."""
|
||||
return _ok(action, client.request(method, path, params={**extra, "device_id": args.get("device_id")}))
|
||||
|
||||
|
||||
_pb_device_cmd = lambda method, path: (lambda c, a, act: _pb_cmd(c, a, act, method, path))
|
||||
|
||||
|
||||
def _pb_required_param(path: str, param: str, convert: Callable[[Any], Any]) -> _Handler:
|
||||
"""PUT *path* with ``param`` (required, converted) plus device_id."""
|
||||
return lambda c, a, act: _pb_cmd(c, a, act, "PUT", path, **{param: convert(_required(a.get(param), f"{param} is required for action='{act}'"))})
|
||||
|
||||
|
||||
def _pb_repeat_state(args: dict) -> str:
|
||||
state = str(args.get("state") or "").strip().lower()
|
||||
if state not in {"track", "context", "off"}:
|
||||
raise SpotifyError("state must be one of: track, context, off")
|
||||
return state
|
||||
|
||||
|
||||
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_required_param("/me/player/seek", "position_ms", int),
|
||||
"set_repeat": lambda c, a, act: _pb_cmd(c, a, act, "PUT", "/me/player/repeat", state=_pb_repeat_state(a)),
|
||||
"set_shuffle": lambda c, a, act: _pb_cmd(c, a, act, "PUT", "/me/player/shuffle", state=str(_coerce_bool(a.get("state"))).lower()),
|
||||
"set_volume": _pb_required_param("/me/player/volume", "volume_percent", lambda v: max(0, min(100, int(v)))),
|
||||
"recently_played": _pb_recently_played,
|
||||
})
|
||||
|
||||
|
||||
# -- spotify_devices / spotify_queue / spotify_search ---------------------------
|
||||
|
||||
_handle_spotify_devices = _dispatcher("spotify_devices", "list", {
|
||||
"list": lambda c, a, act: tool_result(c.request("GET", "/me/player/devices")),
|
||||
"transfer": lambda c, a, act: _ok(act, c.request("PUT", "/me/player", json_body={
|
||||
"device_ids": [_nonblank(a.get("device_id"), "device_id is required for action='transfer'")], "play": _coerce_bool(a.get("play")),
|
||||
})),
|
||||
})
|
||||
|
||||
|
||||
def _queue_add(client: SpotifyClient, args: dict, action: str) -> str:
|
||||
uri = normalize_spotify_uri(str(args.get("uri") or ""), None)
|
||||
return _ok(action, client.request("POST", "/me/player/queue", params={"uri": uri, "device_id": args.get("device_id")}), 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 = 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 _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.request("GET", "/search", params=params))
|
||||
except Exception as exc:
|
||||
return _spotify_tool_error(exc)
|
||||
|
||||
|
||||
# -- spotify_playlists ---------------------------------------------------------
|
||||
|
||||
_playlist_path = lambda args, suffix="": f"/playlists/{normalize_spotify_id(str(args.get('playlist_id') or ''), 'playlist')}{suffix}"
|
||||
|
||||
|
||||
_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": lambda c, a, act: tool_result(c.request("POST", "/me/playlists", json_body={
|
||||
"name": _nonblank(a.get("name"), "name is required for action='create'"), "public": _coerce_bool(a.get("public")),
|
||||
"collaborative": _coerce_bool(a.get("collaborative")), "description": a.get("description")})),
|
||||
"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": lambda c, a, act: tool_result(c.request("DELETE", _playlist_path(a, "/items"), json_body={
|
||||
"items": [{"uri": u} for u in normalize_spotify_uris(_as_list(a.get("uris")))], "snapshot_id": a.get("snapshot_id")})),
|
||||
"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 ------------------------------------------------------------
|
||||
|
||||
_page_params = lambda args: {"limit": _limit(args), "offset": _offset(args), "market": args.get("market")}
|
||||
_prepare_album = lambda args: {**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_remove_uris(args: dict) -> 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"))]
|
||||
return ",".join(f"spotify:{item_type}:{i}" for i in _required(ids or None, "ids/items is required for action='remove'"))
|
||||
|
||||
|
||||
_dispatch_library = _dispatcher("spotify_library", "list", {
|
||||
"list": lambda c, a, act: tool_result(c.request("GET", f"/me/{a['kind']}", params=_page_params(a))),
|
||||
"save": lambda c, a, act: tool_result(c.request("PUT", "/me/library", params={
|
||||
"uris": ",".join(normalize_spotify_uris(_as_list(a.get("uris") or a.get("items")), a["_item_type"]))})),
|
||||
"remove": lambda c, a, act: tool_result(c.request("DELETE", "/me/library", params={"uris": _lib_remove_uris(a)})),
|
||||
})
|
||||
|
||||
|
||||
def _handle_spotify_library(args: dict, **kw) -> str:
|
||||
kind = str(args.get("kind") or "").strip().lower()
|
||||
if kind not in {"tracks", "albums"}:
|
||||
return tool_error("kind must be one of: tracks, albums")
|
||||
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}
|
||||
|
||||
|
||||
_strs = lambda *names: dict.fromkeys(names, COMMON_STRING)
|
||||
_enum = lambda *values: {"type": "string", "enum": list(values)}
|
||||
_idesc = lambda text: {"type": "integer", "description": text}
|
||||
|
||||
|
||||
def _schema(name: str, description: str, properties: dict, required: tuple = ("action",)) -> dict:
|
||||
return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": list(required)}}
|
||||
|
||||
|
||||
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": _idesc("For recently_played: number of tracks (max 50)"),
|
||||
"after": _idesc("For recently_played: Unix ms cursor (after this timestamp)"), "before": _idesc("For recently_played: Unix ms cursor (before this timestamp)"),
|
||||
})
|
||||
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})
|
||||
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")})
|
||||
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})
|
||||
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})
|
||||
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"))
|
||||
Reference in New Issue
Block a user