Plex MCP Server

Instructions

Implementation Reference

• src/plex_mcp/plex_mcp.py:218-287 (handler)

  Primary MCP tool handler for searching movies in Plex library. Decorated with @mcp.tool() for automatic registration. Handles parameter validation, filter construction, Plex search execution, error handling, and result formatting.
  @mcp.tool() async def search_movies( title: Optional[str] = None, year: Optional[int] = None, director: Optional[str] = None, studio: Optional[str] = None, genre: Optional[str] = None, actor: Optional[str] = None, rating: Optional[str] = None, country: Optional[str] = None, language: Optional[str] = None, watched: Optional[bool] = None, min_duration: Optional[int] = None, max_duration: Optional[int] = None, limit: Optional[int] = 5, ) -> str: """ Search for movies in your Plex library using optional filters. Parameters: title: Optional title or substring to match. year: Optional release year to filter by. director: Optional director name to filter by. studio: Optional studio name to filter by. genre: Optional genre tag to filter by. actor: Optional actor name to filter by. rating: Optional rating (e.g., "PG-13") to filter by. country: Optional country of origin to filter by. language: Optional audio or subtitle language to filter by. watched: Optional boolean; True returns only watched movies, False only unwatched. min_duration: Optional minimum duration in minutes. max_duration: Optional maximum duration in minutes. Returns: A formatted string of up to 5 matching movies (with a count of any additional results), or an error message if the search fails or no movies are found. """ # Validate the limit parameter limit = max(1, limit) if limit else 5 # Default to 5 if limit is 0 or negative params = MovieSearchParams( title, year, director, studio, genre, actor, rating, country, language, watched, min_duration, max_duration ) filters = params.to_filters() logger.info("Searching Plex with filters: %r", filters) try: plex = await get_plex_server() movies = await asyncio.to_thread(plex.library.search, **filters) except Exception as e: logger.exception("search_movies failed connecting to Plex") return f"ERROR: Could not search Plex. {e}" if not movies: return f"No movies found matching filters {filters!r}." logger.info("Found %d movies matching filters: %r", len(movies), filters) results: List[str] = [] for i, m in enumerate(movies[:limit], start=1): results.append(f"Result #{i}:\nKey: {m.ratingKey}\n{format_movie(m)}") if len(movies) > limit: results.append(f"\n... and {len(movies)-limit} more results.") return "\n---\n".join(results)
• src/plex_mcp/plex_mcp.py:129-179 (schema)

  Dataclass for input schema of search_movies tool. Defines all filter parameters and provides to_filters() method to map them to Plex API search arguments, handling type conversions and special logic (e.g., watched inversion, duration ms).
  @dataclass class MovieSearchParams: title: Optional[str] = None year: Optional[int] = None director: Optional[str] = None studio: Optional[str] = None genre: Optional[str] = None actor: Optional[str] = None rating: Optional[str] = None country: Optional[str] = None language: Optional[str] = None watched: Optional[bool] = None # True=only watched, False=only unwatched min_duration: Optional[int] = None # in minutes max_duration: Optional[int] = None # in minutes def to_filters(self) -> Dict[str, Any]: FIELD_MAP = { "title": "title", "year": "year", "director": "director", "studio": "studio", "genre": "genre", "actor": "actor", "rating": "rating", "country": "country", "language": "language", "watched": "unwatched", "min_duration": "minDuration", "max_duration": "maxDuration", } filters: Dict[str, Any] = {"libtype": "movie"} for field_name, plex_arg in FIELD_MAP.items(): value = getattr(self, field_name) if value is None: continue if field_name == "watched": # invert for Plex 'unwatched' flag filters["unwatched"] = not value continue if field_name in ("min_duration", "max_duration"): # convert minutes to milliseconds filters[plex_arg] = value * 60_000 continue filters[plex_arg] = value return filters
• src/plex_mcp/plex_mcp.py:33-60 (helper)

  Helper function used by search_movies to format each movie result into a readable string including title, year, rating, duration, studio, directors, actors, and summary.
  def format_movie(movie) -> str: """ Format a movie object into a human-readable string. Parameters: movie: A Plex movie object. Returns: A formatted string containing movie details. """ title = getattr(movie, 'title', 'Unknown Title') year = getattr(movie, 'year', 'Unknown Year') summary = getattr(movie, 'summary', 'No summary available') duration = getattr(movie, 'duration', 0) // 60000 if hasattr(movie, 'duration') else 0 rating = getattr(movie, 'rating', 'Unrated') studio = getattr(movie, 'studio', 'Unknown Studio') directors = [director.tag for director in getattr(movie, 'directors', [])[:3]] actors = [role.tag for role in getattr(movie, 'roles', [])[:5]] return ( f"Title: {title} ({year})\n" f"Rating: {rating}\n" f"Duration: {duration} minutes\n" f"Studio: {studio}\n" f"Directors: {', '.join(directors) if directors else 'Unknown'}\n" f"Starring: {', '.join(actors) if actors else 'Unknown'}\n" f"Summary: {summary}\n" )
• src/plex_mcp/plex_mcp.py:198-215 (helper)

  Helper utility called by search_movies to asynchronously retrieve the PlexServer instance, using a singleton PlexClient for connection management.
  async def get_plex_server() -> PlexServer: """ Asynchronously get a PlexServer instance via the singleton PlexClient. Returns: A PlexServer instance. Raises: Exception: When the Plex server connection fails. """ try: plex_client = get_plex_client() # Singleton accessor plex = await asyncio.to_thread(plex_client.get_server) return plex except Exception as e: logger.exception("Failed to get Plex server instance") raise e
• src/plex_mcp/__init__.py:1-27 (registration)

  Package __init__ re-exports search_movies and other tools, allowing imports like 'from plex_mcp import search_movies' as seen in tests.
  from .plex_mcp import ( search_movies, get_movie_details, list_playlists, get_playlist_items, create_playlist, delete_playlist, add_to_playlist, recent_movies, get_movie_genres, get_plex_server, MovieSearchParams, ) __all__ = [ "search_movies", "get_movie_details", "list_playlists", "get_playlist_items", "create_playlist", "delete_playlist", "add_to_playlist", "recent_movies", "get_movie_genres", "get_plex_server", "MovieSearchParams", ]