feat: add get_thumbnail, list_favorites, add_favorite, delete_favorite (v0.3.1)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,3 @@
|
||||
"""MCP server for Synology FileStation."""
|
||||
|
||||
__version__ = "0.3.0-dev"
|
||||
__version__ = "0.3.1"
|
||||
|
||||
@@ -427,3 +427,63 @@ class FileStationClient:
|
||||
|
||||
code = body.get("error", {}).get("code", 0)
|
||||
raise SynologyError(_error_message(code, api), code=code)
|
||||
|
||||
async def get_thumbnail_bytes(self, path: str, size: str = "large") -> bytes:
|
||||
"""Fetch a thumbnail for a file via SYNO.FileStation.Thumb.
|
||||
|
||||
Args:
|
||||
path: Share-relative path to the image file.
|
||||
size: Thumbnail size — "small", "medium", "large", or "original".
|
||||
|
||||
Returns:
|
||||
Raw JPEG bytes.
|
||||
|
||||
Raises:
|
||||
SynologyError: On HTTP 404 (file not found), non-200 status,
|
||||
or DSM error in the response body.
|
||||
"""
|
||||
api = "SYNO.FileStation.Thumb"
|
||||
await self._ensure_initialized()
|
||||
http = self._get_http()
|
||||
|
||||
if api not in self._api_cache:
|
||||
raise SynologyError(f"API '{api}' not found.", code=102)
|
||||
|
||||
info = self._api_cache[api]
|
||||
url = f"{self._base_url}/webapi/{info['path']}"
|
||||
|
||||
req_params: dict[str, str] = {
|
||||
"api": api,
|
||||
"version": "2",
|
||||
"method": "get",
|
||||
"path": path,
|
||||
"size": size,
|
||||
"rotate": "0",
|
||||
}
|
||||
if self._sid:
|
||||
req_params["_sid"] = self._sid
|
||||
|
||||
logger.debug("DSM POST: %s/get v2 — path=%s size=%s", api, path, size)
|
||||
resp = await http.post(url, data=req_params)
|
||||
|
||||
if resp.status_code == 404:
|
||||
raise SynologyError(f"File not found: {path}", code=404)
|
||||
|
||||
if resp.status_code != 200:
|
||||
raise SynologyError(
|
||||
f"Thumbnail request failed (HTTP {resp.status_code})", code=resp.status_code
|
||||
)
|
||||
|
||||
content_type = resp.headers.get("content-type", "")
|
||||
if not content_type.startswith("image/"):
|
||||
# DSM returned an error envelope instead of image data
|
||||
try:
|
||||
body = resp.json()
|
||||
code = body.get("error", {}).get("code", 0)
|
||||
raise SynologyError(_error_message(code, api), code=code)
|
||||
except (ValueError, KeyError):
|
||||
raise SynologyError(
|
||||
f"Unexpected response content-type: {content_type}"
|
||||
) from None
|
||||
|
||||
return resp.content
|
||||
|
||||
@@ -1138,3 +1138,118 @@ def register_filestation(
|
||||
return f"Error: {e}"
|
||||
|
||||
return f"Deleted sharing link: {link_id}"
|
||||
|
||||
# ── thumbnail + favorites tools ───────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
async def get_thumbnail(path: str, size: str = "large"):
|
||||
"""Fetch a thumbnail for an image. Returns JSON: filename, size_bytes, content_base64."""
|
||||
import base64
|
||||
import json as _json
|
||||
|
||||
from mcp_synology_filestation.client import SynologyError
|
||||
|
||||
valid_sizes = {"small", "medium", "large", "original"}
|
||||
if size not in valid_sizes:
|
||||
return f"Error: size must be one of {sorted(valid_sizes)}"
|
||||
|
||||
try:
|
||||
img_bytes = await client.get_thumbnail_bytes(path, size)
|
||||
except SynologyError as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
filename = path.rsplit("/", 1)[-1]
|
||||
return _json.dumps(
|
||||
{
|
||||
"filename": filename,
|
||||
"size_bytes": len(img_bytes),
|
||||
"content_base64": base64.b64encode(img_bytes).decode(),
|
||||
}
|
||||
)
|
||||
|
||||
@mcp.tool()
|
||||
async def list_favorites():
|
||||
"""List all FileStation favorites. Table: name, path, type, status, real path, modified."""
|
||||
from mcp_synology_filestation.client import SynologyError
|
||||
|
||||
try:
|
||||
data = await client.request(
|
||||
"SYNO.FileStation.Favorite",
|
||||
"list",
|
||||
version=2,
|
||||
params={
|
||||
"offset": 0,
|
||||
"limit": 200,
|
||||
"additional": json.dumps(["real_path", "size", "time"]),
|
||||
},
|
||||
)
|
||||
except SynologyError as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
favorites: list[dict] = data.get("favorites", [])
|
||||
total: int = data.get("total", len(favorites))
|
||||
|
||||
if not favorites:
|
||||
return "No favorites found."
|
||||
|
||||
rows: list[tuple[str, ...]] = []
|
||||
for fav in favorites:
|
||||
name = fav.get("name", "")
|
||||
fpath = fav.get("path", "")
|
||||
ftype = "folder" if fav.get("isdir", False) else "file"
|
||||
status = fav.get("status", "")
|
||||
add = fav.get("additional", {})
|
||||
real_path = add.get("real_path", "")
|
||||
mtime = add.get("time", {}).get("mtime")
|
||||
rows.append((name, fpath, ftype, status, real_path, _fmt_time(mtime)))
|
||||
|
||||
headers = ("Name", "Path", "Type", "Status", "Real Path", "Modified")
|
||||
col_widths = [max(len(h), *(len(r[i]) for r in rows)) for i, h in enumerate(headers)]
|
||||
|
||||
def _sep() -> str:
|
||||
return "+" + "+".join("-" * (w + 2) for w in col_widths) + "+"
|
||||
|
||||
def _row(vals: tuple[str, ...]) -> str:
|
||||
return "| " + " | ".join(f"{v:<{col_widths[i]}}" for i, v in enumerate(vals)) + " |"
|
||||
|
||||
lines = [_sep(), _row(headers), _sep()]
|
||||
for r in rows:
|
||||
lines.append(_row(r))
|
||||
lines.append(_sep())
|
||||
lines.append(f"\n{total} favorite(s)")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
@mcp.tool()
|
||||
async def add_favorite(path: str, name: str):
|
||||
"""Add a path to FileStation favorites. name: display label for the favorite."""
|
||||
from mcp_synology_filestation.client import SynologyError
|
||||
|
||||
try:
|
||||
await client.request(
|
||||
"SYNO.FileStation.Favorite",
|
||||
"add",
|
||||
version=2,
|
||||
params={"path": path, "name": name, "index": -1},
|
||||
)
|
||||
except SynologyError as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
return f"Added favorite '{name}' → {path}"
|
||||
|
||||
@mcp.tool()
|
||||
async def delete_favorite(path: str):
|
||||
"""Delete a FileStation favorite by path."""
|
||||
from mcp_synology_filestation.client import SynologyError
|
||||
|
||||
try:
|
||||
await client.request(
|
||||
"SYNO.FileStation.Favorite",
|
||||
"delete",
|
||||
version=2,
|
||||
params={"path": path},
|
||||
)
|
||||
except SynologyError as e:
|
||||
return f"Error: {e}"
|
||||
|
||||
return f"Deleted favorite for {path}"
|
||||
|
||||
Reference in New Issue
Block a user