当前位置： 首页 > article >正文

Blender-MCP服务源码5-BlenderSocket插件安装

article 2025/3/16 14:56:12

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."
)

http://www.kler.cn/a/587079.html

相关文章：

  • INSERT ... ON DUPLICATE KEY UPDATE
  • 1216走迷宫
  • Keil5下载教程及安装教程（附安装包）
  • Unity AssetBundles资源加载管理器
  • Debain-12.9使用xinference部署音频模型/audio
  • 《灵珠觉醒：从零到算法金仙的C++修炼》卷三·天劫试炼（48）戮魂幡染节点 - 二分图检测（着色法）
  • 【NLP】 9. 处理创造性词汇 词组特征（Creative Words Features Model）， 词袋模型处理未知词，模型得分
  • 【黑马点评|项目】万字总结（下）
  • Python软件和搭建运行环境
  • C++进阶——map和set的使用
  • Python在数据处理中的应用：从入门到精通
  • Linux date 命令使用指南
  • 用Python打造AI玩家：挑战2048，谁与争锋
  • Socket服务器和客户端
  • 安装SQL数据库并且在jupyter中连接，运行
  • 回溯法--力扣第17题“电话号码的字母组合”(java)
  • 【初级篇】如何使用DeepSeek和Dify构建高效的企业级智能客服系统
  • stable Diffusion 中的 VAE是什么
  • Maximize Rating
  • [动手学习深度学习]24. AlexNet