the-nexus/mcp_servers/steam_info_server.py

#!/usr/bin/env python3
"""
MCP Server for Steam Information
Provides Steam Web API integration for game data.
Uses stdio JSON-RPC for MCP protocol.
"""

import json
import sys
import logging
import os
import urllib.request
import urllib.error
from typing import Any, Dict, List, Optional

# Set up logging to stderr (stdout is for JSON-RPC)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    stream=sys.stderr
)
logger = logging.getLogger('steam-info-mcp')

# Steam API configuration
STEAM_API_BASE = "https://api.steampowered.com"
STEAM_API_KEY = os.environ.get('STEAM_API_KEY', '')

# Bannerlord App ID for convenience
BANNERLORD_APP_ID = "261550"


class SteamInfoMCPServer:
    """MCP Server providing Steam information capabilities."""
    
    def __init__(self):
        self.tools = self._define_tools()
        self.mock_mode = not STEAM_API_KEY
        if self.mock_mode:
            logger.warning("No STEAM_API_KEY found - running in mock mode")
    
    def _define_tools(self) -> List[Dict[str, Any]]:
        """Define the available tools for this MCP server."""
        return [
            {
                "name": "steam_recently_played",
                "description": "Get recently played games for a Steam user",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "user_id": {
                            "type": "string",
                            "description": "Steam User ID (64-bit SteamID)"
                        },
                        "count": {
                            "type": "integer",
                            "description": "Number of games to return",
                            "default": 10
                        }
                    },
                    "required": ["user_id"]
                }
            },
            {
                "name": "steam_player_achievements",
                "description": "Get achievement data for a player and game",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "user_id": {
                            "type": "string",
                            "description": "Steam User ID (64-bit SteamID)"
                        },
                        "app_id": {
                            "type": "string",
                            "description": "Steam App ID of the game"
                        }
                    },
                    "required": ["user_id", "app_id"]
                }
            },
            {
                "name": "steam_user_stats",
                "description": "Get user statistics for a specific game",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "user_id": {
                            "type": "string",
                            "description": "Steam User ID (64-bit SteamID)"
                        },
                        "app_id": {
                            "type": "string",
                            "description": "Steam App ID of the game"
                        }
                    },
                    "required": ["user_id", "app_id"]
                }
            },
            {
                "name": "steam_current_players",
                "description": "Get current number of players for a game",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "app_id": {
                            "type": "string",
                            "description": "Steam App ID of the game"
                        }
                    },
                    "required": ["app_id"]
                }
            },
            {
                "name": "steam_news",
                "description": "Get news articles for a game",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "app_id": {
                            "type": "string",
                            "description": "Steam App ID of the game"
                        },
                        "count": {
                            "type": "integer",
                            "description": "Number of news items to return",
                            "default": 5
                        }
                    },
                    "required": ["app_id"]
                }
            },
            {
                "name": "steam_app_details",
                "description": "Get detailed information about a Steam app",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "app_id": {
                            "type": "string",
                            "description": "Steam App ID"
                        }
                    },
                    "required": ["app_id"]
                }
            }
        ]
    
    def _make_steam_api_request(self, endpoint: str, params: Dict[str, str]) -> Dict[str, Any]:
        """Make a request to the Steam Web API."""
        if self.mock_mode:
            raise Exception("Steam API key not configured - running in mock mode")
        
        # Add API key to params
        params['key'] = STEAM_API_KEY
        
        # Build query string
        query = '&'.join(f"{k}={urllib.parse.quote(str(v))}" for k, v in params.items())
        url = f"{STEAM_API_BASE}/{endpoint}?{query}"
        
        try:
            with urllib.request.urlopen(url, timeout=10) as response:
                data = json.loads(response.read().decode('utf-8'))
                return data
        except urllib.error.HTTPError as e:
            logger.error(f"HTTP Error {e.code}: {e.reason}")
            raise Exception(f"Steam API HTTP error: {e.code}")
        except urllib.error.URLError as e:
            logger.error(f"URL Error: {e.reason}")
            raise Exception(f"Steam API connection error: {e.reason}")
        except json.JSONDecodeError as e:
            logger.error(f"JSON decode error: {e}")
            raise Exception("Invalid response from Steam API")
    
    def _get_mock_data(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """Return mock data for testing without API key."""
        app_id = params.get("app_id", BANNERLORD_APP_ID)
        user_id = params.get("user_id", "123456789")
        
        if method == "steam_recently_played":
            return {
                "mock": True,
                "user_id": user_id,
                "total_count": 3,
                "games": [
                    {
                        "appid": 261550,
                        "name": "Mount & Blade II: Bannerlord",
                        "playtime_2weeks": 1425,
                        "playtime_forever": 15230,
                        "img_icon_url": "mock_icon_url"
                    },
                    {
                        "appid": 730,
                        "name": "Counter-Strike 2",
                        "playtime_2weeks": 300,
                        "playtime_forever": 5000,
                        "img_icon_url": "mock_icon_url"
                    }
                ]
            }
        elif method == "steam_player_achievements":
            return {
                "mock": True,
                "player_id": user_id,
                "game_name": "Mock Game",
                "achievements": [
                    {"apiname": "achievement_1", "achieved": 1, "unlocktime": 1700000000},
                    {"apiname": "achievement_2", "achieved": 0},
                    {"apiname": "achievement_3", "achieved": 1, "unlocktime": 1700100000}
                ],
                "success": True
            }
        elif method == "steam_user_stats":
            return {
                "mock": True,
                "player_id": user_id,
                "game_id": app_id,
                "stats": [
                    {"name": "kills", "value": 1250},
                    {"name": "deaths", "value": 450},
                    {"name": "wins", "value": 89}
                ],
                "achievements": [
                    {"name": "first_victory", "achieved": 1}
                ]
            }
        elif method == "steam_current_players":
            return {
                "mock": True,
                "app_id": app_id,
                "player_count": 15432,
                "result": 1
            }
        elif method == "steam_news":
            return {
                "mock": True,
                "appid": app_id,
                "newsitems": [
                    {
                        "gid": "12345",
                        "title": "Major Update Released!",
                        "url": "https://steamcommunity.com/games/261550/announcements/detail/mock",
                        "author": "Developer",
                        "contents": "This is a mock news item for testing purposes.",
                        "feedlabel": "Product Update",
                        "date": 1700000000
                    },
                    {
                        "gid": "12346",
                        "title": "Patch Notes 1.2.3",
                        "url": "https://steamcommunity.com/games/261550/announcements/detail/mock2",
                        "author": "Developer",
                        "contents": "Bug fixes and improvements.",
                        "feedlabel": "Patch Notes",
                        "date": 1699900000
                    }
                ],
                "count": 2
            }
        elif method == "steam_app_details":
            return {
                "mock": True,
                app_id: {
                    "success": True,
                    "data": {
                        "type": "game",
                        "name": "Mock Game Title",
                        "steam_appid": int(app_id),
                        "required_age": 0,
                        "is_free": False,
                        "detailed_description": "This is a mock description.",
                        "about_the_game": "About the mock game.",
                        "short_description": "A short mock description.",
                        "developers": ["Mock Developer"],
                        "publishers": ["Mock Publisher"],
                        "genres": [{"id": "1", "description": "Action"}],
                        "release_date": {"coming_soon": False, "date": "1 Jan, 2024"}
                    }
                }
            }
        return {"mock": True, "message": "Unknown method"}
    
    def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handle the initialize request."""
        logger.info("Received initialize request")
        return {
            "protocolVersion": "2024-11-05",
            "serverInfo": {
                "name": "steam-info-mcp",
                "version": "1.0.0"
            },
            "capabilities": {
                "tools": {}
            }
        }
    
    def handle_tools_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handle the tools/list request."""
        return {"tools": self.tools}
    
    def handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handle the tools/call request."""
        tool_name = params.get("name", "")
        arguments = params.get("arguments", {})
        
        logger.info(f"Tool call: {tool_name} with args: {arguments}")
        
        try:
            result = self._execute_tool(tool_name, arguments)
            return {
                "content": [
                    {
                        "type": "text",
                        "text": json.dumps(result)
                    }
                ],
                "isError": False
            }
        except Exception as e:
            logger.error(f"Error executing tool {tool_name}: {e}")
            return {
                "content": [
                    {
                        "type": "text",
                        "text": json.dumps({"error": str(e)})
                    }
                ],
                "isError": True
            }
    
    def _execute_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the specified tool with the given arguments."""
        if self.mock_mode:
            logger.info(f"Returning mock data for {name}")
            return self._get_mock_data(name, args)
        
        # Real Steam API calls (when API key is configured)
        if name == "steam_recently_played":
            user_id = args.get("user_id")
            count = args.get("count", 10)
            data = self._make_steam_api_request(
                "IPlayerService/GetRecentlyPlayedGames/v1",
                {"steamid": user_id, "count": str(count)}
            )
            return data.get("response", {})
        
        elif name == "steam_player_achievements":
            user_id = args.get("user_id")
            app_id = args.get("app_id")
            data = self._make_steam_api_request(
                "ISteamUserStats/GetPlayerAchievements/v1",
                {"steamid": user_id, "appid": app_id}
            )
            return data.get("playerstats", {})
        
        elif name == "steam_user_stats":
            user_id = args.get("user_id")
            app_id = args.get("app_id")
            data = self._make_steam_api_request(
                "ISteamUserStats/GetUserStatsForGame/v2",
                {"steamid": user_id, "appid": app_id}
            )
            return data.get("playerstats", {})
        
        elif name == "steam_current_players":
            app_id = args.get("app_id")
            data = self._make_steam_api_request(
                "ISteamUserStats/GetNumberOfCurrentPlayers/v1",
                {"appid": app_id}
            )
            return data.get("response", {})
        
        elif name == "steam_news":
            app_id = args.get("app_id")
            count = args.get("count", 5)
            data = self._make_steam_api_request(
                "ISteamNews/GetNewsForApp/v2",
                {"appid": app_id, "count": str(count), "maxlength": "300"}
            )
            return data.get("appnews", {})
        
        elif name == "steam_app_details":
            app_id = args.get("app_id")
            # App details uses a different endpoint
            url = f"https://store.steampowered.com/api/appdetails?appids={app_id}"
            try:
                with urllib.request.urlopen(url, timeout=10) as response:
                    data = json.loads(response.read().decode('utf-8'))
                    return data
            except Exception as e:
                raise Exception(f"Failed to fetch app details: {e}")
        
        else:
            raise ValueError(f"Unknown tool: {name}")
    
    def process_request(self, request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Process an MCP request and return the response."""
        method = request.get("method", "")
        params = request.get("params", {})
        req_id = request.get("id")
        
        if method == "initialize":
            result = self.handle_initialize(params)
        elif method == "tools/list":
            result = self.handle_tools_list(params)
        elif method == "tools/call":
            result = self.handle_tools_call(params)
        else:
            # Unknown method
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "error": {
                    "code": -32601,
                    "message": f"Method not found: {method}"
                }
            }
        
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": result
        }


def main():
    """Main entry point for the MCP server."""
    logger.info("Steam Info MCP Server starting...")
    
    if STEAM_API_KEY:
        logger.info("Steam API key configured - using live API")
    else:
        logger.warning("No STEAM_API_KEY found - running in mock mode")
    
    server = SteamInfoMCPServer()
    
    # Check if running in a TTY (for testing)
    if sys.stdin.isatty():
        logger.info("Running in interactive mode (for testing)")
        print("Steam Info MCP Server", file=sys.stderr)
        print("Enter JSON-RPC requests (one per line):", file=sys.stderr)
    
    try:
        while True:
            # Read line from stdin
            line = sys.stdin.readline()
            if not line:
                break
            
            line = line.strip()
            if not line:
                continue
            
            try:
                request = json.loads(line)
                response = server.process_request(request)
                if response:
                    print(json.dumps(response), flush=True)
            except json.JSONDecodeError as e:
                logger.error(f"Invalid JSON: {e}")
                error_response = {
                    "jsonrpc": "2.0",
                    "id": None,
                    "error": {
                        "code": -32700,
                        "message": "Parse error"
                    }
                }
                print(json.dumps(error_response), flush=True)
    
    except KeyboardInterrupt:
        logger.info("Received keyboard interrupt, shutting down...")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
    
    logger.info("Steam Info MCP Server stopped.")


if __name__ == "__main__":
    main()