""" Seerr skill — connects to Overseerr / Jellyseerr API for media discovery, requests, and issue submission. .env variables: SEERR_URL – base URL (e.g. https://seerr.example.com) SEERR_API_KEY – API key from Seerr settings SEERR_TIMEOUT – optional request timeout in seconds (default 30) """ from __future__ import annotations import json from urllib.parse import quote import httpx from skills import Skill, register, ToolResult, get_config # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- SEERR_URL = (get_config("SEERR_URL") or "").rstrip("/") SEERR_API_KEY = get_config("SEERR_API_KEY") or "" SEERR_TIMEOUT = int(get_config("SEERR_TIMEOUT", "30")) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _client() -> httpx.AsyncClient: return httpx.AsyncClient( base_url=SEERR_URL, headers={"X-Api-Key": SEERR_API_KEY}, timeout=SEERR_TIMEOUT, ) def _fmt_items(items: list[dict], kind: str) -> str: """Format a list of media items for the LLM to present.""" lines = [] for i, item in enumerate(items[:10], 1): title = item.get("title") or item.get("name") or "Unknown" year = ( item.get("releaseDate", "")[:4] or item.get("firstAirDate", "")[:4] or "?" ) overview = (item.get("overview") or "")[:120] lines.append(f"{i}. **{title}** ({year}) — {overview}…") return f"Found {len(items)} {kind}. Top results:\n\n" + "\n".join(lines) # --------------------------------------------------------------------------- # Tool definitions (OpenAI function-calling schema) # --------------------------------------------------------------------------- TOOLS = [ { "type": "function", "function": { "name": "seerr_trending", "description": "Get trending movies and TV shows from Seerr using " "the /discover/trending endpoint. Call this when a user asks what " "is popular, trending, or new.", "parameters": { "type": "object", "properties": { "kind": { "type": "string", "enum": ["movie", "tv", "all"], "description": "What kind of media to fetch. " "Use 'all' when the user doesn't specify.", }, "language": { "type": "string", "description": "Language filter (e.g. 'en', 'nl'). " "Omit for all languages.", }, }, "required": ["kind"], }, }, }, { "type": "function", "function": { "name": "seerr_discover", "description": "Discover movies or TV shows by genre, studio, " "keyword, or language in Seerr. Uses /discover/{movies|tv}/genre/{id} " "for genre queries, /discover/{movies|tv}/studio/{id} for studios, " "and /discover/{movies|tv}?query= for keyword search. " "Call when a user asks 'what movies in category X do you recommend?' " "or 'show me horror movies' or 'find Studio Ghibli movies'.", "parameters": { "type": "object", "properties": { "kind": { "type": "string", "enum": ["movie", "tv"], "description": "Media type to search.", }, "genre": { "type": "string", "description": "Genre name, e.g. 'horror', 'comedy', " "'animation', 'action', 'science fiction'. " "Use this for genre-based discovery.", }, "studio": { "type": "string", "description": "Studio name to filter by, e.g. " "'Studio Ghibli', 'Pixar', 'Marvel'. " "Use this for studio-based discovery.", }, "keyword": { "type": "string", "description": "Free-text keyword search, e.g. " "'space', 'superhero', 'dinosaur'. " "Use this for topic-based discovery.", }, "language": { "type": "string", "description": "Language filter (e.g. 'en', 'ja'). " "Omit for all languages.", }, "page": { "type": "integer", "description": "Page number (default 1).", }, }, "required": ["kind"], }, }, }, { "type": "function", "function": { "name": "seerr_request_media", "description": "Request a movie or TV show to be added to the media " "library via Seerr. Call when a user asks 'can you request movie X?' " "or 'please add show Y'.", "parameters": { "type": "object", "properties": { "kind": { "type": "string", "enum": ["movie", "tv"], "description": "Whether this is a movie or TV show.", }, "title": { "type": "string", "description": "The title of the movie or TV show to request.", }, "tmdb_id": { "type": "integer", "description": "The TMDb ID if known (optional — Seerr will " "search by title if not provided).", }, }, "required": ["kind", "title"], }, }, }, { "type": "function", "function": { "name": "seerr_submit_issue", "description": "Submit a ticket/issue for a specific media item. " "Call when a user wants to report a problem (bad quality, wrong " "language, missing episodes, corrupt file, etc.) or when they want " "an action that only a human operator can perform. " "IMPORTANT: always include the media_title so the system can " "look up the correct mediaId.", "parameters": { "type": "object", "properties": { "subject": { "type": "string", "description": "Short summary of the issue.", }, "description": { "type": "string", "description": "Detailed description of the problem.", }, "media_title": { "type": "string", "description": "The movie or TV show title this issue " "relates to. Always provide this — the system will " "search for the matching mediaId.", }, "issue_type": { "type": "integer", "enum": [1, 2, 3, 4], "description": "Issue category code: " "1 = Video (playback, codec, quality), " "2 = Audio (sync, missing), " "3 = Subtitle (missing, wrong, timing), " "4 = Other (operator-only actions like delete/cancel).", }, }, "required": ["subject", "description", "media_title", "issue_type"], }, }, }, ] # --------------------------------------------------------------------------- # Tool executor # --------------------------------------------------------------------------- async def _execute(tool_name: str, args: dict) -> ToolResult: """Route tool calls to the right handler. Returns ToolResult with success based on HTTP status code (2xx = ok, everything else = fail).""" import logging logger = logging.getLogger("skills.seerr") handlers = { "seerr_trending": _trending, "seerr_discover": _discover, "seerr_request_media": _request_media, "seerr_submit_issue": _submit_issue, } handler = handlers.get(tool_name) if not handler: return ToolResult.fail(f"Unknown tool: {tool_name}") try: result = await handler(args) return result except httpx.HTTPStatusError as exc: status = exc.response.status_code body = exc.response.text[:500] logger.error("Seerr API HTTP %s on %s: %s", status, tool_name, body) return ToolResult.fail( f"Seerr API returned HTTP {status} for '{tool_name}'. " f"Response: {body}" ) except httpx.HTTPError as exc: logger.error("Seerr API network error on %s: %s", tool_name, exc) return ToolResult.fail( f"Seerr API is unreachable for '{tool_name}': {exc}" ) except Exception as exc: logger.exception("Unexpected error in %s", tool_name) return ToolResult.fail(f"Unexpected error in '{tool_name}': {exc}") # --------------------------------------------------------------------------- # API handlers # --------------------------------------------------------------------------- async def _trending(args: dict) -> ToolResult: """Use Jellyseerr's /api/v1/discover/trending endpoint. Query params: language (optional), mediaType (movie | tv, default all). """ media_type = args.get("kind", "all") language = args.get("language", "").strip() or None params: dict = {} if media_type in ("movie", "tv"): params["mediaType"] = media_type if language: params["language"] = language async with _client() as c: r = await c.get("/api/v1/discover/trending", params=params) r.raise_for_status() data = r.json() results = data.get("results", []) label = f"trending {media_type}" if media_type != "all" else "trending items" if language: label += f" ({language})" if not results: return ToolResult.ok(f"No {label} found right now.") return ToolResult.ok(_fmt_items(results, label)) async def _discover(args: dict) -> ToolResult: """Jellyseerr discover endpoints: Genre: /api/v1/discover/{movies|tv}/genre/{genreId} Studio: /api/v1/discover/{movies|tv}/studio/{studioId} Keyword: /api/v1/discover/{movies|tv}?query=keyword Language falls back to the base discover endpoint. """ kind = args["kind"] genre = args.get("genre", "").strip() studio = args.get("studio", "").strip() keyword = args.get("keyword", "").strip() language = args.get("language", "").strip() or None page = args.get("page", 1) # Map common genre names to TMDb genre IDs genre_map = { "action": 28, "adventure": 12, "animation": 16, "comedy": 35, "crime": 80, "documentary": 99, "drama": 18, "family": 10751, "fantasy": 14, "history": 36, "horror": 27, "music": 10402, "mystery": 9648, "romance": 10749, "science fiction": 878, "sci-fi": 878, "scifi": 878, "tv movie": 10770, "thriller": 53, "war": 10752, "western": 37, } base = f"/api/v1/discover/{'movies' if kind == 'movie' else 'tv'}" params: dict = {"page": page} endpoint = base if genre: genre_id = genre_map.get(genre.lower()) if not genre_id: return ToolResult.fail( f"I don't recognise the genre '{genre}'. " f"Try one of: {', '.join(sorted(genre_map.keys()))}." ) endpoint = f"{base}/genre/{genre_id}" elif studio: endpoint = f"{base}/studio/{studio}" elif keyword: params["query"] = keyword endpoint = base if language: params["language"] = language async with _client() as c: r = await c.get(endpoint, params=params) r.raise_for_status() results = r.json().get("results", []) desc = genre or studio or keyword or kind if not results: return ToolResult.ok(f"No {desc} {kind}s found.") return ToolResult.ok(_fmt_items(results, f"{desc} {kind}s")) async def _request_media(args: dict) -> ToolResult: kind = args["kind"] title = args["title"] tmdb_id = args.get("tmdb_id") # Step 1: Search for the media async with _client() as c: r = await c.get("/api/v1/search/", params={"query": quote(title), "page": 1}) r.raise_for_status() results = r.json().get("results", []) # Filter by mediaType if we have results from unified search filtered = [item for item in results if item.get("mediaType") == kind] if results else [] if not filtered: filtered = results # fallback if mediaType not set if not filtered: return ToolResult.fail( f"I couldn't find '{title}' on Seerr. " f"Please double-check the title or provide a TMDb ID." ) # If tmdb_id provided, match it; otherwise use the first result match = None if tmdb_id: match = next( (item for item in filtered if item.get("id") == tmdb_id), None, ) if not match: match = filtered[0] # Seerr's request endpoint expects the local mediaInfo.id media_info = match.get("mediaInfo", {}) media_id = media_info.get("id") or match.get("id") media_title = match.get("title") or match.get("name") or title media_year = ( (match.get("releaseDate") or match.get("firstAirDate") or "?")[:4] ) # Step 2: Submit the request request_body = { "mediaType": kind, "mediaId": media_id, } if kind == "tv": request_body["seasons"] = "all" req_r = await c.post("/api/v1/request", json=request_body) if req_r.status_code == 201: return ToolResult.ok( f"✅ Successfully requested **{media_title}** ({media_year}). " f"It has been submitted to Seerr and will be processed soon." ) elif req_r.status_code == 409: return ToolResult.fail( f"⚠️ **{media_title}** ({media_year}) has already been requested " f"or is already available." ) else: detail = req_r.text return ToolResult.fail( f"❌ Failed to request **{media_title}** ({media_year}). " f"Seerr responded with status {req_r.status_code}: {detail}" ) async def _submit_issue(args: dict) -> ToolResult: subject = args["subject"] description = args["description"] media_title = args.get("media_title", "") issue_type = args.get("issue_type", 4) # numeric code: 1=video, 2=audio, 3=sub, 4=other media_id = args.get("media_id") body: dict = { "issueType": int(issue_type), "subject": subject, "message": description, } if media_title: body["message"] = f"[Media: {media_title}]\n\n{description}" async with _client() as c: # --- Resolve mediaId (Seerr's internal ID, not TMDb) --- if not media_id and media_title: search_r = await c.get("/api/v1/search/", params={"query": quote(media_title), "page": 1}) if search_r.status_code == 200: results = search_r.json().get("results", []) if results: # Seerr's /api/v1/issue expects the local mediaInfo.id, # not the TMDb id at the top level. media_info = results[0].get("mediaInfo", {}) media_id = media_info.get("id") or results[0].get("id") if media_id: body["mediaId"] = int(media_id) r = await c.post("/api/v1/issue", json=body) resp_json = r.json() if r.text else {} if r.status_code in (200, 201): ticket_id = resp_json.get("id", "N/A") return ToolResult.ok( f"✅ Issue submitted successfully (ticket #{ticket_id}). " f"A human operator will review: **{subject}**" ) else: return ToolResult.fail( f"❌ Failed to submit issue. Seerr responded with " f"status {r.status_code}: {r.text[:500]}" ) # --------------------------------------------------------------------------- # Register the skill # --------------------------------------------------------------------------- seerr_skill = Skill( name="seerr", description="Seerr integration — trending, discover, request media, submit issues", prompt_fragment="""## Seerr Media Tools You have access to the Seerr media management system. Use the provided tools to help users with media-related tasks: - **seerr_trending** — when a user asks what is trending/popular/new - **seerr_discover** — when a user asks for recommendations by genre/category - **seerr_request_media** — when a user wants to request a movie or TV show - **seerr_submit_issue** — when a user needs to report a problem or needs an operator-only action (like deleting media or cancelling a request) Always confirm successful actions to the user. If a tool fails, tell the user what went wrong and suggest alternatives.""", tools=TOOLS, execute=_execute, ) register(seerr_skill)