330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""Sonarr condensed action-routed MCP tool.
|
|
|
|
CONCEPT:ECO-4.82 — gitlab-style organized per-service tool surface.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
from agent_utilities.mcp_utilities import dispatch, run_blocking
|
|
from fastmcp import FastMCP
|
|
from pydantic import Field
|
|
|
|
from arr_mcp.auth import get_sonarr_client
|
|
|
|
|
|
READ_ONLY_ACTIONS = frozenset(
|
|
{
|
|
"get_system_status", "get_health", "get_diskspace", "get_ping",
|
|
"get_series", "get_series_id", "get_series_lookup", "lookup_series",
|
|
"get_episode", "get_episode_id", "get_episodefile", "get_episodefile_id",
|
|
"get_calendar", "get_calendar_id", "get_history", "get_history_series",
|
|
"get_history_since", "get_queue", "get_queue_details", "get_queue_status",
|
|
"get_wanted_missing", "get_wanted_missing_id", "get_wanted_cutoff",
|
|
"get_wanted_cutoff_id", "get_qualityprofile", "get_qualityprofile_id",
|
|
"get_languageprofile", "get_languageprofile_id", "get_tag", "get_tag_id",
|
|
"get_tag_detail", "get_tag_detail_id", "get_command", "get_command_id",
|
|
"get_release",
|
|
}
|
|
)
|
|
|
|
PSEUDO_ACTIONS = frozenset({"find_series", "get_season_summary", "search_releases"})
|
|
|
|
WRITE_ACTIONS = frozenset({
|
|
"post_command",
|
|
"post_release",
|
|
"put_episode_id",
|
|
"put_episode_monitor",
|
|
"put_series_id",
|
|
"put_series",
|
|
"put_wanted",
|
|
"post_wanted",
|
|
})
|
|
MAX_COLLECTION_ITEMS = 50
|
|
|
|
|
|
def _plain(value: Any) -> Any:
|
|
if hasattr(value, "model_dump") and callable(value.model_dump):
|
|
return value.model_dump()
|
|
if hasattr(value, "dict") and callable(value.dict):
|
|
return value.dict()
|
|
if isinstance(value, list):
|
|
return [_plain(item) for item in value]
|
|
if isinstance(value, dict):
|
|
return {str(key): _plain(item) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def _unwrap(value: Any) -> Any:
|
|
value = _plain(value)
|
|
if isinstance(value, dict) and set(value) == {"result"}:
|
|
return value["result"]
|
|
return value
|
|
|
|
|
|
def _pick(item: dict[str, Any], fields: tuple[str, ...]) -> dict[str, Any]:
|
|
return {field: item[field] for field in fields if item.get(field) is not None}
|
|
|
|
|
|
def _compact_series(item: dict[str, Any], include_seasons: bool = False) -> dict[str, Any]:
|
|
result = _pick(
|
|
item,
|
|
("id", "title", "sortTitle", "year", "status", "monitored", "path", "tvdbId"),
|
|
)
|
|
statistics = item.get("statistics") or {}
|
|
if isinstance(statistics, dict):
|
|
result["statistics"] = _pick(
|
|
statistics,
|
|
("seasonCount", "episodeFileCount", "episodeCount", "totalEpisodeCount", "sizeOnDisk", "percentOfEpisodes"),
|
|
)
|
|
if include_seasons:
|
|
result["seasons"] = [
|
|
{
|
|
**_pick(season, ("seasonNumber", "monitored")),
|
|
"statistics": _pick(
|
|
season.get("statistics") or {},
|
|
("episodeFileCount", "episodeCount", "totalEpisodeCount", "sizeOnDisk", "percentOfEpisodes"),
|
|
),
|
|
}
|
|
for season in item.get("seasons", [])
|
|
if isinstance(season, dict)
|
|
]
|
|
return result
|
|
|
|
|
|
def _compact_episode(item: dict[str, Any]) -> dict[str, Any]:
|
|
return _pick(
|
|
item,
|
|
("id", "seriesId", "seasonNumber", "episodeNumber", "title", "airDate", "airDateUtc", "monitored", "hasFile", "episodeFileId"),
|
|
)
|
|
|
|
|
|
def _compact_file(item: dict[str, Any]) -> dict[str, Any]:
|
|
quality = item.get("quality") or {}
|
|
quality_name = (quality.get("quality") or {}).get("name") if isinstance(quality, dict) else None
|
|
result = _pick(
|
|
item,
|
|
("id", "seriesId", "seasonNumber", "relativePath", "path", "size", "dateAdded", "releaseGroup"),
|
|
)
|
|
if quality_name:
|
|
result["quality"] = quality_name
|
|
return result
|
|
|
|
|
|
def _compact_release(item: dict[str, Any]) -> dict[str, Any]:
|
|
quality = item.get("quality") or {}
|
|
quality_name = (quality.get("quality") or {}).get("name") if isinstance(quality, dict) else None
|
|
result = _pick(
|
|
item,
|
|
(
|
|
"guid", "title", "indexer", "indexerId", "size", "age", "ageHours",
|
|
"seeders", "leechers", "protocol", "downloadAllowed", "releaseWeight",
|
|
),
|
|
)
|
|
if quality_name:
|
|
result["quality"] = quality_name
|
|
rejections = item.get("rejections")
|
|
if isinstance(rejections, list) and rejections:
|
|
result["rejections"] = [str(reason)[:180] for reason in rejections[:5]]
|
|
return result
|
|
|
|
|
|
def _bounded(items: list[Any], compact) -> dict[str, Any]:
|
|
total = len(items)
|
|
return {
|
|
"total": total,
|
|
"returned": min(total, MAX_COLLECTION_ITEMS),
|
|
"truncated": total > MAX_COLLECTION_ITEMS,
|
|
"items": [compact(item) for item in items[:MAX_COLLECTION_ITEMS] if isinstance(item, dict)],
|
|
"next_step": (
|
|
"Use find_series or narrower Sonarr parameters; do not repeat the same broad request."
|
|
if total > MAX_COLLECTION_ITEMS else None
|
|
),
|
|
}
|
|
|
|
|
|
def _compact_result(action: str, value: Any) -> Any:
|
|
value = _unwrap(value)
|
|
if isinstance(value, list):
|
|
if action in {"get_series", "get_series_lookup", "lookup_series"}:
|
|
return _bounded(value, _compact_series)
|
|
if action in {"get_episode", "get_calendar", "get_wanted_missing", "get_wanted_cutoff"}:
|
|
return _bounded(value, _compact_episode)
|
|
if action == "get_episodefile":
|
|
return _bounded(value, _compact_file)
|
|
if action == "get_release":
|
|
return _bounded(value, _compact_release)
|
|
return _bounded(value, lambda item: item)
|
|
if isinstance(value, dict) and action in {"get_series_id"}:
|
|
return _compact_series(value, include_seasons=True)
|
|
if isinstance(value, dict) and action in {"get_episode_id"}:
|
|
return _compact_episode(value)
|
|
if isinstance(value, dict) and action in {"get_episodefile_id"}:
|
|
return _compact_file(value)
|
|
return value
|
|
|
|
|
|
async def _find_series(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]:
|
|
query = str(kwargs.get("query", "")).strip()
|
|
if len(query) < 2:
|
|
raise ValueError("find_series requires params_json with a query of at least 2 characters")
|
|
limit = max(1, min(int(kwargs.get("limit", 8)), 15))
|
|
raw = _unwrap(await run_blocking(dispatch, client, "get_series", {}, service="arr-sonarr"))
|
|
words = [word for word in re.findall(r"[a-z0-9]+", query.casefold()) if len(word) > 1]
|
|
matches = []
|
|
for item in raw if isinstance(raw, list) else []:
|
|
haystack = " ".join(
|
|
str(item.get(field, "")) for field in ("title", "sortTitle", "originalTitle", "alternateTitles")
|
|
).casefold()
|
|
if all(word in haystack for word in words):
|
|
matches.append(_compact_series(item, include_seasons=True))
|
|
return {
|
|
"query": query,
|
|
"matches": matches[:limit],
|
|
"match_count": len(matches),
|
|
"truncated": len(matches) > limit,
|
|
"task_complete": True,
|
|
"instruction": "Use the returned series id for details. Do not call get_series for discovery.",
|
|
}
|
|
|
|
|
|
async def _season_summary(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]:
|
|
series_id = int(kwargs["series_id"])
|
|
season_number = int(kwargs["season_number"])
|
|
series = _unwrap(
|
|
await run_blocking(dispatch, client, "get_series_id", {"id": series_id}, service="arr-sonarr")
|
|
)
|
|
episodes = _unwrap(
|
|
await run_blocking(
|
|
dispatch,
|
|
client,
|
|
"get_episode",
|
|
{"seriesId": series_id, "seasonNumber": season_number},
|
|
service="arr-sonarr",
|
|
)
|
|
)
|
|
files = _unwrap(
|
|
await run_blocking(
|
|
dispatch,
|
|
client,
|
|
"get_episodefile",
|
|
{"seriesId": series_id},
|
|
service="arr-sonarr",
|
|
)
|
|
)
|
|
selected_episodes = [
|
|
_compact_episode(item) for item in episodes
|
|
if isinstance(item, dict) and item.get("seasonNumber") == season_number
|
|
] if isinstance(episodes, list) else []
|
|
selected_files = [
|
|
_compact_file(item) for item in files
|
|
if isinstance(item, dict) and item.get("seasonNumber") == season_number
|
|
] if isinstance(files, list) else []
|
|
groups = sorted({str(item.get("releaseGroup")) for item in selected_files if item.get("releaseGroup")})
|
|
return {
|
|
"series": _compact_series(series) if isinstance(series, dict) else {"id": series_id},
|
|
"season_number": season_number,
|
|
"episode_count": len(selected_episodes),
|
|
"file_count": len(selected_files),
|
|
"release_groups": groups,
|
|
"episodes": selected_episodes[:30],
|
|
"files": selected_files[:30],
|
|
"task_complete": True,
|
|
"instruction": "This is the complete compact season answer. Do not repeat broad series or episode queries.",
|
|
}
|
|
|
|
|
|
async def _search_releases(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]:
|
|
series_id = kwargs.get("series_id")
|
|
episode_id = kwargs.get("episode_id")
|
|
season_number = kwargs.get("season_number")
|
|
release_group = str(kwargs.get("release_group", "")).strip()
|
|
if series_id is None and episode_id is None:
|
|
raise ValueError("search_releases requires series_id or episode_id")
|
|
query: dict[str, Any] = {}
|
|
if series_id is not None:
|
|
query["seriesId"] = int(series_id)
|
|
if episode_id is not None:
|
|
query["episodeId"] = int(episode_id)
|
|
if season_number is not None:
|
|
query["seasonNumber"] = int(season_number)
|
|
raw = await run_blocking(dispatch, client, "get_release", query, service="arr-sonarr")
|
|
raw = _unwrap(raw)
|
|
if release_group and isinstance(raw, list):
|
|
needle = release_group.casefold()
|
|
raw = [
|
|
item for item in raw
|
|
if isinstance(item, dict)
|
|
and needle in (
|
|
str(item.get("releaseGroup", "")) + " " + str(item.get("title", ""))
|
|
).casefold()
|
|
]
|
|
compact = _compact_result("get_release", raw)
|
|
return {
|
|
"task_complete": True,
|
|
"search_scope": {
|
|
"series_id": series_id,
|
|
"episode_id": episode_id,
|
|
"season_number": season_number,
|
|
"release_group_filter": release_group or None,
|
|
},
|
|
"monitoring_changed": False,
|
|
"download_started": False,
|
|
"results": compact,
|
|
"instruction": "These are Sonarr indexer results. Do not use web search to replace them. Never download unless the user separately approves a write action.",
|
|
}
|
|
|
|
|
|
def register_sonarr_tools(mcp: FastMCP) -> None:
|
|
@mcp.tool(tags={"sonarr"})
|
|
async def sonarr_action(
|
|
action: str = Field(
|
|
description="Read-only Sonarr action. Use find_series {query} for titles, get_season_summary {series_id, season_number} for holdings, and search_releases {series_id, season_number, optional release_group} to query configured Sonarr indexers without downloading or changing monitoring. Avoid broad get_series/get_episode calls."
|
|
),
|
|
params_json: str = Field(
|
|
default="{}",
|
|
description="JSON string of parameters to pass to the action.",
|
|
),
|
|
) -> Any:
|
|
"""Query Sonarr through a server-side allowlist (read-only by default; write actions when ARR_MCP_WRITE=1)."""
|
|
if action in {"list_actions", "help", "actions"}:
|
|
return {
|
|
"service": "sonarr",
|
|
"access_mode": "write" if os.environ.get("ARR_MCP_WRITE", "").strip().lower() in ("1", "true", "yes", "on") else "read-only",
|
|
"actions": sorted(READ_ONLY_ACTIONS),
|
|
"write_actions": sorted(WRITE_ACTIONS) if os.environ.get("ARR_MCP_WRITE", "").strip().lower() in ("1", "true", "yes", "on") else [],
|
|
"preferred_compact_actions": sorted(PSEUDO_ACTIONS),
|
|
}
|
|
allow_write = os.environ.get("ARR_MCP_WRITE", "").strip().lower() in (
|
|
"1", "true", "yes", "on"
|
|
)
|
|
if action in READ_ONLY_ACTIONS | PSEUDO_ACTIONS:
|
|
pass
|
|
elif allow_write and action in WRITE_ACTIONS:
|
|
pass
|
|
else:
|
|
if allow_write:
|
|
raise PermissionError(
|
|
f"Sonarr MCP write mode is enabled, but action '{action}' "
|
|
"is not in the allowed write set. Allowed: "
|
|
f"{sorted(WRITE_ACTIONS)}"
|
|
)
|
|
raise PermissionError(
|
|
f"Sonarr action '{action}' is blocked by the server-side "
|
|
"read-only policy. Set ARR_MCP_WRITE=1 to enable write mode."
|
|
)
|
|
client = get_sonarr_client()
|
|
kwargs = {k: v for k, v in json.loads(params_json).items() if v is not None}
|
|
if action == "find_series":
|
|
return await _find_series(client, kwargs)
|
|
if action == "get_season_summary":
|
|
return await _season_summary(client, kwargs)
|
|
if action == "search_releases":
|
|
return await _search_releases(client, kwargs)
|
|
result = await run_blocking(
|
|
dispatch, client, action, kwargs, service="arr-sonarr"
|
|
)
|
|
return _compact_result(action, result)
|