imdb-mcp-server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/imdb_mcp_server/tools.py:360-372 (handler)

  The primary handler function for the 'get_trending_telugu_movies' tool. It is decorated with @mcp.tool() for automatic registration and includes input schema via type annotations and docstring. Fetches data from the IMDb API endpoint for trending Telugu movies, handles empty response, paginates results using a helper, and returns formatted JSON.
  @mcp.tool() async def get_trending_telugu_movies(start: int, ctx: Context) -> str: """Get the trending Telugu movies on IMDb. Args: start: The starting index (0-based) to retrieve movies from. Returns: JSON object containing 5 trending Telugu movies starting from the specified index. """ trending_telugu_movies_url = f"{BASE_URL}/india/trending-telugu" trending_telugu_movies_data = await make_imdb_request(trending_telugu_movies_url, {}, ctx) if not trending_telugu_movies_data: return "Unable to fetch trending Telugu movies data." return json.dumps(paginated_response(trending_telugu_movies_data, start, len(trending_telugu_movies_data)), indent=4)
• src/imdb_mcp_server/api.py:14-56 (helper)

  Supporting utility function that performs the HTTP request to the IMDb RapidAPI endpoint, handles authentication via MCP context or env var, implements caching, and error handling. Directly called by the tool handler.
  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/api.py:58-77 (helper)

  Utility function to create a standardized paginated JSON response structure with a fixed page size of 5 items, including metadata like totalCount and hasMore. Used by the tool handler to format the API response.
  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:11-11 (helper)

  Constant defining the base URL for all IMDb API requests, used to construct the specific endpoint URL in the tool handler.
  BASE_URL = "https://imdb236.p.rapidapi.com/api/imdb"
• src/imdb_mcp_server/main.py:22-23 (registration)

  The call to register_tools(server) in the MCP server setup, which executes the definition and registration of all tools including 'get_trending_telugu_movies' via decorators.
  # Register all tools with the server register_tools(server)