如何讓 AI 控制你的遊戲 ?

1private void ProcessMessage(string message, NetworkStream stream)
2    {
3        lock (mainThreadActions)
4        {
5            mainThreadActions.Enqueue(() =>
6            {
7                try
8                {
9                    Debug.Log($"[SimpleMCP] 收到消息: {message}");
10
11                    var request = JsonUtility.FromJson<MCPRequest>(message);
12
13                    if (request == null || string.IsNullOrEmpty(request.method))
14                    {
15                        Debug.LogError("[SimpleMCP] 無效的請求格式");
16                        SendResponse(stream, "{\"error\":\"Invalid request format\"}");
17                        return;
18                    }
19
20                    string response = HandleRequest(request);
21                    SendResponse(stream, response);
22                }
23                catch (Exception e)
24                {
25                    Debug.LogError($"[SimpleMCP] 處理消息錯誤: {e.Message}\n消息內容: {message}");
26                    SendResponse(stream, "{\"error\":\"Internal server error\"}");
27                }
28            });
29        }
30    }
31
32    // 新增的 HandleRequest 方法
33    private string HandleRequest(MCPRequest request)
34    {
35        try
36        {
37            Debug.Log($"[SimpleMCP] 處理請求方法: {request.method}");
38
39            switch (request.method.ToLower())
40            {
41                case "get_position":
42                case "getposition":
43                    return GetPlayerPosition();
44
45                case "move_player":
46                case "moveplayer":
47                case "set_position":
48                case "setposition":
49                    return MovePlayer(request.x, request.y, request.z);
50
51                case "ping":
52                    return "{\"success\":true,\"message\":\"pong\"}";
53
54                case "get_player_info":
55                case "getplayerinfo":
56                    return GetPlayerInfo();
57
58                default:
59                    Debug.LogWarning($"[SimpleMCP] 未知的請求方法: {request.method}");
60                    return "{\"error\":\"Unknown method\",\"method\":\"" + request.method + "\"}";
61            }
62        }
63        catch (Exception e)
64        {
65            Debug.LogError($"[SimpleMCP] HandleRequest 錯誤: {e.Message}");
66            return "{\"error\":\"Request handling failed\",\"details\":\"" + e.Message + "\"}";
67        }
68    }

1pip3 install websockets

1async def connect_to_unity():
2    """連接到 Unity WebSocket 伺服器"""
3    global ws_connection
4    
5    log(f"Attempting to connect to Unity at {UNITY_URI}")
6    
7    max_retries = 5
8    retry_delay = 2
9    
10    for attempt in range(max_retries):
11        try:
12            ws_connection = await asyncio.wait_for(
13                websockets.connect(UNITY_URI),
14                timeout=5.0
15            )
16            log(f"✓ Connected to Unity at {UNITY_URI}")
17            return True
18        except asyncio.TimeoutError:
19            log(f"✗ Connection timeout (attempt {attempt + 1}/{max_retries})")
20        except ConnectionRefusedError:
21            log(f"✗ Connection refused - Is Unity running? (attempt {attempt + 1}/{max_retries})")
22        except Exception as e:
23            log(f"✗ Connection error: {type(e).__name__} - {e} (attempt {attempt + 1}/{max_retries})")
24        
25        if attempt < max_retries - 1:
26            log(f"Retrying in {retry_delay} seconds...")
27            await asyncio.sleep(retry_delay)
28    
29    log("✗ Failed to connect to Unity after all retries")
30    log("Please ensure:")
31    log("  1. Unity is running")
32    log("  2. SimpleMCP script is attached to a GameObject")
33    log("  3. The game is in Play mode")
34    log(f"  4. Port {UNITY_PORT} is not blocked by firewall")

1async def send_unity_request(method, **kwargs):
2    """發送請求到 Unity"""
3    global ws_connection
4    
5    if ws_connection is None:
6        log("WebSocket not connected, attempting to connect...")
7        if not await connect_to_unity():
8            return {"error": "Not connected to Unity. Is the game running?"}
9    
10    try:
11        request = {"method": method, **kwargs}
12        log(f"Sending to Unity: {json.dumps(request)}")
13        
14        await ws_connection.send(json.dumps(request))
15        response = await asyncio.wait_for(ws_connection.recv(), timeout=5.0)
16        
17        log(f"Received from Unity: {response}")
18        return json.loads(response)
19        
20    except asyncio.TimeoutError:
21        log("✗ Unity request timeout")
22        return {"error": "Unity request timeout"}
23    except websockets.exceptions.ConnectionClosed as e:
24        log(f"✗ Connection to Unity lost: {e}")
25        ws_connection = None
26        return {"error": "Connection to Unity lost"}
27    except Exception as e:
28        log(f"✗ Error sending request: {type(e).__name__} - {e}")
29        return {"error": str(e)}

1# MCP 初始化握手
2if method == "initialize":
3    log("Handling initialize request")
4    return {
5        "protocolVersion": "2024-11-05",
6        "capabilities": {
7            "tools": {}
8        },
9        "serverInfo": {
10            "name": "unity-game-mcp",
11            "version": "1.0.0"
12        }
13    }

1elif method == "tools/list":
2    return {
3        "tools": [
4            {
5                "name": "get_player_position",
6                "description": "獲取玩家在遊戲中的當前位置座標",
7                "inputSchema": {
8                    "type": "object",
9                    "properties": {},
10                    "required": []
11                }
12            },
13            {
14                "name": "move_player",
15                "description": "將玩家移動到指定的座標位置",
16                "inputSchema": {
17                    "type": "object",
18                    "properties": {
19                        "x": {"type": "number", "description": "X 座標"},
20                        "y": {"type": "number", "description": "Y 座標（高度）"},
21                        "z": {"type": "number", "description": "Z 座標"}
22                    },
23                    "required": ["x", "y", "z"]
24                }
25            }
26        ]
27    }

1elif method == "tools/call":
2        tool_name = params.get("name")
3        arguments = params.get("arguments", {})
4        
5        log(f"Calling tool: {tool_name} with arguments: {arguments}")
6        
7        if tool_name == "get_player_position":
8            result = await send_unity_request("get_position")
9            if "error" in result:
10                return {"error": result["error"]}
11            return {
12                "content": [
13                    {
14                        "type": "text",
15                        "text": f"玩家當前位置：\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
16                    }
17                ]
18            }
19        
20        elif tool_name == "move_player":
21            x = arguments.get("x")
22            y = arguments.get("y")
23            z = arguments.get("z")
24            
25            if x is None or y is None or z is None:
26                return {"error": "Missing required parameters: x, y, z"}
27            
28            result = await send_unity_request("move_player", x=x, y=y, z=z)
29            if "error" in result:
30                return {"error": result["error"]}
31            
32            return {
33                "content": [
34                    {
35                        "type": "text",
36                        "text": f"✓ 玩家已移動到：\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
37                    }
38                ]
39            }

1{
2  "mcpServers": {
3    "unity-game-mcp": {
4      "command": "python",
5      "args": [
6        "Assets/SimpleMCP/PythonServer/simple_mcp_bridge.py"
7      ],
8      "env": {
9        "UNITY_HOST": "localhost",
10        "UNITY_PORT": "8765"
11      },
12      "enabled": true
13    }
14  }
15}