feat: add background_tasks + list_snapshots tools (v0.3.4)

- background_tasks: SYNO.FileStation.BackgroundTask::list (v3) — paginated
  table of active/recent copy/move/delete/extract/compress tasks
- list_snapshots: SYNO.FileStation.Snapshot::list (v2) — Btrfs snapshots
  per share; maps error 400 to a clear Btrfs-required message
- 20 new tests (107 total)
- SPEC.md and CLAUDE.md updated (26 tools)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-14 21:54:03 +02:00
parent b88677a20c
commit 4c6de3bfc7
6 changed files with 456 additions and 13 deletions
@@ -1237,3 +1237,119 @@ def register_filestation(
return f"Error: {e}"
return f"Deleted favorite for {path}"
# ── background task + snapshot tools ──────────────────────────────────
@mcp.tool()
async def background_tasks(offset: int = 0, limit: int = 100):
"""List FileStation background tasks (copy, move, delete, extract, compress, etc.)."""
from mcp_synology_filestation.client import SynologyError
limit = max(1, min(limit, 200))
offset = max(0, offset)
try:
data = await client.request(
"SYNO.FileStation.BackgroundTask",
"list",
version=3,
params={"offset": offset, "limit": limit},
)
except SynologyError as e:
return f"Error: {e}"
tasks: list[dict] = data.get("tasks", [])
total: int = data.get("total", len(tasks))
if not tasks:
return "No background tasks found."
rows: list[tuple[str, ...]] = []
for task in tasks:
taskid = str(task.get("taskid", ""))
ttype = str(task.get("type", ""))
status = str(task.get("status", ""))
path = str(task.get("path", task.get("dest_folder_path", "")))
processed = task.get("processed_num_file", 0)
total_files = task.get("total_num_file", 0)
progress = f"{processed}/{total_files}" if total_files else "-"
rows.append((taskid, ttype, status, path, progress))
headers = ("Task ID", "Type", "Status", "Path", "Progress")
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} task(s) total.")
if total > offset + limit:
lines.append(
f"Showing {offset + 1}{offset + len(tasks)} of {total}. Use offset to page."
)
return "\n".join(lines)
@mcp.tool()
async def list_snapshots(share_path: str, offset: int = 0, limit: int = 100):
"""List Btrfs snapshots for a shared folder (requires Btrfs volume)."""
from mcp_synology_filestation.client import SynologyError
limit = max(1, min(limit, 500))
offset = max(0, offset)
try:
data = await client.request(
"SYNO.FileStation.Snapshot",
"list",
version=2,
params={"folder_path": share_path, "offset": offset, "limit": limit},
)
except SynologyError as e:
if e.code == 400:
return (
f"Error: Snapshots not available for '{share_path}'"
"this feature requires a Btrfs-formatted volume."
)
return f"Error: {e}"
snapshots: list[dict] = data.get("snapshots", [])
total: int = data.get("total", len(snapshots))
if not snapshots:
return f"No snapshots found for '{share_path}'."
rows: list[tuple[str, ...]] = []
for snap in snapshots:
snap_id = str(snap.get("id", snap.get("snapshot_id", "")))
snap_time = _fmt_time(snap.get("time", snap.get("create_time")))
snap_desc = str(snap.get("description", snap.get("desc", "")))
snap_lock = "Yes" if snap.get("lock", False) else "No"
rows.append((snap_id, snap_time, snap_desc, snap_lock))
headers = ("Snapshot ID", "Created", "Description", "Locked")
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} snapshot(s) for '{share_path}'.")
if total > offset + limit:
lines.append(
f"Showing {offset + 1}{offset + len(snapshots)} of {total}. Use offset to page."
)
return "\n".join(lines)