Unreal Engine MCP Bridge

""" Unreal Engine MCP Server A simple MCP server for interacting with Unreal Engine. """ import logging import socket import sys import json from contextlib import asynccontextmanager from typing import AsyncIterator, Dict, Any, Optional from mcp.server.fastmcp import FastMCP # Configure logging with more detailed format logging.basicConfig( level=logging.DEBUG, # Change to DEBUG level for more details format='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s', handlers=[ logging.FileHandler('unreal_mcp.log'), # logging.StreamHandler(sys.stdout) # Remove this handler to unexpected non-whitespace characters in JSON ] ) logger = logging.getLogger("UnrealMCP") # Configuration UNREAL_HOST = "127.0.0.1" UNREAL_PORT = 55557 class UnrealConnection: """Connection to an Unreal Engine instance.""" def __init__(self): """Initialize the connection.""" self.socket = None self.connected = False def connect(self) -> bool: """Connect to the Unreal Engine instance.""" try: # Close any existing socket if self.socket: try: self.socket.close() except: pass self.socket = None logger.info(f"Connecting to Unreal at {UNREAL_HOST}:{UNREAL_PORT}...") self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(5) # 5 second timeout # Set socket options for better stability self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # Set larger buffer sizes self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536) self.socket.connect((UNREAL_HOST, UNREAL_PORT)) self.connected = True logger.info("Connected to Unreal Engine") return True except Exception as e: logger.error(f"Failed to connect to Unreal: {e}") self.connected = False return False def disconnect(self): """Disconnect from the Unreal Engine instance.""" if self.socket: try: self.socket.close() except: pass self.socket = None self.connected = False def receive_full_response(self, sock, buffer_size=4096) -> bytes: """Receive a complete response from Unreal, handling chunked data.""" chunks = [] sock.settimeout(5) # 5 second timeout try: while True: chunk = sock.recv(buffer_size) if not chunk: if not chunks: raise Exception("Connection closed before receiving data") break chunks.append(chunk) # Process the data received so far data = b''.join(chunks) decoded_data = data.decode('utf-8') # Try to parse as JSON to check if complete try: json.loads(decoded_data) logger.info(f"Received complete response ({len(data)} bytes)") return data except json.JSONDecodeError: # Not complete JSON yet, continue reading logger.debug(f"Received partial response, waiting for more data...") continue except Exception as e: logger.warning(f"Error processing response chunk: {str(e)}") continue except socket.timeout: logger.warning("Socket timeout during receive") if chunks: # If we have some data already, try to use it data = b''.join(chunks) try: json.loads(data.decode('utf-8')) logger.info(f"Using partial response after timeout ({len(data)} bytes)") return data except: pass raise Exception("Timeout receiving Unreal response") except Exception as e: logger.error(f"Error during receive: {str(e)}") raise def send_command(self, command: str, params: Dict[str, Any] = None) -> Optional[Dict[str, Any]]: """Send a command to Unreal Engine and get the response.""" # Always reconnect for each command, since Unreal closes the connection after each command # This is different from Unity which keeps connections alive if self.socket: try: self.socket.close() except: pass self.socket = None self.connected = False if not self.connect(): logger.error("Failed to connect to Unreal Engine for command") return None try: # Match Unity's command format exactly command_obj = { "type": command, # Use "type" instead of "command" "params": params or {} # Use Unity's params or {} pattern } # Send without newline, exactly like Unity command_json = json.dumps(command_obj) logger.info(f"Sending command: {command_json}") self.socket.sendall(command_json.encode('utf-8')) # Read response using improved handler response_data = self.receive_full_response(self.socket) response = json.loads(response_data.decode('utf-8')) # Log complete response for debugging logger.info(f"Complete response from Unreal: {response}") # Check for both error formats: {"status": "error", ...} and {"success": false, ...} if response.get("status") == "error": error_message = response.get("error") or response.get("message", "Unknown Unreal error") logger.error(f"Unreal error (status=error): {error_message}") # We want to preserve the original error structure but ensure error is accessible if "error" not in response: response["error"] = error_message elif response.get("success") is False: # This format uses {"success": false, "error": "message"} or {"success": false, "message": "message"} error_message = response.get("error") or response.get("message", "Unknown Unreal error") logger.error(f"Unreal error (success=false): {error_message}") # Convert to the standard format expected by higher layers response = { "status": "error", "error": error_message } # Always close the connection after command is complete # since Unreal will close it on its side anyway try: self.socket.close() except: pass self.socket = None self.connected = False return response except Exception as e: logger.error(f"Error sending command: {e}") # Always reset connection state on any error self.connected = False try: self.socket.close() except: pass self.socket = None return { "status": "error", "error": str(e) } # Global connection state _unreal_connection: UnrealConnection = None def get_unreal_connection() -> Optional[UnrealConnection]: """Get the connection to Unreal Engine.""" global _unreal_connection try: if _unreal_connection is None: _unreal_connection = UnrealConnection() if not _unreal_connection.connect(): logger.warning("Could not connect to Unreal Engine") _unreal_connection = None else: # Verify connection is still valid with a ping-like test try: # Simple test by sending an empty buffer to check if socket is still connected _unreal_connection.socket.sendall(b'\x00') logger.debug("Connection verified with ping test") except Exception as e: logger.warning(f"Existing connection failed: {e}") _unreal_connection.disconnect() _unreal_connection = None # Try to reconnect _unreal_connection = UnrealConnection() if not _unreal_connection.connect(): logger.warning("Could not reconnect to Unreal Engine") _unreal_connection = None else: logger.info("Successfully reconnected to Unreal Engine") return _unreal_connection except Exception as e: logger.error(f"Error getting Unreal connection: {e}") return None @asynccontextmanager async def server_lifespan(server: FastMCP) -> AsyncIterator[Dict[str, Any]]: """Handle server startup and shutdown.""" global _unreal_connection logger.info("UnrealMCP server starting up") try: _unreal_connection = get_unreal_connection() if _unreal_connection: logger.info("Connected to Unreal Engine on startup") else: logger.warning("Could not connect to Unreal Engine on startup") except Exception as e: logger.error(f"Error connecting to Unreal Engine on startup: {e}") _unreal_connection = None try: yield {} finally: if _unreal_connection: _unreal_connection.disconnect() _unreal_connection = None logger.info("Unreal MCP server shut down") # Initialize server mcp = FastMCP( "UnrealMCP", description="Unreal Engine integration via Model Context Protocol", lifespan=server_lifespan ) # Import and register tools from tools.editor_tools import register_editor_tools from tools.blueprint_tools import register_blueprint_tools from tools.node_tools import register_blueprint_node_tools from tools.project_tools import register_project_tools from tools.umg_tools import register_umg_tools # Register tools register_editor_tools(mcp) register_blueprint_tools(mcp) register_blueprint_node_tools(mcp) register_project_tools(mcp) register_umg_tools(mcp) @mcp.prompt() def info(): """Information about available Unreal MCP tools and best practices.""" return """ # Unreal MCP Server Tools and Best Practices ## UMG (Widget Blueprint) Tools - `create_umg_widget_blueprint(widget_name, parent_class="UserWidget", path="/Game/UI")` Create a new UMG Widget Blueprint - `add_text_block_to_widget(widget_name, text_block_name, text="", position=[0,0], size=[200,50], font_size=12, color=[1,1,1,1])` Add a Text Block widget with customizable properties - `add_button_to_widget(widget_name, button_name, text="", position=[0,0], size=[200,50], font_size=12, color=[1,1,1,1], background_color=[0.1,0.1,0.1,1])` Add a Button widget with text and styling - `bind_widget_event(widget_name, widget_component_name, event_name, function_name="")` Bind events like OnClicked to functions - `add_widget_to_viewport(widget_name, z_order=0)` Add widget instance to game viewport - `set_text_block_binding(widget_name, text_block_name, binding_property, binding_type="Text")` Set up dynamic property binding for text blocks ## Editor Tools ### Viewport and Screenshots - `focus_viewport(target, location, distance, orientation)` - Focus viewport - `take_screenshot(filename, show_ui, resolution)` - Capture screenshots ### Actor Management - `get_actors_in_level()` - List all actors in current level - `find_actors_by_name(pattern)` - Find actors by name pattern - `spawn_actor(name, type, location=[0,0,0], rotation=[0,0,0], scale=[1,1,1])` - Create actors - `delete_actor(name)` - Remove actors - `set_actor_transform(name, location, rotation, scale)` - Modify actor transform - `get_actor_properties(name)` - Get actor properties ## Blueprint Management - `create_blueprint(name, parent_class)` - Create new Blueprint classes - `add_component_to_blueprint(blueprint_name, component_type, component_name)` - Add components - `set_static_mesh_properties(blueprint_name, component_name, static_mesh)` - Configure meshes - `set_physics_properties(blueprint_name, component_name)` - Configure physics - `compile_blueprint(blueprint_name)` - Compile Blueprint changes - `set_blueprint_property(blueprint_name, property_name, property_value)` - Set properties - `set_pawn_properties(blueprint_name)` - Configure Pawn settings - `spawn_blueprint_actor(blueprint_name, actor_name)` - Spawn Blueprint actors ## Blueprint Node Management - `add_blueprint_event_node(blueprint_name, event_type)` - Add event nodes - `add_blueprint_input_action_node(blueprint_name, action_name)` - Add input nodes - `add_blueprint_function_node(blueprint_name, target, function_name)` - Add function nodes - `connect_blueprint_nodes(blueprint_name, source_node_id, source_pin, target_node_id, target_pin)` - Connect nodes - `add_blueprint_variable(blueprint_name, variable_name, variable_type)` - Add variables - `add_blueprint_get_self_component_reference(blueprint_name, component_name)` - Add component refs - `add_blueprint_self_reference(blueprint_name)` - Add self references - `find_blueprint_nodes(blueprint_name, node_type, event_type)` - Find nodes ## Project Tools - `create_input_mapping(action_name, key, input_type)` - Create input mappings ## Best Practices ### UMG Widget Development - Create widgets with descriptive names that reflect their purpose - Use consistent naming conventions for widget components - Organize widget hierarchy logically - Set appropriate anchors and alignment for responsive layouts - Use property bindings for dynamic updates instead of direct setting - Handle widget events appropriately with meaningful function names - Clean up widgets when no longer needed - Test widget layouts at different resolutions ### Editor and Actor Management - Use unique names for actors to avoid conflicts - Clean up temporary actors - Validate transforms before applying - Check actor existence before modifications - Take regular viewport screenshots during development - Keep the viewport focused on relevant actors during operations ### Blueprint Development - Compile Blueprints after changes - Use meaningful names for variables and functions - Organize nodes logically - Test functionality in isolation - Consider performance implications - Document complex setups ### Error Handling - Check command responses for success - Handle errors gracefully - Log important operations - Validate parameters - Clean up resources on errors """ # Run the server if __name__ == "__main__": logger.info("Starting MCP server with stdio transport") mcp.run(transport='stdio')