OpenDota MCP Server

Instructions

Implementation Reference

• src/opendota_server/server.py:722-758 (handler)

  The main handler function for the 'search_player' tool. It is decorated with @mcp.tool() which handles both implementation and registration in FastMCP. Calls OpenDota API search endpoint and formats top 10 matching players with account IDs and similarity scores.
  @mcp.tool() async def search_player(query: str) -> str: """Search for players by name. Args: query: Name to search for Returns: List of matching players """ search_results = await make_opendota_request("search", {"q": query}) if "error" in search_results: return f"Error searching for players: {search_results['error']}" if not search_results or len(search_results) == 0: return f"No players found matching '{query}'." formatted_results = [] # Limit to 10 players players_to_show = [] if isinstance(search_results, list): players_to_show = search_results[:10] for i, player in enumerate(players_to_show): account_id = player.get("account_id", "Unknown") name = player.get("personaname", "Anonymous") similarity = player.get("similarity", 0) formatted_results.append( f"{i+1}. {name}\n" f" Account ID: {account_id}\n" f" Similarity: {similarity:.2f}" ) return f"Players matching '{query}':\n\n" + "\n\n".join(formatted_results)
• src/opendota_server/server.py:134-196 (helper)

  Core helper function used by search_player to make HTTP requests to OpenDota API with rate limiting, caching, and error handling. Called as make_opendota_request("search", {"q": query}).
  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)}"}