這一章來分享如何製作自己遊戲內的MCP
讓 AI 可以控製你的遊戲角色, 進行各種操作
要讓 AI 控制你的遊戲, 我們需要在 Unity 中創建一個伺服器,
然後讓 Python 調用裡面的代碼, AI 再聯動到這個 Python 的代碼去查看有什麼工具可以使用
我們只需要兩個簡單的代碼:
完整的專案源代碼放在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 + "\"}";
}
}
現在有兩個功能
知道這個原理之後, 我們就可以繼續在 HandleRequest 繼續擴充新的功能
代碼寫好之後,我們可以創建一個簡單的場景
把玩家 (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 定義可以使用的工具有哪些
像我們目前有的指令是
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
}
}
}
成功的話就會得到這樣的畫面:
Claude 或是別的AI工具也是一樣, 可以到對應的 Config 文件上加上這個 MCP Config 資訊
遊戲測試:
一切都准備好之後, 你可以直接在 Unity Editor 運行遊戲進行測試
或是 Build 出來進行, 我測試過, 兩種方式都是可以運行的
測試方法是先運行 Unity, 再啟動 MCP
然後就讓 AI 來操作遊戲角色:
Build Demo:
以上就是一個簡單的 In Game MCP 效果的實現
你可以繼續延伸擴展更多的應用 :)