Philips Hue MCP Server

""" Main MCP server for Philips Hue control. This module implements the MCP server that exposes Hue control tools. """ import asyncio import json import logging import os import sys from typing import Any from dotenv import load_dotenv from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent from .hue_client import HueClient from .tools import TOOLS logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Environment variable names ENV_VAR_BRIDGE_IP = "HUE_BRIDGE_IP" ENV_VAR_API_KEY = "HUE_API_KEY" # Server name SERVER_NAME = "hue-mcp-server" class HueMCPServer: """MCP Server for Philips Hue control.""" def __init__(self, bridge_ip: str, api_key: str): """ Initialize the Hue MCP server. Args: bridge_ip: IP address of the Hue Bridge api_key: API key for authentication """ self.server = Server(SERVER_NAME) self.client = HueClient(bridge_ip, api_key) self._setup_handlers() def _ensure_client_connected(self) -> bool: """ Check if client is connected to the bridge. Returns: True if connected, False otherwise """ return self.client.is_connected() async def _connect_if_needed(self) -> None: """ Ensure the client is connected to the bridge. Connects if not already connected. """ if not self._ensure_client_connected(): await self.client.connect() def _setup_handlers(self): """Set up MCP server handlers.""" @self.server.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return [ Tool( name=name, description=tool_def["description"], inputSchema=tool_def["parameters"], ) for name, tool_def in TOOLS.items() ] @self.server.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """ Handle tool calls from the MCP client. Args: name: The name of the tool to execute arguments: The arguments to pass to the tool Returns: List of TextContent containing the result or error message """ if name not in TOOLS: error_message = f"Unknown tool: '{name}'. Available tools: {', '.join(TOOLS.keys())}" logger.error(error_message) return [TextContent(type="text", text=error_message)] tool_def = TOOLS[name] tool_function = tool_def["function"] try: # Ensure we're connected to the bridge await self._connect_if_needed() # Call the tool function result = await tool_function(self.client, arguments or {}) # Format the result as JSON for readability result_text = json.dumps(result, indent=2) return [TextContent(type="text", text=result_text)] except ValueError as e: # Validation errors from tool functions error_message = f"Validation error in '{name}': {str(e)}" logger.error(error_message) return [TextContent(type="text", text=error_message)] except RuntimeError as e: # Runtime errors (e.g., connection issues) error_message = f"Runtime error in '{name}': {str(e)}" logger.error(error_message, exc_info=True) return [TextContent(type="text", text=error_message)] except Exception as e: # Catch-all for unexpected errors error_message = f"Unexpected error executing '{name}': {str(e)}" logger.error(error_message, exc_info=True) return [TextContent(type="text", text=error_message)] async def run(self): """Run the MCP server.""" try: # Connect to the Hue Bridge await self.client.connect() logger.info("Hue MCP Server started successfully") # Run the server async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options(), ) except KeyboardInterrupt: logger.info("Server interrupted by user") except Exception as e: logger.error(f"Server error: {e}", exc_info=True) raise finally: await self.client.disconnect() logger.info("Hue MCP Server stopped") def _load_configuration() -> tuple[str, str]: """ Load and validate configuration from environment variables. Returns: Tuple of (bridge_ip, api_key) Raises: SystemExit: If required environment variables are missing """ load_dotenv() bridge_ip = os.getenv(ENV_VAR_BRIDGE_IP) api_key = os.getenv(ENV_VAR_API_KEY) if not bridge_ip: logger.error( f"Missing required environment variable: {ENV_VAR_BRIDGE_IP}. " f"Please set it in your .env file or environment." ) sys.exit(1) if not api_key: logger.error( f"Missing required environment variable: {ENV_VAR_API_KEY}. " f"Please set it in your .env file or environment." ) sys.exit(1) logger.info(f"Configuration loaded: Bridge IP = {bridge_ip}") return bridge_ip, api_key def main(): """ Main entry point for the Hue MCP server. Loads configuration from environment variables and starts the server. """ bridge_ip, api_key = _load_configuration() # Create and run the server server = HueMCPServer(bridge_ip, api_key) asyncio.run(server.run()) if __name__ == "__main__": main()