arr-assistant-mcp

Instructions

Implementation Reference

• src/arr_assistant_mcp/main.py:291-349 (handler)

  The handler function decorated with @mcp.tool, implementing the core logic for searching movies via Radarr API. Handles configuration checks, calls the helper API method, formats results using MediaSearchResult schema.

  @mcp.tool
  async def search_movies(title: str) -> Dict[str, Any]:
      """
      Search for movies by title using Radarr's built-in lookup.
      
      Args:
          title: Movie title only (e.g., "The Matrix" or "Primer")
      
      Returns:
          Dict with search results
      """
      logger.info(f"Searching for movies: '{title}'")
      
      if not config:
          error_msg = "Server not configured. Please set up Radarr API key."
          logger.error(error_msg)
          return {"error": error_msg, "results": []}
      
      if not config.radarr_api_key:
          error_msg = "Radarr API key not configured"
          logger.error(error_msg)
          return {"error": error_msg, "results": []}
      
      api = MediaServerAPI(config)
      
      try:
          logger.info(f"Searching Radarr for: {title}")
          radarr_results = await api.search_radarr_movies(title)
          logger.info(f"Radarr returned {len(radarr_results)} results")
          
          if not radarr_results:
              return {
                  "message": f"No movies found matching '{title}'",
                  "results": [],
                  "searched_query": title
              }
          
          results = []
          for movie in radarr_results[:10]:  # Show more results since we're not auto-adding
              result = MediaSearchResult(
                  title=movie.get("title", "Unknown"),
                  year=movie.get("year"),
                  overview=movie.get("overview", "No overview available"),
                  tmdb_id=movie.get("tmdbId"),
                  poster_path=movie.get("remotePoster"),
                  media_type="movie"
              )
              results.append(result)
          
          return {
              "results": [r.dict() for r in results],
              "total_found": len(results),
              "searched_query": title
          }
          
      except Exception as e:
          error_msg = f"Error during movie search: {str(e)}"
          logger.error(error_msg, exc_info=True)
          return {"error": error_msg, "results": []}

• src/arr_assistant_mcp/main.py:34-42 (schema)

  Pydantic BaseModel schema used to validate and structure individual movie search results returned by the search_movies tool.

  class MediaSearchResult(BaseModel):
      title: str
      year: Optional[int] = None
      overview: str
      tmdb_id: Optional[int] = None
      tvdb_id: Optional[int] = None
      poster_path: Optional[str] = None
      media_type: str  # "movie" or "tv"

• src/arr_assistant_mcp/main.py:93-123 (helper)

  Helper method in MediaServerAPI class that performs the HTTP request to Radarr's /api/v3/movie/lookup endpoint to fetch raw movie search results, which are then processed by the handler.

  async def search_radarr_movies(self, query: str) -> List[Dict[str, Any]]:
      """Search for movies using Radarr's built-in lookup (uses their TMDb access)"""
      url = f"{self.config.radarr_url}/api/v3/movie/lookup"
      headers = {"X-Api-Key": self.config.radarr_api_key}
      params = {"term": query}
      
      logger.info(f"Radarr lookup request: {url} with term='{query}'")
      
      try:
          response = await self.client.get(url, params=params, headers=headers)
          logger.info(f"Radarr response status: {response.status_code}")
          
          if response.status_code == 401:
              logger.error("Radarr authentication failed - check your API key")
              raise Exception("Radarr authentication failed - verify your API key is correct")
          elif response.status_code == 404:
              logger.error("Radarr lookup endpoint not found")
              raise Exception("Radarr lookup endpoint not found")
          
          response.raise_for_status()
          results = response.json()
          logger.info(f"Radarr returned {len(results)} results for query '{query}'")
          
          if results:
              logger.info(f"First result: {results[0].get('title')} ({results[0].get('year', 'No year')})")
          
          return results
      except Exception as e:
          logger.error(f"Radarr lookup error for query '{query}': {e}")
          raise e