OpenDota MCP Server

Instructions

Implementation Reference

• src/opendota_server/server.py:632-720 (handler)

  The primary handler function for the 'get_hero_stats' tool. It fetches hero statistics from the OpenDota API endpoint 'heroStats', optionally filters by hero_id, computes win rates across skill brackets and pro scene, and returns a formatted string summary.
  @mcp.tool() async def get_hero_stats(hero_id: Optional[int] = None) -> str: """Get statistics for heroes. Args: hero_id: Optional hero ID to get stats for a specific hero Returns: Hero statistics including win rates by skill bracket """ hero_stats = await make_opendota_request("heroStats") if "error" in hero_stats: return f"Error retrieving hero stats: {hero_stats['error']}" if hero_id is not None: # Filter for specific hero hero_stats = [ hero for hero in hero_stats if hero.get("id") == hero_id or hero.get("hero_id") == hero_id ] if not hero_stats: return f"No stats found for hero ID {hero_id}." hero = hero_stats[0] localized_name = hero.get("localized_name", f"Hero {hero_id}") # Calculate win rates by bracket brackets = [ "herald", "guardian", "crusader", "archon", "legend", "ancient", "divine", "immortal", ] bracket_stats = [] for i, bracket in enumerate(brackets, 1): picks = hero.get(f"{i}_pick", 0) wins = hero.get(f"{i}_win", 0) win_rate = (wins / picks * 100) if picks > 0 else 0 bracket_stats.append( f"{bracket.capitalize()}: {win_rate:.2f}% ({wins}/{picks})" ) # Pro stats pro_picks = hero.get("pro_pick", 0) pro_wins = hero.get("pro_win", 0) pro_win_rate = (pro_wins / pro_picks * 100) if pro_picks > 0 else 0 pro_ban_rate = hero.get("pro_ban", 0) # Hero attributes roles = hero.get("roles", []) primary_attr = hero.get("primary_attr", "Unknown") attack_type = hero.get("attack_type", "Unknown") return ( f"Hero Stats for {localized_name} (ID: {hero_id}):\n\n" f"Roles: {', '.join(roles)}\n" f"Primary Attribute: {primary_attr}\n" f"Attack Type: {attack_type}\n\n" f"Win Rates by Bracket:\n" f"{', '.join(bracket_stats)}\n\n" f"Pro Scene:\n" f"Pick Rate: {pro_picks} picks\n" f"Win Rate: {pro_win_rate:.2f}% ({pro_wins}/{pro_picks})\n" f"Ban Rate: {pro_ban_rate} bans" ) else: # Return summary of all heroes formatted_heroes = [] for hero in sorted(hero_stats, key=lambda x: x.get("localized_name", "")): localized_name = hero.get("localized_name", f"Hero {hero.get('id')}") # Calculate overall win rate total_picks = sum(hero.get(f"{i}_pick", 0) for i in range(1, 9)) total_wins = sum(hero.get(f"{i}_win", 0) for i in range(1, 9)) win_rate = (total_wins / total_picks * 100) if total_picks > 0 else 0 formatted_heroes.append(f"{localized_name}: {win_rate:.2f}% win rate") return "Hero Win Rates:\n\n" + "\n".join(formatted_heroes)
• src/opendota_server/server.py:632-632 (registration)

  The @mcp.tool() decorator registers the get_hero_stats function as an MCP tool with FastMCP.
  @mcp.tool()
• src/opendota_server/server.py:633-639 (schema)

  Function signature and docstring define the input schema (optional hero_id: int) and output (str description). Schema is inferred by FastMCP from type hints and docstring.
  async def get_hero_stats(hero_id: Optional[int] = None) -> str: """Get statistics for heroes. Args: hero_id: Optional hero ID to get stats for a specific hero Returns:
• src/opendota_server/server.py:134-196 (helper)

  Core helper function used by get_hero_stats to make API requests to OpenDota 'heroStats' endpoint with caching and rate limiting.
  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)}"}