server.py

#!/usr/bin/env python3
"""
Carla MCP Server - Complete audio plugin host control via Model Context Protocol
"""

import asyncio
import json
import logging
import sys
import os
from datetime import datetime
from typing import Dict, Any, Optional

# Add Carla to path using CARLA_PATH environment variable
carla_path = os.environ.get('CARLA_PATH', '/home/gwohl/builds/Carla')
sys.path.append(os.path.join(carla_path, 'source', 'frontend'))

from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
import mcp.types as types

from carla_controller import CarlaController
from tool_registry import create_carla_tool_registry
from tools.session_tools import SessionTools
from tools.plugin_tools import PluginTools
from tools.routing_tools import RoutingTools
from tools.parameter_tools import ParameterTools
from tools.analysis_tools import AnalysisTools
from tools.hardware_tools import HardwareTools
from tools.jack_tools import JackTools
from monitors.event_monitor import EventMonitor
from monitors.audio_monitor import AudioMonitor
from monitors.cpu_monitor import CPUMonitor
from monitors.ambient_stream import AmbientStreamLogger
from mixassist_resources import mixassist_provider

# Configure logging to file (not stderr) when running as MCP server
log_file = os.path.join(os.path.dirname(__file__), 'carla_mcp_server.log')
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(log_file),
    ]
)
logger = logging.getLogger(__name__)


class CarlaMCPServer:
    """Main MCP Server for Carla audio plugin host control"""
    
    def __init__(self, carla_path: str = None):
        """Initialize the Carla MCP Server
        
        Args:
            carla_path: Path to Carla installation (auto-detect if None)
        """
        self.server = Server("carla-mcp")
        
        # Auto-detect Carla path if not provided
        if carla_path is None:
            carla_path = self._find_carla_installation()
        
        # Initialize Carla controller
        self.carla = CarlaController(carla_path)
        
        # Session management
        self.sessions: Dict[str, Any] = {}
        self.active_session_id: Optional[str] = None
        
        # Initialize tool registry
        self.tool_registry = create_carla_tool_registry()

        # Initialize tool modules
        self.session_tools = SessionTools(self.carla)
        self.plugin_tools = PluginTools(self.carla)
        self.routing_tools = RoutingTools(self.carla)
        self.parameter_tools = ParameterTools(self.carla)
        self.analysis_tools = AnalysisTools(self.carla)
        self.hardware_tools = HardwareTools(self.carla)
        self.jack_tools = JackTools(self.carla)
        
        # Initialize monitors
        self.event_monitor = EventMonitor(self.carla)
        self.audio_monitor = AudioMonitor(self.carla)
        self.cpu_monitor = CPUMonitor(self.carla)
        self.ambient_stream = AmbientStreamLogger(self.carla, self.event_monitor)
        
        # Performance metrics
        self.metrics = {
            'tool_calls': 0,
            'errors': 0,
            'avg_response_time': 0,
            'active_plugins': 0,
            'cpu_usage': 0
        }
        
        # Register all tools and resources
        self._register_tools()
        self._register_resources()

        # Setup event handlers
        self._setup_event_handlers()
        
        logger.info(f"Carla MCP Server initialized with path: {carla_path}")
    
    def _find_carla_installation(self) -> str:
        """Auto-detect Carla installation path"""
        possible_paths = [
            "/home/gwohl/builds/Carla",
            "/usr/share/carla",
            "/usr/local/share/carla",
            os.path.expanduser("~/Carla"),
            "/opt/carla"
        ]
        
        for path in possible_paths:
            if os.path.exists(path):
                bin_path = os.path.join(path, "bin", "libcarla_standalone2.so")
                if os.path.exists(bin_path):
                    logger.info(f"Found Carla at: {path}")
                    return path
        
        raise RuntimeError("Could not find Carla installation. Please specify path.")
    
    def _register_tools(self):
        """Register all MCP tools using the tool registry"""

        @self.server.list_tools()
        async def handle_list_tools() -> list[types.Tool]:
            """Return all available tools from the registry"""
            return self.tool_registry.get_mcp_tools()
        
        @self.server.call_tool()
        async def handle_call_tool(
            name: str, arguments: dict | None
        ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
            """Handle tool calls"""
            
            start_time = datetime.now()
            self.metrics['tool_calls'] += 1
            
            try:
                result = await self._execute_tool(name, arguments or {})
                
                # Update metrics
                elapsed = (datetime.now() - start_time).total_seconds()
                self.metrics['avg_response_time'] = (
                    (self.metrics['avg_response_time'] * (self.metrics['tool_calls'] - 1) + elapsed) 
                    / self.metrics['tool_calls']
                )
                
                return [types.TextContent(
                    type="text",
                    text=json.dumps(result, indent=2)
                )]
                
            except Exception as e:
                self.metrics['errors'] += 1
                logger.error(f"Tool {name} failed: {str(e)}")
                
                return [types.TextContent(
                    type="text",
                    text=json.dumps({
                        "error": str(e),
                        "tool": name,
                        "arguments": arguments
                    }, indent=2)
                )]

    def _register_resources(self):
        """Register MCP resources including MixAssist dataset"""

        @self.server.list_resources()
        async def handle_list_resources() -> list[types.Resource]:
            """Return all available resources"""
            resources = []

            # Add MixAssist dataset resources
            resources.extend(mixassist_provider.get_available_resources())

            return resources

        @self.server.read_resource()
        async def handle_read_resource(uri: str) -> str:
            """Read content from a resource"""
            try:
                # Convert AnyUrl to string if needed
                uri_str = str(uri)

                # Route MixAssist resources
                if uri_str.startswith("mixassist://"):
                    return mixassist_provider.get_resource_content(uri_str)
                else:
                    raise ValueError(f"Unknown resource URI: {uri_str}")

            except Exception as e:
                logger.error(f"Failed to read resource {uri}: {str(e)}")
                raise

    async def _execute_tool(self, name: str, arguments: dict) -> dict:
        """Execute a specific tool"""

        # Log user command to ambient stream
        self.ambient_stream.log_user_command(name, arguments)

        # Add context to arguments
        arguments['session_context'] = self.get_active_session()
        arguments['performance_metrics'] = self.get_performance_metrics()
        
        # Route to appropriate tool module using registry
        handler = self.tool_registry.get_handler_for_tool(name)
        if not handler:
            raise ValueError(f"Unknown tool: {name}")

        # Get handler module
        if handler == 'session_tools':
            return await self.session_tools.execute(name, arguments)
        elif handler == 'plugin_tools':
            return await self.plugin_tools.execute(name, arguments)
        elif handler == 'routing_tools':
            return await self.routing_tools.execute(name, arguments)
        elif handler == 'parameter_tools':
            return await self.parameter_tools.execute(name, arguments)
        elif handler == 'analysis_tools':
            return await self.analysis_tools.execute(name, arguments)
        elif handler == 'hardware_tools':
            return await self.hardware_tools.execute(name, arguments)
        elif handler == 'jack_tools':
            return await self.jack_tools.execute(name, arguments)
        else:
            raise ValueError(f"Unknown tool handler: {handler}")
    
    def _setup_event_handlers(self):
        """Setup Carla event callbacks"""
        
        def on_engine_callback(host, action, plugin_id, value1, value2, value3, valuef, value_str):
            """Handle Carla engine callbacks"""
            # Queue event for async processing
            asyncio.create_task(
                self.event_monitor.handle_event({
                    'action': action,
                    'plugin_id': plugin_id,
                    'value1': value1,
                    'value2': value2,
                    'value3': value3,
                    'valuef': valuef,
                    'value_str': value_str,
                    'timestamp': datetime.now().isoformat()
                })
            )
        
        self.carla.set_callback(on_engine_callback)
    
    def get_active_session(self) -> Optional[dict]:
        """Get active session context"""
        if self.active_session_id:
            return self.sessions.get(self.active_session_id)
        return None
    
    def get_performance_metrics(self) -> dict:
        """Get current performance metrics"""
        return {
            **self.metrics,
            'cpu_usage': self.cpu_monitor.get_current_usage(),
            'active_plugins': self.carla.host.get_current_plugin_count()
        }
    
    async def run(self):
        """Run the MCP server"""
        try:
            async with stdio_server() as (read_stream, write_stream):
                await self.server.run(
                    read_stream,
                    write_stream,
                    InitializationOptions(
                        server_name="Carla MCP Server",
                        server_version="1.0.0",
                        capabilities=self.server.get_capabilities(
                            notification_options=NotificationOptions(),
                            experimental_capabilities={},
                        ),
                    ),
                )
        finally:
            # Clean shutdown of ambient stream
            self.ambient_stream.close()


async def main():
    """Main entry point"""
    # Check for custom Carla path in environment
    carla_path = os.environ.get('CARLA_PATH')

    # Create and run server
    server = CarlaMCPServer(carla_path)
    await server.run()


if __name__ == "__main__":
    asyncio.run(main())