diff --git a/CHANGELOG.md b/CHANGELOG.md index 84594b3..a4de0cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **Game Server Status** (`game_servers.py`) + - `get_game_servers` - List game servers for a specific game + - Filter by map, player count, VAC status + - Returns server name, address, map, player count, VAC status + - `query_server_status` - Get detailed status of a specific server + - Lookup by IP:port address + - Returns full server info including OS, version, game type + +### Changed +- Tool count increased from 36 to 38 + ## [v0.9.0] - 2025-12-12 ### Added diff --git a/README.md b/README.md index 163b30e..08f77eb 100644 --- a/README.md +++ b/README.md @@ -17,8 +17,9 @@ Once set up, you can ask Claude things like: - "What's the latest news for Team Fortress 2?" - "Show my pending trade offers" - "What's the current price for an AK-47 Redline?" +- "Show me CS2 servers with players on de_dust2" -The server includes 36 tools covering player profiles, game libraries, achievements, stats, reviews, wishlists, news, community guides, trading/market data, family sharing, and Steam Workshop. +The server includes 38 tools covering player profiles, game libraries, achievements, stats, reviews, wishlists, news, community guides, trading/market data, family sharing, Steam Workshop, and game server status. --- @@ -179,7 +180,7 @@ No registration needed - just drop in the file and restart. --- -## Available Tools (36 total) +## Available Tools (38 total) ### Player Profiles (ISteamUser) - 6 tools @@ -269,6 +270,13 @@ No registration needed - just drop in the file and restart. | `get_workshop_item_details` | Get full details on a Workshop item (description, subscribers, dependencies) | | `get_workshop_collection` | Get items from a Workshop collection | +### Game Servers (IGameServersService) - 2 tools + +| Tool | What it does | +|------|--------------| +| `get_game_servers` | List game servers for a game with filters (map, players, region) | +| `query_server_status` | Get detailed status of a specific server by address | + --- ## Configuration Reference diff --git a/src/steam_mcp/endpoints/game_servers.py b/src/steam_mcp/endpoints/game_servers.py new file mode 100644 index 0000000..22aa414 --- /dev/null +++ b/src/steam_mcp/endpoints/game_servers.py @@ -0,0 +1,240 @@ +"""IGameServersService API endpoints. + +This module provides MCP tools for querying game server information, +including server lists and detailed server status. + +Reference: https://partner.steamgames.com/doc/webapi/IGameServersService +""" + +import json +from typing import Any + +from steam_mcp.endpoints.base import BaseEndpoint, endpoint + + +class IGameServersService(BaseEndpoint): + """IGameServersService API endpoints for game server queries.""" + + @endpoint( + name="get_game_servers", + description=( + "Get a list of game servers for a specific game. " + "Returns server name, map, player count, and address." + ), + supports_json=True, + params={ + "app_id": { + "type": "integer", + "description": "Game App ID (e.g., 730 for CS2, 440 for TF2)", + "required": True, + }, + "filter": { + "type": "string", + "description": ( + "Server filter string. Examples: " + "'\\\\gamedir\\\\tf' for TF2 servers, " + "'\\\\map\\\\de_dust2' for specific map, " + "'\\\\noplayers\\\\1' for empty servers, " + "'\\\\full\\\\1' for full servers. " + "Combine with '\\\\' separator." + ), + "required": False, + }, + "limit": { + "type": "integer", + "description": "Maximum number of servers to return (default: 25, max: 100)", + "required": False, + "default": 25, + "minimum": 1, + "maximum": 100, + }, + }, + ) + async def get_game_servers( + self, + app_id: int, + filter: str | None = None, + limit: int = 25, + format: str = "text", + ) -> str: + """Get game servers for a specific app.""" + # Build the filter string - always include appid + filter_parts = [f"\\appid\\{app_id}"] + if filter: + filter_parts.append(filter) + filter_str = "".join(filter_parts) + + result = await self.client.get( + "IGameServersService", + "GetServerList", + version=1, + params={"filter": filter_str, "limit": limit}, + ) + + response = result.get("response", {}) + servers = response.get("servers", []) + + if not servers: + msg = f"No servers found for App ID {app_id}" + if filter: + msg += f" with filter '{filter}'" + if format == "json": + return json.dumps({"error": msg, "app_id": app_id, "servers": []}) + return msg + + if format == "json": + data = { + "app_id": app_id, + "server_count": len(servers), + "servers": [ + { + "name": s.get("name", "Unknown"), + "address": s.get("addr", "Unknown"), + "map": s.get("map", "Unknown"), + "players": s.get("players", 0), + "max_players": s.get("max_players", 0), + "bots": s.get("bots", 0), + "game_type": s.get("gametype", ""), + "secure": s.get("secure", False), + "dedicated": s.get("dedicated", True), + "os": s.get("os", ""), + "version": s.get("version", ""), + } + for s in servers + ], + } + return json.dumps(data, indent=2) + + # Text format + output = [ + f"Game Servers for App ID {app_id}", + f"Found {len(servers)} server(s)", + "", + ] + + for server in servers: + name = server.get("name", "Unknown Server") + addr = server.get("addr", "Unknown") + map_name = server.get("map", "Unknown") + players = server.get("players", 0) + max_players = server.get("max_players", 0) + bots = server.get("bots", 0) + secure = "VAC" if server.get("secure") else "No VAC" + + player_info = f"{players}/{max_players}" + if bots > 0: + player_info += f" ({bots} bots)" + + output.append(f" {name}") + output.append(f" Address: {addr}") + output.append(f" Map: {map_name} | Players: {player_info} | {secure}") + output.append("") + + return "\n".join(output) + + @endpoint( + name="query_server_status", + description=( + "Get detailed status of a specific game server by address. " + "Returns server info, current players, and game rules." + ), + supports_json=True, + params={ + "server_address": { + "type": "string", + "description": "Server address in IP:port format (e.g., '192.168.1.1:27015')", + "required": True, + }, + }, + ) + async def query_server_status( + self, + server_address: str, + format: str = "text", + ) -> str: + """Query detailed status of a specific server.""" + # Use the filter to find the specific server + result = await self.client.get( + "IGameServersService", + "GetServerList", + version=1, + params={"filter": f"\\addr\\{server_address}", "limit": 1}, + ) + + response = result.get("response", {}) + servers = response.get("servers", []) + + if not servers: + msg = f"Server not found at address: {server_address}" + if format == "json": + return json.dumps({"error": msg, "address": server_address}) + return msg + + server = servers[0] + + if format == "json": + return json.dumps( + { + "address": server.get("addr", server_address), + "name": server.get("name", "Unknown"), + "app_id": server.get("appid"), + "game_dir": server.get("gamedir", ""), + "map": server.get("map", "Unknown"), + "players": server.get("players", 0), + "max_players": server.get("max_players", 0), + "bots": server.get("bots", 0), + "game_type": server.get("gametype", ""), + "secure": server.get("secure", False), + "dedicated": server.get("dedicated", True), + "os": server.get("os", ""), + "version": server.get("version", ""), + "product": server.get("product", ""), + "region": server.get("region", -1), + "steamid": server.get("steamid", ""), + }, + indent=2, + ) + + # Text format + name = server.get("name", "Unknown Server") + addr = server.get("addr", server_address) + map_name = server.get("map", "Unknown") + players = server.get("players", 0) + max_players = server.get("max_players", 0) + bots = server.get("bots", 0) + app_id = server.get("appid", "Unknown") + game_dir = server.get("gamedir", "Unknown") + version = server.get("version", "Unknown") + secure = "Yes" if server.get("secure") else "No" + dedicated = "Dedicated" if server.get("dedicated") else "Listen" + os_type = server.get("os", "Unknown") + game_type = server.get("gametype", "None") + + # Map OS codes + os_names = {"l": "Linux", "w": "Windows", "m": "macOS", "o": "macOS"} + os_display = os_names.get(os_type.lower(), os_type) if os_type else "Unknown" + + output = [ + f"Server Status: {name}", + f"", + f" Address: {addr}", + f" App ID: {app_id}", + f" Game: {game_dir}", + f" Map: {map_name}", + f" Players: {players}/{max_players}", + ] + + if bots > 0: + output.append(f" Bots: {bots}") + + output.extend([ + f" Version: {version}", + f" Server Type: {dedicated}", + f" OS: {os_display}", + f" VAC Secured: {secure}", + ]) + + if game_type: + output.append(f" Game Type: {game_type}") + + return "\n".join(output) diff --git a/tests/test_game_servers.py b/tests/test_game_servers.py new file mode 100644 index 0000000..daa8cb1 --- /dev/null +++ b/tests/test_game_servers.py @@ -0,0 +1,365 @@ +"""Tests for IGameServersService endpoint.""" + +import json +import pytest +from unittest.mock import AsyncMock, MagicMock + +from steam_mcp.endpoints.game_servers import IGameServersService + + +@pytest.fixture +def mock_client(): + """Create mock Steam client.""" + client = MagicMock() + client.owner_steam_id = None + client.get = AsyncMock() + return client + + +@pytest.fixture +def game_servers(mock_client): + """Create IGameServersService instance with mock client.""" + return IGameServersService(mock_client) + + +# --- get_game_servers Tests --- + + +class TestGetGameServers: + """Tests for get_game_servers endpoint.""" + + @pytest.mark.asyncio + async def test_returns_server_list_text_format(self, game_servers, mock_client): + """Should return formatted server list in text format.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Test Server", + "addr": "192.168.1.1:27015", + "map": "de_dust2", + "players": 10, + "max_players": 24, + "bots": 0, + "secure": True, + } + ] + } + } + + result = await game_servers.get_game_servers(app_id=730) + + assert "Test Server" in result + assert "192.168.1.1:27015" in result + assert "de_dust2" in result + assert "10/24" in result + assert "VAC" in result + + @pytest.mark.asyncio + async def test_returns_server_list_json_format(self, game_servers, mock_client): + """Should return structured server list in JSON format.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Test Server", + "addr": "192.168.1.1:27015", + "map": "ctf_2fort", + "players": 5, + "max_players": 32, + "bots": 2, + "secure": False, + } + ] + } + } + + result = await game_servers.get_game_servers(app_id=440, format="json") + data = json.loads(result) + + assert data["app_id"] == 440 + assert data["server_count"] == 1 + assert len(data["servers"]) == 1 + assert data["servers"][0]["name"] == "Test Server" + assert data["servers"][0]["bots"] == 2 + + @pytest.mark.asyncio + async def test_no_servers_found_text(self, game_servers, mock_client): + """Should return appropriate message when no servers found.""" + mock_client.get.return_value = {"response": {}} + + result = await game_servers.get_game_servers(app_id=12345) + + assert "No servers found" in result + assert "12345" in result + + @pytest.mark.asyncio + async def test_no_servers_found_json(self, game_servers, mock_client): + """Should return error JSON when no servers found.""" + mock_client.get.return_value = {"response": {"servers": []}} + + result = await game_servers.get_game_servers(app_id=12345, format="json") + data = json.loads(result) + + assert "error" in data + assert data["servers"] == [] + + @pytest.mark.asyncio + async def test_filter_passed_to_api(self, game_servers, mock_client): + """Should include user filter in API request.""" + mock_client.get.return_value = {"response": {"servers": []}} + + await game_servers.get_game_servers( + app_id=730, filter="\\map\\de_dust2" + ) + + call_args = mock_client.get.call_args + assert "\\appid\\730" in call_args.kwargs["params"]["filter"] + assert "\\map\\de_dust2" in call_args.kwargs["params"]["filter"] + + @pytest.mark.asyncio + async def test_limit_passed_to_api(self, game_servers, mock_client): + """Should pass limit parameter to API.""" + mock_client.get.return_value = {"response": {"servers": []}} + + await game_servers.get_game_servers(app_id=730, limit=50) + + call_args = mock_client.get.call_args + assert call_args.kwargs["params"]["limit"] == 50 + + @pytest.mark.asyncio + async def test_bots_displayed_in_text_output(self, game_servers, mock_client): + """Should show bot count when bots are present.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Bot Server", + "addr": "1.2.3.4:27015", + "map": "test", + "players": 10, + "max_players": 24, + "bots": 5, + "secure": True, + } + ] + } + } + + result = await game_servers.get_game_servers(app_id=730) + + assert "5 bots" in result + + +# --- query_server_status Tests --- + + +class TestQueryServerStatus: + """Tests for query_server_status endpoint.""" + + @pytest.mark.asyncio + async def test_returns_server_status_text(self, game_servers, mock_client): + """Should return detailed server status in text format.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "My Game Server", + "addr": "192.168.1.1:27015", + "appid": 730, + "gamedir": "csgo", + "map": "de_inferno", + "players": 18, + "max_players": 24, + "bots": 0, + "version": "1.38.2.3", + "secure": True, + "dedicated": True, + "os": "l", + "gametype": "competitive", + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="192.168.1.1:27015" + ) + + assert "My Game Server" in result + assert "192.168.1.1:27015" in result + assert "730" in result + assert "de_inferno" in result + assert "18/24" in result + assert "Linux" in result + assert "Yes" in result # VAC secured + + @pytest.mark.asyncio + async def test_returns_server_status_json(self, game_servers, mock_client): + """Should return structured server status in JSON format.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Test Server", + "addr": "10.0.0.1:27015", + "appid": 440, + "gamedir": "tf", + "map": "cp_dustbowl", + "players": 24, + "max_players": 32, + "bots": 8, + "version": "7.0.0", + "secure": True, + "dedicated": True, + "os": "w", + "gametype": "payload", + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="10.0.0.1:27015", format="json" + ) + data = json.loads(result) + + assert data["address"] == "10.0.0.1:27015" + assert data["name"] == "Test Server" + assert data["app_id"] == 440 + assert data["map"] == "cp_dustbowl" + assert data["players"] == 24 + assert data["bots"] == 8 + assert data["secure"] is True + + @pytest.mark.asyncio + async def test_server_not_found_text(self, game_servers, mock_client): + """Should return error message when server not found.""" + mock_client.get.return_value = {"response": {}} + + result = await game_servers.query_server_status( + server_address="1.1.1.1:27015" + ) + + assert "not found" in result.lower() + assert "1.1.1.1:27015" in result + + @pytest.mark.asyncio + async def test_server_not_found_json(self, game_servers, mock_client): + """Should return error JSON when server not found.""" + mock_client.get.return_value = {"response": {"servers": []}} + + result = await game_servers.query_server_status( + server_address="1.1.1.1:27015", format="json" + ) + data = json.loads(result) + + assert "error" in data + assert data["address"] == "1.1.1.1:27015" + + @pytest.mark.asyncio + async def test_address_filter_sent_to_api(self, game_servers, mock_client): + """Should filter by server address in API request.""" + mock_client.get.return_value = {"response": {"servers": []}} + + await game_servers.query_server_status(server_address="5.5.5.5:27015") + + call_args = mock_client.get.call_args + assert "\\addr\\5.5.5.5:27015" in call_args.kwargs["params"]["filter"] + + @pytest.mark.asyncio + async def test_os_mapping_windows(self, game_servers, mock_client): + """Should map 'w' OS code to Windows.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Win Server", + "addr": "1.2.3.4:27015", + "os": "w", + "map": "test", + "players": 0, + "max_players": 10, + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="1.2.3.4:27015" + ) + + assert "Windows" in result + + @pytest.mark.asyncio + async def test_os_mapping_macos(self, game_servers, mock_client): + """Should map 'm' and 'o' OS codes to macOS.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Mac Server", + "addr": "1.2.3.4:27015", + "os": "m", + "map": "test", + "players": 0, + "max_players": 10, + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="1.2.3.4:27015" + ) + + assert "macOS" in result + + @pytest.mark.asyncio + async def test_bots_shown_when_present(self, game_servers, mock_client): + """Should display bot count when bots are present.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "Bot Server", + "addr": "1.2.3.4:27015", + "map": "test", + "players": 16, + "max_players": 24, + "bots": 4, + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="1.2.3.4:27015" + ) + + assert "Bots:" in result + assert "4" in result + + @pytest.mark.asyncio + async def test_bots_hidden_when_zero(self, game_servers, mock_client): + """Should not display bot line when no bots.""" + mock_client.get.return_value = { + "response": { + "servers": [ + { + "name": "No Bot Server", + "addr": "1.2.3.4:27015", + "map": "test", + "players": 16, + "max_players": 24, + "bots": 0, + } + ] + } + } + + result = await game_servers.query_server_status( + server_address="1.2.3.4:27015" + ) + + assert "Bots:" not in result