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

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

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

  • 执行通讯效果如下
    在这里插入图片描述


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

相关文章:

  • Linux find 命令完全指南
  • 接口测试中常见的bug有哪些?
  • 使用elementplus的table表格遇到的问题
  • ubuntu ollama+dify实践
  • 关于修改 Ollama 及其模型默认路径、迁移已安装的 Ollama 程序和模型以及重启 Ollama 的操作指南
  • 计算机视觉——深入理解卷积神经网络与使用卷积神经网络创建图像分类算法
  • 在线 SQL 转 Flask-SQLAlchemy
  • 高版本node(17+)环境下VUE2项目启动报错
  • Android 7 及以上夜神模拟器,Fiddler 抓 https 包
  • DDS:保障物联网系统的稳定运行和高效协作
  • 提升 React 应用性能:使用 React Profiler 进行性能调优
  • Assembly语言的自然语言处理
  • Spring Boot项目中成功集成了JWT
  • C++学习内存管理
  • 【NLP 38、实践 ⑩ NER 命名实体识别任务 Bert 实现】
  • STM32 - 在机器人领域,LL库相比HAL优势明显
  • RAG数据嵌入和重排序:如何选择合适的模型
  • .NET_Prism基本项目创建
  • 嵌入式工程师春招面试高频题与参考回答
  • PyTorch 深度学习实战(15):Twin Delayed DDPG (TD3) 算法