#!/usr/bin/env python3 """romarr-mcp — an MCP server for ROMarr, the *arr for games. Talks to ROMarr's REST API (documented at /api/v1/openapi.json). ROMarr generates its spec from the routes it actually serves, but it carries no OpenAPI tags, so a generic spec-to-MCP bridge exposes all 133 operations with names like ``postApiV1AuditQuarantineMove``. This server names the twenty that are actually useful and gives them arguments a model can reason about. Standard library only: no httpx, no mcp SDK. That keeps the Nix package a single file with no vendored wheels, and the protocol surface this server needs (initialize, tools/list, tools/call, ping) is small. Environment: ROMARR_URL base URL, default http://127.0.0.1:6868 ROMARR_API_KEY required; Settings -> General, or ROMARR_API_KEY upstream ROMARR_READ_ONLY when "1"/"true", refuses every write tool ROMARR_TIMEOUT seconds, default 60 ROMARR_VERIFY_TLS set "0" to skip certificate verification """ from __future__ import annotations import json import os import ssl import sys import urllib.error import urllib.parse import urllib.request SERVER_NAME = "romarr" SERVER_VERSION = "1.0.0" PROTOCOL_VERSION = "2024-11-05" BASE_URL = os.environ.get("ROMARR_URL", "http://127.0.0.1:6868").rstrip("/") API_KEY = os.environ.get("ROMARR_API_KEY", "").strip() READ_ONLY = os.environ.get("ROMARR_READ_ONLY", "").strip().lower() in ("1", "true", "yes", "on") TIMEOUT = float(os.environ.get("ROMARR_TIMEOUT", "60") or 60) VERIFY_TLS = os.environ.get("ROMARR_VERIFY_TLS", "1").strip().lower() not in ("0", "false", "no", "off") class ToolError(Exception): """A tool failed in a way the caller should see as an error, not a crash.""" # --------------------------------------------------------------------------- http def _ssl_context(): if VERIFY_TLS: return None ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE return ctx def call_api(method, path, params=None, body=None): """Call ROMarr and return the decoded JSON body.""" if not API_KEY: raise ToolError( "ROMARR_API_KEY is not set. Find the key under ROMarr -> Settings -> General " "(or GET /api/v1/system/apikey with an existing credential)." ) url = BASE_URL + path clean = {k: v for k, v in (params or {}).items() if v is not None and v != ""} if clean: url += "?" + urllib.parse.urlencode(clean, doseq=True) data = None headers = {"X-Api-Key": API_KEY, "Accept": "application/json"} if body is not None: data = json.dumps(body).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=TIMEOUT, context=_ssl_context()) as resp: raw = resp.read().decode("utf-8", "replace") status = resp.status except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", "replace") status = exc.code detail = raw try: parsed = json.loads(raw) detail = parsed.get("error") or parsed.get("detail") or raw except Exception: pass raise ToolError(f"ROMarr {method} {path} -> HTTP {status}: {str(detail)[:400]}") except urllib.error.URLError as exc: raise ToolError(f"cannot reach ROMarr at {BASE_URL}: {exc.reason}") if not raw.strip(): return {"status": status, "ok": True} try: return json.loads(raw) except json.JSONDecodeError: return {"status": status, "body": raw[:4000]} def get(endpoint, **params): return call_api("GET", endpoint, params=params) def post(endpoint, body=None): if READ_ONLY: raise ToolError("ROMARR_READ_ONLY is set; this server only reads right now.") return call_api("POST", endpoint, body=body if body is not None else {}) # -------------------------------------------------------------------------- tools def t_system_status(_): return get("/api/v1/system/status") def t_system_counts(_): return get("/api/v1/system/counts") def t_tasks(a): return get("/api/v1/system/tasks", limit=a.get("limit"), platform=a.get("platform")) def t_library(a): return get( "/api/v1/game", platform=a.get("platform"), shelf=a.get("shelf"), limit=a.get("limit", 50), offset=a.get("offset"), decade=a.get("decade"), genre=a.get("genre"), region=a.get("region"), origin=a.get("origin"), ) def t_wanted(a): return get("/api/v1/wanted/missing", platform=a.get("platform"), limit=a.get("limit"), view=a.get("view")) def t_queue(_): return get("/api/v1/queue") def t_history(a): return get("/api/v1/history", limit=a.get("limit"), platform=a.get("platform")) def t_search_releases(a): if not a.get("game"): raise ToolError("game is required") return get("/api/v1/release", game=a["game"], platform=a.get("platform")) def t_platforms(_): return get("/api/platforms") def t_audit(a): return get("/api/v1/audit", platform=a.get("platform"), status=a.get("status"), game=a.get("game")) def t_duplicates(_): return get("/api/v1/audit/quarantine") def t_indexers(_): return get("/api/v1/indexer") def t_download_clients(_): return get("/api/v1/downloadclient") def t_request_game(a): game, platform = (a.get("game") or "").strip(), (a.get("platform") or "").strip() if not game or not platform: raise ToolError("both game and platform are required") return post("/api/request", {"game": game, "platform": platform}) def t_grab_release(a): release_id = (a.get("id") or "").strip() if not release_id: raise ToolError("id is required (from search_releases)") return post("/api/v1/release/grab", {"id": release_id}) def t_queue_action(a): action = (a.get("action") or "").strip() if action not in ("retry", "remove"): raise ToolError("action must be 'retry' or 'remove'") if a.get("index") is None: raise ToolError("index is required (row number from queue)") return post("/api/v1/queue/action", {"index": int(a["index"]), "action": action}) def t_audit_run(a): return post("/api/v1/audit", {"platform": a.get("platform") or ""}) def t_quarantine(a): if a.get("confirm") is not True: raise ToolError("pass confirm=true; this moves files into ROMarr's quarantine") files = a.get("files") or [] if not files: raise ToolError("files is required (the paths from the duplicates list)") return post("/api/v1/audit/quarantine/move", {"platform": a.get("platform") or "", "files": files}) def t_rename_preview(a): return post( "/api/v1/audit/renames/preview", {"platform": a.get("platform") or "", "scope": a.get("scope") or "", "offset": int(a.get("offset") or 0)}, ) def t_import_finished(_): return post("/api/import", {}) def t_manual_scan(a): if not a.get("path"): raise ToolError("path is required (a directory on ROMarr's host)") return get("/api/v1/manualimport", path=a["path"], platform=a.get("platform")) def t_indexer_schema(_): return get("/api/v1/indexer/schema") def _secret_from(body, field, literal, env_name): """Set a credential field, optionally resolved from the server's own env. Passing a key by *reference* means the value never enters the transcript: the gateway already exposes Prowlarr's and SABnzbd's keys to this process, so the caller names the variable instead of quoting the secret. """ if env_name: value = os.environ.get(env_name, "") if not value: raise ToolError(f"{env_name} is not set in this server's environment") body[field] = value elif literal: body[field] = literal def t_indexer_save(a): kind = (a.get("type") or "").strip() if not kind: raise ToolError("type is required (see indexer_schema), e.g. 'prowlarr'") body = dict(a.get("extra") or {}) body["type"] = kind for field in ("name", "url", "enable", "id"): if a.get(field) not in (None, ""): body[field] = a[field] _secret_from(body, "api_key", a.get("api_key"), a.get("api_key_env")) return post("/api/v1/indexer", body) def t_client_schema(_): return get("/api/v1/downloadclient/schema") def t_client_save(a): kind = (a.get("type") or "").strip() if not kind: raise ToolError("type is required (see download_client_schema), e.g. 'qbittorrent'") body = dict(a.get("extra") or {}) body["type"] = kind for field in ("name", "host", "port", "use_ssl", "url_base", "username", "category", "id"): if a.get(field) not in (None, ""): body[field] = a[field] _secret_from(body, "password", a.get("password"), a.get("password_env")) _secret_from(body, "api_key", a.get("api_key"), a.get("api_key_env")) return post("/api/v1/downloadclient", body) def _obj(props, required=None): return { "type": "object", "properties": props, "required": required or [], "additionalProperties": False, } P_PLATFORM = {"type": "string", "description": "Platform folder slug, e.g. 'psx', 'snes', 'gbc'."} P_LIMIT = {"type": "integer", "description": "Maximum rows to return.", "minimum": 1, "maximum": 500} TOOLS = [ { "name": "system_status", "description": "Health of every ROMarr dependency (library server, indexers, download clients) plus play-route counts.", "schema": _obj({}), "write": False, "fn": t_system_status, }, { "name": "system_counts", "description": "Library and queue sizes: how many games, how many on disk, how many catalogued elsewhere.", "schema": _obj({}), "write": False, "fn": t_system_counts, }, { "name": "tasks", "description": "The scheduled jobs with interval, last run and last result — use it to see whether a scan or audit is in flight.", "schema": _obj({"limit": P_LIMIT, "platform": P_PLATFORM}), "write": False, "fn": t_tasks, }, { "name": "library", "description": "The games ROMarr holds, filterable by platform, shelf, decade, genre, region. Paged by limit/offset.", "schema": _obj( { "platform": P_PLATFORM, "shelf": {"type": "string", "description": "Shelf filter such as 'top-rated' or 'recent'."}, "limit": P_LIMIT, "offset": {"type": "integer", "minimum": 0}, "decade": {"type": "string", "description": "e.g. '1990'."}, "genre": {"type": "string"}, "region": {"type": "string", "description": "e.g. 'USA', 'Europe', 'Japan'."}, "origin": {"type": "string", "description": "Where the entry came from: on-disk, elsewhere, or wanted."}, } ), "write": False, "fn": t_library, }, { "name": "wanted_missing", "description": "Games that are wanted but not yet found — the acquisition backlog.", "schema": _obj({"platform": P_PLATFORM, "limit": P_LIMIT, "view": {"type": "string"}}), "write": False, "fn": t_wanted, }, { "name": "queue", "description": "Active downloads with their state, progress and error text.", "schema": _obj({}), "write": False, "fn": t_queue, }, { "name": "history", "description": "What ROMarr has done: grabs, imports, failures, with timestamps.", "schema": _obj({"limit": P_LIMIT, "platform": P_PLATFORM}), "write": False, "fn": t_history, }, { "name": "search_releases", "description": "Interactive search across every configured indexer for one game+platform; returns scored candidates with a rejection reason per candidate and the id to pass to grab_release.", "schema": _obj({"game": {"type": "string"}, "platform": P_PLATFORM}, ["game"]), "write": False, "fn": t_search_releases, }, { "name": "platforms", "description": "Every platform ROMarr knows, with its media, file extensions, size ceiling and how many games it holds.", "schema": _obj({}), "write": False, "fn": t_platforms, }, { "name": "audit", "description": "DAT verification state of the library: verified, bad dump, unknown, per platform. Read-only view of the last audit.", "schema": _obj({"platform": P_PLATFORM, "status": {"type": "string"}, "game": {"type": "string"}}), "write": False, "fn": t_audit, }, { "name": "duplicates", "description": "Preview exact byte-identical duplicates and list the reversible quarantine manifests already recorded.", "schema": _obj({}), "write": False, "fn": t_duplicates, }, { "name": "indexers", "description": "Configured indexers with their keys masked, plus whether each is enabled.", "schema": _obj({}), "write": False, "fn": t_indexers, }, { "name": "download_clients", "description": "Configured download clients (torrent and usenet), keys masked.", "schema": _obj({}), "write": False, "fn": t_download_clients, }, { "name": "request_game", "description": "Request a game: ROMarr searches the indexers, scores the releases, grabs the best and imports it. This starts a download.", "schema": _obj({"game": {"type": "string"}, "platform": P_PLATFORM}, ["game", "platform"]), "write": True, "fn": t_request_game, }, { "name": "grab_release", "description": "Grab one specific candidate returned by search_releases, by its id.", "schema": _obj({"id": {"type": "string", "description": "Release id from search_releases."}}, ["id"]), "write": True, "fn": t_grab_release, }, { "name": "queue_action", "description": "Retry or remove one queue row by its index.", "schema": _obj( { "index": {"type": "integer", "minimum": 0, "description": "Row index from queue."}, "action": {"type": "string", "enum": ["retry", "remove"]}, }, ["index", "action"], ), "write": True, "fn": t_queue_action, }, { "name": "audit_run", "description": "Re-verify one platform (or all) against the loaded No-Intro/Redump DATs. Long-running.", "schema": _obj({"platform": P_PLATFORM}), "write": True, "fn": t_audit_run, }, { "name": "quarantine_duplicates", "description": "Move selected byte-identical duplicates into ROMarr's quarantine. Reversible via the manifest, but it moves files: requires confirm=true.", "schema": _obj( { "files": {"type": "array", "items": {"type": "string"}, "description": "File paths from duplicates."}, "platform": P_PLATFORM, "confirm": {"type": "boolean", "description": "Must be true; acknowledgement that files are moved."}, }, ["files", "confirm"], ), "write": True, "fn": t_quarantine, }, { "name": "rename_preview", "description": "Build a read-only, DAT-verified rename preview (e.g. scope='missing-dat-names'). Nothing is renamed.", "schema": _obj({"platform": P_PLATFORM, "scope": {"type": "string"}, "offset": {"type": "integer", "minimum": 0}}), "write": True, "fn": t_rename_preview, }, { "name": "import_finished", "description": "Run an import pass over finished downloads, so completed items land in the library.", "schema": _obj({}), "write": True, "fn": t_import_finished, }, { "name": "manual_scan", "description": "Scan a directory on ROMarr's host for files it could adopt, without importing them.", "schema": _obj({"path": {"type": "string"}, "platform": P_PLATFORM}, ["path"]), "write": False, "fn": t_manual_scan, }, { "name": "indexer_schema", "description": "Every indexer type ROMarr supports, with the fields each one needs — read this before indexer_save.", "schema": _obj({}), "write": False, "fn": t_indexer_schema, }, { "name": "indexer_save", "description": "Create or update an indexer. Prefer api_key_env over api_key: the server resolves the variable from its own environment, so the secret never appears in the call.", "schema": _obj( { "type": {"type": "string", "description": "Indexer type, e.g. 'prowlarr', 'torznab', 'newznab'."}, "name": {"type": "string"}, "url": {"type": "string", "description": "Base URL, e.g. http://127.0.0.1:9696 for Prowlarr."}, "api_key": {"type": "string", "description": "Inline key. Avoid: prefer api_key_env."}, "api_key_env": {"type": "string", "description": "Name of an env var holding the key, e.g. 'PROWLARR_API_KEY'."}, "enable": {"type": "boolean"}, "id": {"type": "string", "description": "Existing indexer id to update."}, "extra": {"type": "object", "description": "Any other fields the type needs, verbatim."}, }, ["type"], ), "write": True, "fn": t_indexer_save, }, { "name": "download_client_schema", "description": "Every download client type ROMarr supports, with its fields and default port.", "schema": _obj({}), "write": False, "fn": t_client_schema, }, { "name": "download_client_save", "description": "Create or update a torrent/usenet download client. Credentials may be passed by reference (password_env / api_key_env) instead of inline.", "schema": _obj( { "type": {"type": "string", "description": "Client type, e.g. 'qbittorrent', 'sabnzbd', 'transmission'."}, "name": {"type": "string"}, "host": {"type": "string"}, "port": {"type": "integer"}, "use_ssl": {"type": "boolean"}, "url_base": {"type": "string"}, "username": {"type": "string"}, "password": {"type": "string", "description": "Inline. Avoid: prefer password_env."}, "password_env": {"type": "string"}, "api_key": {"type": "string", "description": "Inline. Avoid: prefer api_key_env."}, "api_key_env": {"type": "string", "description": "e.g. 'SABNZBD_API_KEY'."}, "category": {"type": "string"}, "id": {"type": "string", "description": "Existing client id to update."}, "extra": {"type": "object", "description": "Any other fields the type needs, verbatim."}, }, ["type"], ), "write": True, "fn": t_client_save, }, ] BY_NAME = {t["name"]: t for t in TOOLS} def tool_descriptors(): out = [] for t in TOOLS: d = {"name": t["name"], "description": t["description"], "inputSchema": t["schema"]} d["annotations"] = { "title": t["name"].replace("_", " ").title(), "readOnlyHint": not t["write"], "destructiveHint": t["name"] in ("quarantine_duplicates", "queue_action"), "openWorldHint": True, } out.append(d) return out # ----------------------------------------------------------------------- protocol def handle(req): method = req.get("method") rid = req.get("id") params = req.get("params") or {} if method == "initialize": return { "protocolVersion": params.get("protocolVersion") or PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, } if method == "ping": return {} if method == "tools/list": return {"tools": tool_descriptors()} if method == "tools/call": name = params.get("name") tool = BY_NAME.get(name) if tool is None: raise ToolError(f"unknown tool: {name}") if tool["write"] and READ_ONLY: raise ToolError("ROMARR_READ_ONLY is set; this server only reads right now.") result = tool["fn"](params.get("arguments") or {}) return { "content": [{"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)[:100000]}], "structuredContent": result if isinstance(result, dict) else {"result": result}, "isError": False, } if method in ("notifications/initialized", "notifications/cancelled"): return None if method == "resources/list": return {"resources": []} if method == "prompts/list": return {"prompts": []} raise _UnknownMethod(method) class _UnknownMethod(Exception): def __init__(self, method): self.method = method def _read_message(stream): """Newline-delimited JSON, tolerating the LSP-style Content-Length framing.""" while True: line = stream.readline() if not line: return None line = line.strip() if not line: continue if line.lower().startswith("content-length:"): try: length = int(line.split(":", 1)[1].strip()) except ValueError: continue while True: # skip remaining headers up to the blank line hdr = stream.readline() if not hdr or not hdr.strip(): break payload = stream.read(length) if not payload: return None try: return json.loads(payload) except json.JSONDecodeError: continue try: return json.loads(line) except json.JSONDecodeError: continue def _write_message(obj): sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n") sys.stdout.flush() def main(): if not API_KEY: sys.stderr.write("romarr-mcp: ROMARR_API_KEY is not set; every tool will fail.\n") sys.stderr.write(f"romarr-mcp {SERVER_VERSION}: stdio, target {BASE_URL}, {len(TOOLS)} tools\n") sys.stderr.flush() while True: msg = _read_message(sys.stdin) if msg is None: return 0 if not isinstance(msg, dict) or "method" not in msg: continue is_notification = "id" not in msg try: result = handle(msg) except _UnknownMethod as exc: if not is_notification: _write_message({"jsonrpc": "2.0", "id": msg.get("id"), "error": {"code": -32601, "message": f"method not found: {exc.method}"}}) continue except ToolError as exc: if not is_notification: _write_message( { "jsonrpc": "2.0", "id": msg.get("id"), "result": {"content": [{"type": "text", "text": f"error: {exc}"}], "isError": True}, } ) continue except Exception as exc: # noqa: BLE001 - never kill the server on one bad call if not is_notification: _write_message( { "jsonrpc": "2.0", "id": msg.get("id"), "result": {"content": [{"type": "text", "text": f"error: {type(exc).__name__}: {exc}"}], "isError": True}, } ) continue if is_notification or result is None: continue _write_message({"jsonrpc": "2.0", "id": msg.get("id"), "result": result}) if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: sys.exit(0)