#!/usr/bin/env python3 """ Test script for MCP servers. Validates that both desktop-control and steam-info servers respond correctly to MCP requests. """ import json import subprocess import sys from typing import Dict, Any, Tuple, List def send_request(server_script: str, request: Dict[str, Any]) -> Tuple[bool, Dict[str, Any], str]: """Send a JSON-RPC request to an MCP server and return the response.""" try: proc = subprocess.run( ["python3", server_script], input=json.dumps(request) + "\n", capture_output=True, text=True, timeout=10 ) # Parse stdout for JSON-RPC response for line in proc.stdout.strip().split("\n"): line = line.strip() if line and line.startswith("{"): try: response = json.loads(line) if "jsonrpc" in response: return True, response, "" except json.JSONDecodeError: continue return False, {}, f"No valid JSON-RPC response found. stderr: {proc.stderr}" except subprocess.TimeoutExpired: return False, {}, "Server timed out" except Exception as e: return False, {}, str(e) def test_desktop_control_server() -> List[str]: """Test the desktop control MCP server.""" errors = [] server = "mcp_servers/desktop_control_server.py" print("\n=== Testing Desktop Control Server ===") # Test initialize print(" Testing initialize...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {} }) if not success: errors.append(f"initialize failed: {error}") elif "error" in response: errors.append(f"initialize returned error: {response['error']}") else: print(" ✓ initialize works") # Test tools/list print(" Testing tools/list...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }) if not success: errors.append(f"tools/list failed: {error}") elif "error" in response: errors.append(f"tools/list returned error: {response['error']}") else: tools = response.get("result", {}).get("tools", []) expected_tools = [ "take_screenshot", "get_screen_size", "get_mouse_position", "pixel_color", "click", "right_click", "move_to", "drag_to", "type_text", "press_key", "hotkey", "scroll", "get_os" ] tool_names = [t["name"] for t in tools] missing = [t for t in expected_tools if t not in tool_names] if missing: errors.append(f"Missing tools: {missing}") else: print(f" ✓ tools/list works ({len(tools)} tools available)") # Test get_os (works without display) print(" Testing tools/call get_os...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "get_os", "arguments": {}} }) if not success: errors.append(f"get_os failed: {error}") elif "error" in response: errors.append(f"get_os returned error: {response['error']}") else: content = response.get("result", {}).get("content", []) if content and not response["result"].get("isError"): result_data = json.loads(content[0]["text"]) if "system" in result_data: print(f" ✓ get_os works (system: {result_data['system']})") else: errors.append("get_os response missing system info") else: errors.append("get_os returned error content") return errors def test_steam_info_server() -> List[str]: """Test the Steam info MCP server.""" errors = [] server = "mcp_servers/steam_info_server.py" print("\n=== Testing Steam Info Server ===") # Test initialize print(" Testing initialize...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {} }) if not success: errors.append(f"initialize failed: {error}") elif "error" in response: errors.append(f"initialize returned error: {response['error']}") else: print(" ✓ initialize works") # Test tools/list print(" Testing tools/list...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }) if not success: errors.append(f"tools/list failed: {error}") elif "error" in response: errors.append(f"tools/list returned error: {response['error']}") else: tools = response.get("result", {}).get("tools", []) expected_tools = [ "steam_recently_played", "steam_player_achievements", "steam_user_stats", "steam_current_players", "steam_news", "steam_app_details" ] tool_names = [t["name"] for t in tools] missing = [t for t in expected_tools if t not in tool_names] if missing: errors.append(f"Missing tools: {missing}") else: print(f" ✓ tools/list works ({len(tools)} tools available)") # Test steam_current_players (mock mode) print(" Testing tools/call steam_current_players...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "steam_current_players", "arguments": {"app_id": "261550"}} }) if not success: errors.append(f"steam_current_players failed: {error}") elif "error" in response: errors.append(f"steam_current_players returned error: {response['error']}") else: content = response.get("result", {}).get("content", []) if content and not response["result"].get("isError"): result_data = json.loads(content[0]["text"]) if "player_count" in result_data: mode = "mock" if result_data.get("mock") else "live" print(f" ✓ steam_current_players works ({mode} mode, {result_data['player_count']} players)") else: errors.append("steam_current_players response missing player_count") else: errors.append("steam_current_players returned error content") # Test steam_recently_played (mock mode) print(" Testing tools/call steam_recently_played...") success, response, error = send_request(server, { "jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "steam_recently_played", "arguments": {"user_id": "12345"}} }) if not success: errors.append(f"steam_recently_played failed: {error}") elif "error" in response: errors.append(f"steam_recently_played returned error: {response['error']}") else: content = response.get("result", {}).get("content", []) if content and not response["result"].get("isError"): result_data = json.loads(content[0]["text"]) if "games" in result_data: print(f" ✓ steam_recently_played works ({len(result_data['games'])} games)") else: errors.append("steam_recently_played response missing games") else: errors.append("steam_recently_played returned error content") return errors def main(): """Run all tests.""" print("=" * 60) print("MCP Server Test Suite") print("=" * 60) all_errors = [] all_errors.extend(test_desktop_control_server()) all_errors.extend(test_steam_info_server()) print("\n" + "=" * 60) if all_errors: print(f"FAILED: {len(all_errors)} error(s)") for err in all_errors: print(f" - {err}") sys.exit(1) else: print("ALL TESTS PASSED") print("=" * 60) sys.exit(0) if __name__ == "__main__": main()