Blender-MCP服务源码5-BlenderSocket插件安装
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 socket
def 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 = host
self.port = port
self.running = False
self.socket = None
self.client = None
self.server_thread = None
def start(self):
self.running = True
self.server_thread = threading.Thread(target=self._run_server)
self.server_thread.daemon = True
self.server_thread.start()
print(f"BlenderMCP server started on {self.host}:{self.port}")
def stop(self):
self.running = False
if 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 accept
while self.running:
try:
self.client, address = self.socket.accept()
print(f"Connected to client: {address}")
while self.running:
try:
# Set a timeout for receiving data
self.client.settimeout(15.0)
data = self.client.recv(4096)
if not data:
print("Empty data received, client may have disconnected")
break
try:
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 log
self.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 traceback
traceback.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")
continue
except Exception as e:
print(f"Error receiving data: {str(e)}")
break
self.client.close()
self.client = None
except socket.timeout:
# This is normal - just continue the loop
continue
except Exception as e:
print(f"Connection error: {str(e)}")
if self.client:
self.client.close()
self.client = None
time.sleep(1) # Prevent busy waiting
except 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 handler
if 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'}
@classmethod
def poll(cls, context: bpy.types.Context):
return context.active_object is not None
def 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 server
bpy.types.blender_mcp_server = BlenderMCPServer(port=9876)
bpy.types.blender_mcp_server.start()
print("---------------------Start MCP Server后重置状态")
scene.blender_mcp_server_running = True
return {'FINISHED'}
class BlenderMCPOperatorStop(bpy.types.Operator):
'''BlenderMCPOperatorStop'''
bl_idname = "object.blender_mcp_operator_stop"
bl_label = "Now_Start_Click_Stop"
@classmethod
def poll(cls, context: bpy.types.Context):
return context.active_object is not None
def 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_server
print("---------------------Stop MCP Server后重置状态")
scene.blender_mcp_server_running = False
return {'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.layout
layout.label(text="BlenderMCP Panel")
# 当前只是为了测试常驻按钮的点击测试-点击对图形缩小0.8
layout.operator(StaticButtonOperator.bl_idname)
# 测试服务器的装
scene = context.scene
if 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."
)