Blender MCP Server

bl_info = { "name": "MCP Connector", "blender": (3, 0, 0), "category": "Interface", "author": "Erge", "description": "Connects Blender to an MCP Server via Sockets", } import bpy import socket import threading import json import queue import traceback # Configuration HOST = '127.0.0.1' PORT = 9876 # Queue for thread-safe execution execution_queue = queue.Queue() # Global state server_thread = None server_socket = None is_server_running = False def execute_command(command_data): """Execute command in the main Blender thread""" try: cmd_type = command_data.get('type') if cmd_type == 'run_script': script = command_data.get('script') # Create a shared dictionary for local variables local_vars = {} exec(script, globals(), local_vars) return {"status": "success", "output": "Script executed"} elif cmd_type == 'get_version': return {"status": "success", "version": bpy.app.version_string} elif cmd_type == 'apply_texture': import os target_name = command_data.get('object_name') img_path = command_data.get('texture_path') if target_name not in bpy.data.objects: return {"status": "error", "message": f"Object '{target_name}' not found"} obj = bpy.data.objects[target_name] # Auto-Join Logic bpy.context.view_layer.objects.active = obj obj.select_set(True) if obj.children: for child in obj.children: child.select_set(True) bpy.ops.object.join() obj = bpy.context.active_object # Auto-UV if obj.type == 'MESH': bpy.ops.object.mode_set(mode='EDIT') bpy.ops.mesh.select_all(action='SELECT') bpy.ops.uv.smart_project(island_margin=0.02) bpy.ops.object.mode_set(mode='OBJECT') # Material Setup if not os.path.exists(img_path): return {"status": "error", "message": f"Image not found: {img_path}"} mat_name = f"Mat_{obj.name}" if mat_name in bpy.data.materials: mat = bpy.data.materials[mat_name] else: mat = bpy.data.materials.new(name=mat_name) mat.use_nodes = True nodes = mat.node_tree.nodes nodes.clear() bsdf = nodes.new(type='ShaderNodeBsdfPrincipled') bsdf.location = (0, 0) output = nodes.new(type='ShaderNodeOutputMaterial') output.location = (300, 0) tex_image = nodes.new(type='ShaderNodeTexImage') tex_image.location = (-300, 0) try: img = bpy.data.images.load(img_path) tex_image.image = img except Exception as e: return {"status": "error", "message": f"Load image failed: {str(e)}"} mat.node_tree.links.new(bsdf.outputs['BSDF'], output.inputs['Surface']) mat.node_tree.links.new(tex_image.outputs['Color'], bsdf.inputs['Base Color']) if obj.data.materials: obj.data.materials[0] = mat else: obj.data.materials.append(mat) return {"status": "success", "output": f"Texture applied to {obj.name}"} else: return {"status": "error", "message": f"Unknown command: {cmd_type}"} except Exception as e: traceback.print_exc() return {"status": "error", "message": str(e)} def handle_client(conn): """Handle individual client connection""" try: while is_server_running: try: conn.settimeout(1.0) # Allow checking is_server_running periodically data = conn.recv(4096) except socket.timeout: continue except Exception: break if not data: break try: # Parse request request = json.loads(data.decode('utf-8')) # Create a result container result_container = {} # Define a wrapper to run and capture result def run_job(): result_container['data'] = execute_command(request) # Put job in queue execution_queue.put(run_job) # Wait for result (simple polling for this example) # In a real app, we might use an event or condition variable while 'data' not in result_container and is_server_running: pass if 'data' in result_container: # Send response response = json.dumps(result_container['data']) conn.sendall(response.encode('utf-8')) except json.JSONDecodeError: conn.sendall(json.dumps({"status": "error", "message": "Invalid JSON"}).encode('utf-8')) except Exception as e: print(f"Connection error: {e}") finally: conn.close() def start_server_loop(): """Start the socket server in a background thread""" global server_socket, is_server_running server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: server_socket.bind((HOST, PORT)) server_socket.listen(1) server_socket.settimeout(1.0) # Non-blocking accept to allow shutdown print(f"MCP Connector listening on {HOST}:{PORT}") while is_server_running: try: conn, addr = server_socket.accept() print(f"Connected by {addr}") client_thread = threading.Thread(target=handle_client, args=(conn,)) client_thread.daemon = True client_thread.start() except socket.timeout: continue except OSError: break except Exception as e: print(f"Server error: {e}") finally: if server_socket: server_socket.close() print("MCP Server Stopped") def process_queue(): """Process the execution queue on the main thread""" while not execution_queue.empty(): job = execution_queue.get() job() return 0.1 # Run every 0.1 seconds # --- UI & Operators --- class MCP_OT_StartServer(bpy.types.Operator): """Start the MCP Server""" bl_idname = "mcp.start_server" bl_label = "Start Server" def execute(self, context): global server_thread, is_server_running if not is_server_running: is_server_running = True server_thread = threading.Thread(target=start_server_loop) server_thread.daemon = True server_thread.start() if not bpy.app.timers.is_registered(process_queue): bpy.app.timers.register(process_queue) self.report({'INFO'}, "MCP Server Started") return {'FINISHED'} class MCP_OT_StopServer(bpy.types.Operator): """Stop the MCP Server""" bl_idname = "mcp.stop_server" bl_label = "Stop Server" def execute(self, context): global is_server_running, server_socket if is_server_running: is_server_running = False # Socket close will trigger exception in accept loop if server_socket: try: server_socket.close() except: pass if bpy.app.timers.is_registered(process_queue): bpy.app.timers.unregister(process_queue) self.report({'INFO'}, "MCP Server Stopping...") return {'FINISHED'} class MCP_PT_Panel(bpy.types.Panel): """Creates a Panel in the 3D View Sidebar""" bl_label = "MCP Connector" bl_idname = "MCP_PT_main_panel" bl_space_type = 'VIEW_3D' bl_region_type = 'UI' bl_category = "MCP" def draw(self, context): layout = self.layout global is_server_running box = layout.box() row = box.row() if is_server_running: row.label(text="Status: Connected", icon='CHECKBOX_HLT') row = box.row() row.operator("mcp.stop_server", icon='CANCEL') row = box.row() row.label(text=f"Listening on {HOST}:{PORT}") else: row.label(text="Status: Disconnected", icon='CHECKBOX_DEHLT') row = box.row() row.operator("mcp.start_server", icon='PLAY') classes = ( MCP_OT_StartServer, MCP_OT_StopServer, MCP_PT_Panel, ) def register(): for cls in classes: bpy.utils.register_class(cls) # Auto-start optional, but let's default to manual for safety as requested # bpy.ops.mcp.start_server() def unregister(): global is_server_running is_server_running = False # Signal threads to stop for cls in reversed(classes): bpy.utils.unregister_class(cls) if bpy.app.timers.is_registered(process_queue): bpy.app.timers.unregister(process_queue) if __name__ == "__main__": register()