今天來分享如何製作自己遊戲內的 MCP
讓 AI 可以控製你的遊戲角色, 進行各種操作
原理
我們先來看看一般要讓 AI 可以操控 Unity 編輯器的 MCP 原理的實現方式:
其實就是透過在 Unity 中建立一個持續監聽的伺服器
然後再准備一個可以跟 AI 互相溝通的橋接器
進行各種 Function 的呼叫操作
根據這個核心原理
我們需要在遊戲中創建一個本地端的伺服器
然後讓 Python 調用裡面的代碼
AI 再調用到這個 Python 的代碼去查看
我們的遊戲裡面都有什麼工具可以讓 MCP 使用
準備工作
為了不要把整個專案弄得太複雜
我們只需要兩個簡單的代碼來驗證邏輯:
- simpleMCP.cs // 用來開啟伺服器, 和定義遊戲裡可以執行的方法
- simple_mcp_bridge.py // 讓 AI MCP 可以執行的 Python 橋接器
完整的專案源代碼放在 Github : https://github.com/yayapipi/InGameMCP
simpleMCP.cs
主要的功能室在遊戲啟動的時候先開啟一個 Server:
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, 當然這個地方也可以帶一些參數進來
執行完之後, 就把結果包成 Json 回傳回去給 AI :
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 + "\"}";
}
}
現在有兩個功能：
- get_position - 取得玩家的位置
- set_position - 設定玩家的位置
代碼寫好之後，我們可以創建一個簡單的場景
把玩家 (Capsule) 放進去, 執行遊戲進行測試
看 Console 的顯示會不會正常打開 Server, 開始監聽
simple_mcp_bridge.py
接下來我們需要一個 Python 的橋接器代碼
讓 AI 可以透過這個代碼跟我們的遊戲進行溝通
一開始啟動的時候需要先安裝 Python 和 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 這個方法裡
一開始的時候 AI 會先進行初始化 Initialize
這個時候要回傳這個工具的相關資訊給 AI
# 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 定義可以使用的工具有哪些
像我們目前有的指令是
- get_player_position - 取得玩家的位置
- 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 Config
架設好代碼和環境之後
我們要在想要使用的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 出來進行
Build 的時候 Player Setting 那邊可以把 Run In Background 打鉤
我測試過兩種方式都是可以正常運行的
測試的時候記得先運行 Unity, 再啟動 MCP 的工具:
0:00
/0:36
以上就是一個簡單的 In Game MCP 效果的實現
你可以繼續延伸擴展更多的應用
如果對你有收獲的話歡迎訂閱支持 Yapi Flow 平台 :)
Discussion