# Directory Structure
```
├── .coverage
├── .gitignore
├── config.json.example
├── pyproject.toml
├── radarr_sonarr_mcp
│ ├── __init__.py
│ ├── cli.py
│ ├── server.py
│ └── services
│ ├── __init__.py
│ ├── jellyfin_service.py
│ ├── plex_service.py
│ ├── radarr_service.py
│ └── sonarr_service.py
├── README.md
├── requirements.txt
├── run_tests.py
├── run.py
├── setup.py
├── tests
│ ├── __init__.py
│ └── test_server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# Distribution / packaging
dist/
build/
*.egg-info/
# Virtual environments
venv/
env/
ENV/
# Local development settings
.env
config.json
# Editor settings
.vscode/
.idea/
# Logs
*.log
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Radarr and Sonarr MCP Server
A Python-based Model Context Protocol (MCP) server that provides AI assistants like Claude with access to your Radarr (movies) and Sonarr (TV series) data.
## Overview
This MCP server allows AI assistants to query your movie and TV show collection via Radarr and Sonarr APIs. Built with FastMCP, it implements the standardized protocol for AI context that Claude Desktop and other MCP-compatible clients can use.
## Features
- **Native MCP Implementation**: Built with FastMCP for seamless AI integration
- **Radarr Integration**: Access your movie collection
- **Sonarr Integration**: Access your TV show and episode data
- **Rich Filtering**: Filter by year, watched status, actors, and more
- **Claude Desktop Compatible**: Works seamlessly with Claude's MCP client
- **Easy Setup**: Interactive configuration wizard
- **Well-tested**: Comprehensive test suite for reliability
## Installation
### From Source
1. Clone this repository:
```bash
git clone https://github.com/yourusername/radarr-sonarr-mcp.git
cd radarr-sonarr-mcp-python
```
2. Install the package:
```bash
pip install -e .
```
### Using pip (coming soon)
```bash
pip install radarr-sonarr-mcp
```
## Quick Start
1. Configure the server:
```bash
radarr-sonarr-mcp configure
```
Follow the prompts to enter your Radarr/Sonarr API keys and other settings.
2. Start the server:
```bash
radarr-sonarr-mcp start
```
3. Connect Claude Desktop:
- In Claude Desktop, go to Settings > MCP Servers
- Add a new server with URL: `http://localhost:3000` (or your configured port)
## Configuration
The configuration wizard will guide you through setting up:
- NAS/Server IP address
- Radarr API key and port
- Sonarr API key and port
- MCP server port
You can also manually edit the `config.json` file:
```json
{
"nasConfig": {
"ip": "10.0.0.23",
"port": "7878"
},
"radarrConfig": {
"apiKey": "YOUR_RADARR_API_KEY",
"basePath": "/api/v3",
"port": "7878"
},
"sonarrConfig": {
"apiKey": "YOUR_SONARR_API_KEY",
"basePath": "/api/v3",
"port": "8989"
},
"server": {
"port": 3000
}
}
```
## Available MCP Tools
This server provides the following tools to Claude:
### Movies
- `get_available_movies` - Get a list of movies with optional filters
- `lookup_movie` - Search for a movie by title
- `get_movie_details` - Get detailed information about a specific movie
### Series
- `get_available_series` - Get a list of TV series with optional filters
- `lookup_series` - Search for a TV series by title
- `get_series_details` - Get detailed information about a specific series
- `get_series_episodes` - Get episodes for a specific series
### Resources
The server also provides standard MCP resources:
- `/movies` - Browse all available movies
- `/series` - Browse all available TV series
### Filtering Options
Most tools support various filtering options:
- `year` - Filter by release year
- `watched` - Filter by watched status (true/false)
- `downloaded` - Filter by download status (true/false)
- `watchlist` - Filter by watchlist status (true/false)
- `actors` - Filter by actor/cast name
- `actresses` - Filter by actress name (movies only)
## Example Queries for Claude
Once your MCP server is connected to Claude Desktop, you can ask questions like:
- "What sci-fi movies from 2023 do I have?"
- "Show me TV shows starring Pedro Pascal"
- "Do I have any unwatched episodes of The Mandalorian?"
- "Find movies with Tom Hanks that I haven't watched yet"
- "How many episodes of Stranger Things do I have downloaded?"
## Finding API Keys
### Radarr API Key
1. Open Radarr in your browser
2. Go to Settings > General
3. Find the "API Key" section
4. Copy the API Key
### Sonarr API Key
1. Open Sonarr in your browser
2. Go to Settings > General
3. Find the "API Key" section
4. Copy the API Key
## Command-Line Interface
The package provides a command-line interface:
- `radarr-sonarr-mcp configure` - Run configuration wizard
- `radarr-sonarr-mcp start` - Start the MCP server
- `radarr-sonarr-mcp status` - Show the current configuration
## Development
### Running Tests
To run the test suite:
```bash
# Install development dependencies
pip install -e ".[dev]"
# Run tests
pytest
# Run tests with coverage
pytest --cov=radarr_sonarr_mcp
```
### Local Development
For quick development and testing:
```bash
# Run directly without installation
python run.py
```
## Requirements
- Python 3.7+
- FastMCP
- Requests
- Pydantic
## Notes
- The watched/watchlist status functionality assumes these are tracked using specific mechanisms in Radarr/Sonarr. You may need to adapt this to your specific setup.
- For security reasons, it's recommended to run this server only on your local network.
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
"""Test package for radarr-sonarr-mcp."""
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp>=0.1.0
requests>=2.28.0
pydantic>=2.0.0
pytest>=7.0.0
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/__init__.py:
--------------------------------------------------------------------------------
```python
"""Services for interacting with Radarr and Sonarr APIs."""
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""Radarr and Sonarr MCP Server - Model Context Protocol server for media management."""
__version__ = "1.0.0"
```
--------------------------------------------------------------------------------
/run.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Simple script to run the MCP server directly without installation.
Useful for development.
"""
import sys
import os
# Add the project directory to the path
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
from radarr_sonarr_mcp.server import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/run_tests.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Simple script to run the test suite.
"""
import sys
import unittest
import os
# Add the project directory to the path
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
# Import the test modules
from tests.test_server import TestRadarrSonarrMCPServer
if __name__ == "__main__":
# Run all tests
unittest.main(module=None, argv=['run_tests.py'])
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "radarr-sonarr-mcp"
version = "0.1.0"
description = "MCP Server for Radarr and Sonarr media management"
authors = [
{ name = "Berry", email = "[email protected]" }
]
requires-python = ">=3.10"
readme = "README.md"
classifiers = [
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent"
]
scripts = { radarr-sonarr-mcp = "radarr_sonarr_mcp.cli:main" }
dependencies = [
"mcp>=0.1.0",
"requests>=2.28.0",
"pydantic>=2.0.0"
]
optional-dependencies = { dev = ["pytest>=7.0.0"] }
[build-system]
requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
"""Setup script for the radarr-sonarr-mcp package."""
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="radarr-sonarr-mcp",
version="1.0.0",
packages=find_packages(),
install_requires=[
"fastmcp>=0.4.1",
"requests>=2.28.0",
"pydantic>=2.0.0",
],
entry_points={
"console_scripts": [
"radarr-sonarr-mcp=radarr_sonarr_mcp.cli:main",
],
},
author="Berry",
author_email="",
description="Model Context Protocol server for Radarr and Sonarr",
long_description=long_description,
long_description_content_type="text/markdown",
url="",
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
python_requires=">=3.7",
extras_require={
"dev": [
"pytest>=7.0.0",
"pytest-cov>=3.0.0",
],
},
)
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/jellyfin_service.py:
--------------------------------------------------------------------------------
```python
import requests
from typing import Any, Dict, List
class JellyfinService:
"""
Service for interacting with the Jellyfin API.
This service searches for a series by title and retrieves its episodes to check the watch status.
"""
def __init__(self, config: Dict[str, Any]):
self.base_url = config.get("baseUrl") # e.g., "http://10.0.0.23:5055"
self.api_key = config.get("apiKey")
self.user_id = config.get("userId") # The user ID to check watch status for
def search_series(self, title: str) -> List[Dict[str, Any]]:
"""
Search for a series in Jellyfin by title.
"""
url = f"{self.base_url}/Users/{self.user_id}/Items"
params = {
"IncludeItemTypes": "Series",
"SearchTerm": title,
"api_key": self.api_key
}
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json().get("Items", [])
def get_episodes_for_series(self, series_id: str) -> List[Dict[str, Any]]:
"""
Retrieve episodes for a given series ID from Jellyfin.
"""
url = f"{self.base_url}/Users/{self.user_id}/Items"
params = {
"ParentId": series_id,
"IncludeItemTypes": "Episode",
"api_key": self.api_key
}
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json().get("Items", [])
def is_series_watched(self, series_title: str) -> bool:
"""
Determine if the series is watched.
A series is considered watched if all episodes have a PlayCount > 0.
"""
items = self.search_series(series_title)
if not items:
return False
series_item = items[0] # take the first match
series_id = series_item.get("Id")
episodes = self.get_episodes_for_series(series_id)
if not episodes:
return False
# Consider the series watched if every episode has a PlayCount > 0
return all(ep.get("UserData", {}).get("PlayCount", 0) > 0 for ep in episodes)
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/plex_service.py:
--------------------------------------------------------------------------------
```python
import requests
from typing import Any, Dict, List
class PlexService:
"""
Service for interacting with the Plex API.
Note: Plex’s API typically returns XML, but here we assume a JSON endpoint
(or you can use an XML parser). This is a simplified example.
"""
def __init__(self, config: Dict[str, Any]):
self.base_url = config.get("baseUrl") # e.g., "http://10.0.0.23:32400"
self.token = config.get("token")
def search_series(self, title: str) -> List[Dict[str, Any]]:
url = f"{self.base_url}/library/search"
params = {
"query": title,
"type": 4 # type 4 usually indicates a TV series
}
headers = {"X-Plex-Token": self.token}
response = requests.get(url, params=params, headers=headers, timeout=30)
# In a real implementation, parse XML; here we assume JSON for simplicity.
try:
return response.json().get("MediaContainer", {}).get("Metadata", [])
except Exception:
return []
def get_episodes_for_series(self, rating_key: str) -> List[Dict[str, Any]]:
url = f"{self.base_url}/library/metadata/{rating_key}/children"
headers = {"X-Plex-Token": self.token}
response = requests.get(url, headers=headers, timeout=30)
try:
return response.json().get("MediaContainer", {}).get("Metadata", [])
except Exception:
return []
def is_series_watched(self, series_title: str) -> bool:
items = self.search_series(series_title)
if not items:
return False
# Assume the first matching series is our target.
series_item = items[0]
rating_key = series_item.get("ratingKey")
episodes = self.get_episodes_for_series(rating_key)
if not episodes:
return False
# Consider the series watched if every episode's UserData indicates it was played.
return all(ep.get("UserData", {}).get("viewCount", 0) > 0 for ep in episodes)
def search_movie(self, title: str) -> List[Dict[str, Any]]:
url = f"{self.base_url}/library/search"
params = {
"query": title,
"type": 2 # type 2 for movies
}
headers = {"X-Plex-Token": self.token}
response = requests.get(url, params=params, headers=headers, timeout=30)
try:
return response.json().get("MediaContainer", {}).get("Metadata", [])
except Exception:
return []
def is_movie_watched(self, movie_title: str) -> bool:
items = self.search_movie(movie_title)
if not items:
return False
movie_item = items[0]
return movie_item.get("UserData", {}).get("viewCount", 0) > 0
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/radarr_service.py:
--------------------------------------------------------------------------------
```python
"""Service for interacting with Radarr API."""
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import requests
from ..config import RadarrConfig
@dataclass
class Movie:
"""Movie data class."""
id: int
title: str
year: int
overview: str
has_file: bool
status: str
tags: List[int] = None
genres: List[str] = None
data: Dict[str, Any] = None
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Movie':
"""Create a Movie object from a dictionary."""
return cls(
id=data['id'],
title=data['title'],
year=data.get('year', 0),
overview=data.get('overview', ''),
has_file=data.get('hasFile', False),
status=data.get('status', ''),
tags=data.get('tags', []),
genres=data.get('genres', []),
data=data
)
class RadarrService:
"""Service for interacting with Radarr API."""
def __init__(self, config: RadarrConfig):
"""Initialize the Radarr service with configuration."""
self.config = config
def get_all_movies(self) -> List[Movie]:
"""Fetch all movies from Radarr."""
try:
response = requests.get(
f"{self.config.base_url}/movie",
params={"apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
movies = []
for movie_data in response.json():
movies.append(Movie.from_dict(movie_data))
return movies
except requests.RequestException as e:
import logging
logging.error(f"Error fetching movies from Radarr: {e}")
raise Exception(f"Failed to fetch movies from Radarr: {e}")
def lookup_movie(self, term: str) -> List[Movie]:
"""Look up movies by search term."""
try:
response = requests.get(
f"{self.config.base_url}/movie/lookup",
params={"term": term, "apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
movies = []
for movie_data in response.json():
movies.append(Movie.from_dict(movie_data))
return movies
except requests.RequestException as e:
import logging
logging.error(f"Error looking up movie in Radarr: {e}")
raise Exception(f"Failed to lookup movie in Radarr: {e}")
def get_movie_file(self, movie_id: int) -> Dict[str, Any]:
"""Get the file information for a movie."""
try:
response = requests.get(
f"{self.config.base_url}/moviefile",
params={"movieId": movie_id, "apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
import logging
logging.error(f"Error fetching movie file for ID {movie_id}: {e}")
raise Exception(f"Failed to fetch movie file: {e}")
def is_movie_watched(self, movie: Movie) -> bool:
"""Check if a movie is watched based on tags."""
# This is an assumption - actual implementation may vary based on how
# watched status is tracked in your Radarr setup
return movie.data.get('movieFile', {}).get('mediaInfo', {}).get('watched', False)
def is_movie_in_watchlist(self, movie: Movie) -> bool:
"""Check if a movie is in the watchlist based on tags."""
# This is an assumption - implementation may vary
# Assuming 'watchlist' tag with ID 1 (adjust as needed)
return 1 in (movie.tags or [])
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/cli.py:
--------------------------------------------------------------------------------
```python
"""Command-line interface for the Radarr/Sonarr MCP server."""
import argparse
import logging
from .config import Config, NasConfig, RadarrConfig, SonarrConfig, ServerConfig, load_config, save_config
from .server import create_server
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def configure():
"""Run the configuration wizard."""
logging.info("==== Radarr/Sonarr MCP Server Configuration Wizard ====")
# Try to load existing config
config = None
try:
config = load_config()
logging.info("Loaded existing configuration. Press Enter to keep current values.")
except Exception:
# No existing config or error loading
pass
# NAS configuration
nas_ip = input(f"NAS/Server IP address [{config.nas_config.ip if config else '10.0.0.23'}]: ")
nas_ip = nas_ip or (config.nas_config.ip if config else '10.0.0.23')
nas_port = input(f"Default port [{config.nas_config.port if config else '7878'}]: ")
nas_port = nas_port or (config.nas_config.port if config else '7878')
# Radarr configuration
radarr_api_key = input(f"Radarr API key [{config.radarr_config.api_key if config else ''}]: ")
radarr_api_key = radarr_api_key or (config.radarr_config.api_key if config else '')
if not radarr_api_key:
logging.warning("Warning: Radarr API key is required for movie functionality!")
radarr_port = input(f"Radarr port [{config.radarr_config.port if config else '7878'}]: ")
radarr_port = radarr_port or (config.radarr_config.port if config else '7878')
radarr_base_path = input(f"Radarr API base path [{config.radarr_config.base_path if config else '/api/v3'}]: ")
radarr_base_path = radarr_base_path or (config.radarr_config.base_path if config else '/api/v3')
# Sonarr configuration
sonarr_api_key = input(f"Sonarr API key [{config.sonarr_config.api_key if config else ''}]: ")
sonarr_api_key = sonarr_api_key or (config.sonarr_config.api_key if config else '')
if not sonarr_api_key:
logging.warning("Warning: Sonarr API key is required for TV show functionality!")
sonarr_port = input(f"Sonarr port [{config.sonarr_config.port if config else '8989'}]: ")
sonarr_port = sonarr_port or (config.sonarr_config.port if config else '8989')
sonarr_base_path = input(f"Sonarr API base path [{config.sonarr_config.base_path if config else '/api/v3'}]: ")
sonarr_base_path = sonarr_base_path or (config.sonarr_config.base_path if config else '/api/v3')
# Server configuration
server_port = input(f"MCP server port [{config.server_config.port if config else '3000'}]: ")
if server_port:
try:
server_port = int(server_port)
except ValueError:
logging.warning("Invalid port number, using default.")
server_port = config.server_config.port if config else 3000
else:
server_port = config.server_config.port if config else 3000
# Create new config
new_config = Config(
nas_config=NasConfig(
ip=nas_ip,
port=nas_port
),
radarr_config=RadarrConfig(
api_key=radarr_api_key,
base_path=radarr_base_path,
port=radarr_port
),
sonarr_config=SonarrConfig(
api_key=sonarr_api_key,
base_path=sonarr_base_path,
port=sonarr_port
),
server_config=ServerConfig(
port=server_port
)
)
# Save config
save_config(new_config)
logging.info("Configuration saved successfully!")
logging.info(f"To start the server, run: radarr-sonarr-mcp start")
return new_config
def start(config_path=None):
"""Start the MCP server."""
server = create_server(config_path)
server.start()
def show_status():
"""Show the current status of the server."""
try:
config = load_config()
logging.info("==== Radarr/Sonarr MCP Server Status ====")
logging.info(f"NAS IP: {config.nas_config.ip}")
logging.info(f"Radarr Port: {config.radarr_config.port or config.nas_config.port}")
logging.info(f"Sonarr Port: {config.sonarr_config.port or config.nas_config.port}")
logging.info(f"MCP Server Port: {config.server_config.port}")
logging.info(f"MCP Endpoint URL: http://localhost:{config.server_config.port}")
logging.info(f"Server is configured. Use 'radarr-sonarr-mcp start' to run the server.")
except Exception as e:
logging.error(f"Server is not configured: {e}")
logging.info("Run 'radarr-sonarr-mcp configure' to set up the server.")
def main():
"""Main CLI entry point."""
parser = argparse.ArgumentParser(description="Radarr/Sonarr MCP Server")
subparsers = parser.add_subparsers(dest="command", help="Command to execute")
# Configure command
configure_parser = subparsers.add_parser("configure", help="Configure the MCP server")
# Start command
start_parser = subparsers.add_parser("start", help="Start the MCP server")
start_parser.add_argument("--config", help="Path to config.json file")
# Status command
status_parser = subparsers.add_parser("status", help="Show the server status")
args = parser.parse_args()
if args.command == "configure":
configure()
elif args.command == "start":
start(args.config)
elif args.command == "status":
show_status()
else:
parser.print_help()
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/sonarr_service.py:
--------------------------------------------------------------------------------
```python
"""Service for interacting with Sonarr API."""
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import requests
from ..config import SonarrConfig
@dataclass
class Statistics:
"""Statistics for a TV series."""
episode_file_count: int
episode_count: int
total_episode_count: int
size_on_disk: int
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Statistics':
"""Create a Statistics object from a dictionary."""
return cls(
episode_file_count=data.get('episodeFileCount', 0),
episode_count=data.get('episodeCount', 0),
total_episode_count=data.get('totalEpisodeCount', 0),
size_on_disk=data.get('sizeOnDisk', 0)
)
@dataclass
class Series:
"""TV Series data class."""
id: int
title: str
year: Optional[int]
overview: str
status: str
network: str
tags: List[int]
genres: List[str]
statistics: Optional[Statistics]
data: Dict[str, Any] # Store original data for reference
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Series':
"""Create a Series object from a dictionary."""
statistics = None
if 'statistics' in data:
statistics = Statistics.from_dict(data['statistics'])
return cls(
id=data['id'],
title=data['title'],
year=data.get('year'),
overview=data.get('overview', ''),
status=data.get('status', ''),
network=data.get('network', ''),
tags=data.get('tags', []),
genres=data.get('genres', []),
statistics=statistics,
data=data
)
@dataclass
class Episode:
"""TV Episode data class."""
id: int
series_id: int
episode_file_id: Optional[int]
season_number: int
episode_number: int
title: str
air_date: Optional[str]
has_file: bool
monitored: bool
overview: str
data: Dict[str, Any] # Store original data for reference
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Episode':
"""Create an Episode object from a dictionary."""
return cls(
id=data['id'],
series_id=data['seriesId'],
episode_file_id=data.get('episodeFileId'),
season_number=data['seasonNumber'],
episode_number=data['episodeNumber'],
title=data.get('title', ''),
air_date=data.get('airDate'),
has_file=data.get('hasFile', False),
monitored=data.get('monitored', True),
overview=data.get('overview', ''),
data=data
)
class SonarrService:
"""Service for interacting with Sonarr API."""
def __init__(self, config: SonarrConfig):
"""Initialize the Sonarr service with configuration."""
self.config = config
def get_all_series(self) -> List[Series]:
"""Fetch all TV series from Sonarr."""
try:
response = requests.get(
f"{self.config.base_url}/series",
params={"apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
series_list = []
for series_data in response.json():
series_list.append(Series.from_dict(series_data))
return series_list
except requests.RequestException as e:
import logging
logging.error(f"Error fetching series from Sonarr: {e}")
raise Exception(f"Failed to fetch series from Sonarr: {e}")
def lookup_series(self, term: str) -> List[Series]:
"""Look up TV series by search term."""
try:
response = requests.get(
f"{self.config.base_url}/series/lookup",
params={"term": term, "apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
series_list = []
for series_data in response.json():
series_list.append(Series.from_dict(series_data))
return series_list
except requests.RequestException as e:
import logging
logging.error(f"Error looking up series in Sonarr: {e}")
raise Exception(f"Failed to lookup series from Sonarr: {e}")
def get_episodes(self, series_id: int) -> List[Episode]:
"""Fetch episodes for a TV series."""
try:
response = requests.get(
f"{self.config.base_url}/episode",
params={"seriesId": series_id, "apikey": self.config.api_key},
timeout=30
)
response.raise_for_status()
episodes = []
for episode_data in response.json():
episodes.append(Episode.from_dict(episode_data))
return episodes
except requests.RequestException as e:
import logging
logging.error(f"Error fetching episodes for series ID {series_id}: {e}")
raise Exception(f"Failed to fetch episodes: {e}")
def is_series_watched(self, series: Series) -> bool:
"""Check if a series is watched based on tags."""
# This is an assumption - actual implementation may vary based on how
# watched status is tracked in your Sonarr setup
if not series.statistics:
return False
# Consider series watched if all episodes are downloaded
return (series.statistics.episode_file_count >=
series.statistics.episode_count)
def is_series_in_watchlist(self, series: Series) -> bool:
"""Check if a series is in the watchlist based on tags."""
# This is an assumption - implementation may vary
# Assuming 'watchlist' tag with ID 1 (adjust as needed)
return 1 in (series.tags or [])
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
"""Tests for the MCP server implementation."""
import unittest
import json
import os
import sys
import tempfile
from unittest.mock import patch, MagicMock
# Add parent directory to path to import the package
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from radarr_sonarr_mcp.config import Config, NasConfig, RadarrConfig, SonarrConfig, ServerConfig
from radarr_sonarr_mcp.server import RadarrSonarrMCPServer, create_server
from radarr_sonarr_mcp.services.radarr_service import Movie
from radarr_sonarr_mcp.services.sonarr_service import Series, Statistics
class TestRadarrSonarrMCPServer(unittest.TestCase):
"""Test suite for the RadarrSonarrMCPServer class."""
def setUp(self):
"""Set up test environment."""
# Create a temporary config file
self.temp_file = tempfile.NamedTemporaryFile(delete=False)
config_data = {
"nasConfig": {
"ip": "127.0.0.1",
"port": "7878"
},
"radarrConfig": {
"apiKey": "test_radarr_api_key",
"basePath": "/api/v3",
"port": "7878"
},
"sonarrConfig": {
"apiKey": "test_sonarr_api_key",
"basePath": "/api/v3",
"port": "8989"
},
"server": {
"port": 5000
}
}
with open(self.temp_file.name, 'w') as f:
json.dump(config_data, f)
# Create sample movie and series data
self.sample_movies = [
Movie(
id=1,
title="Test Movie 1",
year=2022,
overview="A test movie",
has_file=True,
status="downloaded",
genres=["Action", "Comedy"],
tags=[1, 2],
data={
"id": 1,
"title": "Test Movie 1",
"year": 2022,
"overview": "A test movie",
"hasFile": True,
"status": "downloaded",
"genres": ["Action", "Comedy"],
"tags": [1, 2],
"credits": {
"cast": [
{"name": "Actor One", "character": "Character One"},
{"name": "Actress One", "character": "Character Two"}
]
}
}
),
Movie(
id=2,
title="Test Movie 2",
year=2023,
overview="Another test movie",
has_file=False,
status="wanted",
genres=["Drama", "Thriller"],
tags=[2],
data={
"id": 2,
"title": "Test Movie 2",
"year": 2023,
"overview": "Another test movie",
"hasFile": False,
"status": "wanted",
"genres": ["Drama", "Thriller"],
"tags": [2],
"credits": {
"cast": [
{"name": "Actor Two", "character": "Character Three"}
]
}
}
)
]
self.sample_series = [
Series(
id=1,
title="Test Series 1",
year=2022,
overview="A test series",
status="continuing",
network="Test Network",
tags=[1],
genres=["Comedy"],
statistics=Statistics.from_dict({
"episodeFileCount": 10,
"episodeCount": 10,
"totalEpisodeCount": 20,
"sizeOnDisk": 10000
}),
data={
"id": 1,
"title": "Test Series 1",
"year": 2022,
"overview": "A test series",
"status": "continuing",
"network": "Test Network",
"tags": [1],
"genres": ["Comedy"],
"statistics": {
"episodeFileCount": 10,
"episodeCount": 10,
"totalEpisodeCount": 20,
"sizeOnDisk": 10000
},
"credits": {
"cast": [
{"name": "Actor Three", "character": "Character Four"}
]
}
}
)
]
def tearDown(self):
"""Clean up after tests."""
self.temp_file.close()
os.unlink(self.temp_file.name)
@patch('radarr_sonarr_mcp.server.FastMCP')
def test_server_initialization(self, mock_fastmcp):
"""Test server initialization with config file."""
server = create_server(self.temp_file.name)
self.assertEqual(server.config.radarr_config.api_key, "test_radarr_api_key")
self.assertEqual(server.config.sonarr_config.api_key, "test_sonarr_api_key")
self.assertEqual(server.config.server_config.port, 5000)
# Check that FastMCP was initialized correctly
mock_fastmcp.assert_called_once()
self.assertEqual(mock_fastmcp.call_args[1]['name'], "radarr-sonarr-mcp-server")
@patch('radarr_sonarr_mcp.server.RadarrService')
@patch('radarr_sonarr_mcp.server.SonarrService')
@patch('radarr_sonarr_mcp.server.FastMCP')
def test_get_available_movies(self, mock_fastmcp, mock_sonarr_service, mock_radarr_service):
"""Test the get_available_movies tool."""
# Setup mocks
mock_radarr_instance = mock_radarr_service.return_value
mock_radarr_instance.get_all_movies.return_value = self.sample_movies
mock_radarr_instance.is_movie_watched.return_value = True
mock_radarr_instance.is_movie_in_watchlist.return_value = False
mock_server = mock_fastmcp.return_value
# Create server and register tools
server = create_server(self.temp_file.name)
# Extract the registered tool function
tool_decorator = mock_server.tool.return_value
get_movies_func = None
for call in tool_decorator.call_args_list:
# The decorated function is passed to the decorator
if call.args and call.args[0].__name__ == 'get_available_movies':
get_movies_func = call.args[0]
break
self.assertIsNotNone(get_movies_func, "get_available_movies tool not registered")
# Test the tool function
result = get_movies_func(year=2022)
result_data = json.loads(result)
# Check results
self.assertEqual(result_data['count'], 1)
self.assertEqual(result_data['movies'][0]['title'], "Test Movie 1")
self.assertEqual(result_data['movies'][0]['year'], 2022)
@patch('radarr_sonarr_mcp.server.SonarrService')
@patch('radarr_sonarr_mcp.server.RadarrService')
@patch('radarr_sonarr_mcp.server.FastMCP')
def test_get_available_series(self, mock_fastmcp, mock_radarr_service, mock_sonarr_service):
"""Test the get_available_series tool."""
# Setup mocks
mock_sonarr_instance = mock_sonarr_service.return_value
mock_sonarr_instance.get_all_series.return_value = self.sample_series
mock_sonarr_instance.is_series_watched.return_value = True
mock_sonarr_instance.is_series_in_watchlist.return_value = False
mock_server = mock_fastmcp.return_value
# Create server and register tools
server = create_server(self.temp_file.name)
# Extract the registered tool function
tool_decorator = mock_server.tool.return_value
get_series_func = None
for call in tool_decorator.call_args_list:
# The decorated function is passed to the decorator
if call.args and call.args[0].__name__ == 'get_available_series':
get_series_func = call.args[0]
break
self.assertIsNotNone(get_series_func, "get_available_series tool not registered")
# Test the tool function
result = get_series_func()
result_data = json.loads(result)
# Check results
self.assertEqual(result_data['count'], 1)
self.assertEqual(result_data['series'][0]['title'], "Test Series 1")
self.assertEqual(result_data['series'][0]['year'], 2022)
@patch('radarr_sonarr_mcp.server.RadarrService')
@patch('radarr_sonarr_mcp.server.SonarrService')
@patch('radarr_sonarr_mcp.server.FastMCP')
def test_server_resources(self, mock_fastmcp, mock_sonarr_service, mock_radarr_service):
"""Test registered resources."""
# Setup mocks
mock_radarr_instance = mock_radarr_service.return_value
mock_radarr_instance.get_all_movies.return_value = self.sample_movies
mock_sonarr_instance = mock_sonarr_service.return_value
mock_sonarr_instance.get_all_series.return_value = self.sample_series
mock_server = mock_fastmcp.return_value
# Create server and register resources
server = create_server(self.temp_file.name)
# Mock get_resource_handler to return MagicMock objects
mock_server_instance = mock_fastmcp.return_value
mock_server_instance.get_resource_handler.side_effect = lambda path: MagicMock(return_value={
'/movies': {"count": 2, "movies": self.sample_movies},
'/series': {"count": 1, "series": self.sample_series}
}.get(path))
# Test movies resource
movies_resource = server.server.get_resource_handler('/movies')
self.assertIsNotNone(movies_resource, "Movies resource not registered")
result_movies = movies_resource()
self.assertEqual(result_movies['count'], 2)
self.assertEqual(len(result_movies['movies']), 2)
# Test series resource
series_resource = server.server.get_resource_handler('/series')
self.assertIsNotNone(series_resource, "Series resource not registered")
result_series = series_resource()
self.assertEqual(result_series['count'], 1)
self.assertEqual(len(result_series['series']), 1)
if __name__ == '__main__':
unittest.main()
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""Main MCP server implementation for Radarr/Sonarr."""
import os
import json
import sys
import logging
from typing import Optional
import argparse
from fastmcp import FastMCP
import requests
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
# Configuration handling
# -----------------------------------------------------------------------------
def load_config():
"""Load configuration from environment variables or config file."""
if os.environ.get('RADARR_API_KEY') or os.environ.get('SONARR_API_KEY'):
logger.info("Loading configuration from environment variables...")
nas_ip = os.environ.get('NAS_IP', '10.0.0.23')
return {
"nasConfig": {
"ip": nas_ip,
"port": os.environ.get('RADARR_PORT', '7878')
},
"radarrConfig": {
"apiKey": os.environ.get('RADARR_API_KEY', ''),
"basePath": os.environ.get('RADARR_BASE_PATH', '/api/v3'),
"port": os.environ.get('RADARR_PORT', '7878')
},
"sonarrConfig": {
"apiKey": os.environ.get('SONARR_API_KEY', ''),
"basePath": os.environ.get('SONARR_BASE_PATH', '/api/v3'),
"port": os.environ.get('SONARR_PORT', '8989')
},
# Optionally, include Jellyfin and Plex configuration if set in env
"jellyfinConfig": {
"baseUrl": os.environ.get('JELLYFIN_BASE_URL', ''), # e.g., "http://10.0.0.23:5055"
"apiKey": os.environ.get('JELLYFIN_API_KEY', ''),
"userId": os.environ.get('JELLYFIN_USER_ID', '')
},
"plexConfig": {
"baseUrl": os.environ.get('PLEX_BASE_URL', ''), # e.g., "http://10.0.0.23:32400"
"token": os.environ.get('PLEX_TOKEN', '')
},
"server": {
"port": int(os.environ.get('MCP_SERVER_PORT', '3000'))
}
}
else:
config_path = 'config.json'
try:
with open(config_path, 'r') as f:
return json.load(f)
except Exception as e:
logger.error(f"Error loading config: {e}")
logger.info("Using default configuration")
return {
"nasConfig": {"ip": "10.0.0.23", "port": "7878"},
"radarrConfig": {"apiKey": "", "basePath": "/api/v3", "port": "7878"},
"sonarrConfig": {"apiKey": "", "basePath": "/api/v3", "port": "8989"},
"server": {"port": 3000}
}
# -----------------------------------------------------------------------------
# API Service functions
# -----------------------------------------------------------------------------
def get_radarr_url(config):
nas_ip = config["nasConfig"]["ip"]
port = config["radarrConfig"]["port"]
base_path = config["radarrConfig"]["basePath"]
return f"http://{nas_ip}:{port}{base_path}"
def get_sonarr_url(config):
nas_ip = config["nasConfig"]["ip"]
port = config["sonarrConfig"]["port"]
base_path = config["sonarrConfig"]["basePath"]
return f"http://{nas_ip}:{port}{base_path}"
def make_radarr_request(config, endpoint, params=None):
api_key = config["radarrConfig"]["apiKey"]
base_url = get_radarr_url(config)
url = f"{base_url}/{endpoint}"
if params is None:
params = {}
params['apikey'] = api_key
try:
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error making request to {url}: {e}")
return []
def make_sonarr_request(config, endpoint, params=None):
api_key = config["sonarrConfig"]["apiKey"]
base_url = get_sonarr_url(config)
url = f"{base_url}/{endpoint}"
if params is None:
params = {}
params['apikey'] = api_key
try:
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error making request to {url}: {e}")
return []
def get_all_series(config):
from radarr_sonarr_mcp.services.sonarr_service import SonarrService
service = SonarrService(config["sonarrConfig"])
return service.get_all_series()
# -----------------------------------------------------------------------------
# Helper function to check watched status from multiple sources
# -----------------------------------------------------------------------------
def is_watched_series(title: str, fallback: bool, config: dict, sonarr_service) -> bool:
"""
Check if a series is watched using available media services.
Returns True if any service reports the series as watched.
"""
statuses = []
if config.get("jellyfinConfig", {}).get("baseUrl"):
from radarr_sonarr_mcp.services.jellyfin_service import JellyfinService
jellyfin = JellyfinService(config["jellyfinConfig"])
try:
statuses.append(jellyfin.is_series_watched(title))
except Exception as e:
logger.error(f"Jellyfin check failed for {title}: {e}")
if config.get("plexConfig", {}).get("baseUrl"):
from radarr_sonarr_mcp.services.plex_service import PlexService
plex = PlexService(config["plexConfig"])
try:
statuses.append(plex.is_series_watched(title))
except Exception as e:
logger.error(f"Plex check failed for {title}: {e}")
if statuses:
return any(statuses)
# Fallback to Sonarr's own logic if no external services are configured.
return sonarr_service.is_series_watched(title)
def is_watched_movie(title: str, config: dict) -> bool:
"""
Check if a movie is watched using available media services.
Returns True if any service reports the movie as watched.
"""
statuses = []
if config.get("jellyfinConfig", {}).get("baseUrl"):
from radarr_sonarr_mcp.services.jellyfin_service import JellyfinService
jellyfin = JellyfinService(config["jellyfinConfig"])
try:
# For movies, you could implement a similar method in JellyfinService.
statuses.append(jellyfin.is_movie_watched(title))
except Exception as e:
logger.error(f"Jellyfin movie check failed for {title}: {e}")
if config.get("plexConfig", {}).get("baseUrl"):
from radarr_sonarr_mcp.services.plex_service import PlexService
plex = PlexService(config["plexConfig"])
try:
statuses.append(plex.is_movie_watched(title))
except Exception as e:
logger.error(f"Plex movie check failed for {title}: {e}")
# If no external services configured, default to unwatched.
return any(statuses)
# -----------------------------------------------------------------------------
# MCP Server implementation
# -----------------------------------------------------------------------------
from radarr_sonarr_mcp.services.sonarr_service import SonarrService
class RadarrSonarrMCP:
"""MCP Server for Radarr and Sonarr."""
def __init__(self):
self.config = load_config()
self.server = FastMCP(
name="radarr-sonarr-mcp-server",
description="MCP Server for Radarr and Sonarr media management"
)
self.sonarr_service = SonarrService(self.config["sonarrConfig"])
self._register_tools()
self._register_resources()
# Optionally, register prompts.
def _register_tools(self):
@self.server.tool()
def get_available_series(year: Optional[int] = None,
downloaded: Optional[bool] = None,
watched: Optional[bool] = None,
actors: Optional[str] = None) -> dict:
"""
Get a list of available TV series with optional filters.
Watched status is determined using Plex and/or Jellyfin; if either reports watched, the series is considered watched.
"""
all_series = get_all_series(self.config) # List of Series objects
filtered_series = all_series
if year is not None:
filtered_series = [s for s in filtered_series if s.year == year]
if downloaded is not None:
filtered_series = [
s for s in filtered_series
if (s.statistics and s.statistics.episode_file_count > 0) == downloaded
]
if watched is not None:
if watched:
filtered_series = [
s for s in filtered_series
if is_watched_series(s.title, False, self.config, self.sonarr_service)
]
else:
filtered_series = [
s for s in filtered_series
if not is_watched_series(s.title, False, self.config, self.sonarr_service)
]
if actors:
filtered_series = [
s for s in filtered_series
if s.data.get("credits") and any(
actors.lower() in cast.get("name", "").lower()
for cast in s.data.get("credits", {}).get("cast", [])
)
]
return {
"count": len(filtered_series),
"series": [
{
"id": s.id,
"title": s.title,
"year": s.year,
"overview": s.overview,
"status": s.status,
"network": s.network,
"genres": s.genres,
"watched": is_watched_series(s.title, False, self.config, self.sonarr_service)
}
for s in filtered_series
]
}
@self.server.tool()
def lookup_series(term: str) -> dict:
service = SonarrService(self.config["sonarrConfig"])
results = service.lookup_series(term)
return {
"count": len(results),
"series": [
{
"id": s.id,
"title": s.title,
"year": s.year,
"overview": s.overview
}
for s in results
]
}
# Similarly, for movies you can define a tool:
@self.server.tool()
def get_available_movies(year: Optional[int] = None,
downloaded: Optional[bool] = None,
watched: Optional[bool] = None,
actors: Optional[str] = None) -> dict:
"""
Get a list of all available movies with optional filters.
Watched status is determined using Plex and/or Jellyfin.
"""
# For movies, assume you have a function get_all_movies similar to get_all_series.
from radarr_sonarr_mcp.services.radarr_service import RadarrService
# You would need to instantiate a RadarrService and fetch movies.
radarr_service = RadarrService(self.config["radarrConfig"])
all_movies = radarr_service.get_all_movies() # Assuming this returns a list of dicts
filtered_movies = all_movies
if year is not None:
filtered_movies = [m for m in filtered_movies if m.get("year") == year]
if downloaded is not None:
filtered_movies = [m for m in filtered_movies if m.get("hasFile") == downloaded]
if watched is not None:
if watched:
filtered_movies = [
m for m in filtered_movies
if is_watched_movie(m.get("title", ""), self.config)
]
else:
filtered_movies = [
m for m in filtered_movies
if not is_watched_movie(m.get("title", ""), self.config)
]
if actors:
filtered_movies = [
m for m in filtered_movies
if m.get("credits") and any(
actors.lower() in cast.get("name", "").lower()
for cast in m.get("credits", {}).get("cast", [])
)
]
return {
"count": len(filtered_movies),
"movies": [
{
"id": m.get("id"),
"title": m.get("title"),
"year": m.get("year"),
"overview": m.get("overview"),
"hasFile": m.get("hasFile"),
"status": m.get("status"),
"genres": m.get("genres", []),
"watched": is_watched_movie(m.get("title", ""), self.config)
}
for m in filtered_movies
]
}
def _register_resources(self):
@self.server.resource("http://example.com/series", description="TV series collection from Sonarr")
def series() -> dict:
series_list = get_all_series(self.config)
return {
"count": len(series_list),
"series": [
{
"id": s.id,
"title": s.title,
"year": s.year
}
for s in series_list
]
}
@self.server.resource("http://example.com/movies", description="Movie collection from Radarr")
def movies() -> dict:
from radarr_sonarr_mcp.services.radarr_service import RadarrService
radarr_service = RadarrService(self.config["radarrConfig"])
movies_list = radarr_service.get_all_movies() # Assuming list of dicts
return {
"count": len(movies_list),
"movies": [
{
"id": m.get("id"),
"title": m.get("title"),
"year": m.get("year")
}
for m in movies_list
]
}
def run(self):
port = self.config["server"]["port"]
logger.info(f"Starting Radarr-Sonarr MCP Server on port {port}")
logger.info(f"Connect Claude Desktop to: http://localhost:{port}")
self.server.run()
if __name__ == "__main__":
server = RadarrSonarrMCP()
server.run()
```