UnrealBlueprintMCP

""" Unreal Blueprint MCP Server - Phase 1 WebSocket Bridge Implementation 언리얼 엔진 블루프린트 조작을 위한 MCP 서버 (WebSocket 브리지 방식) FastMCP를 사용하여 MCP 표준을 준수하면서 기존 언리얼 플러그인과 호환성을 보장합니다. """ import asyncio import logging import time import threading from typing import Any, Dict from fastmcp import FastMCP from starlette.websockets import WebSocket, WebSocketDisconnect from starlette.routing import WebSocketRoute import websockets import json from models import ( BlueprintCreateParams, BlueprintCreateResponse, BlueprintPropertyParams, BlueprintPropertyResponse, BlueprintCompileParams, BlueprintCompileResponse, BlueprintNodeAddParams, BlueprintNodeAddResponse, ConnectionTestParams, ConnectionTestResponse, ServerStatusResponse, SupportedClassesResponse, ErrorResponse ) from connection_manager import connection_manager # 로깅 설정 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # FastMCP 인스턴스 생성 mcp = FastMCP("UnrealBlueprintMCPServer") # 서버 시작 시간 (상태 조회용) SERVER_START_TIME = time.time() SERVER_VERSION = "1.0.0" @mcp.tool() async def create_blueprint(params: BlueprintCreateParams) -> BlueprintCreateResponse: """ 새로운 블루프린트를 생성합니다. Args: params: 블루프린트 생성 파라미터 Returns: BlueprintCreateResponse: 생성 결과 """ try: logger.info(f"Creating blueprint: {params.blueprint_name}") # 언리얼로 명령 전송 unreal_params = { "blueprint_name": params.blueprint_name, "parent_class": params.parent_class, "asset_path": params.asset_path } result = await connection_manager.send_to_unreal("create_blueprint", unreal_params) # 응답 생성 blueprint_path = f"{params.asset_path.rstrip('/')}/{params.blueprint_name}" return BlueprintCreateResponse( success=True, message=f"Blueprint '{params.blueprint_name}' created successfully.", blueprint_path=blueprint_path, unreal_response=result ) except ConnectionError as e: logger.error(f"Connection error creating blueprint: {e}") return BlueprintCreateResponse( success=False, message=f"Connection error: {str(e)}", blueprint_path=None, unreal_response=None ) except Exception as e: logger.error(f"Error creating blueprint: {e}") return BlueprintCreateResponse( success=False, message=f"Error: {str(e)}", blueprint_path=None, unreal_response=None ) @mcp.tool() async def set_blueprint_property(params: BlueprintPropertyParams) -> BlueprintPropertyResponse: """ 블루프린트의 속성을 설정합니다. Args: params: 속성 설정 파라미터 Returns: BlueprintPropertyResponse: 설정 결과 """ try: logger.info(f"Setting blueprint property: {params.blueprint_path} -> {params.property_name}") # 언리얼로 명령 전송 unreal_params = { "blueprint_path": params.blueprint_path, "property_name": params.property_name, "property_value": params.property_value, "property_type": params.property_type } result = await connection_manager.send_to_unreal("set_blueprint_property", unreal_params) return BlueprintPropertyResponse( success=True, message=f"Property '{params.property_name}' set successfully.", property_name=params.property_name, old_value=result.get("old_value"), new_value=params.property_value ) except ConnectionError as e: logger.error(f"Connection error setting property: {e}") return BlueprintPropertyResponse( success=False, message=f"Connection error: {str(e)}", property_name=params.property_name, old_value=None, new_value=params.property_value ) except Exception as e: logger.error(f"Error setting property: {e}") return BlueprintPropertyResponse( success=False, message=f"Error: {str(e)}", property_name=params.property_name, old_value=None, new_value=params.property_value ) @mcp.tool() async def compile_blueprint(params: BlueprintCompileParams) -> BlueprintCompileResponse: """ 블루프린트를 컴파일합니다. Args: params: 컴파일 파라미터 Returns: BlueprintCompileResponse: 컴파일 결과 """ try: logger.info(f"Compiling blueprint: {params.blueprint_path}") unreal_params = {"blueprint_path": params.blueprint_path} result = await connection_manager.send_to_unreal("compile_blueprint", unreal_params) return BlueprintCompileResponse( success=result.get("success", True), message=result.get("message", "Blueprint compiled successfully."), compile_errors=result.get("errors", []), compile_warnings=result.get("warnings", []) ) except ConnectionError as e: logger.error(f"Connection error compiling blueprint: {e}") return BlueprintCompileResponse( success=False, message=f"Connection error: {str(e)}", compile_errors=[str(e)], compile_warnings=[] ) except Exception as e: logger.error(f"Error compiling blueprint: {e}") return BlueprintCompileResponse( success=False, message=f"Error: {str(e)}", compile_errors=[str(e)], compile_warnings=[] ) @mcp.tool() async def add_blueprint_node(params: BlueprintNodeAddParams) -> BlueprintNodeAddResponse: """ 블루프린트에 노드를 추가합니다. Args: params: 노드 추가 파라미터 Returns: BlueprintNodeAddResponse: 노드 추가 결과 """ try: logger.info(f"Adding node to blueprint: {params.blueprint_path} -> {params.node_class}") unreal_params = { "blueprint_path": params.blueprint_path, "node_class": params.node_class, "position_x": params.position_x, "position_y": params.position_y, "node_name": params.node_name } result = await connection_manager.send_to_unreal("add_blueprint_node", unreal_params) return BlueprintNodeAddResponse( success=True, message=f"Node '{params.node_class}' added successfully.", node_id=result.get("node_id"), node_name=result.get("node_name", params.node_name) ) except ConnectionError as e: logger.error(f"Connection error adding node: {e}") return BlueprintNodeAddResponse( success=False, message=f"Connection error: {str(e)}", node_id=None, node_name=params.node_name ) except Exception as e: logger.error(f"Error adding node: {e}") return BlueprintNodeAddResponse( success=False, message=f"Error: {str(e)}", node_id=None, node_name=params.node_name ) @mcp.tool() async def test_unreal_connection(params: ConnectionTestParams) -> ConnectionTestResponse: """ 언리얼 엔진과의 연결을 테스트합니다. Args: params: 연결 테스트 파라미터 Returns: ConnectionTestResponse: 테스트 결과 """ try: start_time = time.time() logger.info("Testing Unreal Engine connection") if not connection_manager.is_connected(): return ConnectionTestResponse( success=False, message="No Unreal Engine client is connected", response_time_ms=0.0, unreal_version=None ) # ping 명령으로 연결 테스트 result = await connection_manager.send_to_unreal("ping", {}, timeout=params.timeout) response_time = (time.time() - start_time) * 1000 # 밀리초로 변환 return ConnectionTestResponse( success=True, message="Connection test successful", response_time_ms=response_time, unreal_version=result.get("unreal_version", "Unknown") ) except Exception as e: response_time = (time.time() - start_time) * 1000 logger.error(f"Connection test failed: {e}") return ConnectionTestResponse( success=False, message=f"Connection test failed: {str(e)}", response_time_ms=response_time, unreal_version=None ) @mcp.tool() async def get_server_status() -> ServerStatusResponse: """ MCP 서버의 상태를 조회합니다. Returns: ServerStatusResponse: 서버 상태 정보 """ uptime = time.time() - SERVER_START_TIME return ServerStatusResponse( success=True, server_version=SERVER_VERSION, unreal_connected=connection_manager.is_connected(), active_connections=1 if connection_manager.is_connected() else 0, uptime_seconds=uptime ) @mcp.tool() async def list_supported_blueprint_classes() -> SupportedClassesResponse: """ 지원되는 블루프린트 부모 클래스 목록을 조회합니다. Returns: SupportedClassesResponse: 지원 클래스 목록 """ try: logger.info("Listing supported blueprint classes") result = await connection_manager.send_to_unreal("list_supported_classes", {}) classes = result.get("classes", []) return SupportedClassesResponse( success=True, classes=classes, total_count=len(classes) ) except ConnectionError as e: logger.error(f"Connection error listing classes: {e}") # 기본 클래스 목록 반환 default_classes = ["Actor", "Pawn", "Character", "PlayerController", "GameMode", "HUD"] return SupportedClassesResponse( success=False, classes=default_classes, total_count=len(default_classes) ) except Exception as e: logger.error(f"Error listing classes: {e}") default_classes = ["Actor", "Pawn", "Character", "PlayerController", "GameMode", "HUD"] return SupportedClassesResponse( success=False, classes=default_classes, total_count=len(default_classes) ) # WebSocket 엔드포인트 (기존 언리얼 플러그인과 통신) - Starlette 버전 async def websocket_endpoint(websocket: WebSocket): """ 기존 언리얼 플러그인과 호환되는 WebSocket 엔드포인트 (Starlette) JSON-RPC 2.0 메시지를 처리합니다. """ await websocket.accept() logger.info("Unreal Engine client connected to WebSocket (Starlette)") try: # 연결 등록 await connection_manager.connect(websocket) # 메시지 수신 루프 while True: try: # 언리얼 플러그인으로부터 메시지 수신 data = await websocket.receive_text() logger.debug(f"Received from Unreal: {data}") # JSON 파싱 message = json.loads(data) # 응답 처리 await connection_manager.handle_unreal_response(message) except json.JSONDecodeError as e: logger.error(f"Invalid JSON from Unreal client: {e}") # 잘못된 JSON은 무시하고 연결 유지 except WebSocketDisconnect: logger.info("Unreal Engine client disconnected") except Exception as e: logger.error(f"WebSocket error: {e}") finally: # 연결 해제 await connection_manager.disconnect() # 독립적인 WebSocket 서버 (websockets 라이브러리 사용) class WebSocketWrapper: """websockets 라이브러리의 WebSocket을 Starlette WebSocket처럼 사용하기 위한 래퍼""" def __init__(self, websocket): self.websocket = websocket async def send_text(self, text: str): await self.websocket.send(text) async def close(self): await self.websocket.close() async def standalone_websocket_handler(websocket, path): """ 독립적인 WebSocket 서버 핸들러 (websockets 라이브러리 사용) fastmcp run 모드에서 사용됩니다. """ logger.info("Unreal Engine client connected to WebSocket (standalone)") # WebSocket 래퍼 생성 wrapped_websocket = WebSocketWrapper(websocket) try: # 연결 등록 await connection_manager.connect(wrapped_websocket) # 메시지 수신 루프 async for message_str in websocket: try: logger.debug(f"Received from Unreal: {message_str}") # JSON 파싱 message = json.loads(message_str) # 응답 처리 await connection_manager.handle_unreal_response(message) except json.JSONDecodeError as e: logger.error(f"Invalid JSON from Unreal client: {e}") # 잘못된 JSON은 무시하고 연결 유지 except websockets.exceptions.ConnectionClosed: logger.info("Unreal Engine client disconnected (standalone)") except Exception as e: logger.error(f"WebSocket error: {e}") finally: # 연결 해제 await connection_manager.disconnect() # 독립적인 WebSocket 서버 시작 def start_standalone_websocket_server(): """별도 스레드에서 독립적인 WebSocket 서버를 시작합니다.""" async def run_server(): logger.info("Starting standalone WebSocket server on ws://localhost:8001/ws") # websockets 서버 시작 server = await websockets.serve( standalone_websocket_handler, "localhost", 8001, subprotocols=[] ) logger.info("Standalone WebSocket server started successfully") # 서버 실행 유지 await server.wait_closed() def run_in_thread(): # 새 이벤트 루프에서 실행 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(run_server()) except Exception as e: logger.error(f"Standalone WebSocket server error: {e}") finally: loop.close() # 백그라운드 스레드에서 실행 thread = threading.Thread(target=run_in_thread, daemon=True) thread.start() return thread # FastMCP에 WebSocket 라우트 추가 def add_websocket_route(): """FastMCP 서버에 WebSocket 라우트를 추가합니다.""" websocket_route = WebSocketRoute("/ws", endpoint=websocket_endpoint) # FastMCP 서버에 추가 HTTP 라우트로 등록 if not hasattr(mcp, '_additional_http_routes'): mcp._additional_http_routes = [] mcp._additional_http_routes.append(websocket_route) logger.info("WebSocket route added to FastMCP server at /ws") # 서버 초기화 def initialize_server(): """서버 초기화 작업을 수행합니다.""" logger.info(f"Starting Unreal Blueprint MCP Server v{SERVER_VERSION}") # 개발 모드 (fastmcp dev)에서는 Starlette 라우트 추가 add_websocket_route() # 운영 모드 (fastmcp run)에서는 독립적인 WebSocket 서버 시작 websocket_thread = start_standalone_websocket_server() logger.info("Server initialization completed") logger.info("WebSocket endpoints available:") logger.info(" - ws://localhost:8000/ws (fastmcp dev mode)") logger.info(" - ws://localhost:8001/ws (fastmcp run mode)") return websocket_thread # 메인 실행부 if __name__ == "__main__": initialize_server() logger.info("Use 'fastmcp run unreal_blueprint_mcp_server.py' to start the server")