Maya MCP Server

""" Maya MCP Server - Main server implementation Entry point for the Maya Model Context Protocol server """ import asyncio import argparse import logging import signal import sys import atexit from pathlib import Path from typing import List, Dict, Any, Optional # Add src directory to Python path for imports sys.path.insert(0, str(Path(__file__).parent)) from models import MAYA_TOOLS, MayaRequest, MayaResponse from config import config_manager, ServerConfig from mcp_server import MayaMCPServer from maya_bridge import MayaBridge # Global server instance for cleanup _server_instance: Optional[MayaMCPServer] = None def setup_logging(config: ServerConfig) -> None: """Configure logging based on configuration""" log_level = getattr(logging, config.log_level.upper(), logging.INFO) # Configure root logger logging.basicConfig( level=log_level, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('maya-mcp-server.log') ] ) # Set specific logger levels logger = logging.getLogger('maya-mcp-server') logger.setLevel(log_level) if config.debug: # Enable debug logging for all components logging.getLogger('maya-mcp-server').setLevel(logging.DEBUG) logging.getLogger('mcp').setLevel(logging.DEBUG) def parse_arguments() -> argparse.Namespace: """Parse command line arguments""" parser = argparse.ArgumentParser( description='Maya MCP Server - Model Context Protocol server for Autodesk Maya' ) parser.add_argument( '--host', type=str, help='Host to bind the server to (default: localhost)' ) parser.add_argument( '--port', type=int, help='Port to bind the server to (default: 8765)' ) parser.add_argument( '--maya-port', type=int, help='Port for Maya command port (default: 7022)' ) parser.add_argument( '--debug', action='store_true', help='Enable debug mode' ) parser.add_argument( '--config', type=str, help='Path to configuration file' ) parser.add_argument( '--log-level', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], help='Set logging level' ) parser.add_argument( '--version', action='version', version='Maya MCP Server 1.1.0' ) return parser.parse_args() def cleanup_resources() -> None: """Cleanup resources on exit""" global _server_instance if _server_instance: try: # Run cleanup in event loop if available loop = asyncio.get_event_loop() if loop.is_running(): loop.create_task(_server_instance.shutdown()) else: asyncio.run(_server_instance.shutdown()) except Exception: pass # Ignore errors during cleanup def setup_signal_handlers(server: MayaMCPServer) -> None: """Setup signal handlers for graceful shutdown""" global _server_instance _server_instance = server def signal_handler(signum, frame): """Handle shutdown signals""" logger = logging.getLogger('maya-mcp-server') logger.info(f"Received signal {signum}, initiating graceful shutdown...") # Create shutdown task loop = asyncio.get_event_loop() if loop.is_running(): loop.create_task(server.shutdown()) # Register signal handlers if hasattr(signal, 'SIGTERM'): signal.signal(signal.SIGTERM, signal_handler) if hasattr(signal, 'SIGINT'): signal.signal(signal.SIGINT, signal_handler) # Register cleanup function for normal exit atexit.register(cleanup_resources) async def main() -> None: """Main server entry point""" global _server_instance # Parse command line arguments args = parse_arguments() # Load configuration if args.config: config_manager.config_path = args.config config = config_manager.load_config() # Override config with command line arguments if args.host: config.host = args.host if args.port: config.port = args.port if args.maya_port: config.maya_port = args.maya_port if args.debug: config.debug = True if args.log_level: config.log_level = args.log_level # Setup logging setup_logging(config) logger = logging.getLogger('maya-mcp-server') logger.info(f"Starting Maya MCP Server v1.0.0") logger.info(f"Server configuration: host={config.host}, port={config.port}") logger.info(f"Maya port: {config.maya_port}") logger.debug(f"Debug mode: {config.debug}") mcp_server = None try: # Initialize Maya bridge maya_bridge = MayaBridge(config) # Initialize MCP server mcp_server = MayaMCPServer(config, maya_bridge) _server_instance = mcp_server # Setup signal handlers for graceful shutdown setup_signal_handlers(mcp_server) # Start the server logger.info("Initializing MCP server...") await mcp_server.start() logger.info(f"Maya MCP Server started successfully on {config.host}:{config.port}") logger.info("Available tools: " + ", ".join([tool.name for tool in MAYA_TOOLS])) logger.info("Press Ctrl+C to stop the server") # Keep the server running try: await mcp_server.serve_forever() except KeyboardInterrupt: logger.info("Received shutdown signal") except asyncio.CancelledError: logger.info("Server operation cancelled") except Exception as e: logger.error(f"Failed to start Maya MCP Server: {e}") if config.debug: logger.exception("Full error traceback:") sys.exit(1) finally: logger.info("Shutting down Maya MCP Server...") if mcp_server: try: await mcp_server.shutdown() except Exception as e: logger.error(f"Error during shutdown: {e}") logger.info("Maya MCP Server stopped") if __name__ == "__main__": # Run the async main function try: asyncio.run(main()) except KeyboardInterrupt: print("\nMaya MCP Server stopped by user") sys.exit(0) except Exception as e: print(f"Fatal error: {e}") sys.exit(1)