iT邦幫忙

2025 iThome 鐵人賽

DAY 22
0
Software Development

Vibe Unity - AI時代的遊戲開發工作流系列 第 22

Day 22 - 讓 AI 控制你的遊戲角色 - In Game MCP

  • 分享至 

  • xImage
  •  

這一章來分享如何製作自己遊戲內的MCP

讓 AI 可以控製你的遊戲角色, 進行各種操作

%E8%9E%A2%E5%B9%95%E6%93%B7%E5%8F%96%E7%95%AB%E9%9D%A2_2025-10-05_172613.png

要讓 AI 控制你的遊戲, 我們需要在 Unity 中創建一個伺服器,

然後讓 Python 調用裡面的代碼, AI 再聯動到這個 Python 的代碼去查看有什麼工具可以使用

我們只需要兩個簡單的代碼:

  1. simpleMCP.cs
  2. simple_mcp_bridge.py

完整的專案源代碼放在Github: https://github.com/yayapipi/InGameMCP


simpleMCP.cs

https://github.com/yayapipi/InGameMCP/blob/main/Assets/SimpleMCP/SimpleMCP.cs

主要是會在 Unity 啟動的時候開啟一個 Server, 設定 Port:

private void StartServer()
{
    try
    {
        listener = new TcpListener(IPAddress.Any, port);
        listener.Start();
        isRunning = true;

        serverThread = new Thread(AcceptClients);
        serverThread.IsBackground = true;
        serverThread.Start();

        Debug.Log($"[SimpleMCP] 伺服器已啟動在端口 {port}");
        Debug.Log("[SimpleMCP] 建議透過 MCP Bridge 連接(Cursor 將依 .cursor/mcp.json 自動啟動 Python Bridge)");
    }
    catch (Exception e)
    {
        Debug.LogError($"[SimpleMCP] 啟動失敗: {e.Message}");
    }
}

接著就會不斷的監聽這個 Port 有沒有新的訊息進來,

當檢測到有人連進來的時候, 就會跟它握手, 然後建立長連線 WebSocket

開始不斷地從他那邊接受指令:

 private void HandleClient(TcpClient client)
    {
        NetworkStream stream = client.GetStream();
        byte[] buffer = new byte[4096];

        try
        {
            // WebSocket 握手
            int bytesRead = stream.Read(buffer, 0, buffer.Length);
            string request = Encoding.UTF8.GetString(buffer, 0, bytesRead);

            if (request.Contains("Upgrade: websocket"))
            {
                string response = PerformHandshake(request);
                byte[] responseBytes = Encoding.UTF8.GetBytes(response);
                stream.Write(responseBytes, 0, responseBytes.Length);

                Debug.Log("[SimpleMCP] 客戶端已連接");

                // 處理消息
                while (isRunning && client.Connected)
                {
                    if (stream.DataAvailable)
                    {
                        bytesRead = stream.Read(buffer, 0, buffer.Length);
                        if (bytesRead <= 0)
                        {
                            break; // 連線關閉
                        }
                        if (bytesRead > 0)
                        {
                            string message = DecodeFrame(buffer, bytesRead);
                            if (!string.IsNullOrEmpty(message))
                            {
                                ProcessMessage(message, stream);
                            }
                        }
                    }

                    Thread.Sleep(10);
                }
            }
        }
        catch (Exception e)
        {
            Debug.LogError($"[SimpleMCP] 客戶端錯誤: {e.Message}");
        }
        finally
        {
            client.Close();
            Debug.Log("[SimpleMCP] 客戶端已斷開");
        }
    }

當收到新的指令的時候, 會在 ProcessMessage/HandleRequest 裡面進行判斷:

這裡就是收到什麼指令, 就觸發什麼 Function, 也可以帶一些參數進來

執行完之後, 就把結果回傳回去

private void ProcessMessage(string message, NetworkStream stream)
    {
        lock (mainThreadActions)
        {
            mainThreadActions.Enqueue(() =>
            {
                try
                {
                    Debug.Log($"[SimpleMCP] 收到消息: {message}");

                    var request = JsonUtility.FromJson<MCPRequest>(message);

                    if (request == null || string.IsNullOrEmpty(request.method))
                    {
                        Debug.LogError("[SimpleMCP] 無效的請求格式");
                        SendResponse(stream, "{\"error\":\"Invalid request format\"}");
                        return;
                    }

                    string response = HandleRequest(request);
                    SendResponse(stream, response);
                }
                catch (Exception e)
                {
                    Debug.LogError($"[SimpleMCP] 處理消息錯誤: {e.Message}\n消息內容: {message}");
                    SendResponse(stream, "{\"error\":\"Internal server error\"}");
                }
            });
        }
    }

    // 新增的 HandleRequest 方法
    private string HandleRequest(MCPRequest request)
    {
        try
        {
            Debug.Log($"[SimpleMCP] 處理請求方法: {request.method}");

            switch (request.method.ToLower())
            {
                case "get_position":
                case "getposition":
                    return GetPlayerPosition();

                case "move_player":
                case "moveplayer":
                case "set_position":
                case "setposition":
                    return MovePlayer(request.x, request.y, request.z);

                case "ping":
                    return "{\"success\":true,\"message\":\"pong\"}";

                case "get_player_info":
                case "getplayerinfo":
                    return GetPlayerInfo();

                default:
                    Debug.LogWarning($"[SimpleMCP] 未知的請求方法: {request.method}");
                    return "{\"error\":\"Unknown method\",\"method\":\"" + request.method + "\"}";
            }
        }
        catch (Exception e)
        {
            Debug.LogError($"[SimpleMCP] HandleRequest 錯誤: {e.Message}");
            return "{\"error\":\"Request handling failed\",\"details\":\"" + e.Message + "\"}";
        }
    }

現在有兩個功能

  1. get_position - 取得玩家的位置
  2. set_position - 設定玩家的位置

知道這個原理之後, 我們就可以繼續在 HandleRequest 繼續擴充新的功能

image.png

代碼寫好之後,我們可以創建一個簡單的場景

把玩家 (Capsule) 放進去, 執行遊戲進行測試, 看會不會打開 Server


simple_mcp_bridge.py

https://github.com/yayapipi/InGameMCP/blob/main/Assets/SimpleMCP/PythonServer/simple_mcp_bridge.py

接下來我們需要一個 Python 的橋接器代碼

讓 AI 可以透過這個代碼跟我們的遊戲進行溝通

一開始啟動的時候需要先安裝 Websockets

pip3 install websockets

這個代碼一開始會想連接到 Unity WebSocket 的伺服器

async def connect_to_unity():
    """連接到 Unity WebSocket 伺服器"""
    global ws_connection
    
    log(f"Attempting to connect to Unity at {UNITY_URI}")
    
    max_retries = 5
    retry_delay = 2
    
    for attempt in range(max_retries):
        try:
            ws_connection = await asyncio.wait_for(
                websockets.connect(UNITY_URI),
                timeout=5.0
            )
            log(f"✓ Connected to Unity at {UNITY_URI}")
            return True
        except asyncio.TimeoutError:
            log(f"✗ Connection timeout (attempt {attempt + 1}/{max_retries})")
        except ConnectionRefusedError:
            log(f"✗ Connection refused - Is Unity running? (attempt {attempt + 1}/{max_retries})")
        except Exception as e:
            log(f"✗ Connection error: {type(e).__name__} - {e} (attempt {attempt + 1}/{max_retries})")
        
        if attempt < max_retries - 1:
            log(f"Retrying in {retry_delay} seconds...")
            await asyncio.sleep(retry_delay)
    
    log("✗ Failed to connect to Unity after all retries")
    log("Please ensure:")
    log("  1. Unity is running")
    log("  2. SimpleMCP script is attached to a GameObject")
    log("  3. The game is in Play mode")
    log(f"  4. Port {UNITY_PORT} is not blocked by firewall")

收到 AI 的指令的時候就給我們的遊戲發送指令:

async def send_unity_request(method, **kwargs):
    """發送請求到 Unity"""
    global ws_connection
    
    if ws_connection is None:
        log("WebSocket not connected, attempting to connect...")
        if not await connect_to_unity():
            return {"error": "Not connected to Unity. Is the game running?"}
    
    try:
        request = {"method": method, **kwargs}
        log(f"Sending to Unity: {json.dumps(request)}")
        
        await ws_connection.send(json.dumps(request))
        response = await asyncio.wait_for(ws_connection.recv(), timeout=5.0)
        
        log(f"Received from Unity: {response}")
        return json.loads(response)
        
    except asyncio.TimeoutError:
        log("✗ Unity request timeout")
        return {"error": "Unity request timeout"}
    except websockets.exceptions.ConnectionClosed as e:
        log(f"✗ Connection to Unity lost: {e}")
        ws_connection = None
        return {"error": "Connection to Unity lost"}
    except Exception as e:
        log(f"✗ Error sending request: {type(e).__name__} - {e}")
        return {"error": str(e)}

那麼 AI 是怎麼跟這個 Python 代碼進行溝通的呢?

關鍵的代碼在 handle_mcp_request 這個 function 裡:

一開始的時候 AI 會先進行初始化 Initialize

這個時候要回傳一些相關資訊給 AI, 讓他知道這個 MCP 是幹嘛的

# MCP 初始化握手
if method == "initialize":
    log("Handling initialize request")
    return {
        "protocolVersion": "2024-11-05",
        "capabilities": {
            "tools": {}
        },
        "serverInfo": {
            "name": "unity-game-mcp",
            "version": "1.0.0"
        }
    }

接著, 我們會在 tools/list 定義可以使用的工具有哪些

像我們目前有的指令是

  1. get_player_position - 取得玩家的位置
  2. move_player - 移動玩家坐標, 然後要帶入參數進來
elif method == "tools/list":
    return {
        "tools": [
            {
                "name": "get_player_position",
                "description": "獲取玩家在遊戲中的當前位置座標",
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            },
            {
                "name": "move_player",
                "description": "將玩家移動到指定的座標位置",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "x": {"type": "number", "description": "X 座標"},
                        "y": {"type": "number", "description": "Y 座標(高度)"},
                        "z": {"type": "number", "description": "Z 座標"}
                    },
                    "required": ["x", "y", "z"]
                }
            }
        ]
    }

最後 AI 會根據 使用者的 Prompt 來決定要使用哪個工具

然後會自己調用 tools/call 這個地方

這裡會直接 send_unity_request , 發送指令到 Unity 的遊戲中

然後等待回復, Unity 回復之後, 就把他列印出來

elif method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        
        log(f"Calling tool: {tool_name} with arguments: {arguments}")
        
        if tool_name == "get_player_position":
            result = await send_unity_request("get_position")
            if "error" in result:
                return {"error": result["error"]}
            return {
                "content": [
                    {
                        "type": "text",
                        "text": f"玩家當前位置:\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
                    }
                ]
            }
        
        elif tool_name == "move_player":
            x = arguments.get("x")
            y = arguments.get("y")
            z = arguments.get("z")
            
            if x is None or y is None or z is None:
                return {"error": "Missing required parameters: x, y, z"}
            
            result = await send_unity_request("move_player", x=x, y=y, z=z)
            if "error" in result:
                return {"error": result["error"]}
            
            return {
                "content": [
                    {
                        "type": "text",
                        "text": f"✓ 玩家已移動到:\nX: {result.get('x', 0):.2f}\nY: {result.get('y', 0):.2f}\nZ: {result.get('z', 0):.2f}"
                    }
                ]
            }

基於這點, 我們就可以無限的擴展我們的 MCP 遊戲工具了


那麼架設好之後, 我們要在想要使用的AI工具上設定 MCP Config

以 Cursor 舉例, 你要到 Cursor 的MCP.json 中加入執行 Python 代碼的指令

{
  "mcpServers": {
    "unity-game-mcp": {
      "command": "python",
      "args": [
        "Assets/SimpleMCP/PythonServer/simple_mcp_bridge.py"
      ],
      "env": {
        "UNITY_HOST": "localhost",
        "UNITY_PORT": "8765"
      },
      "enabled": true
    }
  }
}

成功的話就會得到這樣的畫面:

image.png

Claude 或是別的AI工具也是一樣, 可以到對應的 Config 文件上加上這個 MCP Config 資訊


遊戲測試:

一切都准備好之後, 你可以直接在 Unity Editor 運行遊戲進行測試

或是 Build 出來進行, 我測試過, 兩種方式都是可以運行的

測試方法是先運行 Unity, 再啟動 MCP

然後就讓 AI 來操作遊戲角色:

20251005-1007-33.7080333.mp4

Build Demo:

20251005-1009-47.7816913.mp4


以上就是一個簡單的 In Game MCP 效果的實現

你可以繼續延伸擴展更多的應用 :)


上一篇
Day 21 - MCP Unity 的技術開發介紹
下一篇
Day 23 - 如何讓 AI 控制遊戲裡的天氣 ?
系列文
Vibe Unity - AI時代的遊戲開發工作流25
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言