OpenDota MCP Server

Instructions

Implementation Reference

• src/opendota_server/server.py:760-806 (handler)

  The handler function implementing the get_pro_players tool. Decorated with @mcp.tool() for automatic registration in FastMCP. Fetches pro players from OpenDota API, sorts them, formats output with name, team, country, and account ID.
  @mcp.tool() async def get_pro_players(limit: int = 10) -> str: """Get list of professional players. Args: limit: Number of players to retrieve (default: 10) Returns: List of professional players """ if limit > 30: limit = 30 # Cap for reasonable response size pro_players = await make_opendota_request("proPlayers") if "error" in pro_players: return f"Error retrieving pro players: {pro_players['error']}" if not pro_players or not isinstance(pro_players, list) or len(pro_players) == 0: return "No professional players found." # Sort by name for consistency sorted_players = sorted( pro_players, key=lambda x: ( x.get("team_name", ""), x.get("name", ""), x.get("account_id", 0), ), ) formatted_players = [] for i, player in enumerate(sorted_players[:limit]): account_id = player.get("account_id", "Unknown") name = player.get("name", "Anonymous") team_name = player.get("team_name", "No Team") country_code = player.get("country_code", "Unknown") formatted_players.append( f"{i+1}. {name}\n" f" Team: {team_name}\n" f" Country: {country_code}\n" f" Account ID: {account_id}" ) return "Professional Players:\n\n" + "\n\n".join(formatted_players)
• src/opendota_server/server.py:134-196 (helper)

  Key helper function used by get_pro_players (and other tools) to make HTTP requests to the OpenDota API endpoint "proPlayers", including caching, rate limiting, and error handling.
  async def make_opendota_request( endpoint: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """Make a request to the OpenDota API with proper error handling and caching.""" # Apply rate limiting await apply_rate_limit() url = f"{OPENDOTA_API_BASE}/{endpoint}" request_params = API_PARAMS.copy() if params: request_params.update(params) # Create a cache key manually cache_key = endpoint if request_params: param_str = "&".join(f"{k}={v}" for k, v in sorted(request_params.items())) cache_key = f"{endpoint}?{param_str}" # Check cache cache_entry = api_cache.get(cache_key) if cache_entry: timestamp, data = cache_entry if time.time() - timestamp < CACHE_TTL: logger.debug(f"Cache hit for {cache_key}") return data logger.info(f"Making request to {endpoint} with params {request_params}") async with httpx.AsyncClient() as client: try: response = await client.get( url, params=request_params, headers={"User-Agent": USER_AGENT}, timeout=10.0, ) response.raise_for_status() data = response.json() # Cache the response api_cache[cache_key] = (time.time(), data) return data except httpx.HTTPStatusError as e: if e.response.status_code == 429: logger.error(f"Rate limit exceeded for {endpoint}") return { "error": "Rate limit exceeded. Consider using an API key for more requests." } if e.response.status_code == 404: logger.error(f"Resource not found: {endpoint}") return {"error": "Not found. The requested resource doesn't exist."} if e.response.status_code >= 500: logger.error(f"OpenDota API server error: {e.response.status_code}") return {"error": "OpenDota API server error. Please try again later."} logger.error( f"HTTP error {e.response.status_code} for {endpoint}: {e.response.text}" ) return {"error": f"HTTP error {e.response.status_code}: {e.response.text}"} except Exception as e: logger.error(f"Unexpected error for {endpoint}: {str(e)}") return {"error": f"Unexpected error: {str(e)}"}