Blender-MCP服務源碼5-BlenderSocket插件安裝

上一篇講述了Blender是基于Socket進行本地和遠程進行通訊，現在嘗試將BlenderSocket插件安裝到Blender中進行功能調試

---

1-核心知識點

• 將開發的BlenderSocket插件安裝到Blender中

---

2-思路整理

• 1）將SocketServer部署到Blender啟動SocketServer
• 2）本地使用SocketClient連接SocketServer嘗試發送指令
• 3）驗證交互結果->如果該邏輯通->后續就可以完善MCP業務指令

---

3-參考網址

• Blender-MCP-Github地址：https://github.com/ahujasid/blender-mcp
• B站大佬開源Blender開發框架:https://github.com/xzhuah/BlenderAddonPackageTool
• B站大佬開源Blender開發框架教程
• 個人實現代碼倉庫1：https://gitee.com/enzoism/python_blender_socket
• 個人實現代碼倉庫2：https://gitee.com/enzoism/python_blender_mcp

---

4-上手實操

1-部署Socket到Blender

代碼地址：https://gitee.com/enzoism/python_blender_mcp

• 部署Socket到Blender
  
• 本地SocketClient與SocketServer通訊驗證
  

---

2-本地項目調試

運行test.py文件即可

• SocketClient代碼

代碼地址：https://gitee.com/enzoism/python_blender_socket

import json
import socketdef send_command(command):"""發送命令到服務器并接收響應"""sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 創建TCP套接字try:print("Connecting to server...")  # 連接到服務器sock.connect(('localhost', 9876))  # 連接到本地9876端口的服務器# 發送命令json_command = json.dumps(command).encode('utf-8')  # 將命令轉換為JSON格式并編碼為字節sock.sendall(json_command)  # 發送命令print(f"Sent command: {command}")  # 打印發送的命令# 接收響應sock.settimeout(10)  # 設置超時時間為10秒response_data = sock.recv(65536)  # 接收響應數據，最大長度為65536字節if response_data:  # 如果接收到數據response = json.loads(response_data.decode('utf-8'))  # 將響應數據解碼為JSON格式return response  # 返回響應else:return {"status": "error", "result": "Empty response"}  # 如果沒有接收到數據，返回錯誤信息except Exception as e:  # 捕獲所有異常return {"status": "error", "result": str(e)}  # 返回異常信息finally:sock.close()  # 關閉套接字if __name__ == "__main__":# 測試ping命令ping_command = {"type": "ping","params": {}}response = send_command(ping_command)print("Server response:", response)  # 打印服務器響應

---

3-代碼結構

• 1）clazz類中業務類->比如創建一個Person類，有構造方法，有eat方法
• 2）operator中放要clazz類中業務類的操作邏輯
• 3）panels中放看板的點擊Operator類->但是不要在這里也業務邏輯（解耦）
• 4）panels中防止看板的判斷屬性->根據狀態更換按鈕文字和點擊事件

---

• 1）clazz代碼示例

import json
import socket
import threading
import time# 添加自己的業務Operator
class BlenderMCPServer:def __init__(self, host='localhost', port=9876):self.host = hostself.port = portself.running = Falseself.socket = Noneself.client = Noneself.server_thread = Nonedef start(self):self.running = Trueself.server_thread = threading.Thread(target=self._run_server)self.server_thread.daemon = Trueself.server_thread.start()print(f"BlenderMCP server started on {self.host}:{self.port}")def stop(self):self.running = Falseif self.socket:self.socket.close()if self.client:self.client.close()print("BlenderMCP server stopped")def _run_server(self):self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)try:self.socket.bind((self.host, self.port))self.socket.listen(1)self.socket.settimeout(1.0)  # Add a timeout for acceptwhile self.running:try:self.client, address = self.socket.accept()print(f"Connected to client: {address}")while self.running:try:# Set a timeout for receiving dataself.client.settimeout(15.0)data = self.client.recv(4096)if not data:print("Empty data received, client may have disconnected")breaktry:print(f"Received data: {data.decode('utf-8')}")command = json.loads(data.decode('utf-8'))response = self.execute_command(command)print(f"Sending response: {json.dumps(response)[:100]}...")  # Truncate long responses in logself.client.sendall(json.dumps(response).encode('utf-8'))except json.JSONDecodeError:print(f"Invalid JSON received: {data.decode('utf-8')}")self.client.sendall(json.dumps({"status": "error","message": "Invalid JSON format"}).encode('utf-8'))except Exception as e:print(f"Error executing command: {str(e)}")import tracebacktraceback.print_exc()self.client.sendall(json.dumps({"status": "error","message": str(e)}).encode('utf-8'))except socket.timeout:print("Socket timeout while waiting for data")continueexcept Exception as e:print(f"Error receiving data: {str(e)}")breakself.client.close()self.client = Noneexcept socket.timeout:# This is normal - just continue the loopcontinueexcept Exception as e:print(f"Connection error: {str(e)}")if self.client:self.client.close()self.client = Nonetime.sleep(1)  # Prevent busy waitingexcept Exception as e:print(f"Server error: {str(e)}")finally:if self.socket:self.socket.close()def execute_command(self, command):"""Execute a Blender command received from the MCP server"""cmd_type = command.get("type")params = command.get("params", {})# Add a simple ping handlerif cmd_type == "ping":print("Handling ping command")return {"status": "success", "result": {"pong": True}}else:return {"status": "error","result": f"Unknown command type: {command.get('type')}"}

---

• 2）operator代碼示例

class BlenderMCPOperatorStart(bpy.types.Operator):'''BlenderMCPOperatorStart'''bl_idname = "object.blender_mcp_operator_start"bl_label = "Now_Stop_Click_Start"# 確保在操作之前備份數據，用戶撤銷操作時可以恢復bl_options = {'REGISTER', 'UNDO'}@classmethoddef poll(cls, context: bpy.types.Context):return context.active_object is not Nonedef execute(self, context: bpy.types.Context):scene = context.scene# 創建socket實例if not hasattr(bpy.types, "blender_mcp_server") or not bpy.types.blender_mcp_server:# Start the serverbpy.types.blender_mcp_server = BlenderMCPServer(port=9876)bpy.types.blender_mcp_server.start()print("---------------------Start MCP Server后重置狀態")scene.blender_mcp_server_running = Truereturn {'FINISHED'}class BlenderMCPOperatorStop(bpy.types.Operator):'''BlenderMCPOperatorStop'''bl_idname = "object.blender_mcp_operator_stop"bl_label = "Now_Start_Click_Stop"@classmethoddef poll(cls, context: bpy.types.Context):return context.active_object is not Nonedef execute(self, context: bpy.types.Context):scene = context.scene# 銷毀socket實例if hasattr(bpy.types, "blender_mcp_server") or bpy.types.blender_mcp_server:bpy.types.blender_mcp_server.stop()del bpy.types.blender_mcp_serverprint("---------------------Stop MCP Server后重置狀態")scene.blender_mcp_server_running = Falsereturn {'FINISHED'}

---

• 3）panels中Operator代碼示例

@reg_order(0)
class ExampleAddonPanel2(BasePanel, bpy.types.Panel):bl_label = "Example Addon Side Bar Panel"bl_idname = "SCENE_PT_sample2"def draw(self, context: bpy.types.Context):layout = self.layoutlayout.label(text="BlenderMCP Panel")# 當前只是為了測試常駐按鈕的點擊測試-點擊對圖形縮小0.8layout.operator(StaticButtonOperator.bl_idname)# 測試服務器的裝scene = context.sceneif not scene.blender_mcp_server_running:layout.operator(BlenderMCPOperatorStart.bl_idname)print("Start MCP Server")else:layout.operator(BlenderMCPOperatorStop.bl_idname)print("Stop MCP Server")

---

• 4）panels中屬性代碼示例

import bpy# 添加屬性到 Scene 類型
bpy.types.Scene.blender_mcp_server = bpy.props.BoolProperty(name="Blender MCP Server Running",default=False,description="Indicates whether the Blender MCP server is running."
)
bpy.types.Scene.blender_mcp_server_running = bpy.props.BoolProperty(name="Blender MCP Server Running",default=False,description="Indicates whether the Blender MCP server is running."
)