297 lines
10 KiB
Python
297 lines
10 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Maya MCP Server (FastAPI Implementation)
|
|
Core implementation of the Model Context Protocol server for Maya using FastAPI.
|
|
|
|
Version: 1.0.0
|
|
Author: Jeffrey Tsai
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import traceback
|
|
import maya.cmds as cmds
|
|
import maya.api.OpenMaya as om
|
|
from port_config import SERVER_PORT
|
|
from log_config import get_logger, initialize_logging
|
|
|
|
# Initialize logging
|
|
initialize_logging()
|
|
logger = get_logger("Server")
|
|
|
|
# Try to import FastAPI server, fall back to HTTP handler if not available
|
|
USE_FASTAPI = True # Default to True
|
|
|
|
try:
|
|
from fastapi_server import start_server as fastapi_start_server
|
|
from fastapi_server import stop_server as fastapi_stop_server
|
|
from fastapi_server import is_server_running as fastapi_is_server_running
|
|
from fastapi_server import broadcast_event_sync as fastapi_broadcast_event_sync
|
|
logger.info("FastAPI server imported successfully")
|
|
# Check if FastAPI is actually working by trying to create a simple app
|
|
try:
|
|
import fastapi
|
|
test_app = fastapi.FastAPI()
|
|
logger.info("FastAPI test successful")
|
|
USE_FASTAPI = True
|
|
except Exception as e:
|
|
logger.error(f"FastAPI test failed: {str(e)}")
|
|
logger.info("HTTP handler imported as fallback")
|
|
USE_FASTAPI = False
|
|
except ImportError as e:
|
|
logger.error(f"Error importing FastAPI server: {str(e)}")
|
|
logger.error(traceback.format_exc())
|
|
logger.info("HTTP handler imported as fallback")
|
|
USE_FASTAPI = False
|
|
|
|
# Fallback to HTTP handler if FastAPI is not available
|
|
try:
|
|
from http_handler import start_http_server, stop_http_server, is_http_server_running, broadcast_event
|
|
logger.info("HTTP handler imported as fallback")
|
|
# Set flags to indicate we're using the fallback
|
|
_using_fastapi = False
|
|
except Exception as e2:
|
|
logger.error(f"Error importing HTTP handler: {e2}")
|
|
logger.error(traceback.format_exc())
|
|
else:
|
|
# Set flags to indicate we're using FastAPI
|
|
_using_fastapi = True
|
|
# Alias the functions for compatibility
|
|
start_http_server = fastapi_start_server
|
|
stop_http_server = fastapi_stop_server
|
|
is_http_server_running = fastapi_is_server_running
|
|
broadcast_event = fastapi_broadcast_event_sync
|
|
|
|
# Global variables
|
|
_server_running = False
|
|
|
|
# Public functions
|
|
def start_server(port=SERVER_PORT):
|
|
"""
|
|
Start the MCP server
|
|
|
|
Args:
|
|
port (int): Server port number
|
|
|
|
Returns:
|
|
int: Port number if server started successfully, None otherwise
|
|
"""
|
|
global _server_running, USE_FASTAPI
|
|
|
|
try:
|
|
# Check if server is already running
|
|
if is_server_running():
|
|
logger.info(f"Server already running on port {port}")
|
|
return port
|
|
|
|
logger.info(f"Starting MCP server on port {port}...")
|
|
|
|
# Start the server based on available implementation
|
|
if USE_FASTAPI:
|
|
try:
|
|
# Try to start FastAPI server
|
|
port = fastapi_start_server(port=port)
|
|
if port:
|
|
_server_running = True
|
|
# Log Maya info
|
|
maya_info = {
|
|
'maya_version': cmds.about(version=True),
|
|
'maya_api_version': om.MGlobal.apiVersion(),
|
|
'os_name': cmds.about(os=True),
|
|
'product_name': cmds.about(product=True)
|
|
}
|
|
logger.info(f"Maya info: {maya_info}")
|
|
|
|
try:
|
|
# Broadcast Maya info to clients
|
|
fastapi_broadcast_event_sync("maya_info", maya_info)
|
|
except Exception as e:
|
|
logger.error(f"Error broadcasting event: {str(e)}")
|
|
# This is not a critical error, so we can continue
|
|
|
|
logger.info(f"MCP server successfully started on port {port}")
|
|
print("\n==================================================")
|
|
print(f"MCP SERVER STARTED SUCCESSFULLY ON PORT {port}")
|
|
print("==================================================\n")
|
|
return port
|
|
else:
|
|
logger.error("Failed to start FastAPI server")
|
|
# Fall back to HTTP handler
|
|
logger.info("Falling back to HTTP handler")
|
|
USE_FASTAPI = False
|
|
except Exception as e:
|
|
logger.error(f"Error starting FastAPI server: {str(e)}")
|
|
logger.error(traceback.format_exc())
|
|
# Fall back to HTTP handler
|
|
logger.info("Falling back to HTTP handler")
|
|
USE_FASTAPI = False
|
|
|
|
# If FastAPI failed or not available, use HTTP handler
|
|
if not USE_FASTAPI:
|
|
from http_handler import start_http_server, stop_http_server, is_http_server_running, broadcast_event
|
|
# Ensure port value is valid, if port becomes None, use SERVER_PORT
|
|
http_port = port if port is not None else SERVER_PORT
|
|
print(f"Attempting to start HTTP server on 127.0.0.1:{http_port}")
|
|
port = start_http_server(port=http_port)
|
|
if port:
|
|
_server_running = True
|
|
# Log Maya info
|
|
maya_info = {
|
|
'maya_version': cmds.about(version=True),
|
|
'maya_api_version': om.MGlobal.apiVersion(),
|
|
'os_name': cmds.about(os=True),
|
|
'product_name': cmds.about(product=True)
|
|
}
|
|
logger.info(f"Maya info: {maya_info}")
|
|
|
|
# Broadcast Maya info to clients
|
|
broadcast_event("maya_info", maya_info)
|
|
|
|
logger.info(f"MCP server successfully started on port {port}")
|
|
print("\n==================================================")
|
|
print(f"MCP SERVER STARTED SUCCESSFULLY ON PORT {port}")
|
|
print("==================================================\n")
|
|
return port
|
|
else:
|
|
logger.error("Failed to start HTTP server")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error starting server: {e}")
|
|
logger.error(traceback.format_exc())
|
|
return None
|
|
|
|
def stop_server():
|
|
"""
|
|
Stop the MCP server
|
|
|
|
Returns:
|
|
bool: Whether server was successfully stopped
|
|
"""
|
|
global _server_running
|
|
|
|
try:
|
|
if _server_running:
|
|
logger.info("Stopping MCP server...")
|
|
|
|
# Stop HTTP server
|
|
success = stop_http_server()
|
|
|
|
if success:
|
|
_server_running = False
|
|
logger.info("MCP server stopped successfully")
|
|
else:
|
|
logger.error("Failed to stop HTTP server")
|
|
|
|
return success
|
|
else:
|
|
logger.info("Server is not running")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"Error stopping server: {e}")
|
|
logger.error(traceback.format_exc())
|
|
return False
|
|
|
|
def is_server_running():
|
|
"""
|
|
Check if server is running
|
|
|
|
Returns:
|
|
bool: Whether server is running
|
|
"""
|
|
global _server_running
|
|
|
|
if _server_running:
|
|
return is_http_server_running()
|
|
return False
|
|
|
|
def get_scene_info():
|
|
"""
|
|
Get Maya scene information
|
|
|
|
Returns:
|
|
dict: Scene information
|
|
"""
|
|
try:
|
|
logger.debug("Getting scene information...")
|
|
|
|
# Get current file
|
|
try:
|
|
current_file = cmds.file(query=True, sceneName=True) or "untitled"
|
|
logger.debug(f"Current file: {current_file}")
|
|
except Exception as e:
|
|
logger.warning(f"Error getting current file: {e}")
|
|
logger.debug(traceback.format_exc())
|
|
current_file = "unknown"
|
|
|
|
# Get selection
|
|
try:
|
|
# Use OpenMaya API to get selection objects, avoid cmds.ls parameter issues
|
|
selection_list = om.MGlobal.getActiveSelectionList()
|
|
selection = []
|
|
for i in range(selection_list.length()):
|
|
try:
|
|
dep_node = selection_list.getDependNode(i)
|
|
if not dep_node.isNull():
|
|
node_fn = om.MFnDependencyNode(dep_node)
|
|
selection.append(node_fn.name())
|
|
except Exception as e:
|
|
logger.debug(f"Error getting selection item {i}: {e}")
|
|
logger.debug(f"Selection: {selection}")
|
|
except Exception as e:
|
|
logger.warning(f"Error getting selection: {e}")
|
|
logger.debug(traceback.format_exc())
|
|
selection = []
|
|
|
|
# Get scene objects
|
|
try:
|
|
objects = cmds.ls(type="transform") or []
|
|
logger.debug(f"Found {len(objects)} transform objects")
|
|
except Exception as e:
|
|
logger.warning(f"Error getting objects: {e}")
|
|
logger.debug(traceback.format_exc())
|
|
objects = []
|
|
|
|
# Get cameras
|
|
try:
|
|
cameras = cmds.ls(type="camera") or []
|
|
logger.debug(f"Found {len(cameras)} cameras")
|
|
except Exception as e:
|
|
logger.warning(f"Error getting cameras: {e}")
|
|
logger.debug(traceback.format_exc())
|
|
cameras = []
|
|
|
|
# Get lights
|
|
try:
|
|
lights = cmds.ls(type="light") or []
|
|
logger.debug(f"Found {len(lights)} lights")
|
|
except Exception as e:
|
|
logger.warning(f"Error getting lights: {e}")
|
|
logger.debug(traceback.format_exc())
|
|
lights = []
|
|
|
|
# Create scene info
|
|
scene_info = {
|
|
"file": current_file,
|
|
"selection": selection,
|
|
"objects": objects,
|
|
"cameras": cameras,
|
|
"lights": lights
|
|
}
|
|
|
|
logger.debug(f"Scene info: {scene_info}")
|
|
return scene_info
|
|
except Exception as e:
|
|
logger.error(f"Error getting scene info: {e}")
|
|
logger.error(traceback.format_exc())
|
|
return {
|
|
"file": "unknown",
|
|
"selection": [],
|
|
"objects": [],
|
|
"cameras": [],
|
|
"lights": []
|
|
}
|