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

Blender-MCP服务源码1-项目解读

article 2025/3/17 16:53:56

Blender-MCP服务源码

有个大佬做了一个Blender-MCP源码，第一次提交代码是【2025年3月7号】今天是【2025年月15日】也就是刚过去一周的时间，所以想从0开始学习这个代码，了解一下大佬们的开发思路

1-核心知识点

1）第一版：作者貌似就实现了一个Socket的远程连接
2）有一个分支：polyhaven-integration（对接3D资源库）

2-思路整理

1）polyhaven是一家3D资源库->作者希望从3D资源库获取对应的免费资源

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测试

1-原作者代码

import socket
import json
import time


def test_simple_command():
    """
    测试向Blender发送简单命令并接收响应的功能。
    """
    # 创建一个TCP/IP套接字
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        # 尝试连接到Blender服务器
        print("Connecting to Blender...")
        sock.connect(('localhost', 9876))
        print("Connected!")

        # 构造一个简单的ping命令
        command = {
            "type": "ping",
            "params": {}
        }

        # 发送命令到Blender服务器
        print(f"Sending command: {json.dumps(command)}")
        sock.sendall(json.dumps(command).encode('utf-8'))

        # 设置套接字超时时间为10秒
        print(f"Setting socket timeout: 10 seconds")
        sock.settimeout(10)

        # 等待接收响应
        print("Waiting for response...")
        try:
            # 接收响应数据
            response_data = sock.recv(65536)
            print(f"Received {len(response_data)} bytes")

            # 如果接收到数据，则解析并打印响应
            if response_data:
                response = json.loads(response_data.decode('utf-8'))
                print(f"Response: {response}")
            else:
                print("Received empty response")
        except socket.timeout:
            # 如果超时，则打印超时信息
            print("Socket timeout while waiting for response")

    except Exception as e:
        # 如果发生异常，则打印错误信息
        print(f"Error: {type(e).__name__}: {str(e)}")
    finally:
        # 关闭套接字连接
        sock.close()


if __name__ == "__main__":
    # 运行测试函数
    test_simple_command()

2-代码联想

1）作者期望在Blender中部署一个socket服务
2）本地MCP服务尝试通过socket发送指令->调用blender的指令进行页面操作

1- 开发一个服务端

import json
import socket

def handle_command(command):
    """处理客户端命令并返回响应结果"""
    if command.get('type') == 'ping':
        return {
            "status": "success",
            "result": "pong"
        }
    else:
        return {
            "status": "error",
            "result": f"Unknown command type: {command.get('type')}"
        }

def run_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 9876))
    server_socket.listen(1)
    print("Server is listening on port 9876...")

    try:
        while True:
            client_socket, addr = server_socket.accept()
            print(f"\nAccepted connection from {addr}")

            try:
                # 接收客户端数据
                data = client_socket.recv(65536)
                if not data:
                    print("Client sent empty data")
                    continue

                # 解析JSON命令
                try:
                    command = json.loads(data.decode('utf-8'))
                    print(f"Received command: {command}")

                    # 处理命令并发送响应
                    response = handle_command(command)
                    response_data = json.dumps(response).encode('utf-8')
                    client_socket.sendall(response_data)
                    print("Sent response:", response)

                except json.JSONDecodeError:
                    error_response = {
                        "status": "error",
                        "result": "Invalid JSON format"
                    }
                    client_socket.sendall(json.dumps(error_response).encode('utf-8'))
                    print("Sent JSON decode error response")

            except Exception as e:
                print(f"Error handling client: {e}")
            finally:
                client_socket.close()
                print("Client connection closed")

    except KeyboardInterrupt:
        print("\nServer shutting down...")
    finally:
        server_socket.close()

if __name__ == "__main__":
    run_server()

2- 开发一个客户端

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)  # 打印服务器响应