imdb-mcp-server

Instructions

Implementation Reference

• src/imdb_mcp_server/tools.py:231-243 (handler)

  The core handler function decorated with @mcp.tool() that fetches most popular movies from IMDb API endpoint, handles errors, paginates results, and returns JSON.

  @mcp.tool()
  async def get_most_popular_movies(start: int, ctx: Context) -> str:
      """Get the most popular movies from IMDb with pagination.
      Args:
          start: The starting index (0-based) to retrieve movies from.
      Returns:
          JSON object containing 5 most popular movies starting from the specified index.
      """
      most_popular_movies_url = f"{BASE_URL}/most-popular-movies"
      most_popular_movies_data = await make_imdb_request(most_popular_movies_url, {}, ctx) 
      if not most_popular_movies_data:
          return "Unable to fetch most popular movies data."
      return json.dumps(paginated_response(most_popular_movies_data, start, len(most_popular_movies_data)), indent=4)

• src/imdb_mcp_server/api.py:58-77 (helper)

  Helper utility function used by the tool to format the API response into a standard paginated JSON structure with page size 5.

  def paginated_response(items, start, total_count=None):
      """Format a paginated response with a fixed page size of 5."""
      if total_count is None:
          total_count = len(items)
      
      # Validate starting index
      start = max(0, min(total_count - 1 if total_count > 0 else 0, start))
      
      # Fixed page size of 5
      page_size = 5
      end = min(start + page_size, total_count)
      
      return {
          "items": items[start:end],
          "start": start,
          "count": end - start,
          "totalCount": total_count,
          "hasMore": end < total_count,
          "nextStart": end if end < total_count else None
      }

• src/imdb_mcp_server/api.py:14-56 (helper)

  Core helper function that makes HTTP requests to the IMDb RapidAPI, handles caching, API key management, and errors.

  async def make_imdb_request(url: str, querystring: dict[str, Any], ctx: Optional[Context] = None) -> Optional[Dict[str, Any]]:
      """Make a request to the IMDb API with proper error handling and caching."""
      
      # Check if it's time to clean the cache
      cache_manager.cleanup_if_needed()
      
      # Create a cache key from the URL and querystring
      cache_key = f"{url}_{str(querystring)}"
      
      # Try to get from cache first
      cached_data = cache_manager.cache.get(cache_key)
      if cached_data:
          return cached_data
      
      # Get API key from session config or fallback to environment variable
      api_key = None
      if ctx and hasattr(ctx, 'session_config') and ctx.session_config:
          api_key = ctx.session_config.rapidApiKeyImdb
      
      if not api_key:
          api_key = os.getenv("RAPID_API_KEY_IMDB")
      
      # Not in cache, make the request
      headers = {
          "x-rapidapi-key": api_key,
          "x-rapidapi-host": "imdb236.p.rapidapi.com",
      }
      
      if not api_key:
          raise ValueError("API key not found. Please set the RAPID_API_KEY_IMDB environment variable or provide rapidApiKeyImdb in the request.")
      
      try:
          response = requests.get(url, headers=headers, params=querystring, timeout=30.0)
          response.raise_for_status()
          data = response.json()
          
          # Cache the response
          cache_manager.cache.set(cache_key, data)
              
          return data
      except Exception as e:
          raise ValueError(f"Unable to fetch data from IMDb. Please try again later. Error: {e}")

• src/imdb_mcp_server/main.py:19-25 (registration)

  Server creation and tool registration via register_tools(server), which defines and registers the get_most_popular_movies handler among others.

  # Create your FastMCP server as usual
  server = FastMCP("IMDb MCP Server")
  
  # Register all tools with the server
  register_tools(server)
  
  return server

• src/imdb_mcp_server/main.py:11-13 (schema)

  Pydantic schema for server configuration, providing the required rapidApiKeyImdb for API requests.

  class ConfigSchema(BaseModel):
      rapidApiKeyImdb: str = Field(..., description="RapidAPI API key for accessing the IMDb API")