MCP Play Sound Server

""" MCP Server implementation for audio playback notifications. This module implements the Model Context Protocol server using FastMCP, providing the play_notification_sound tool for AI agents. """ import logging from typing import Optional, Dict, Any from mcp.server.fastmcp import FastMCP from .config import ServerConfig from .audio_player import AudioPlayer, PlaybackStatus, AFPlayBackend logger = logging.getLogger(__name__) class PlaySoundServer: """MCP Server for audio playback notifications.""" def __init__(self, config: ServerConfig): """Initialize the MCP server with configuration.""" self.config = config self.audio_player = AudioPlayer(config) # Initialize FastMCP server self.app = FastMCP("play-sound-server") self._setup_tools() logger.info("PlaySoundServer initialized") def _setup_tools(self) -> None: """Set up MCP tools.""" @self.app.tool() async def play_notification_sound( custom_sound_path: Optional[str] = None, message: Optional[str] = None ) -> Dict[str, Any]: """ Play a notification sound to alert the user. This tool plays an audio notification to get the user's attention, typically used when an AI coding task has been completed. Args: custom_sound_path: Optional path to a custom audio file to play. If not provided, uses the configured default or built-in sound. message: Optional message to include in the response for context. Returns: Dictionary containing playback status, message, and details. """ logger.info(f"play_notification_sound called: custom_path={custom_sound_path}, message={message}") try: # Play the notification sound result = await self.audio_player.play_notification(custom_sound_path) # Prepare response response = { "success": result.status in [PlaybackStatus.SUCCESS, PlaybackStatus.FALLBACK_USED], "status": result.status.value, "message": result.message, "backend_used": result.backend_used, "fallback_used": result.fallback_used, } # Add user message if provided if message: response["user_message"] = message # Add duration if available if result.duration_ms: response["duration_ms"] = result.duration_ms # Log the result if result.status == PlaybackStatus.SUCCESS: logger.info(f"Notification played successfully using {result.backend_used}") elif result.status == PlaybackStatus.FALLBACK_USED: logger.info(f"Notification played using fallback with {result.backend_used}") else: logger.warning(f"Notification playback failed: {result.message}") return response except Exception as e: logger.error(f"Error in play_notification_sound: {e}") return { "success": False, "status": "error", "message": f"Unexpected error: {str(e)}", "backend_used": None, "fallback_used": False, } @self.app.tool() async def get_audio_status() -> Dict[str, Any]: """ Get the current audio system status and configuration. Returns information about available audio backends, configuration, and system capabilities. Returns: Dictionary containing audio system status and configuration. """ logger.debug("get_audio_status called") try: # Get backend information backend_info = [] for backend in self.audio_player.backends: backend_info.append({ "name": backend.name, "available": backend.is_available(), }) # Prepare status response status = { "backends_available": len(self.audio_player.backends), "backends": backend_info, "configuration": { "volume_level": self.config.volume_level, "enable_fallback": self.config.enable_fallback, "custom_sound_configured": bool(self.config.custom_sound_path), "audio_backend": self.config.audio_backend, "playback_timeout_seconds": self.config.playback_timeout_seconds, }, "default_sound_exists": self.audio_player._default_sound_path.exists(), } # Add custom sound info if configured if self.config.custom_sound_path: from pathlib import Path custom_path = Path(self.config.custom_sound_path) status["custom_sound"] = { "path": str(custom_path), "exists": custom_path.exists(), "size_mb": round(custom_path.stat().st_size / (1024 * 1024), 2) if custom_path.exists() else None, } logger.debug(f"Audio status: {len(backend_info)} backends available") return status except Exception as e: logger.error(f"Error in get_audio_status: {e}") return { "error": f"Failed to get audio status: {str(e)}", "backends_available": 0, "backends": [], } @self.app.tool() async def test_audio_playback( use_custom: bool = False ) -> Dict[str, Any]: """ Test audio playback functionality. Performs a test of the audio system to verify it's working correctly. Useful for troubleshooting audio issues. Args: use_custom: If True, test with custom sound (if configured). If False, test with default sound. Returns: Dictionary containing test results and diagnostic information. """ logger.info(f"test_audio_playback called: use_custom={use_custom}") try: # Determine which sound to test test_path = None if use_custom and self.config.custom_sound_path: test_path = self.config.custom_sound_path # Run the test result = await self.audio_player.play_notification(test_path) # Prepare test results test_result = { "test_passed": result.status in [PlaybackStatus.SUCCESS, PlaybackStatus.FALLBACK_USED], "status": result.status.value, "message": result.message, "backend_used": result.backend_used, "fallback_used": result.fallback_used, "test_type": "custom" if use_custom else "default", } logger.info(f"Audio test completed: {result.status.value}") return test_result except Exception as e: logger.error(f"Error in test_audio_playback: {e}") return { "test_passed": False, "status": "error", "message": f"Test failed with error: {str(e)}", "backend_used": None, "fallback_used": False, "test_type": "custom" if use_custom else "default", } @self.app.tool() async def list_audio_devices() -> Dict[str, Any]: """ List available audio output devices on the system. Returns information about all available audio output devices, including which one is currently the default. Returns: Dictionary containing list of available audio devices with their properties. """ logger.info("list_audio_devices called") try: from .audio_player import AFPlayBackend devices = await AFPlayBackend.get_available_audio_devices() return { "success": True, "devices": devices, "device_count": len(devices), "current_configured_device": self.config.audio_device, "message": f"Found {len(devices)} audio output devices" } except Exception as e: logger.error(f"Error in list_audio_devices: {e}") return { "success": False, "devices": [], "device_count": 0, "current_configured_device": self.config.audio_device, "error": str(e), "message": f"Failed to list audio devices: {str(e)}" } def run(self) -> None: """Run the MCP server.""" logger.info("Starting MCP Play Sound Server...") try: # Run the FastMCP server self.app.run() except KeyboardInterrupt: logger.info("Server stopped by user") except Exception as e: logger.error(f"Server error: {e}") raise finally: logger.info("MCP Play Sound Server stopped") def get_server_info(self) -> Dict[str, Any]: """Get server information for debugging.""" return { "name": "MCP Play Sound Server", "version": "0.1.0", "tools_count": 4, # We have 4 tools: play_notification_sound, get_audio_status, test_audio_playback, list_audio_devices "backends_available": len(self.audio_player.backends), "config": { "volume_level": self.config.volume_level, "enable_fallback": self.config.enable_fallback, "custom_sound_configured": bool(self.config.custom_sound_path), } }