512 lines
20 KiB
Python
512 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""mike-ai MCP: Deemix.
|
|
|
|
Drives the existing Deemix instance on the Unraid home server through its
|
|
Web-UI HTTP API (http://<unraid>:6595/deemix/api/*). Deemix ships no CLI and
|
|
no documented public REST API; the endpoints used here were verified against
|
|
the route files of webui 4.6.0 / deemix 3.13.7 (dist/routes/api/{get,post}).
|
|
|
|
Credential handling (honest documentation):
|
|
* No credential is stored on Athena or in this container on disk.
|
|
* In single-user mode the Deemix app itself serves the saved Deezer ARL from
|
|
GET /connect (singleUser.arl). The MCP reads it into process memory ONLY
|
|
for the duration of a (re-)login and never writes it to disk, logs, tool
|
|
output or error messages. DEEMIX_ARL is only a fallback for multi-user
|
|
setups and is read from the environment.
|
|
* The ARL is therefore transiently present in the MCP's working memory
|
|
between a /connect read and the corresponding /loginArl call, and in the
|
|
cookie jar afterwards. It is redacted from anything that leaves the
|
|
process.
|
|
|
|
Session handling:
|
|
* Every tool funnels through Session.call(), which guarantees a login before
|
|
the first request (so a direct first call right after a fresh container
|
|
start works) and, if Deemix reports an expired/absent session
|
|
(errid NotLoggedIn) or answers HTTP 401/403, performs AT MOST ONE
|
|
automatic re-login and retries the request once. There is no retry loop.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
import re
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any, Optional
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
DEEMIX_URL = os.environ.get("DEEMIX_URL", "http://192.168.1.2:6595").rstrip("/")
|
|
API = DEEMIX_URL + "/deemix/api"
|
|
FALLBACK_ARL = os.environ.get("DEEMIX_ARL", "").strip()
|
|
TIMEOUT = float(os.environ.get("DEEMIX_TIMEOUT", "30"))
|
|
|
|
# --- Limits (kept small and bounded so tool output stays compact) ---------
|
|
MAX_QUEUE_ROWS = 50 # max queue entries returned per call
|
|
MAX_SEARCH_RESULTS = 20 # hard cap on search result rows
|
|
MAX_SEARCH_TERM = 100 # max characters in a search term
|
|
MAX_URLS_PER_CALL = 10 # max Deezer URLs accepted by deemix_add_to_queue
|
|
MAX_ERRORS_PER_ENTRY = 3
|
|
|
|
# --- Bitrate: verified against the deployed webui/deezer-sdk code ---------
|
|
# `addToQueue.js` does `bitrate = Number(bitrate)` and falls back to the
|
|
# Deemix `maxBitrate` setting; it is NOT a kbit/s figure. The values are
|
|
# deezer-sdk `TrackFormats` enum ids. The Deemix settings UI offers exactly:
|
|
TRACKFORMATS: dict[int, str] = {
|
|
9: "FLAC (lossless, ~1411 kbps)",
|
|
3: "MP3 320 kbps",
|
|
1: "MP3 128 kbps",
|
|
}
|
|
VALID_BITRATES = tuple(sorted(TRACKFORMATS)) # (1, 3, 9)
|
|
|
|
# --- Secret redaction ------------------------------------------------------
|
|
# Long hex runs (Deezer ARL is a long lowercase-hex cookie) are masked, as is
|
|
# any value that matches a secret we actually hold, plus cookie headers.
|
|
_HEX_RUN = re.compile(r"[0-9a-fA-F]{32,}")
|
|
_COOKIE_LINE = re.compile(r"(?im)^\s*(set-?cookie|cookie)\s*:\s*.*$")
|
|
_KNOWN_SECRETS: list[str] = [s for s in {FALLBACK_ARL, os.environ.get("DEEMIX_ARL", "").strip()} if s]
|
|
|
|
|
|
def redact(value: Any) -> Any:
|
|
"""Recursively strip anything that could carry a credential."""
|
|
if isinstance(value, dict):
|
|
out: dict = {}
|
|
for k, v in value.items():
|
|
if isinstance(k, str) and k.lower() in ("arl", "cookie", "cookies", "set-cookie", "token", "authorization"):
|
|
out[k] = "[REDACTED]"
|
|
else:
|
|
out[k] = redact(v)
|
|
return out
|
|
if isinstance(value, (list, tuple)):
|
|
return [redact(v) for v in value]
|
|
if isinstance(value, str):
|
|
s = value
|
|
for sec in _KNOWN_SECRETS:
|
|
s = s.replace(sec, "[REDACTED]")
|
|
s = _HEX_RUN.sub("[REDACTED]", s)
|
|
s = _COOKIE_LINE.sub("cookie: [REDACTED]", s)
|
|
return s
|
|
return value
|
|
|
|
|
|
def _clean_error(message: str) -> str:
|
|
"""Trim + redact an error so no credential/cookie/raw body escapes."""
|
|
s = str(message).replace("\n", " ")
|
|
for sec in _KNOWN_SECRETS:
|
|
s = s.replace(sec, "[REDACTED]")
|
|
s = _HEX_RUN.sub("[REDACTED]", s)
|
|
return s[:300]
|
|
|
|
|
|
mcp = FastMCP(
|
|
"deemix",
|
|
instructions=(
|
|
"Control the Deemix download server on the Unraid home server: "
|
|
"search Deezer, queue albums/tracks, monitor download progress, "
|
|
"retry failures. Files land on Unraid at the configured download "
|
|
"location (see deemix_status). No credentials are stored here."
|
|
),
|
|
host="0.0.0.0",
|
|
port=int(os.environ.get("PORT", "8000")),
|
|
stateless_http=True,
|
|
)
|
|
|
|
|
|
class DeemixError(RuntimeError):
|
|
"""User-facing error with a short, actionable, credential-free message."""
|
|
|
|
|
|
class AuthError(DeemixError):
|
|
"""The Deemix session is missing or expired; a re-login may help."""
|
|
|
|
|
|
class Session:
|
|
"""One HTTP session against Deemix with bounded automatic (re-)login."""
|
|
|
|
def __init__(self) -> None:
|
|
self.jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(self.jar)
|
|
)
|
|
self.user: dict = {}
|
|
self._logged_in = False
|
|
|
|
# -- low level ---------------------------------------------------------
|
|
def _headers(self) -> dict:
|
|
return {
|
|
"Accept": "application/json",
|
|
"User-Agent": "mike-ai-mcp-deemix/1.1",
|
|
}
|
|
|
|
def _raw(self, method: str, path: str, body: Optional[dict] = None):
|
|
"""Return (http_status, parsed). Network failures raise DeemixError.
|
|
|
|
HTTP 401/403 are returned (not raised) so the caller can decide on a
|
|
re-login. Any non-JSON body is turned into a clean, redacted error.
|
|
"""
|
|
url = API + path
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
headers = self._headers()
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
resp = self.opener.open(req, timeout=TIMEOUT)
|
|
status = resp.getcode()
|
|
raw = resp.read().decode("utf-8", "replace")
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (401, 403):
|
|
return e.code, None
|
|
raise DeemixError(
|
|
_clean_error(f"Deemix {path} -> HTTP {e.code}")
|
|
) from None
|
|
except (urllib.error.URLError, TimeoutError, OSError) as e:
|
|
raise DeemixError(
|
|
_clean_error(f"Deemix unreachable at {API}: {getattr(e, 'reason', e)}")
|
|
) from None
|
|
if not raw.strip():
|
|
return status, {}
|
|
try:
|
|
return status, json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
raise DeemixError(
|
|
_clean_error(f"Deemix {path} returned non-JSON ({raw[:80]!r})")
|
|
) from None
|
|
|
|
# -- login -------------------------------------------------------------
|
|
def _jar_has_session(self) -> bool:
|
|
return any(c.name == "deemix_session" for c in self.jar)
|
|
|
|
def login(self) -> None:
|
|
"""Establish a session. Raises DeemixError on failure (no retry)."""
|
|
status, info = self._raw("GET", "/connect")
|
|
if status in (401, 403):
|
|
# Not authenticated yet; /connect should still answer, but if it
|
|
# refuses we treat it as "needs login" and fall through.
|
|
info = info or {}
|
|
if not isinstance(info, dict):
|
|
info = {}
|
|
if str(info.get("deezerAvailable")) != "yes":
|
|
raise DeemixError(
|
|
"Deemix reports Deezer unavailable "
|
|
f"(deezerAvailable={info.get('deezerAvailable')!r}) - "
|
|
"check the Unraid VPN tunnel"
|
|
)
|
|
if not info.get("autologin", True) and self._jar_has_session():
|
|
# autologin=false is only trusted while a live session cookie is
|
|
# actually held; after a cookie loss a re-login is required even
|
|
# though the server still reports autologin=false.
|
|
self._logged_in = True
|
|
self.user = info.get("currentUser") or {}
|
|
return
|
|
arl = ((info.get("singleUser") or {}).get("arl") or "") or FALLBACK_ARL
|
|
if not arl:
|
|
raise DeemixError(
|
|
"No ARL available: Deemix is not in single-user mode and "
|
|
"DEEMIX_ARL is not set in the MCP environment"
|
|
)
|
|
status, res = self._raw("POST", "/loginArl", {"arl": arl})
|
|
if status in (401, 403) or not isinstance(res, dict) or res.get("status") not in (1, 2, 3):
|
|
raise DeemixError(
|
|
f"Deezer ARL login failed (status="
|
|
f"{(res or {}).get('status') if isinstance(res, dict) else status}, "
|
|
f"error={(res or {}).get('error') if isinstance(res, dict) else 'http'})"
|
|
)
|
|
self._logged_in = True
|
|
self.user = res.get("user") or {}
|
|
|
|
# -- central call with at most ONE re-login ----------------------------
|
|
def call(self, method: str, path: str, body: Optional[dict] = None) -> Any:
|
|
for _attempt in (1, 2):
|
|
if not self._logged_in:
|
|
try:
|
|
self.login()
|
|
except DeemixError as e:
|
|
# The login (or the automatic re-login) failed. Abort
|
|
# cleanly and boundedly: we never loop, we raise after
|
|
# this single failed attempt.
|
|
raise DeemixError(
|
|
"Deemix session could not be established: "
|
|
f"login/re-login attempt failed - {_clean_error(e)}"
|
|
) from None
|
|
status, res = self._raw(method, path, body)
|
|
if status in (401, 403):
|
|
# Expiring session: reset and allow exactly one re-login.
|
|
self._logged_in = False
|
|
self.jar.clear()
|
|
continue
|
|
if (
|
|
isinstance(res, dict)
|
|
and res.get("result") is False
|
|
and res.get("errid") == "NotLoggedIn"
|
|
):
|
|
self._logged_in = False
|
|
self.jar.clear()
|
|
continue
|
|
return res
|
|
raise DeemixError(
|
|
"Deemix session could not be established: re-login was attempted "
|
|
"once and the request still reports an expired/absent session."
|
|
)
|
|
|
|
def request(self, method: str, path: str, body: Optional[dict] = None) -> Any:
|
|
"""Back-compat alias used by tools that only need a read."""
|
|
return self.call(method, path, body)
|
|
|
|
|
|
SESSION = Session()
|
|
|
|
|
|
def call(method: str, path: str, body: Optional[dict] = None) -> Any:
|
|
return SESSION.call(method, path, body)
|
|
|
|
|
|
def _out(payload: Any) -> str:
|
|
return json.dumps(redact(payload), ensure_ascii=False, indent=2)
|
|
|
|
|
|
def _compact_entry(e: dict) -> dict:
|
|
size = e.get("size") or 0
|
|
row = {
|
|
"uuid": e.get("uuid"),
|
|
"type": e.get("type"),
|
|
"id": e.get("id"),
|
|
"title": e.get("title"),
|
|
"artist": e.get("artist"),
|
|
"progress": e.get("progress"),
|
|
"downloaded_mb": round((e.get("downloaded") or 0) / 1048576, 1),
|
|
"total_mb": round(size / 1048576, 1) if size else None,
|
|
"failed": bool(e.get("failed")),
|
|
}
|
|
errs = e.get("errors")
|
|
if errs:
|
|
row["errors"] = redact(errs)[:MAX_ERRORS_PER_ENTRY]
|
|
return row
|
|
|
|
|
|
def _compact_search_row(item: dict, kind: str) -> dict:
|
|
if kind == "album":
|
|
artists = ", ".join(
|
|
a.get("name", "") for a in (item.get("related") or {}).get("artists", [])[:2]
|
|
)
|
|
return {
|
|
"id": item.get("id"),
|
|
"title": item.get("title"),
|
|
"artist": artists,
|
|
"year": ((item.get("release_date") or "")[:4] or None),
|
|
"tracks": item.get("nb_tracks"),
|
|
"link": f"https://www.deezer.com/album/{item.get('id')}",
|
|
}
|
|
if kind == "track":
|
|
return {
|
|
"id": item.get("id"),
|
|
"title": item.get("title"),
|
|
"artist": (item.get("artist") or {}).get("name"),
|
|
"album": (item.get("album") or {}).get("title"),
|
|
"duration": item.get("duration"),
|
|
"link": f"https://www.deezer.com/track/{item.get('id')}",
|
|
}
|
|
if kind == "artist":
|
|
return {
|
|
"id": item.get("id"),
|
|
"name": item.get("name"),
|
|
"link": f"https://www.deezer.com/artist/{item.get('id')}",
|
|
}
|
|
if kind == "playlist":
|
|
return {
|
|
"id": item.get("id"),
|
|
"title": item.get("name") or item.get("title"),
|
|
"nb_tracks": item.get("nb_tracks"),
|
|
"link": f"https://www.deezer.com/playlist/{item.get('id')}",
|
|
}
|
|
return {
|
|
k: item.get(k)
|
|
for k in ("id", "name", "title", "link", "url")
|
|
if item.get(k) is not None
|
|
}
|
|
|
|
|
|
def _valid_bitrate(bitrate: Optional[int]) -> Optional[int]:
|
|
if bitrate is None:
|
|
return None
|
|
if bitrate not in TRACKFORMATS:
|
|
raise DeemixError(
|
|
f"bitrate {bitrate} is not a valid TrackFormats value. "
|
|
f"Use one of: {', '.join(f'{b}={TRACKFORMATS[b]}' for b in VALID_BITRATES)}. "
|
|
"These are deezer-sdk TrackFormats enum ids, not kbit/s; "
|
|
"omit bitrate to use Deemix's own maxBitrate setting."
|
|
)
|
|
return bitrate
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_status() -> str:
|
|
"""Status of the Deemix server: versions, Deezer login, download location and a compact queue summary."""
|
|
info = call("GET", "/connect") # guarantees a session (also first call after fresh start)
|
|
queue = call("GET", "/getQueue")
|
|
entries = list((queue.get("queue") or {}).values())
|
|
order = queue.get("queueOrder") or []
|
|
by_id = {e.get("uuid"): e for e in entries}
|
|
ordered = [by_id[u] for u in order if u in by_id]
|
|
for e in entries:
|
|
if e.get("uuid") not in order:
|
|
ordered.append(e)
|
|
rows = [_compact_entry(e) for e in ordered[:MAX_QUEUE_ROWS]]
|
|
u = SESSION.user or {}
|
|
settings = (info.get("settingsData") or {}).get("settings") or {}
|
|
user_view = {
|
|
k: u.get(k)
|
|
for k in ("id", "name", "lastName", "country", "subscription")
|
|
if u.get(k) not in (None, "")
|
|
}
|
|
return _out(
|
|
{
|
|
"deemix_version": (info.get("update") or {}).get("deemixVersion"),
|
|
"webui_version": (info.get("update") or {}).get("webuiVersion"),
|
|
"deezer_available": info.get("deezerAvailable"),
|
|
"logged_in_user": user_view or "not logged in",
|
|
"spotify_enabled": info.get("spotifyEnabled"),
|
|
"download_location": settings.get("downloadLocation"),
|
|
"max_bitrate_setting": settings.get("maxBitrate"),
|
|
"queue": {
|
|
"total": len(entries),
|
|
"failed": sum(1 for e in entries if e.get("failed")),
|
|
"rows": rows,
|
|
"truncated": len(entries) > MAX_QUEUE_ROWS,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_search(term: str, type: str = "album", start: int = 0, nb: int = 10) -> str:
|
|
"""Search Deezer via Deemix. type: album, track, artist, playlist or radio.
|
|
Returns compact rows with Deezer links usable for deemix_add_to_queue.
|
|
term is limited to 100 chars and nb is capped at 20 results."""
|
|
term = (term or "").strip()
|
|
if not term:
|
|
raise DeemixError("term is empty")
|
|
if len(term) > MAX_SEARCH_TERM:
|
|
term = term[:MAX_SEARCH_TERM]
|
|
nb = max(1, min(int(nb), MAX_SEARCH_RESULTS))
|
|
start = max(0, int(start))
|
|
q = urllib.parse.urlencode({"term": term, "type": type, "start": start, "nb": nb})
|
|
data = SESSION.call("GET", "/search?" + q)
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
if data.get("error"):
|
|
raise DeemixError(_clean_error(f"Deezer search failed: {data.get('error')}"))
|
|
rows = [_compact_search_row(i, type) for i in (data.get("data") or [])[:nb]]
|
|
return _out({"type": type, "total": data.get("total"), "results": rows})
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_add_to_queue(urls: str, bitrate: Optional[int] = None) -> str:
|
|
"""Queue Deezer URLs for download. urls: one or several Deezer links,
|
|
space separated (album/track/artist/playlist), max 10 per call.
|
|
bitrate (optional) is a deezer-sdk TrackFormats enum id, NOT kbit/s:
|
|
1=MP3 128kbps, 3=MP3 320kbps, 9=FLAC lossless. Omit to use Deemix's
|
|
own maxBitrate setting."""
|
|
url_list = [u.strip() for u in (urls or "").split() if u.strip()]
|
|
if not url_list:
|
|
raise DeemixError("urls is empty")
|
|
if len(url_list) > MAX_URLS_PER_CALL:
|
|
raise DeemixError(
|
|
f"too many urls ({len(url_list)}); max {MAX_URLS_PER_CALL} per call"
|
|
)
|
|
body: dict = {"url": " ".join(url_list)}
|
|
br = _valid_bitrate(bitrate)
|
|
if br is not None:
|
|
body["bitrate"] = br
|
|
res = SESSION.call("POST", "/addToQueue", body)
|
|
if not isinstance(res, dict) or res.get("result") is False:
|
|
errid = (res or {}).get("errid") if isinstance(res, dict) else "http"
|
|
raise DeemixError(_clean_error(f"addToQueue rejected (errid={errid!r})"))
|
|
data = res.get("data") or {}
|
|
obj = data.get("obj")
|
|
# Real Deemix behaviour: an unknown/invalid ID is accepted but resolves
|
|
# to zero items (obj == []) - nothing is queued. Report that honestly.
|
|
resolved = obj not in ([], None, "")
|
|
payload = {
|
|
"queued": resolved,
|
|
"bitrate_requested": br,
|
|
"bitrate_used": data.get("bitrate"),
|
|
"items": obj if isinstance(obj, (list, dict)) else str(obj),
|
|
}
|
|
if not resolved:
|
|
payload["warning"] = (
|
|
"Deemix accepted the URL(s) but resolved NO items - the ID(s) "
|
|
"are probably invalid. Nothing was queued."
|
|
)
|
|
return _out(payload)
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_queue() -> str:
|
|
"""Current download queue with per-entry progress, size and errors (compact)."""
|
|
queue = SESSION.call("GET", "/getQueue")
|
|
entries = list((queue.get("queue") or {}).values())
|
|
order = queue.get("queueOrder") or []
|
|
by_id = {e.get("uuid"): e for e in entries}
|
|
ordered = [by_id[u] for u in order if u in by_id]
|
|
for e in entries:
|
|
if e.get("uuid") not in order:
|
|
ordered.append(e)
|
|
rows = [_compact_entry(e) for e in ordered[:MAX_QUEUE_ROWS]]
|
|
return _out(
|
|
{
|
|
"total": len(entries),
|
|
"failed": sum(1 for e in entries if e.get("failed")),
|
|
"rows": rows,
|
|
"truncated": len(entries) > MAX_QUEUE_ROWS,
|
|
}
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_retry_download(uuid: str) -> str:
|
|
"""Retry a failed queue entry (uuid from deemix_queue). Re-queues the whole album/track; existing files are not overwritten."""
|
|
res = SESSION.call("POST", "/retryDownload", {"uuid": uuid})
|
|
if not isinstance(res, dict) or res.get("result") is False:
|
|
raise DeemixError(_clean_error(f"retryDownload failed (errid={(res or {}).get('errid')!r})"))
|
|
return _out({"retried": uuid, "result": True})
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_remove_from_queue(uuid: str) -> str:
|
|
"""Cancel one queue entry by uuid (from deemix_queue)."""
|
|
q = urllib.parse.urlencode({"uuid": uuid})
|
|
res = SESSION.call("POST", "/removeFromQueue?" + q)
|
|
if not isinstance(res, dict) or res.get("result") is False:
|
|
raise DeemixError(_clean_error("removeFromQueue failed - uuid unknown?"))
|
|
return _out({"removed": uuid, "result": True})
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_remove_finished() -> str:
|
|
"""Remove all finished (non-failed) entries from the queue. Does not touch files on disk."""
|
|
res = SESSION.call("POST", "/removeFinishedDownloads")
|
|
if not isinstance(res, dict) or res.get("result") is False:
|
|
raise DeemixError(_clean_error("removeFinishedDownloads failed"))
|
|
return _out({"result": True})
|
|
|
|
|
|
@mcp.tool()
|
|
def deemix_cancel_all() -> str:
|
|
"""Cancel ALL active downloads in Deemix. Use only when the user explicitly asked to stop downloading."""
|
|
res = SESSION.call("POST", "/cancelAllDownloads")
|
|
if not isinstance(res, dict) or res.get("result") is False:
|
|
raise DeemixError(_clean_error("cancelAllDownloads failed"))
|
|
return _out({"result": True})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# MCPHub manages local servers as stdio subprocesses. The standalone
|
|
# Athena container can keep using Streamable HTTP by setting
|
|
# MCP_TRANSPORT=streamable-http, so the same source works in both places
|
|
# during the migration.
|
|
mcp.run(transport=os.environ.get("MCP_TRANSPORT", "stdio"))
|