Fixing MCPServer startup for claude code
This commit is contained in:
ange committed
1 parent
afabda86cf
commit
cb3e50d7c7
33 files changed
+7090
-1
No files matched your search
@@ -0,0 +1,366 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Test script for blueprint input mapping via MCP.
|
||||
This script creates input mappings and connects them to functions in a Blueprint.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import socket
|
||||
import json
|
||||
import logging
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
# Add the parent directory to the path so we can import the server module
|
||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger("TestInputMapping")
|
||||
|
||||
def send_command(sock: socket.socket, command: str, params: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
"""Send a command to the Unreal MCP server and get the response."""
|
||||
try:
|
||||
# Create command object
|
||||
command_obj = {
|
||||
"type": command,
|
||||
"params": params
|
||||
}
|
||||
|
||||
# Convert to JSON and send
|
||||
command_json = json.dumps(command_obj)
|
||||
logger.info(f"Sending command: {command_json}")
|
||||
sock.sendall(command_json.encode('utf-8'))
|
||||
|
||||
# Receive response
|
||||
chunks = []
|
||||
while True:
|
||||
chunk = sock.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
|
||||
# Try parsing to see if we have a complete response
|
||||
try:
|
||||
data = b''.join(chunks)
|
||||
json.loads(data.decode('utf-8'))
|
||||
# If we can parse it, we have the complete response
|
||||
break
|
||||
except json.JSONDecodeError:
|
||||
# Not a complete JSON object yet, continue receiving
|
||||
continue
|
||||
|
||||
# Parse response
|
||||
data = b''.join(chunks)
|
||||
response = json.loads(data.decode('utf-8'))
|
||||
logger.info(f"Received response: {response}")
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending command: {e}")
|
||||
return None
|
||||
|
||||
def setup_input_mapping(sock: socket.socket, action_name: str, key: str, input_type: str = "Action") -> bool:
|
||||
"""Helper function to set up an input mapping."""
|
||||
input_params = {
|
||||
"action_name": action_name,
|
||||
"key": key,
|
||||
"input_type": input_type
|
||||
}
|
||||
|
||||
response = send_command(sock, "create_input_mapping", input_params)
|
||||
|
||||
success = (response and
|
||||
response.get("status") == "success" and
|
||||
response.get("result", {}).get("success"))
|
||||
|
||||
if success:
|
||||
logger.info(f"Input mapping '{action_name}' created with key '{key}'")
|
||||
else:
|
||||
logger.error(f"Failed to create input mapping: {response}")
|
||||
|
||||
return success
|
||||
|
||||
def main():
|
||||
"""Main function to test input mappings in blueprints."""
|
||||
try:
|
||||
# Step 1: Create a controller blueprint
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
bp_params = {
|
||||
"name": "InputControllerBP",
|
||||
"parent_class": "Actor"
|
||||
}
|
||||
|
||||
response = send_command(sock, "create_blueprint", bp_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to create blueprint: {response}")
|
||||
return
|
||||
|
||||
# Check if blueprint already existed
|
||||
if response.get("result", {}).get("already_exists"):
|
||||
logger.info(f"Blueprint 'InputControllerBP' already exists, reusing it")
|
||||
else:
|
||||
logger.info("Controller blueprint created successfully!")
|
||||
|
||||
# Close and reopen connection for each command
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Step 2: Add variables to track state
|
||||
var_params_list = [
|
||||
{
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"variable_name": "Score",
|
||||
"variable_type": "Integer",
|
||||
"default_value": 0,
|
||||
"is_exposed": True
|
||||
},
|
||||
{
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"variable_name": "IsGameActive",
|
||||
"variable_type": "Boolean",
|
||||
"default_value": True,
|
||||
"is_exposed": True
|
||||
},
|
||||
{
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"variable_name": "PlayerName",
|
||||
"variable_type": "String",
|
||||
"default_value": "Player1",
|
||||
"is_exposed": True
|
||||
}
|
||||
]
|
||||
|
||||
for var_params in var_params_list:
|
||||
response = send_command(sock, "add_blueprint_variable", var_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to add variable: {response}")
|
||||
return
|
||||
|
||||
logger.info(f"Variable {var_params['variable_name']} added successfully!")
|
||||
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Step 3: Set up input mappings for a simple game controller
|
||||
input_mappings = [
|
||||
("Jump", "SpaceBar", "Action"),
|
||||
("Pause", "P", "Action"),
|
||||
("Restart", "R", "Action"),
|
||||
("MoveForward", "W", "Axis"),
|
||||
("MoveRight", "D", "Axis")
|
||||
]
|
||||
|
||||
for action_name, key, input_type in input_mappings:
|
||||
success = setup_input_mapping(sock, action_name, key, input_type)
|
||||
if not success:
|
||||
return
|
||||
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Step 4: Add event nodes for BeginPlay and input actions
|
||||
event_node_ids = {}
|
||||
|
||||
# BeginPlay event
|
||||
begin_play_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"event_type": "BeginPlay",
|
||||
"node_position": [0, 0]
|
||||
}
|
||||
|
||||
response = send_command(sock, "add_blueprint_event_node", begin_play_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to add BeginPlay event node: {response}")
|
||||
return
|
||||
|
||||
logger.info("BeginPlay event node added successfully!")
|
||||
event_node_ids["BeginPlay"] = response.get("result", {}).get("node_id")
|
||||
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Step 5: Add function nodes for different actions
|
||||
function_node_ids = {}
|
||||
|
||||
# Print String function for BeginPlay
|
||||
function_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"target": "self",
|
||||
"function_name": "PrintString",
|
||||
"params": {
|
||||
"InString": "Input Controller Initialized",
|
||||
"Duration": 5.0
|
||||
},
|
||||
"node_position": [250, 0]
|
||||
}
|
||||
|
||||
response = send_command(sock, "add_blueprint_function_node", function_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to add PrintString function node: {response}")
|
||||
return
|
||||
|
||||
logger.info("PrintString function node added successfully!")
|
||||
function_node_ids["PrintInit"] = response.get("result", {}).get("node_id")
|
||||
|
||||
# For each action, add a event node and function node
|
||||
action_positions = {
|
||||
"Jump": [0, 150],
|
||||
"Pause": [0, 300],
|
||||
"Restart": [0, 450],
|
||||
"MoveForward": [0, 600],
|
||||
"MoveRight": [0, 750]
|
||||
}
|
||||
|
||||
function_positions = {
|
||||
"Jump": [250, 150],
|
||||
"Pause": [250, 300],
|
||||
"Restart": [250, 450],
|
||||
"MoveForward": [250, 600],
|
||||
"MoveRight": [250, 750]
|
||||
}
|
||||
|
||||
# Since we can't directly create InputAction nodes, we'll simulate with BeginPlay
|
||||
for action_name in ["Jump", "Pause", "Restart"]:
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Create placeholder event node
|
||||
event_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"event_type": "BeginPlay", # Using BeginPlay as a placeholder for InputAction
|
||||
"node_position": action_positions[action_name]
|
||||
}
|
||||
|
||||
response = send_command(sock, "add_blueprint_event_node", event_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to add event node for {action_name}: {response}")
|
||||
return
|
||||
|
||||
logger.info(f"Event node for {action_name} added (simulated)")
|
||||
event_node_ids[action_name] = response.get("result", {}).get("node_id")
|
||||
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Create function node to print what action was performed
|
||||
function_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"target": "self",
|
||||
"function_name": "PrintString",
|
||||
"params": {
|
||||
"InString": f"{action_name} Action Triggered",
|
||||
"Duration": 2.0
|
||||
},
|
||||
"node_position": function_positions[action_name]
|
||||
}
|
||||
|
||||
response = send_command(sock, "add_blueprint_function_node", function_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to add function node for {action_name}: {response}")
|
||||
return
|
||||
|
||||
logger.info(f"Function node for {action_name} added successfully!")
|
||||
function_node_ids[action_name] = response.get("result", {}).get("node_id")
|
||||
|
||||
# Step 6: Connect nodes
|
||||
for action_name in ["BeginPlay"] + list(action_positions.keys())[:3]: # BeginPlay + first 3 actions
|
||||
# Close and reopen connection
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
# Connect appropriate function based on event type
|
||||
if action_name == "BeginPlay":
|
||||
target_function = "PrintInit"
|
||||
else:
|
||||
target_function = action_name
|
||||
|
||||
connect_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"source_node_id": event_node_ids[action_name],
|
||||
"source_pin": "Then", # Execute pin on event
|
||||
"target_node_id": function_node_ids[target_function],
|
||||
"target_pin": "execute" # Execute pin on function
|
||||
}
|
||||
|
||||
response = send_command(sock, "connect_blueprint_nodes", connect_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to connect nodes for {action_name}: {response}")
|
||||
return
|
||||
|
||||
logger.info(f"Connected {action_name} event to function successfully!")
|
||||
|
||||
# Step 7: Compile the blueprint
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
compile_params = {
|
||||
"blueprint_name": "InputControllerBP"
|
||||
}
|
||||
|
||||
response = send_command(sock, "compile_blueprint", compile_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to compile blueprint: {response}")
|
||||
return
|
||||
|
||||
logger.info("Blueprint compiled successfully!")
|
||||
|
||||
# Step 8: Spawn the controller in the level
|
||||
sock.close()
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sock.connect(("127.0.0.1", 55557))
|
||||
|
||||
spawn_params = {
|
||||
"blueprint_name": "InputControllerBP",
|
||||
"actor_name": "InputController",
|
||||
"location": [0.0, 0.0, 100.0],
|
||||
"rotation": [0.0, 0.0, 0.0],
|
||||
"scale": [1.0, 1.0, 1.0]
|
||||
}
|
||||
|
||||
response = send_command(sock, "spawn_blueprint_actor", spawn_params)
|
||||
|
||||
if not response or response.get("status") != "success" or not response.get("result", {}).get("success"):
|
||||
logger.error(f"Failed to spawn blueprint actor: {response}")
|
||||
return
|
||||
|
||||
logger.info("Input controller spawned successfully!")
|
||||
logger.info("The controller will run the BeginPlay event and show a message.")
|
||||
logger.info("The following input mappings have been set up:")
|
||||
|
||||
for action_name, key, input_type in input_mappings:
|
||||
logger.info(f" - {action_name}: {key} ({input_type})")
|
||||
|
||||
# Close final connection
|
||||
sock.close()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user