# 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:
--------------------------------------------------------------------------------
```
1 | # Byte-compiled / optimized / DLL files
2 | __pycache__/
3 | *.py[cod]
4 | *$py.class
5 |
6 | # Distribution / packaging
7 | dist/
8 | build/
9 | *.egg-info/
10 |
11 | # Virtual environments
12 | venv/
13 | env/
14 | ENV/
15 |
16 | # Local development settings
17 | .env
18 | config.json
19 |
20 | # Editor settings
21 | .vscode/
22 | .idea/
23 |
24 | # Logs
25 | *.log
26 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Radarr and Sonarr MCP Server
2 |
3 | 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.
4 |
5 | ## Overview
6 |
7 | 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.
8 |
9 | ## Features
10 |
11 | - **Native MCP Implementation**: Built with FastMCP for seamless AI integration
12 | - **Radarr Integration**: Access your movie collection
13 | - **Sonarr Integration**: Access your TV show and episode data
14 | - **Rich Filtering**: Filter by year, watched status, actors, and more
15 | - **Claude Desktop Compatible**: Works seamlessly with Claude's MCP client
16 | - **Easy Setup**: Interactive configuration wizard
17 | - **Well-tested**: Comprehensive test suite for reliability
18 |
19 | ## Installation
20 |
21 | ### From Source
22 |
23 | 1. Clone this repository:
24 | ```bash
25 | git clone https://github.com/yourusername/radarr-sonarr-mcp.git
26 | cd radarr-sonarr-mcp-python
27 | ```
28 |
29 | 2. Install the package:
30 | ```bash
31 | pip install -e .
32 | ```
33 |
34 | ### Using pip (coming soon)
35 |
36 | ```bash
37 | pip install radarr-sonarr-mcp
38 | ```
39 |
40 | ## Quick Start
41 |
42 | 1. Configure the server:
43 | ```bash
44 | radarr-sonarr-mcp configure
45 | ```
46 | Follow the prompts to enter your Radarr/Sonarr API keys and other settings.
47 |
48 | 2. Start the server:
49 | ```bash
50 | radarr-sonarr-mcp start
51 | ```
52 |
53 | 3. Connect Claude Desktop:
54 | - In Claude Desktop, go to Settings > MCP Servers
55 | - Add a new server with URL: `http://localhost:3000` (or your configured port)
56 |
57 | ## Configuration
58 |
59 | The configuration wizard will guide you through setting up:
60 |
61 | - NAS/Server IP address
62 | - Radarr API key and port
63 | - Sonarr API key and port
64 | - MCP server port
65 |
66 | You can also manually edit the `config.json` file:
67 |
68 | ```json
69 | {
70 | "nasConfig": {
71 | "ip": "10.0.0.23",
72 | "port": "7878"
73 | },
74 | "radarrConfig": {
75 | "apiKey": "YOUR_RADARR_API_KEY",
76 | "basePath": "/api/v3",
77 | "port": "7878"
78 | },
79 | "sonarrConfig": {
80 | "apiKey": "YOUR_SONARR_API_KEY",
81 | "basePath": "/api/v3",
82 | "port": "8989"
83 | },
84 | "server": {
85 | "port": 3000
86 | }
87 | }
88 | ```
89 |
90 | ## Available MCP Tools
91 |
92 | This server provides the following tools to Claude:
93 |
94 | ### Movies
95 | - `get_available_movies` - Get a list of movies with optional filters
96 | - `lookup_movie` - Search for a movie by title
97 | - `get_movie_details` - Get detailed information about a specific movie
98 |
99 | ### Series
100 | - `get_available_series` - Get a list of TV series with optional filters
101 | - `lookup_series` - Search for a TV series by title
102 | - `get_series_details` - Get detailed information about a specific series
103 | - `get_series_episodes` - Get episodes for a specific series
104 |
105 | ### Resources
106 |
107 | The server also provides standard MCP resources:
108 |
109 | - `/movies` - Browse all available movies
110 | - `/series` - Browse all available TV series
111 |
112 | ### Filtering Options
113 |
114 | Most tools support various filtering options:
115 |
116 | - `year` - Filter by release year
117 | - `watched` - Filter by watched status (true/false)
118 | - `downloaded` - Filter by download status (true/false)
119 | - `watchlist` - Filter by watchlist status (true/false)
120 | - `actors` - Filter by actor/cast name
121 | - `actresses` - Filter by actress name (movies only)
122 |
123 | ## Example Queries for Claude
124 |
125 | Once your MCP server is connected to Claude Desktop, you can ask questions like:
126 |
127 | - "What sci-fi movies from 2023 do I have?"
128 | - "Show me TV shows starring Pedro Pascal"
129 | - "Do I have any unwatched episodes of The Mandalorian?"
130 | - "Find movies with Tom Hanks that I haven't watched yet"
131 | - "How many episodes of Stranger Things do I have downloaded?"
132 |
133 | ## Finding API Keys
134 |
135 | ### Radarr API Key
136 | 1. Open Radarr in your browser
137 | 2. Go to Settings > General
138 | 3. Find the "API Key" section
139 | 4. Copy the API Key
140 |
141 | ### Sonarr API Key
142 | 1. Open Sonarr in your browser
143 | 2. Go to Settings > General
144 | 3. Find the "API Key" section
145 | 4. Copy the API Key
146 |
147 | ## Command-Line Interface
148 |
149 | The package provides a command-line interface:
150 |
151 | - `radarr-sonarr-mcp configure` - Run configuration wizard
152 | - `radarr-sonarr-mcp start` - Start the MCP server
153 | - `radarr-sonarr-mcp status` - Show the current configuration
154 |
155 | ## Development
156 |
157 | ### Running Tests
158 |
159 | To run the test suite:
160 |
161 | ```bash
162 | # Install development dependencies
163 | pip install -e ".[dev]"
164 |
165 | # Run tests
166 | pytest
167 |
168 | # Run tests with coverage
169 | pytest --cov=radarr_sonarr_mcp
170 | ```
171 |
172 | ### Local Development
173 |
174 | For quick development and testing:
175 |
176 | ```bash
177 | # Run directly without installation
178 | python run.py
179 | ```
180 |
181 | ## Requirements
182 |
183 | - Python 3.7+
184 | - FastMCP
185 | - Requests
186 | - Pydantic
187 |
188 | ## Notes
189 |
190 | - 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.
191 | - For security reasons, it's recommended to run this server only on your local network.
192 |
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Test package for radarr-sonarr-mcp."""
2 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | mcp>=0.1.0
2 | requests>=2.28.0
3 | pydantic>=2.0.0
4 | pytest>=7.0.0
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Services for interacting with Radarr and Sonarr APIs."""
2 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Radarr and Sonarr MCP Server - Model Context Protocol server for media management."""
2 |
3 | __version__ = "1.0.0"
4 |
```
--------------------------------------------------------------------------------
/run.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Simple script to run the MCP server directly without installation.
4 | Useful for development.
5 | """
6 |
7 | import sys
8 | import os
9 |
10 | # Add the project directory to the path
11 | sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
12 |
13 | from radarr_sonarr_mcp.server import main
14 |
15 | if __name__ == "__main__":
16 | main()
17 |
```
--------------------------------------------------------------------------------
/run_tests.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Simple script to run the test suite.
4 | """
5 |
6 | import sys
7 | import unittest
8 | import os
9 |
10 | # Add the project directory to the path
11 | sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
12 |
13 | # Import the test modules
14 | from tests.test_server import TestRadarrSonarrMCPServer
15 |
16 | if __name__ == "__main__":
17 | # Run all tests
18 | unittest.main(module=None, argv=['run_tests.py'])
19 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "radarr-sonarr-mcp"
3 | version = "0.1.0"
4 | description = "MCP Server for Radarr and Sonarr media management"
5 | authors = [
6 | { name = "Berry", email = "[email protected]" }
7 | ]
8 | requires-python = ">=3.10"
9 | readme = "README.md"
10 | classifiers = [
11 | "Programming Language :: Python :: 3",
12 | "License :: OSI Approved :: MIT License",
13 | "Operating System :: OS Independent"
14 | ]
15 | scripts = { radarr-sonarr-mcp = "radarr_sonarr_mcp.cli:main" }
16 | dependencies = [
17 | "mcp>=0.1.0",
18 | "requests>=2.28.0",
19 | "pydantic>=2.0.0"
20 | ]
21 | optional-dependencies = { dev = ["pytest>=7.0.0"] }
22 |
23 | [build-system]
24 | requires = ["setuptools>=42", "wheel"]
25 | build-backend = "setuptools.build_meta"
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
1 | """Setup script for the radarr-sonarr-mcp package."""
2 |
3 | from setuptools import setup, find_packages
4 |
5 | with open("README.md", "r", encoding="utf-8") as fh:
6 | long_description = fh.read()
7 |
8 | setup(
9 | name="radarr-sonarr-mcp",
10 | version="1.0.0",
11 | packages=find_packages(),
12 | install_requires=[
13 | "fastmcp>=0.4.1",
14 | "requests>=2.28.0",
15 | "pydantic>=2.0.0",
16 | ],
17 | entry_points={
18 | "console_scripts": [
19 | "radarr-sonarr-mcp=radarr_sonarr_mcp.cli:main",
20 | ],
21 | },
22 | author="Berry",
23 | author_email="",
24 | description="Model Context Protocol server for Radarr and Sonarr",
25 | long_description=long_description,
26 | long_description_content_type="text/markdown",
27 | url="",
28 | classifiers=[
29 | "Programming Language :: Python :: 3",
30 | "License :: OSI Approved :: MIT License",
31 | "Operating System :: OS Independent",
32 | ],
33 | python_requires=">=3.7",
34 | extras_require={
35 | "dev": [
36 | "pytest>=7.0.0",
37 | "pytest-cov>=3.0.0",
38 | ],
39 | },
40 | )
41 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/jellyfin_service.py:
--------------------------------------------------------------------------------
```python
1 | import requests
2 | from typing import Any, Dict, List
3 |
4 | class JellyfinService:
5 | """
6 | Service for interacting with the Jellyfin API.
7 | This service searches for a series by title and retrieves its episodes to check the watch status.
8 | """
9 | def __init__(self, config: Dict[str, Any]):
10 | self.base_url = config.get("baseUrl") # e.g., "http://10.0.0.23:5055"
11 | self.api_key = config.get("apiKey")
12 | self.user_id = config.get("userId") # The user ID to check watch status for
13 |
14 | def search_series(self, title: str) -> List[Dict[str, Any]]:
15 | """
16 | Search for a series in Jellyfin by title.
17 | """
18 | url = f"{self.base_url}/Users/{self.user_id}/Items"
19 | params = {
20 | "IncludeItemTypes": "Series",
21 | "SearchTerm": title,
22 | "api_key": self.api_key
23 | }
24 | response = requests.get(url, params=params, timeout=30)
25 | response.raise_for_status()
26 | return response.json().get("Items", [])
27 |
28 | def get_episodes_for_series(self, series_id: str) -> List[Dict[str, Any]]:
29 | """
30 | Retrieve episodes for a given series ID from Jellyfin.
31 | """
32 | url = f"{self.base_url}/Users/{self.user_id}/Items"
33 | params = {
34 | "ParentId": series_id,
35 | "IncludeItemTypes": "Episode",
36 | "api_key": self.api_key
37 | }
38 | response = requests.get(url, params=params, timeout=30)
39 | response.raise_for_status()
40 | return response.json().get("Items", [])
41 |
42 | def is_series_watched(self, series_title: str) -> bool:
43 | """
44 | Determine if the series is watched.
45 | A series is considered watched if all episodes have a PlayCount > 0.
46 | """
47 | items = self.search_series(series_title)
48 | if not items:
49 | return False
50 | series_item = items[0] # take the first match
51 | series_id = series_item.get("Id")
52 | episodes = self.get_episodes_for_series(series_id)
53 | if not episodes:
54 | return False
55 | # Consider the series watched if every episode has a PlayCount > 0
56 | return all(ep.get("UserData", {}).get("PlayCount", 0) > 0 for ep in episodes)
57 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/plex_service.py:
--------------------------------------------------------------------------------
```python
1 | import requests
2 | from typing import Any, Dict, List
3 |
4 | class PlexService:
5 | """
6 | Service for interacting with the Plex API.
7 |
8 | Note: Plex’s API typically returns XML, but here we assume a JSON endpoint
9 | (or you can use an XML parser). This is a simplified example.
10 | """
11 | def __init__(self, config: Dict[str, Any]):
12 | self.base_url = config.get("baseUrl") # e.g., "http://10.0.0.23:32400"
13 | self.token = config.get("token")
14 |
15 | def search_series(self, title: str) -> List[Dict[str, Any]]:
16 | url = f"{self.base_url}/library/search"
17 | params = {
18 | "query": title,
19 | "type": 4 # type 4 usually indicates a TV series
20 | }
21 | headers = {"X-Plex-Token": self.token}
22 | response = requests.get(url, params=params, headers=headers, timeout=30)
23 | # In a real implementation, parse XML; here we assume JSON for simplicity.
24 | try:
25 | return response.json().get("MediaContainer", {}).get("Metadata", [])
26 | except Exception:
27 | return []
28 |
29 | def get_episodes_for_series(self, rating_key: str) -> List[Dict[str, Any]]:
30 | url = f"{self.base_url}/library/metadata/{rating_key}/children"
31 | headers = {"X-Plex-Token": self.token}
32 | response = requests.get(url, headers=headers, timeout=30)
33 | try:
34 | return response.json().get("MediaContainer", {}).get("Metadata", [])
35 | except Exception:
36 | return []
37 |
38 | def is_series_watched(self, series_title: str) -> bool:
39 | items = self.search_series(series_title)
40 | if not items:
41 | return False
42 | # Assume the first matching series is our target.
43 | series_item = items[0]
44 | rating_key = series_item.get("ratingKey")
45 | episodes = self.get_episodes_for_series(rating_key)
46 | if not episodes:
47 | return False
48 | # Consider the series watched if every episode's UserData indicates it was played.
49 | return all(ep.get("UserData", {}).get("viewCount", 0) > 0 for ep in episodes)
50 |
51 | def search_movie(self, title: str) -> List[Dict[str, Any]]:
52 | url = f"{self.base_url}/library/search"
53 | params = {
54 | "query": title,
55 | "type": 2 # type 2 for movies
56 | }
57 | headers = {"X-Plex-Token": self.token}
58 | response = requests.get(url, params=params, headers=headers, timeout=30)
59 | try:
60 | return response.json().get("MediaContainer", {}).get("Metadata", [])
61 | except Exception:
62 | return []
63 |
64 | def is_movie_watched(self, movie_title: str) -> bool:
65 | items = self.search_movie(movie_title)
66 | if not items:
67 | return False
68 | movie_item = items[0]
69 | return movie_item.get("UserData", {}).get("viewCount", 0) > 0
70 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/radarr_service.py:
--------------------------------------------------------------------------------
```python
1 | """Service for interacting with Radarr API."""
2 |
3 | from dataclasses import dataclass
4 | from typing import List, Dict, Any, Optional
5 | import requests
6 | from ..config import RadarrConfig
7 |
8 |
9 | @dataclass
10 | class Movie:
11 | """Movie data class."""
12 | id: int
13 | title: str
14 | year: int
15 | overview: str
16 | has_file: bool
17 | status: str
18 | tags: List[int] = None
19 | genres: List[str] = None
20 | data: Dict[str, Any] = None
21 |
22 | @classmethod
23 | def from_dict(cls, data: Dict[str, Any]) -> 'Movie':
24 | """Create a Movie object from a dictionary."""
25 | return cls(
26 | id=data['id'],
27 | title=data['title'],
28 | year=data.get('year', 0),
29 | overview=data.get('overview', ''),
30 | has_file=data.get('hasFile', False),
31 | status=data.get('status', ''),
32 | tags=data.get('tags', []),
33 | genres=data.get('genres', []),
34 | data=data
35 | )
36 |
37 |
38 | class RadarrService:
39 | """Service for interacting with Radarr API."""
40 |
41 | def __init__(self, config: RadarrConfig):
42 | """Initialize the Radarr service with configuration."""
43 | self.config = config
44 |
45 | def get_all_movies(self) -> List[Movie]:
46 | """Fetch all movies from Radarr."""
47 | try:
48 | response = requests.get(
49 | f"{self.config.base_url}/movie",
50 | params={"apikey": self.config.api_key},
51 | timeout=30
52 | )
53 | response.raise_for_status()
54 |
55 | movies = []
56 | for movie_data in response.json():
57 | movies.append(Movie.from_dict(movie_data))
58 |
59 | return movies
60 | except requests.RequestException as e:
61 | import logging
62 | logging.error(f"Error fetching movies from Radarr: {e}")
63 | raise Exception(f"Failed to fetch movies from Radarr: {e}")
64 |
65 | def lookup_movie(self, term: str) -> List[Movie]:
66 | """Look up movies by search term."""
67 | try:
68 | response = requests.get(
69 | f"{self.config.base_url}/movie/lookup",
70 | params={"term": term, "apikey": self.config.api_key},
71 | timeout=30
72 | )
73 | response.raise_for_status()
74 |
75 | movies = []
76 | for movie_data in response.json():
77 | movies.append(Movie.from_dict(movie_data))
78 |
79 | return movies
80 | except requests.RequestException as e:
81 | import logging
82 | logging.error(f"Error looking up movie in Radarr: {e}")
83 | raise Exception(f"Failed to lookup movie in Radarr: {e}")
84 |
85 | def get_movie_file(self, movie_id: int) -> Dict[str, Any]:
86 | """Get the file information for a movie."""
87 | try:
88 | response = requests.get(
89 | f"{self.config.base_url}/moviefile",
90 | params={"movieId": movie_id, "apikey": self.config.api_key},
91 | timeout=30
92 | )
93 | response.raise_for_status()
94 |
95 | return response.json()
96 | except requests.RequestException as e:
97 | import logging
98 | logging.error(f"Error fetching movie file for ID {movie_id}: {e}")
99 | raise Exception(f"Failed to fetch movie file: {e}")
100 |
101 | def is_movie_watched(self, movie: Movie) -> bool:
102 | """Check if a movie is watched based on tags."""
103 | # This is an assumption - actual implementation may vary based on how
104 | # watched status is tracked in your Radarr setup
105 | return movie.data.get('movieFile', {}).get('mediaInfo', {}).get('watched', False)
106 |
107 | def is_movie_in_watchlist(self, movie: Movie) -> bool:
108 | """Check if a movie is in the watchlist based on tags."""
109 | # This is an assumption - implementation may vary
110 | # Assuming 'watchlist' tag with ID 1 (adjust as needed)
111 | return 1 in (movie.tags or [])
112 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/cli.py:
--------------------------------------------------------------------------------
```python
1 | """Command-line interface for the Radarr/Sonarr MCP server."""
2 |
3 | import argparse
4 | import logging
5 |
6 | from .config import Config, NasConfig, RadarrConfig, SonarrConfig, ServerConfig, load_config, save_config
7 | from .server import create_server
8 |
9 | logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
10 |
11 | def configure():
12 | """Run the configuration wizard."""
13 | logging.info("==== Radarr/Sonarr MCP Server Configuration Wizard ====")
14 |
15 | # Try to load existing config
16 | config = None
17 | try:
18 | config = load_config()
19 | logging.info("Loaded existing configuration. Press Enter to keep current values.")
20 | except Exception:
21 | # No existing config or error loading
22 | pass
23 |
24 | # NAS configuration
25 | nas_ip = input(f"NAS/Server IP address [{config.nas_config.ip if config else '10.0.0.23'}]: ")
26 | nas_ip = nas_ip or (config.nas_config.ip if config else '10.0.0.23')
27 |
28 | nas_port = input(f"Default port [{config.nas_config.port if config else '7878'}]: ")
29 | nas_port = nas_port or (config.nas_config.port if config else '7878')
30 |
31 | # Radarr configuration
32 | radarr_api_key = input(f"Radarr API key [{config.radarr_config.api_key if config else ''}]: ")
33 | radarr_api_key = radarr_api_key or (config.radarr_config.api_key if config else '')
34 | if not radarr_api_key:
35 | logging.warning("Warning: Radarr API key is required for movie functionality!")
36 |
37 | radarr_port = input(f"Radarr port [{config.radarr_config.port if config else '7878'}]: ")
38 | radarr_port = radarr_port or (config.radarr_config.port if config else '7878')
39 |
40 | radarr_base_path = input(f"Radarr API base path [{config.radarr_config.base_path if config else '/api/v3'}]: ")
41 | radarr_base_path = radarr_base_path or (config.radarr_config.base_path if config else '/api/v3')
42 |
43 | # Sonarr configuration
44 | sonarr_api_key = input(f"Sonarr API key [{config.sonarr_config.api_key if config else ''}]: ")
45 | sonarr_api_key = sonarr_api_key or (config.sonarr_config.api_key if config else '')
46 | if not sonarr_api_key:
47 | logging.warning("Warning: Sonarr API key is required for TV show functionality!")
48 |
49 | sonarr_port = input(f"Sonarr port [{config.sonarr_config.port if config else '8989'}]: ")
50 | sonarr_port = sonarr_port or (config.sonarr_config.port if config else '8989')
51 |
52 | sonarr_base_path = input(f"Sonarr API base path [{config.sonarr_config.base_path if config else '/api/v3'}]: ")
53 | sonarr_base_path = sonarr_base_path or (config.sonarr_config.base_path if config else '/api/v3')
54 |
55 | # Server configuration
56 | server_port = input(f"MCP server port [{config.server_config.port if config else '3000'}]: ")
57 | if server_port:
58 | try:
59 | server_port = int(server_port)
60 | except ValueError:
61 | logging.warning("Invalid port number, using default.")
62 | server_port = config.server_config.port if config else 3000
63 | else:
64 | server_port = config.server_config.port if config else 3000
65 |
66 | # Create new config
67 | new_config = Config(
68 | nas_config=NasConfig(
69 | ip=nas_ip,
70 | port=nas_port
71 | ),
72 | radarr_config=RadarrConfig(
73 | api_key=radarr_api_key,
74 | base_path=radarr_base_path,
75 | port=radarr_port
76 | ),
77 | sonarr_config=SonarrConfig(
78 | api_key=sonarr_api_key,
79 | base_path=sonarr_base_path,
80 | port=sonarr_port
81 | ),
82 | server_config=ServerConfig(
83 | port=server_port
84 | )
85 | )
86 |
87 | # Save config
88 | save_config(new_config)
89 | logging.info("Configuration saved successfully!")
90 | logging.info(f"To start the server, run: radarr-sonarr-mcp start")
91 |
92 | return new_config
93 |
94 |
95 | def start(config_path=None):
96 | """Start the MCP server."""
97 | server = create_server(config_path)
98 | server.start()
99 |
100 |
101 | def show_status():
102 | """Show the current status of the server."""
103 | try:
104 | config = load_config()
105 | logging.info("==== Radarr/Sonarr MCP Server Status ====")
106 | logging.info(f"NAS IP: {config.nas_config.ip}")
107 | logging.info(f"Radarr Port: {config.radarr_config.port or config.nas_config.port}")
108 | logging.info(f"Sonarr Port: {config.sonarr_config.port or config.nas_config.port}")
109 | logging.info(f"MCP Server Port: {config.server_config.port}")
110 | logging.info(f"MCP Endpoint URL: http://localhost:{config.server_config.port}")
111 | logging.info(f"Server is configured. Use 'radarr-sonarr-mcp start' to run the server.")
112 | except Exception as e:
113 | logging.error(f"Server is not configured: {e}")
114 | logging.info("Run 'radarr-sonarr-mcp configure' to set up the server.")
115 |
116 |
117 | def main():
118 | """Main CLI entry point."""
119 | parser = argparse.ArgumentParser(description="Radarr/Sonarr MCP Server")
120 | subparsers = parser.add_subparsers(dest="command", help="Command to execute")
121 |
122 | # Configure command
123 | configure_parser = subparsers.add_parser("configure", help="Configure the MCP server")
124 |
125 | # Start command
126 | start_parser = subparsers.add_parser("start", help="Start the MCP server")
127 | start_parser.add_argument("--config", help="Path to config.json file")
128 |
129 | # Status command
130 | status_parser = subparsers.add_parser("status", help="Show the server status")
131 |
132 | args = parser.parse_args()
133 |
134 | if args.command == "configure":
135 | configure()
136 | elif args.command == "start":
137 | start(args.config)
138 | elif args.command == "status":
139 | show_status()
140 | else:
141 | parser.print_help()
142 |
143 |
144 | if __name__ == "__main__":
145 | main()
146 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/services/sonarr_service.py:
--------------------------------------------------------------------------------
```python
1 | """Service for interacting with Sonarr API."""
2 |
3 | from dataclasses import dataclass
4 | from typing import List, Dict, Any, Optional
5 | import requests
6 | from ..config import SonarrConfig
7 |
8 |
9 | @dataclass
10 | class Statistics:
11 | """Statistics for a TV series."""
12 | episode_file_count: int
13 | episode_count: int
14 | total_episode_count: int
15 | size_on_disk: int
16 |
17 | @classmethod
18 | def from_dict(cls, data: Dict[str, Any]) -> 'Statistics':
19 | """Create a Statistics object from a dictionary."""
20 | return cls(
21 | episode_file_count=data.get('episodeFileCount', 0),
22 | episode_count=data.get('episodeCount', 0),
23 | total_episode_count=data.get('totalEpisodeCount', 0),
24 | size_on_disk=data.get('sizeOnDisk', 0)
25 | )
26 |
27 |
28 | @dataclass
29 | class Series:
30 | """TV Series data class."""
31 | id: int
32 | title: str
33 | year: Optional[int]
34 | overview: str
35 | status: str
36 | network: str
37 | tags: List[int]
38 | genres: List[str]
39 | statistics: Optional[Statistics]
40 | data: Dict[str, Any] # Store original data for reference
41 |
42 | @classmethod
43 | def from_dict(cls, data: Dict[str, Any]) -> 'Series':
44 | """Create a Series object from a dictionary."""
45 | statistics = None
46 | if 'statistics' in data:
47 | statistics = Statistics.from_dict(data['statistics'])
48 |
49 | return cls(
50 | id=data['id'],
51 | title=data['title'],
52 | year=data.get('year'),
53 | overview=data.get('overview', ''),
54 | status=data.get('status', ''),
55 | network=data.get('network', ''),
56 | tags=data.get('tags', []),
57 | genres=data.get('genres', []),
58 | statistics=statistics,
59 | data=data
60 | )
61 |
62 |
63 | @dataclass
64 | class Episode:
65 | """TV Episode data class."""
66 | id: int
67 | series_id: int
68 | episode_file_id: Optional[int]
69 | season_number: int
70 | episode_number: int
71 | title: str
72 | air_date: Optional[str]
73 | has_file: bool
74 | monitored: bool
75 | overview: str
76 | data: Dict[str, Any] # Store original data for reference
77 |
78 | @classmethod
79 | def from_dict(cls, data: Dict[str, Any]) -> 'Episode':
80 | """Create an Episode object from a dictionary."""
81 | return cls(
82 | id=data['id'],
83 | series_id=data['seriesId'],
84 | episode_file_id=data.get('episodeFileId'),
85 | season_number=data['seasonNumber'],
86 | episode_number=data['episodeNumber'],
87 | title=data.get('title', ''),
88 | air_date=data.get('airDate'),
89 | has_file=data.get('hasFile', False),
90 | monitored=data.get('monitored', True),
91 | overview=data.get('overview', ''),
92 | data=data
93 | )
94 |
95 |
96 | class SonarrService:
97 | """Service for interacting with Sonarr API."""
98 |
99 | def __init__(self, config: SonarrConfig):
100 | """Initialize the Sonarr service with configuration."""
101 | self.config = config
102 |
103 | def get_all_series(self) -> List[Series]:
104 | """Fetch all TV series from Sonarr."""
105 | try:
106 | response = requests.get(
107 | f"{self.config.base_url}/series",
108 | params={"apikey": self.config.api_key},
109 | timeout=30
110 | )
111 | response.raise_for_status()
112 |
113 | series_list = []
114 | for series_data in response.json():
115 | series_list.append(Series.from_dict(series_data))
116 |
117 | return series_list
118 | except requests.RequestException as e:
119 | import logging
120 | logging.error(f"Error fetching series from Sonarr: {e}")
121 | raise Exception(f"Failed to fetch series from Sonarr: {e}")
122 |
123 | def lookup_series(self, term: str) -> List[Series]:
124 | """Look up TV series by search term."""
125 | try:
126 | response = requests.get(
127 | f"{self.config.base_url}/series/lookup",
128 | params={"term": term, "apikey": self.config.api_key},
129 | timeout=30
130 | )
131 | response.raise_for_status()
132 |
133 | series_list = []
134 | for series_data in response.json():
135 | series_list.append(Series.from_dict(series_data))
136 |
137 | return series_list
138 | except requests.RequestException as e:
139 | import logging
140 | logging.error(f"Error looking up series in Sonarr: {e}")
141 | raise Exception(f"Failed to lookup series from Sonarr: {e}")
142 |
143 | def get_episodes(self, series_id: int) -> List[Episode]:
144 | """Fetch episodes for a TV series."""
145 | try:
146 | response = requests.get(
147 | f"{self.config.base_url}/episode",
148 | params={"seriesId": series_id, "apikey": self.config.api_key},
149 | timeout=30
150 | )
151 | response.raise_for_status()
152 |
153 | episodes = []
154 | for episode_data in response.json():
155 | episodes.append(Episode.from_dict(episode_data))
156 |
157 | return episodes
158 | except requests.RequestException as e:
159 | import logging
160 | logging.error(f"Error fetching episodes for series ID {series_id}: {e}")
161 | raise Exception(f"Failed to fetch episodes: {e}")
162 |
163 | def is_series_watched(self, series: Series) -> bool:
164 | """Check if a series is watched based on tags."""
165 | # This is an assumption - actual implementation may vary based on how
166 | # watched status is tracked in your Sonarr setup
167 | if not series.statistics:
168 | return False
169 |
170 | # Consider series watched if all episodes are downloaded
171 | return (series.statistics.episode_file_count >=
172 | series.statistics.episode_count)
173 |
174 | def is_series_in_watchlist(self, series: Series) -> bool:
175 | """Check if a series is in the watchlist based on tags."""
176 | # This is an assumption - implementation may vary
177 | # Assuming 'watchlist' tag with ID 1 (adjust as needed)
178 | return 1 in (series.tags or [])
179 |
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
1 | """Tests for the MCP server implementation."""
2 |
3 | import unittest
4 | import json
5 | import os
6 | import sys
7 | import tempfile
8 | from unittest.mock import patch, MagicMock
9 |
10 | # Add parent directory to path to import the package
11 | sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
12 |
13 | from radarr_sonarr_mcp.config import Config, NasConfig, RadarrConfig, SonarrConfig, ServerConfig
14 | from radarr_sonarr_mcp.server import RadarrSonarrMCPServer, create_server
15 | from radarr_sonarr_mcp.services.radarr_service import Movie
16 | from radarr_sonarr_mcp.services.sonarr_service import Series, Statistics
17 |
18 |
19 | class TestRadarrSonarrMCPServer(unittest.TestCase):
20 | """Test suite for the RadarrSonarrMCPServer class."""
21 |
22 | def setUp(self):
23 | """Set up test environment."""
24 | # Create a temporary config file
25 | self.temp_file = tempfile.NamedTemporaryFile(delete=False)
26 | config_data = {
27 | "nasConfig": {
28 | "ip": "127.0.0.1",
29 | "port": "7878"
30 | },
31 | "radarrConfig": {
32 | "apiKey": "test_radarr_api_key",
33 | "basePath": "/api/v3",
34 | "port": "7878"
35 | },
36 | "sonarrConfig": {
37 | "apiKey": "test_sonarr_api_key",
38 | "basePath": "/api/v3",
39 | "port": "8989"
40 | },
41 | "server": {
42 | "port": 5000
43 | }
44 | }
45 | with open(self.temp_file.name, 'w') as f:
46 | json.dump(config_data, f)
47 |
48 | # Create sample movie and series data
49 | self.sample_movies = [
50 | Movie(
51 | id=1,
52 | title="Test Movie 1",
53 | year=2022,
54 | overview="A test movie",
55 | has_file=True,
56 | status="downloaded",
57 | genres=["Action", "Comedy"],
58 | tags=[1, 2],
59 | data={
60 | "id": 1,
61 | "title": "Test Movie 1",
62 | "year": 2022,
63 | "overview": "A test movie",
64 | "hasFile": True,
65 | "status": "downloaded",
66 | "genres": ["Action", "Comedy"],
67 | "tags": [1, 2],
68 | "credits": {
69 | "cast": [
70 | {"name": "Actor One", "character": "Character One"},
71 | {"name": "Actress One", "character": "Character Two"}
72 | ]
73 | }
74 | }
75 | ),
76 | Movie(
77 | id=2,
78 | title="Test Movie 2",
79 | year=2023,
80 | overview="Another test movie",
81 | has_file=False,
82 | status="wanted",
83 | genres=["Drama", "Thriller"],
84 | tags=[2],
85 | data={
86 | "id": 2,
87 | "title": "Test Movie 2",
88 | "year": 2023,
89 | "overview": "Another test movie",
90 | "hasFile": False,
91 | "status": "wanted",
92 | "genres": ["Drama", "Thriller"],
93 | "tags": [2],
94 | "credits": {
95 | "cast": [
96 | {"name": "Actor Two", "character": "Character Three"}
97 | ]
98 | }
99 | }
100 | )
101 | ]
102 |
103 | self.sample_series = [
104 | Series(
105 | id=1,
106 | title="Test Series 1",
107 | year=2022,
108 | overview="A test series",
109 | status="continuing",
110 | network="Test Network",
111 | tags=[1],
112 | genres=["Comedy"],
113 | statistics=Statistics.from_dict({
114 | "episodeFileCount": 10,
115 | "episodeCount": 10,
116 | "totalEpisodeCount": 20,
117 | "sizeOnDisk": 10000
118 | }),
119 | data={
120 | "id": 1,
121 | "title": "Test Series 1",
122 | "year": 2022,
123 | "overview": "A test series",
124 | "status": "continuing",
125 | "network": "Test Network",
126 | "tags": [1],
127 | "genres": ["Comedy"],
128 | "statistics": {
129 | "episodeFileCount": 10,
130 | "episodeCount": 10,
131 | "totalEpisodeCount": 20,
132 | "sizeOnDisk": 10000
133 | },
134 | "credits": {
135 | "cast": [
136 | {"name": "Actor Three", "character": "Character Four"}
137 | ]
138 | }
139 | }
140 | )
141 | ]
142 |
143 | def tearDown(self):
144 | """Clean up after tests."""
145 | self.temp_file.close()
146 | os.unlink(self.temp_file.name)
147 |
148 | @patch('radarr_sonarr_mcp.server.FastMCP')
149 | def test_server_initialization(self, mock_fastmcp):
150 | """Test server initialization with config file."""
151 | server = create_server(self.temp_file.name)
152 | self.assertEqual(server.config.radarr_config.api_key, "test_radarr_api_key")
153 | self.assertEqual(server.config.sonarr_config.api_key, "test_sonarr_api_key")
154 | self.assertEqual(server.config.server_config.port, 5000)
155 |
156 | # Check that FastMCP was initialized correctly
157 | mock_fastmcp.assert_called_once()
158 | self.assertEqual(mock_fastmcp.call_args[1]['name'], "radarr-sonarr-mcp-server")
159 |
160 | @patch('radarr_sonarr_mcp.server.RadarrService')
161 | @patch('radarr_sonarr_mcp.server.SonarrService')
162 | @patch('radarr_sonarr_mcp.server.FastMCP')
163 | def test_get_available_movies(self, mock_fastmcp, mock_sonarr_service, mock_radarr_service):
164 | """Test the get_available_movies tool."""
165 | # Setup mocks
166 | mock_radarr_instance = mock_radarr_service.return_value
167 | mock_radarr_instance.get_all_movies.return_value = self.sample_movies
168 | mock_radarr_instance.is_movie_watched.return_value = True
169 | mock_radarr_instance.is_movie_in_watchlist.return_value = False
170 |
171 | mock_server = mock_fastmcp.return_value
172 |
173 | # Create server and register tools
174 | server = create_server(self.temp_file.name)
175 |
176 | # Extract the registered tool function
177 | tool_decorator = mock_server.tool.return_value
178 | get_movies_func = None
179 | for call in tool_decorator.call_args_list:
180 | # The decorated function is passed to the decorator
181 | if call.args and call.args[0].__name__ == 'get_available_movies':
182 | get_movies_func = call.args[0]
183 | break
184 |
185 | self.assertIsNotNone(get_movies_func, "get_available_movies tool not registered")
186 |
187 | # Test the tool function
188 | result = get_movies_func(year=2022)
189 | result_data = json.loads(result)
190 |
191 | # Check results
192 | self.assertEqual(result_data['count'], 1)
193 | self.assertEqual(result_data['movies'][0]['title'], "Test Movie 1")
194 | self.assertEqual(result_data['movies'][0]['year'], 2022)
195 |
196 | @patch('radarr_sonarr_mcp.server.SonarrService')
197 | @patch('radarr_sonarr_mcp.server.RadarrService')
198 | @patch('radarr_sonarr_mcp.server.FastMCP')
199 | def test_get_available_series(self, mock_fastmcp, mock_radarr_service, mock_sonarr_service):
200 | """Test the get_available_series tool."""
201 | # Setup mocks
202 | mock_sonarr_instance = mock_sonarr_service.return_value
203 | mock_sonarr_instance.get_all_series.return_value = self.sample_series
204 | mock_sonarr_instance.is_series_watched.return_value = True
205 | mock_sonarr_instance.is_series_in_watchlist.return_value = False
206 |
207 | mock_server = mock_fastmcp.return_value
208 |
209 | # Create server and register tools
210 | server = create_server(self.temp_file.name)
211 |
212 | # Extract the registered tool function
213 | tool_decorator = mock_server.tool.return_value
214 | get_series_func = None
215 | for call in tool_decorator.call_args_list:
216 | # The decorated function is passed to the decorator
217 | if call.args and call.args[0].__name__ == 'get_available_series':
218 | get_series_func = call.args[0]
219 | break
220 |
221 | self.assertIsNotNone(get_series_func, "get_available_series tool not registered")
222 |
223 | # Test the tool function
224 | result = get_series_func()
225 | result_data = json.loads(result)
226 |
227 | # Check results
228 | self.assertEqual(result_data['count'], 1)
229 | self.assertEqual(result_data['series'][0]['title'], "Test Series 1")
230 | self.assertEqual(result_data['series'][0]['year'], 2022)
231 |
232 | @patch('radarr_sonarr_mcp.server.RadarrService')
233 | @patch('radarr_sonarr_mcp.server.SonarrService')
234 | @patch('radarr_sonarr_mcp.server.FastMCP')
235 | def test_server_resources(self, mock_fastmcp, mock_sonarr_service, mock_radarr_service):
236 | """Test registered resources."""
237 | # Setup mocks
238 | mock_radarr_instance = mock_radarr_service.return_value
239 | mock_radarr_instance.get_all_movies.return_value = self.sample_movies
240 |
241 | mock_sonarr_instance = mock_sonarr_service.return_value
242 | mock_sonarr_instance.get_all_series.return_value = self.sample_series
243 |
244 | mock_server = mock_fastmcp.return_value
245 |
246 | # Create server and register resources
247 | server = create_server(self.temp_file.name)
248 |
249 | # Mock get_resource_handler to return MagicMock objects
250 | mock_server_instance = mock_fastmcp.return_value
251 | mock_server_instance.get_resource_handler.side_effect = lambda path: MagicMock(return_value={
252 | '/movies': {"count": 2, "movies": self.sample_movies},
253 | '/series': {"count": 1, "series": self.sample_series}
254 | }.get(path))
255 |
256 | # Test movies resource
257 | movies_resource = server.server.get_resource_handler('/movies')
258 | self.assertIsNotNone(movies_resource, "Movies resource not registered")
259 | result_movies = movies_resource()
260 | self.assertEqual(result_movies['count'], 2)
261 | self.assertEqual(len(result_movies['movies']), 2)
262 |
263 | # Test series resource
264 | series_resource = server.server.get_resource_handler('/series')
265 | self.assertIsNotNone(series_resource, "Series resource not registered")
266 | result_series = series_resource()
267 | self.assertEqual(result_series['count'], 1)
268 | self.assertEqual(len(result_series['series']), 1)
269 |
270 |
271 | if __name__ == '__main__':
272 | unittest.main()
273 |
```
--------------------------------------------------------------------------------
/radarr_sonarr_mcp/server.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """Main MCP server implementation for Radarr/Sonarr."""
3 |
4 | import os
5 | import json
6 | import sys
7 | import logging
8 | from typing import Optional
9 | import argparse
10 |
11 | from fastmcp import FastMCP
12 | import requests
13 |
14 | # Set up logging
15 | logging.basicConfig(level=logging.INFO)
16 | logger = logging.getLogger(__name__)
17 |
18 | # -----------------------------------------------------------------------------
19 | # Configuration handling
20 | # -----------------------------------------------------------------------------
21 |
22 | def load_config():
23 | """Load configuration from environment variables or config file."""
24 | if os.environ.get('RADARR_API_KEY') or os.environ.get('SONARR_API_KEY'):
25 | logger.info("Loading configuration from environment variables...")
26 | nas_ip = os.environ.get('NAS_IP', '10.0.0.23')
27 | return {
28 | "nasConfig": {
29 | "ip": nas_ip,
30 | "port": os.environ.get('RADARR_PORT', '7878')
31 | },
32 | "radarrConfig": {
33 | "apiKey": os.environ.get('RADARR_API_KEY', ''),
34 | "basePath": os.environ.get('RADARR_BASE_PATH', '/api/v3'),
35 | "port": os.environ.get('RADARR_PORT', '7878')
36 | },
37 | "sonarrConfig": {
38 | "apiKey": os.environ.get('SONARR_API_KEY', ''),
39 | "basePath": os.environ.get('SONARR_BASE_PATH', '/api/v3'),
40 | "port": os.environ.get('SONARR_PORT', '8989')
41 | },
42 | # Optionally, include Jellyfin and Plex configuration if set in env
43 | "jellyfinConfig": {
44 | "baseUrl": os.environ.get('JELLYFIN_BASE_URL', ''), # e.g., "http://10.0.0.23:5055"
45 | "apiKey": os.environ.get('JELLYFIN_API_KEY', ''),
46 | "userId": os.environ.get('JELLYFIN_USER_ID', '')
47 | },
48 | "plexConfig": {
49 | "baseUrl": os.environ.get('PLEX_BASE_URL', ''), # e.g., "http://10.0.0.23:32400"
50 | "token": os.environ.get('PLEX_TOKEN', '')
51 | },
52 | "server": {
53 | "port": int(os.environ.get('MCP_SERVER_PORT', '3000'))
54 | }
55 | }
56 | else:
57 | config_path = 'config.json'
58 | try:
59 | with open(config_path, 'r') as f:
60 | return json.load(f)
61 | except Exception as e:
62 | logger.error(f"Error loading config: {e}")
63 | logger.info("Using default configuration")
64 | return {
65 | "nasConfig": {"ip": "10.0.0.23", "port": "7878"},
66 | "radarrConfig": {"apiKey": "", "basePath": "/api/v3", "port": "7878"},
67 | "sonarrConfig": {"apiKey": "", "basePath": "/api/v3", "port": "8989"},
68 | "server": {"port": 3000}
69 | }
70 |
71 | # -----------------------------------------------------------------------------
72 | # API Service functions
73 | # -----------------------------------------------------------------------------
74 |
75 | def get_radarr_url(config):
76 | nas_ip = config["nasConfig"]["ip"]
77 | port = config["radarrConfig"]["port"]
78 | base_path = config["radarrConfig"]["basePath"]
79 | return f"http://{nas_ip}:{port}{base_path}"
80 |
81 | def get_sonarr_url(config):
82 | nas_ip = config["nasConfig"]["ip"]
83 | port = config["sonarrConfig"]["port"]
84 | base_path = config["sonarrConfig"]["basePath"]
85 | return f"http://{nas_ip}:{port}{base_path}"
86 |
87 | def make_radarr_request(config, endpoint, params=None):
88 | api_key = config["radarrConfig"]["apiKey"]
89 | base_url = get_radarr_url(config)
90 | url = f"{base_url}/{endpoint}"
91 | if params is None:
92 | params = {}
93 | params['apikey'] = api_key
94 | try:
95 | response = requests.get(url, params=params, timeout=30)
96 | response.raise_for_status()
97 | return response.json()
98 | except Exception as e:
99 | logger.error(f"Error making request to {url}: {e}")
100 | return []
101 |
102 | def make_sonarr_request(config, endpoint, params=None):
103 | api_key = config["sonarrConfig"]["apiKey"]
104 | base_url = get_sonarr_url(config)
105 | url = f"{base_url}/{endpoint}"
106 | if params is None:
107 | params = {}
108 | params['apikey'] = api_key
109 | try:
110 | response = requests.get(url, params=params, timeout=30)
111 | response.raise_for_status()
112 | return response.json()
113 | except Exception as e:
114 | logger.error(f"Error making request to {url}: {e}")
115 | return []
116 |
117 | def get_all_series(config):
118 | from radarr_sonarr_mcp.services.sonarr_service import SonarrService
119 | service = SonarrService(config["sonarrConfig"])
120 | return service.get_all_series()
121 |
122 | # -----------------------------------------------------------------------------
123 | # Helper function to check watched status from multiple sources
124 | # -----------------------------------------------------------------------------
125 |
126 | def is_watched_series(title: str, fallback: bool, config: dict, sonarr_service) -> bool:
127 | """
128 | Check if a series is watched using available media services.
129 | Returns True if any service reports the series as watched.
130 | """
131 | statuses = []
132 | if config.get("jellyfinConfig", {}).get("baseUrl"):
133 | from radarr_sonarr_mcp.services.jellyfin_service import JellyfinService
134 | jellyfin = JellyfinService(config["jellyfinConfig"])
135 | try:
136 | statuses.append(jellyfin.is_series_watched(title))
137 | except Exception as e:
138 | logger.error(f"Jellyfin check failed for {title}: {e}")
139 | if config.get("plexConfig", {}).get("baseUrl"):
140 | from radarr_sonarr_mcp.services.plex_service import PlexService
141 | plex = PlexService(config["plexConfig"])
142 | try:
143 | statuses.append(plex.is_series_watched(title))
144 | except Exception as e:
145 | logger.error(f"Plex check failed for {title}: {e}")
146 | if statuses:
147 | return any(statuses)
148 | # Fallback to Sonarr's own logic if no external services are configured.
149 | return sonarr_service.is_series_watched(title)
150 |
151 | def is_watched_movie(title: str, config: dict) -> bool:
152 | """
153 | Check if a movie is watched using available media services.
154 | Returns True if any service reports the movie as watched.
155 | """
156 | statuses = []
157 | if config.get("jellyfinConfig", {}).get("baseUrl"):
158 | from radarr_sonarr_mcp.services.jellyfin_service import JellyfinService
159 | jellyfin = JellyfinService(config["jellyfinConfig"])
160 | try:
161 | # For movies, you could implement a similar method in JellyfinService.
162 | statuses.append(jellyfin.is_movie_watched(title))
163 | except Exception as e:
164 | logger.error(f"Jellyfin movie check failed for {title}: {e}")
165 | if config.get("plexConfig", {}).get("baseUrl"):
166 | from radarr_sonarr_mcp.services.plex_service import PlexService
167 | plex = PlexService(config["plexConfig"])
168 | try:
169 | statuses.append(plex.is_movie_watched(title))
170 | except Exception as e:
171 | logger.error(f"Plex movie check failed for {title}: {e}")
172 | # If no external services configured, default to unwatched.
173 | return any(statuses)
174 |
175 | # -----------------------------------------------------------------------------
176 | # MCP Server implementation
177 | # -----------------------------------------------------------------------------
178 |
179 | from radarr_sonarr_mcp.services.sonarr_service import SonarrService
180 |
181 | class RadarrSonarrMCP:
182 | """MCP Server for Radarr and Sonarr."""
183 |
184 | def __init__(self):
185 | self.config = load_config()
186 | self.server = FastMCP(
187 | name="radarr-sonarr-mcp-server",
188 | description="MCP Server for Radarr and Sonarr media management"
189 | )
190 | self.sonarr_service = SonarrService(self.config["sonarrConfig"])
191 | self._register_tools()
192 | self._register_resources()
193 | # Optionally, register prompts.
194 |
195 | def _register_tools(self):
196 | @self.server.tool()
197 | def get_available_series(year: Optional[int] = None,
198 | downloaded: Optional[bool] = None,
199 | watched: Optional[bool] = None,
200 | actors: Optional[str] = None) -> dict:
201 | """
202 | Get a list of available TV series with optional filters.
203 | Watched status is determined using Plex and/or Jellyfin; if either reports watched, the series is considered watched.
204 | """
205 | all_series = get_all_series(self.config) # List of Series objects
206 | filtered_series = all_series
207 |
208 | if year is not None:
209 | filtered_series = [s for s in filtered_series if s.year == year]
210 |
211 | if downloaded is not None:
212 | filtered_series = [
213 | s for s in filtered_series
214 | if (s.statistics and s.statistics.episode_file_count > 0) == downloaded
215 | ]
216 |
217 | if watched is not None:
218 | if watched:
219 | filtered_series = [
220 | s for s in filtered_series
221 | if is_watched_series(s.title, False, self.config, self.sonarr_service)
222 | ]
223 | else:
224 | filtered_series = [
225 | s for s in filtered_series
226 | if not is_watched_series(s.title, False, self.config, self.sonarr_service)
227 | ]
228 |
229 | if actors:
230 | filtered_series = [
231 | s for s in filtered_series
232 | if s.data.get("credits") and any(
233 | actors.lower() in cast.get("name", "").lower()
234 | for cast in s.data.get("credits", {}).get("cast", [])
235 | )
236 | ]
237 |
238 | return {
239 | "count": len(filtered_series),
240 | "series": [
241 | {
242 | "id": s.id,
243 | "title": s.title,
244 | "year": s.year,
245 | "overview": s.overview,
246 | "status": s.status,
247 | "network": s.network,
248 | "genres": s.genres,
249 | "watched": is_watched_series(s.title, False, self.config, self.sonarr_service)
250 | }
251 | for s in filtered_series
252 | ]
253 | }
254 |
255 | @self.server.tool()
256 | def lookup_series(term: str) -> dict:
257 | service = SonarrService(self.config["sonarrConfig"])
258 | results = service.lookup_series(term)
259 | return {
260 | "count": len(results),
261 | "series": [
262 | {
263 | "id": s.id,
264 | "title": s.title,
265 | "year": s.year,
266 | "overview": s.overview
267 | }
268 | for s in results
269 | ]
270 | }
271 |
272 | # Similarly, for movies you can define a tool:
273 | @self.server.tool()
274 | def get_available_movies(year: Optional[int] = None,
275 | downloaded: Optional[bool] = None,
276 | watched: Optional[bool] = None,
277 | actors: Optional[str] = None) -> dict:
278 | """
279 | Get a list of all available movies with optional filters.
280 | Watched status is determined using Plex and/or Jellyfin.
281 | """
282 | # For movies, assume you have a function get_all_movies similar to get_all_series.
283 | from radarr_sonarr_mcp.services.radarr_service import RadarrService
284 | # You would need to instantiate a RadarrService and fetch movies.
285 | radarr_service = RadarrService(self.config["radarrConfig"])
286 | all_movies = radarr_service.get_all_movies() # Assuming this returns a list of dicts
287 | filtered_movies = all_movies
288 |
289 | if year is not None:
290 | filtered_movies = [m for m in filtered_movies if m.get("year") == year]
291 |
292 | if downloaded is not None:
293 | filtered_movies = [m for m in filtered_movies if m.get("hasFile") == downloaded]
294 |
295 | if watched is not None:
296 | if watched:
297 | filtered_movies = [
298 | m for m in filtered_movies
299 | if is_watched_movie(m.get("title", ""), self.config)
300 | ]
301 | else:
302 | filtered_movies = [
303 | m for m in filtered_movies
304 | if not is_watched_movie(m.get("title", ""), self.config)
305 | ]
306 |
307 | if actors:
308 | filtered_movies = [
309 | m for m in filtered_movies
310 | if m.get("credits") and any(
311 | actors.lower() in cast.get("name", "").lower()
312 | for cast in m.get("credits", {}).get("cast", [])
313 | )
314 | ]
315 |
316 | return {
317 | "count": len(filtered_movies),
318 | "movies": [
319 | {
320 | "id": m.get("id"),
321 | "title": m.get("title"),
322 | "year": m.get("year"),
323 | "overview": m.get("overview"),
324 | "hasFile": m.get("hasFile"),
325 | "status": m.get("status"),
326 | "genres": m.get("genres", []),
327 | "watched": is_watched_movie(m.get("title", ""), self.config)
328 | }
329 | for m in filtered_movies
330 | ]
331 | }
332 |
333 | def _register_resources(self):
334 | @self.server.resource("http://example.com/series", description="TV series collection from Sonarr")
335 | def series() -> dict:
336 | series_list = get_all_series(self.config)
337 | return {
338 | "count": len(series_list),
339 | "series": [
340 | {
341 | "id": s.id,
342 | "title": s.title,
343 | "year": s.year
344 | }
345 | for s in series_list
346 | ]
347 | }
348 | @self.server.resource("http://example.com/movies", description="Movie collection from Radarr")
349 | def movies() -> dict:
350 | from radarr_sonarr_mcp.services.radarr_service import RadarrService
351 | radarr_service = RadarrService(self.config["radarrConfig"])
352 | movies_list = radarr_service.get_all_movies() # Assuming list of dicts
353 | return {
354 | "count": len(movies_list),
355 | "movies": [
356 | {
357 | "id": m.get("id"),
358 | "title": m.get("title"),
359 | "year": m.get("year")
360 | }
361 | for m in movies_list
362 | ]
363 | }
364 |
365 | def run(self):
366 | port = self.config["server"]["port"]
367 | logger.info(f"Starting Radarr-Sonarr MCP Server on port {port}")
368 | logger.info(f"Connect Claude Desktop to: http://localhost:{port}")
369 | self.server.run()
370 |
371 | if __name__ == "__main__":
372 | server = RadarrSonarrMCP()
373 | server.run()
374 |
```