787 lines
32 KiB
Python
Executable File
787 lines
32 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Jelly Media MCP Server & CLI Tool
|
|
Integrates Jellyfin and Jellyseerr with OpenClaw via MCP stdio or direct CLI commands.
|
|
Supports username & password authentication with automatic session management and renewal.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import ssl
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
import urllib.error
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
# Default Configurations (Can be overridden via environment variables)
|
|
JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "https://stream.paschke.dev").rstrip("/")
|
|
JELLYFIN_USERNAME = os.environ.get("JELLYFIN_USERNAME", "mop")
|
|
JELLYFIN_PASSWORD = os.environ.get("JELLYFIN_PASSWORD", "Hintenlinks1#")
|
|
|
|
JELLYSEERR_URL = os.environ.get("JELLYSEERR_URL", "https://jellyseer.paschke.dev").rstrip("/")
|
|
JELLYSEERR_USERNAME = os.environ.get("JELLYSEERR_USERNAME", "mop")
|
|
JELLYSEERR_PASSWORD = os.environ.get("JELLYSEERR_PASSWORD", "Hintenlinks1#")
|
|
|
|
CACHE_DIR = os.environ.get("JELLY_CACHE_DIR", "/tmp")
|
|
JF_CACHE_FILE = os.path.join(CACHE_DIR, "jellyfin_session.json")
|
|
JS_CACHE_FILE = os.path.join(CACHE_DIR, "jellyseerr_session.json")
|
|
|
|
# SSL Context
|
|
SSL_CTX = ssl.create_default_context()
|
|
|
|
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,
|
|
"sci-fi": 878,
|
|
"science fiction": 878,
|
|
"thriller": 53,
|
|
"war": 10752,
|
|
"western": 37,
|
|
}
|
|
|
|
|
|
class JellyClient:
|
|
def __init__(self):
|
|
self.jf_token: Optional[str] = None
|
|
self.jf_user_id: Optional[str] = None
|
|
self.jf_server_id: Optional[str] = None
|
|
self.js_cookie: Optional[str] = None
|
|
self.load_cache()
|
|
|
|
def load_cache(self):
|
|
if os.path.exists(JF_CACHE_FILE):
|
|
try:
|
|
with open(JF_CACHE_FILE, "r") as f:
|
|
data = json.load(f)
|
|
self.jf_token = data.get("token")
|
|
self.jf_user_id = data.get("user_id")
|
|
self.jf_server_id = data.get("server_id")
|
|
except Exception:
|
|
pass
|
|
|
|
if os.path.exists(JS_CACHE_FILE):
|
|
try:
|
|
with open(JS_CACHE_FILE, "r") as f:
|
|
data = json.load(f)
|
|
self.js_cookie = data.get("cookie")
|
|
except Exception:
|
|
pass
|
|
|
|
def save_cache(self):
|
|
try:
|
|
with open(JF_CACHE_FILE, "w") as f:
|
|
json.dump({
|
|
"token": self.jf_token,
|
|
"user_id": self.jf_user_id,
|
|
"server_id": self.jf_server_id,
|
|
"timestamp": time.time()
|
|
}, f)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
with open(JS_CACHE_FILE, "w") as f:
|
|
json.dump({
|
|
"cookie": self.js_cookie,
|
|
"timestamp": time.time()
|
|
}, f)
|
|
except Exception:
|
|
pass
|
|
|
|
# --- JELLYFIN AUTH & REQUESTS ---
|
|
def auth_jellyfin(self) -> bool:
|
|
login_url = f"{JELLYFIN_URL}/Users/AuthenticateByName"
|
|
auth_header = "MediaBrowser Client=\"OpenClaw\", Device=\"OpenClaw\", DeviceId=\"openclaw-bot-1\", Version=\"1.0.0\""
|
|
payload = json.dumps({"Username": JELLYFIN_USERNAME, "Pw": JELLYFIN_PASSWORD}).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
login_url,
|
|
data=payload,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Authorization": auth_header,
|
|
"User-Agent": "OpenClaw/1.0"
|
|
}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
self.jf_token = data.get("AccessToken")
|
|
self.jf_user_id = data.get("User", {}).get("Id")
|
|
self.jf_server_id = data.get("ServerId")
|
|
self.save_cache()
|
|
return True
|
|
except Exception as e:
|
|
sys.stderr.write(f"Jellyfin auth failed: {e}\n")
|
|
return False
|
|
|
|
def request_jellyfin(self, path: str, method: str = "GET", payload: Optional[Dict[str, Any]] = None) -> Any:
|
|
if not self.jf_token or not self.jf_user_id:
|
|
if not self.auth_jellyfin():
|
|
raise RuntimeError("Failed to authenticate with Jellyfin")
|
|
|
|
url = f"{JELLYFIN_URL}{path}"
|
|
auth_header = (
|
|
f"MediaBrowser Client=\"OpenClaw\", Device=\"OpenClaw\", DeviceId=\"openclaw-bot-1\", "
|
|
f"Version=\"1.0.0\", Token=\"{self.jf_token}\""
|
|
)
|
|
data_bytes = json.dumps(payload).encode("utf-8") if payload else None
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=data_bytes,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Authorization": auth_header,
|
|
"X-Emby-Token": self.jf_token or "",
|
|
"User-Agent": "OpenClaw/1.0"
|
|
},
|
|
method=method
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
return json.loads(body) if body else {}
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 401:
|
|
# Token expired, retry once
|
|
if self.auth_jellyfin():
|
|
auth_header = (
|
|
f"MediaBrowser Client=\"OpenClaw\", Device=\"OpenClaw\", DeviceId=\"openclaw-bot-1\", "
|
|
f"Version=\"1.0.0\", Token=\"{self.jf_token}\""
|
|
)
|
|
req.headers["Authorization"] = auth_header
|
|
req.headers["X-Emby-Token"] = self.jf_token or ""
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
return json.loads(body) if body else {}
|
|
raise
|
|
|
|
# --- JELLYSEERR AUTH & REQUESTS ---
|
|
def auth_jellyseerr(self) -> bool:
|
|
login_url = f"{JELLYSEERR_URL}/api/v1/auth/jellyfin"
|
|
payload = json.dumps({"username": JELLYSEERR_USERNAME, "password": JELLYSEERR_PASSWORD}).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
login_url,
|
|
data=payload,
|
|
headers={"Content-Type": "application/json", "User-Agent": "OpenClaw/1.0"}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
cookies = resp.info().get_all("Set-Cookie") or []
|
|
for c in cookies:
|
|
if "connect.sid=" in c:
|
|
self.js_cookie = c.split(";")[0]
|
|
break
|
|
self.save_cache()
|
|
return bool(self.js_cookie)
|
|
except Exception as e:
|
|
sys.stderr.write(f"Jellyseerr auth failed: {e}\n")
|
|
return False
|
|
|
|
def request_jellyseerr(self, path: str, method: str = "GET", payload: Optional[Dict[str, Any]] = None) -> Any:
|
|
if not self.js_cookie:
|
|
if not self.auth_jellyseerr():
|
|
raise RuntimeError("Failed to authenticate with Jellyseerr")
|
|
|
|
url = f"{JELLYSEERR_URL}{path}"
|
|
data_bytes = json.dumps(payload).encode("utf-8") if payload else None
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Cookie": self.js_cookie or "",
|
|
"User-Agent": "OpenClaw/1.0"
|
|
}
|
|
req = urllib.request.Request(url, data=data_bytes, headers=headers, method=method)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
return json.loads(body) if body else {}
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (401, 403):
|
|
# Session expired, retry once
|
|
if self.auth_jellyseerr():
|
|
req.headers["Cookie"] = self.js_cookie or ""
|
|
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
return json.loads(body) if body else {}
|
|
raise
|
|
|
|
# --- HIGH LEVEL TOOLS ---
|
|
|
|
def search_media(self, query: str, media_type: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""Search Jellyseerr and cross-reference with Jellyfin library."""
|
|
results: List[Dict[str, Any]] = []
|
|
|
|
# 1. Search Jellyseerr
|
|
encoded_query = urllib.parse.quote(query)
|
|
js_data = self.request_jellyseerr(f"/api/v1/search?query={encoded_query}")
|
|
raw_results = js_data.get("results", [])
|
|
|
|
# 2. Search Jellyfin directly to match any items in library
|
|
jf_matches = {}
|
|
try:
|
|
encoded_jf = urllib.parse.quote(query)
|
|
jf_items = self.request_jellyfin(
|
|
f"/Users/{self.jf_user_id}/Items?searchTerm={encoded_jf}&Limit=20&Recursive=true&IncludeItemTypes=Movie,Series"
|
|
).get("Items", [])
|
|
for item in jf_items:
|
|
name_clean = item.get("Name", "").strip().lower()
|
|
jf_matches[name_clean] = item
|
|
except Exception as e:
|
|
sys.stderr.write(f"Jellyfin search error: {e}\n")
|
|
|
|
for item in raw_results:
|
|
mtype = item.get("mediaType")
|
|
if media_type and media_type != "all" and mtype != media_type:
|
|
continue
|
|
|
|
title = item.get("title") or item.get("name") or "Unknown"
|
|
year = (item.get("releaseDate") or item.get("firstAirDate") or "")[:4]
|
|
rating = item.get("voteAverage")
|
|
minfo = item.get("mediaInfo") or {}
|
|
status_code = minfo.get("status")
|
|
|
|
# Check availability
|
|
is_available = status_code == 5
|
|
is_processing = status_code in (2, 3, 4)
|
|
|
|
jf_item = jf_matches.get(title.strip().lower())
|
|
jf_item_id = None
|
|
if jf_item:
|
|
is_available = True
|
|
jf_item_id = jf_item.get("Id")
|
|
|
|
entry: Dict[str, Any] = {
|
|
"title": title,
|
|
"year": year,
|
|
"media_type": mtype,
|
|
"tmdb_id": item.get("id"),
|
|
"rating": rating,
|
|
"overview": item.get("overview"),
|
|
"status": "AVAILABLE" if is_available else ("PROCESSING" if is_processing else "NOT_AVAILABLE"),
|
|
"poster_url": f"https://image.tmdb.org/t/p/w500{item.get("posterPath")}" if item.get("posterPath") else None,
|
|
}
|
|
|
|
if is_available:
|
|
if not jf_item_id and minfo.get("jellyfinMediaId"):
|
|
jf_item_id = minfo.get("jellyfinMediaId")
|
|
|
|
if jf_item_id:
|
|
entry["jellyfin_item_id"] = jf_item_id
|
|
entry["playback"] = {
|
|
"web_details_url": f"{JELLYFIN_URL}/web/index.html#!/details?id={jf_item_id}&serverId={self.jf_server_id}",
|
|
"stream_url": f"{JELLYFIN_URL}/Videos/{jf_item_id}/stream?static=true&api_key={self.jf_token}",
|
|
"download_url": f"{JELLYFIN_URL}/Items/{jf_item_id}/Download?api_key={self.jf_token}"
|
|
}
|
|
|
|
results.append(entry)
|
|
|
|
# Sort: AVAILABLE first, then PROCESSING, then NOT_AVAILABLE
|
|
status_priority = {"AVAILABLE": 0, "PROCESSING": 1, "NOT_AVAILABLE": 2}
|
|
results.sort(key=lambda x: (status_priority.get(x["status"], 3), -(x.get("rating") or 0)))
|
|
return results
|
|
|
|
def discover_media(
|
|
self,
|
|
genre: Optional[str] = None,
|
|
min_rating: Optional[float] = None,
|
|
year: Optional[int] = None,
|
|
sort_by: str = "popularity.desc",
|
|
only_available: bool = False,
|
|
media_type: str = "movie"
|
|
) -> List[Dict[str, Any]]:
|
|
"""Discover movies or TV shows with rich filters, prioritizing available items."""
|
|
genre_id = None
|
|
if genre:
|
|
genre_lower = genre.lower().strip()
|
|
genre_id = GENRE_MAP.get(genre_lower)
|
|
if not genre_id and genre.isdigit():
|
|
genre_id = int(genre)
|
|
|
|
endpoint = "/api/v1/discover/movies" if media_type != "tv" else "/api/v1/discover/tv"
|
|
params: List[str] = [f"sortBy={sort_by}"]
|
|
if genre_id:
|
|
params.append(f"genre={genre_id}")
|
|
if year:
|
|
params.append(f"primaryReleaseDateGte={year}-01-01")
|
|
params.append(f"primaryReleaseDateLte={year}-12-31")
|
|
|
|
query_str = "&".join(params)
|
|
js_data = self.request_jellyseerr(f"{endpoint}?{query_str}")
|
|
raw_results = js_data.get("results", [])
|
|
|
|
results: List[Dict[str, Any]] = []
|
|
for item in raw_results:
|
|
rating = item.get("voteAverage") or 0.0
|
|
if min_rating and rating < min_rating:
|
|
continue
|
|
|
|
title = item.get("title") or item.get("name") or "Unknown"
|
|
rel_date = item.get("releaseDate") or item.get("firstAirDate") or ""
|
|
minfo = item.get("mediaInfo") or {}
|
|
status_code = minfo.get("status")
|
|
|
|
is_available = status_code == 5
|
|
is_processing = status_code in (2, 3, 4)
|
|
|
|
if only_available and not is_available:
|
|
continue
|
|
|
|
entry = {
|
|
"title": title,
|
|
"year": rel_date[:4] if rel_date else None,
|
|
"media_type": media_type,
|
|
"tmdb_id": item.get("id"),
|
|
"rating": rating,
|
|
"overview": item.get("overview"),
|
|
"status": "AVAILABLE" if is_available else ("PROCESSING" if is_processing else "NOT_AVAILABLE"),
|
|
"poster_url": f"https://image.tmdb.org/t/p/w500{item.get("posterPath")}" if item.get("posterPath") else None,
|
|
}
|
|
|
|
if is_available and minfo.get("jellyfinMediaId"):
|
|
jf_id = minfo.get("jellyfinMediaId")
|
|
entry["jellyfin_item_id"] = jf_id
|
|
entry["playback"] = {
|
|
"web_details_url": f"{JELLYFIN_URL}/web/index.html#!/details?id={jf_id}&serverId={self.jf_server_id}",
|
|
"stream_url": f"{JELLYFIN_URL}/Videos/{jf_id}/stream?static=true&api_key={self.jf_token}",
|
|
"download_url": f"{JELLYFIN_URL}/Items/{jf_id}/Download?api_key={self.jf_token}"
|
|
}
|
|
|
|
results.append(entry)
|
|
|
|
# Sort: AVAILABLE first
|
|
status_priority = {"AVAILABLE": 0, "PROCESSING": 1, "NOT_AVAILABLE": 2}
|
|
results.sort(key=lambda x: (status_priority.get(x["status"], 3), -(x.get("rating") or 0)))
|
|
return results
|
|
|
|
def get_playback_info(self, item_id: str) -> Dict[str, Any]:
|
|
"""Fetch full details, web link, stream link, and download link for a Jellyfin item."""
|
|
data = self.request_jellyfin(f"/Users/{self.jf_user_id}/Items/{item_id}")
|
|
name = data.get("Name", "Unknown")
|
|
year = data.get("ProductionYear")
|
|
overview = data.get("Overview")
|
|
media_type = data.get("Type")
|
|
genres = data.get("Genres", [])
|
|
community_rating = data.get("CommunityRating")
|
|
runtime_ticks = data.get("RunTimeTicks", 0)
|
|
runtime_mins = int(runtime_ticks / (10_000_000 * 60)) if runtime_ticks else None
|
|
|
|
return {
|
|
"title": name,
|
|
"year": year,
|
|
"type": media_type,
|
|
"genres": genres,
|
|
"rating": community_rating,
|
|
"runtime_mins": runtime_mins,
|
|
"overview": overview,
|
|
"jellyfin_item_id": item_id,
|
|
"web_details_url": f"{JELLYFIN_URL}/web/index.html#!/details?id={item_id}&serverId={self.jf_server_id}",
|
|
"stream_url": f"{JELLYFIN_URL}/Videos/{item_id}/stream?static=true&api_key={self.jf_token}",
|
|
"download_url": f"{JELLYFIN_URL}/Items/{item_id}/Download?api_key={self.jf_token}",
|
|
}
|
|
|
|
def request_media(self, media_id: int, media_type: str = "movie", seasons: Optional[List[int]] = None) -> Dict[str, Any]:
|
|
"""Submit a request to Jellyseerr."""
|
|
payload: Dict[str, Any] = {
|
|
"mediaId": media_id,
|
|
"mediaType": media_type,
|
|
}
|
|
if media_type == "tv":
|
|
payload["seasons"] = seasons if seasons is not None else [1]
|
|
|
|
res = self.request_jellyseerr("/api/v1/request", method="POST", payload=payload)
|
|
return {
|
|
"success": True,
|
|
"request_id": res.get("id"),
|
|
"status": res.get("status"),
|
|
"media": res.get("media"),
|
|
"message": "Media request successfully submitted to Jellyseerr!"
|
|
}
|
|
|
|
def list_sessions(self) -> List[Dict[str, Any]]:
|
|
"""List active playback sessions and devices on Jellyfin."""
|
|
raw_sessions = self.request_jellyfin("/Sessions")
|
|
output = []
|
|
for s in raw_sessions:
|
|
np = s.get("NowPlayingItem")
|
|
output.append({
|
|
"session_id": s.get("Id"),
|
|
"device_name": s.get("DeviceName"),
|
|
"client": s.get("Client"),
|
|
"user_name": s.get("UserName"),
|
|
"supports_remote_control": s.get("SupportsRemoteControl", False),
|
|
"is_active": s.get("IsActive", False),
|
|
"now_playing": {
|
|
"name": np.get("Name"),
|
|
"type": np.get("Type"),
|
|
"id": np.get("Id")
|
|
} if np else None
|
|
})
|
|
return output
|
|
|
|
def remote_play(self, session_id: str, item_id: str) -> Dict[str, Any]:
|
|
"""Trigger remote playback on a Jellyfin client session."""
|
|
path = f"/Sessions/{session_id}/Play?playCommand=PlayNow&ItemIds={item_id}"
|
|
self.request_jellyfin(path, method="POST")
|
|
return {"success": True, "message": f"Playback triggered for item {item_id} on session {session_id}"}
|
|
|
|
def get_resume(self) -> List[Dict[str, Any]]:
|
|
"""Fetch items currently in progress for the user."""
|
|
data = self.request_jellyfin(f"/Users/{self.jf_user_id}/Items/Resume?Recursive=true")
|
|
items = data.get("Items", [])
|
|
output = []
|
|
for it in items:
|
|
ticks = it.get("UserData", {}).get("PlaybackPositionTicks", 0)
|
|
total_ticks = it.get("RunTimeTicks", 1) or 1
|
|
pct = round((ticks / total_ticks) * 100, 1)
|
|
mins = round(ticks / (10_000_000 * 60), 1)
|
|
total_mins = round(total_ticks / (10_000_000 * 60), 1)
|
|
|
|
iid = it.get("Id")
|
|
output.append({
|
|
"title": it.get("Name"),
|
|
"series": it.get("SeriesName"),
|
|
"type": it.get("Type"),
|
|
"item_id": iid,
|
|
"progress_percent": pct,
|
|
"position_mins": mins,
|
|
"total_mins": total_mins,
|
|
"web_details_url": f"{JELLYFIN_URL}/web/index.html#!/details?id={iid}&serverId={self.jf_server_id}",
|
|
"stream_url": f"{JELLYFIN_URL}/Videos/{iid}/stream?static=true&api_key={self.jf_token}"
|
|
})
|
|
return output
|
|
|
|
def get_latest(self, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""Fetch newly added items in the Jellyfin library."""
|
|
data = self.request_jellyfin(f"/Users/{self.jf_user_id}/Items/Latest?Limit={limit}")
|
|
output = []
|
|
for it in data:
|
|
iid = it.get("Id")
|
|
output.append({
|
|
"title": it.get("Name"),
|
|
"series": it.get("SeriesName"),
|
|
"type": it.get("Type"),
|
|
"year": it.get("ProductionYear"),
|
|
"item_id": iid,
|
|
"web_details_url": f"{JELLYFIN_URL}/web/index.html#!/details?id={iid}&serverId={self.jf_server_id}",
|
|
"stream_url": f"{JELLYFIN_URL}/Videos/{iid}/stream?static=true&api_key={self.jf_token}",
|
|
"download_url": f"{JELLYFIN_URL}/Items/{iid}/Download?api_key={self.jf_token}"
|
|
})
|
|
return output
|
|
|
|
|
|
# --- MCP SERVER STDIO IMPLEMENTATION ---
|
|
|
|
MCP_TOOLS = [
|
|
{
|
|
"name": "jelly_search",
|
|
"description": "Search for movies or series across Jellyseerr and Jellyfin. Returns availability status (AVAILABLE, PROCESSING, NOT_AVAILABLE) favoring existing library media with instant playback and download URLs.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "The search term, title, or keywords"},
|
|
"media_type": {"type": "string", "enum": ["all", "movie", "tv"], "description": "Optional filter for media type", "default": "all"}
|
|
},
|
|
"required": ["query"]
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_discover",
|
|
"description": "Discover movies or TV shows based on genre, minimum rating, release year, or popularity. Automatically checks and highlights items already available to stream on Jellyfin.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"genre": {"type": "string", "description": "Genre name (e.g. action, comedy, sci-fi, thriller, drama, animation, horror)"},
|
|
"min_rating": {"type": "number", "description": "Minimum TMDB score out of 10 (e.g. 7.0)"},
|
|
"year": {"type": "integer", "description": "Release year filter (e.g. 2024)"},
|
|
"sort_by": {"type": "string", "description": "Sort order (e.g. popularity.desc, vote_average.desc)", "default": "popularity.desc"},
|
|
"only_available": {"type": "boolean", "description": "If true, returns only items already downloaded on Jellyfin", "default": False},
|
|
"media_type": {"type": "string", "enum": ["movie", "tv"], "description": "Type of media", "default": "movie"}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_get_playback",
|
|
"description": "Get web details URL, direct streaming URL, and offline direct download URL for an item in the Jellyfin library.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"item_id": {"type": "string", "description": "The Jellyfin item ID"}
|
|
},
|
|
"required": ["item_id"]
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_request",
|
|
"description": "Submit a download request for a movie or TV series in Jellyseerr on behalf of user mop.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"media_id": {"type": "integer", "description": "The TMDB ID of the movie or series"},
|
|
"media_type": {"type": "string", "enum": ["movie", "tv"], "description": "Type of media", "default": "movie"},
|
|
"seasons": {"type": "array", "items": {"type": "integer"}, "description": "For TV series: list of season numbers to request (e.g. [1, 2])"}
|
|
},
|
|
"required": ["media_id"]
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_list_sessions",
|
|
"description": "List all active playback devices and sessions connected to Jellyfin (e.g. Apple TV in Living Room, iPad, Chrome Web).",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_remote_play",
|
|
"description": "Send a command to a specific Jellyfin client session to immediately play an item.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"session_id": {"type": "string", "description": "The Jellyfin session ID of the target device"},
|
|
"item_id": {"type": "string", "description": "The Jellyfin item ID to play"}
|
|
},
|
|
"required": ["session_id", "item_id"]
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_get_resume",
|
|
"description": "Get in-progress movies and TV series episodes that user mop is currently watching, with timestamps and progress percentage.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
},
|
|
{
|
|
"name": "jelly_get_latest",
|
|
"description": "Get the latest movies or series episodes newly added to the Jellyfin library.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"limit": {"type": "integer", "description": "Maximum number of items to return", "default": 10}
|
|
}
|
|
}
|
|
}
|
|
]
|
|
|
|
|
|
def run_mcp_server():
|
|
client = JellyClient()
|
|
|
|
while True:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
break
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
try:
|
|
req = json.loads(line)
|
|
except Exception as e:
|
|
sys.stderr.write(f"Invalid JSON received: {e}\n")
|
|
continue
|
|
|
|
msg_id = req.get("id")
|
|
method = req.get("method")
|
|
params = req.get("params", {})
|
|
|
|
if method == "initialize":
|
|
resp = {
|
|
"jsonrpc": "2.0",
|
|
"id": msg_id,
|
|
"result": {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": {"name": "jelly-media-mcp", "version": "1.0.0"}
|
|
}
|
|
}
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
elif method == "notifications/initialized":
|
|
pass
|
|
|
|
elif method == "ping":
|
|
resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}}
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
elif method == "tools/list":
|
|
resp = {
|
|
"jsonrpc": "2.0",
|
|
"id": msg_id,
|
|
"result": {"tools": MCP_TOOLS}
|
|
}
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
elif method == "tools/call":
|
|
tool_name = params.get("name")
|
|
args = params.get("arguments", {})
|
|
try:
|
|
if tool_name == "jelly_search":
|
|
data = client.search_media(args.get("query", ""), args.get("media_type"))
|
|
elif tool_name == "jelly_discover":
|
|
data = client.discover_media(
|
|
genre=args.get("genre"),
|
|
min_rating=args.get("min_rating"),
|
|
year=args.get("year"),
|
|
sort_by=args.get("sort_by", "popularity.desc"),
|
|
only_available=args.get("only_available", False),
|
|
media_type=args.get("media_type", "movie")
|
|
)
|
|
elif tool_name == "jelly_get_playback":
|
|
data = client.get_playback_info(args.get("item_id", ""))
|
|
elif tool_name == "jelly_request":
|
|
data = client.request_media(
|
|
media_id=args.get("media_id"),
|
|
media_type=args.get("media_type", "movie"),
|
|
seasons=args.get("seasons")
|
|
)
|
|
elif tool_name == "jelly_list_sessions":
|
|
data = client.list_sessions()
|
|
elif tool_name == "jelly_remote_play":
|
|
data = client.remote_play(args.get("session_id", ""), args.get("item_id", ""))
|
|
elif tool_name == "jelly_get_resume":
|
|
data = client.get_resume()
|
|
elif tool_name == "jelly_get_latest":
|
|
data = client.get_latest(args.get("limit", 10))
|
|
else:
|
|
raise ValueError(f"Unknown tool: {tool_name}")
|
|
|
|
text_result = json.dumps(data, indent=2, ensure_ascii=False)
|
|
resp = {
|
|
"jsonrpc": "2.0",
|
|
"id": msg_id,
|
|
"result": {
|
|
"content": [{"type": "text", "text": text_result}]
|
|
}
|
|
}
|
|
except Exception as e:
|
|
resp = {
|
|
"jsonrpc": "2.0",
|
|
"id": msg_id,
|
|
"result": {
|
|
"isError": True,
|
|
"content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}]
|
|
}
|
|
}
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
else:
|
|
if msg_id is not None:
|
|
resp = {
|
|
"jsonrpc": "2.0",
|
|
"id": msg_id,
|
|
"error": {"code": -32601, "message": f"Method not found: {method}"}
|
|
}
|
|
sys.stdout.write(json.dumps(resp) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def run_cli():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="Jellyfin & Jellyseerr Media Tool")
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
|
|
# Search
|
|
sp_search = subparsers.add_parser("search", help="Search media")
|
|
sp_search.add_argument("query", help="Search query")
|
|
sp_search.add_argument("--type", choices=["all", "movie", "tv"], default="all")
|
|
|
|
# Discover
|
|
sp_disc = subparsers.add_parser("discover", help="Discover movies/shows")
|
|
sp_disc.add_argument("--genre", help="Genre (action, comedy, etc.)")
|
|
sp_disc.add_argument("--min-rating", type=float, help="Minimum TMDB rating")
|
|
sp_disc.add_argument("--year", type=int, help="Year")
|
|
sp_disc.add_argument("--sort-by", default="popularity.desc")
|
|
sp_disc.add_argument("--only-available", action="store_true")
|
|
sp_disc.add_argument("--type", choices=["movie", "tv"], default="movie")
|
|
|
|
# Playback
|
|
sp_play = subparsers.add_parser("playback", help="Get playback links for item ID")
|
|
sp_play.add_argument("item_id", help="Jellyfin Item ID")
|
|
|
|
# Request
|
|
sp_req = subparsers.add_parser("request", help="Request media")
|
|
sp_req.add_argument("media_id", type=int, help="TMDB ID")
|
|
sp_req.add_argument("--type", choices=["movie", "tv"], default="movie")
|
|
sp_req.add_argument("--seasons", type=int, nargs="*", help="Seasons for TV")
|
|
|
|
# Sessions
|
|
subparsers.add_parser("sessions", help="List active Jellyfin sessions")
|
|
|
|
# Remote Play
|
|
sp_rem = subparsers.add_parser("remote-play", help="Trigger remote play")
|
|
sp_rem.add_argument("session_id", help="Session ID")
|
|
sp_rem.add_argument("item_id", help="Item ID")
|
|
|
|
# Resume
|
|
subparsers.add_parser("resume", help="Get in-progress items")
|
|
|
|
# Latest
|
|
sp_latest = subparsers.add_parser("latest", help="Get latest items")
|
|
sp_latest.add_argument("--limit", type=int, default=10)
|
|
|
|
# Test
|
|
subparsers.add_parser("test", help="Test connectivity to Jellyfin & Jellyseerr")
|
|
|
|
args = parser.parse_args()
|
|
client = JellyClient()
|
|
|
|
if args.command == "search":
|
|
print(json.dumps(client.search_media(args.query, args.type), indent=2, ensure_ascii=False))
|
|
elif args.command == "discover":
|
|
print(json.dumps(client.discover_media(
|
|
genre=args.genre,
|
|
min_rating=args.min_rating,
|
|
year=args.year,
|
|
sort_by=args.sort_by,
|
|
only_available=args.only_available,
|
|
media_type=args.type
|
|
), indent=2, ensure_ascii=False))
|
|
elif args.command == "playback":
|
|
print(json.dumps(client.get_playback_info(args.item_id), indent=2, ensure_ascii=False))
|
|
elif args.command == "request":
|
|
print(json.dumps(client.request_media(args.media_id, args.type, args.seasons), indent=2, ensure_ascii=False))
|
|
elif args.command == "sessions":
|
|
print(json.dumps(client.list_sessions(), indent=2, ensure_ascii=False))
|
|
elif args.command == "remote-play":
|
|
print(json.dumps(client.remote_play(args.session_id, args.item_id), indent=2, ensure_ascii=False))
|
|
elif args.command == "resume":
|
|
print(json.dumps(client.get_resume(), indent=2, ensure_ascii=False))
|
|
elif args.command == "latest":
|
|
print(json.dumps(client.get_latest(args.limit), indent=2, ensure_ascii=False))
|
|
elif args.command == "test":
|
|
print("Testing Jellyfin Auth...")
|
|
jf_ok = client.auth_jellyfin()
|
|
print(" Jellyfin:", "[OK]" if jf_ok else "[FAILED]")
|
|
print("Testing Jellyseerr Auth...")
|
|
js_ok = client.auth_jellyseerr()
|
|
print(" Jellyseerr:", "[OK]" if js_ok else "[FAILED]")
|
|
else:
|
|
parser.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"):
|
|
run_mcp_server()
|
|
else:
|
|
run_cli()
|