"""Sonarr condensed action-routed MCP tool. CONCEPT:ECO-4.82 — gitlab-style organized per-service tool surface. """ import os import json import re import secrets import time 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", "preview_episode_search", } ) WRITE_ACTIONS = frozenset({"start_episode_search"}) MAX_COLLECTION_ITEMS = 50 APPROVAL_TTL_SECONDS = 600 _APPROVALS: dict[str, tuple[float, str]] = {} 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 _episode_numbers(value: Any) -> list[int]: if value is None: return [] if not isinstance(value, list) or len(value) > 100: raise ValueError("episode_numbers must be a JSON list with at most 100 entries") numbers = sorted({int(item) for item in value}) if any(item < 0 or item > 9999 for item in numbers): raise ValueError("episode_numbers contains an invalid episode number") return numbers async def _resolve_episode_search(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]: series_id = int(kwargs["series_id"]) season_number = int(kwargs["season_number"]) requested_numbers = _episode_numbers(kwargs.get("episode_numbers")) if series_id < 1 or season_number < 0: raise ValueError("series_id and season_number must be non-negative identifiers") 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", ) ) candidates = [ item for item in episodes if isinstance(item, dict) and int(item.get("seasonNumber", -1)) == season_number and (not requested_numbers or int(item.get("episodeNumber", -1)) in requested_numbers) ] if isinstance(episodes, list) else [] if requested_numbers: found_numbers = {int(item.get("episodeNumber", -1)) for item in candidates} missing_metadata = sorted(set(requested_numbers) - found_numbers) if missing_metadata: raise ValueError(f"Sonarr has no episode metadata for episode numbers: {missing_metadata}") missing = [item for item in candidates if not bool(item.get("hasFile"))] if not candidates: raise ValueError("No Sonarr episodes match the requested scope") if len(missing) > 100: raise ValueError("Refusing to search more than 100 missing episodes at once") compact = [_compact_episode(item) for item in missing] scope = { "series_id": series_id, "series_title": str(series.get("title", "")) if isinstance(series, dict) else "", "season_number": season_number, "requested_episode_numbers": requested_numbers, "missing_episode_ids": [int(item["id"]) for item in missing], "missing_episode_numbers": [int(item["episodeNumber"]) for item in missing], } fingerprint = json.dumps(scope, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return { "scope": scope, "fingerprint": fingerprint, "selected_episode_count": len(candidates), "already_present_count": len(candidates) - len(missing), "unmonitored_missing_count": sum(not bool(item.get("monitored")) for item in missing), "missing_episodes": compact, } async def _preview_episode_search(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]: resolved = await _resolve_episode_search(client, kwargs) ticket = secrets.token_urlsafe(24) now = time.monotonic() for old_ticket, (expires, _) in list(_APPROVALS.items()): if expires <= now: _APPROVALS.pop(old_ticket, None) _APPROVALS[ticket] = (now + APPROVAL_TTL_SECONDS, resolved["fingerprint"]) return { "action": "preview-only", **{key: value for key, value in resolved.items() if key != "fingerprint"}, "monitoring_changed": False, "download_started": False, "approval_ticket": ticket, "approval_expires_in_seconds": APPROVAL_TTL_SECONDS, "next_step": ( "Review series, season and episode list. Only after explicit approval call " "start_episode_search with exactly the same scope, confirm=true and this ticket." ), } async def _start_episode_search(client: Any, kwargs: dict[str, Any]) -> dict[str, Any]: if kwargs.get("confirm") is not True: raise PermissionError("confirm=true is required after reviewing preview_episode_search") ticket = str(kwargs.get("approval_ticket", "")) if not ticket: raise PermissionError("approval_ticket is required") resolved = await _resolve_episode_search(client, kwargs) approval = _APPROVALS.pop(ticket, None) if approval is None or approval[0] <= time.monotonic(): raise PermissionError("Approval ticket is missing, expired or already used") if not secrets.compare_digest(approval[1], resolved["fingerprint"]): raise PermissionError("Approval ticket does not match this exact episode search") episode_ids = resolved["scope"]["missing_episode_ids"] if not episode_ids: return { "ok": True, "command_started": False, "reason": "All selected episodes already have files", "scope": resolved["scope"], } command = _unwrap( await run_blocking( dispatch, client, "post_command", {"name": "EpisodeSearch", "episodeIds": episode_ids}, service="arr-sonarr", ) ) return { "ok": True, "command_started": True, "sonarr_command": _pick(command, ("id", "name", "status", "queued", "startedOn")) if isinstance(command, dict) else command, "scope": resolved["scope"], "monitoring_changed": False, "instruction": "Sonarr is now searching its configured indexers for the approved missing episodes.", } def register_sonarr_tools(mcp: FastMCP) -> None: @mcp.tool(tags={"sonarr"}) async def sonarr_action( action: str = Field( description="Sonarr action. Read with find_series {query}, get_season_summary {series_id, season_number}, or search_releases {series_id, season_number}. To download missing episodes, first call preview_episode_search {series_id, season_number, episode_numbers:[2,3,...]}, show its exact preview to the user, then only after explicit approval call start_episode_search with the same scope plus confirm:true and approval_ticket. Monitoring is never changed." ), 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) if action == "preview_episode_search": return await _preview_episode_search(client, kwargs) if action == "start_episode_search": return await _start_episode_search(client, kwargs) result = await run_blocking( dispatch, client, action, kwargs, service="arr-sonarr" ) return _compact_result(action, result)