Philips Hue MCP Server

""" FastMCP-based HTTP server for Philips Hue control. This module provides a modern HTTP/Streamable HTTP transport implementation using the FastMCP framework for network-based access. """ import asyncio import logging import os import sys from functools import lru_cache from typing import List, Dict, Any, Optional from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP import uvicorn from .hue_client import HueClient from .validation import ( validate_light_id, validate_group_id, validate_scene_id, validate_brightness, validate_color_temperature, validate_xy_coordinates, ) 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" ENV_VAR_HOST = "MCP_HOST" ENV_VAR_PORT = "MCP_PORT" # Default network configuration DEFAULT_HOST = "0.0.0.0" DEFAULT_PORT = 8080 # Global client instance (initialized on startup) hue_client: Optional[HueClient] = None # Create FastMCP instance mcp = FastMCP("hue-mcp-server") # ============================================================================ # Client Accessor # ============================================================================ @lru_cache(maxsize=1) def get_hue_client() -> HueClient: """ Get the global Hue client singleton, ensuring it's initialized. This accessor eliminates the need for repeated null checks in every tool function. Uses lru_cache to return the same instance on subsequent calls. Returns: The initialized HueClient instance Raises: RuntimeError: If the Hue client has not been initialized """ if hue_client is None: raise RuntimeError("Hue client not initialized") return hue_client # ============================================================================ # Tool Implementations # ============================================================================ @mcp.tool() async def list_lights() -> List[Dict[str, Any]]: """List all lights connected to the Hue Bridge with their current state.""" client = get_hue_client() lights = await client.get_lights() return [ { "id": light_id, "name": light_data["name"], "on": light_data["on"], "brightness": light_data["brightness"], "color_temp": light_data["color_temp"], "reachable": light_data["reachable"], } for light_id, light_data in lights.items() ] @mcp.tool() async def get_light_state(light_id: str) -> Dict[str, Any]: """ Get detailed state information for a specific light. Args: light_id: The unique identifier of the light Returns: Detailed state of the light """ validate_light_id(light_id) client = get_hue_client() light = await client.get_light(light_id) if not light: raise ValueError(f"Light with ID '{light_id}' not found") return light @mcp.tool() async def turn_light_on(light_id: str, brightness: Optional[int] = None) -> Dict[str, Any]: """ Turn a light on, optionally setting brightness. Args: light_id: The unique identifier of the light brightness: Optional brightness level (0-254) Returns: Success message """ validate_light_id(light_id) if brightness is not None: validate_brightness(brightness) client = get_hue_client() await client.set_light_state(light_id, on=True, brightness=brightness) brightness_message = f" with brightness {brightness}" if brightness is not None else "" return { "success": True, "message": f"Light {light_id} turned on{brightness_message}", } @mcp.tool() async def turn_light_off(light_id: str) -> Dict[str, Any]: """ Turn a light off. Args: light_id: The unique identifier of the light Returns: Success message """ validate_light_id(light_id) client = get_hue_client() await client.set_light_state(light_id, on=False) return {"success": True, "message": f"Light {light_id} turned off"} @mcp.tool() async def set_brightness(light_id: str, brightness: int) -> Dict[str, Any]: """ Set the brightness level of a light. Args: light_id: The unique identifier of the light brightness: Brightness level (0-254, where 0 is minimum and 254 is maximum) Returns: Success message """ validate_light_id(light_id) validate_brightness(brightness) client = get_hue_client() await client.set_light_state(light_id, brightness=int(brightness)) return {"success": True, "message": f"Light {light_id} brightness set to {brightness}"} @mcp.tool() async def set_color_temp(light_id: str, color_temp: int) -> Dict[str, Any]: """ Set the color temperature of a light in mireds (153=cold, 500=warm). Args: light_id: The unique identifier of the light color_temp: Color temperature in mireds (153-500, where 153 is coldest and 500 is warmest) Returns: Success message """ validate_light_id(light_id) validate_color_temperature(color_temp) client = get_hue_client() await client.set_light_state(light_id, color_temp=int(color_temp)) return { "success": True, "message": f"Light {light_id} color temperature set to {color_temp} mireds", } @mcp.tool() async def set_color(light_id: str, xy: List[float]) -> Dict[str, Any]: """ Set the color of a light using CIE xy color space coordinates. Args: light_id: The unique identifier of the light xy: CIE xy color coordinates as [x, y], each value between 0 and 1 Returns: Success message """ validate_light_id(light_id) validate_xy_coordinates(xy) client = get_hue_client() await client.set_light_state(light_id, xy=xy) return {"success": True, "message": f"Light {light_id} color set to xy={xy}"} @mcp.tool() async def list_groups() -> List[Dict[str, Any]]: """List all groups/rooms with their properties.""" client = get_hue_client() groups = await client.get_groups() return [ { "id": group_id, "name": group_data["name"], "type": group_data["type"], "lights": group_data["lights"], } for group_id, group_data in groups.items() ] @mcp.tool() async def control_group( group_id: str, on: Optional[bool] = None, brightness: Optional[int] = None, color_temp: Optional[int] = None, xy: Optional[List[float]] = None, ) -> Dict[str, Any]: """ Control all lights in a group/room at once. Args: group_id: The unique identifier of the group on: Turn lights on (true) or off (false) brightness: Optional brightness level (0-254) color_temp: Optional color temperature in mireds (153-500) xy: Optional CIE xy color coordinates as [x, y] Returns: Success message """ # Validate all parameters validate_group_id(group_id) if brightness is not None: validate_brightness(brightness) if color_temp is not None: validate_color_temperature(color_temp) if xy is not None: validate_xy_coordinates(xy) # Execute group control client = get_hue_client() await client.set_group_state( group_id=group_id, on=on, brightness=int(brightness) if brightness is not None else None, color_temp=int(color_temp) if color_temp is not None else None, xy=xy, ) return {"success": True, "message": f"Group {group_id} state updated"} @mcp.tool() async def list_scenes() -> List[Dict[str, Any]]: """List all available scenes.""" client = get_hue_client() scenes = await client.get_scenes() return [ { "id": scene_id, "name": scene_data["name"], "group": scene_data["group"], } for scene_id, scene_data in scenes.items() ] @mcp.tool() async def activate_scene(scene_id: str) -> Dict[str, Any]: """ Activate a predefined scene. Args: scene_id: The unique identifier of the scene Returns: Success message """ validate_scene_id(scene_id) client = get_hue_client() await client.activate_scene(scene_id) return {"success": True, "message": f"Scene {scene_id} activated"} # ============================================================================ # Configuration and Server Setup # ============================================================================ def _load_configuration() -> tuple[str, str, str, int]: """ Load and validate configuration from environment variables. Returns: Tuple of (bridge_ip, api_key, host, port) 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) host = os.getenv(ENV_VAR_HOST, DEFAULT_HOST) port = int(os.getenv(ENV_VAR_PORT, str(DEFAULT_PORT))) 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}, Host = {host}, Port = {port}") return bridge_ip, api_key, host, port def main(): """ Main entry point for the FastMCP HTTP server. Loads configuration and starts the HTTP server with Streamable HTTP support. """ global hue_client # Load configuration bridge_ip, api_key, host, port = _load_configuration() # Initialize Hue client logger.info("Initializing Hue Bridge connection...") hue_client = HueClient(bridge_ip, api_key) # Connect to bridge synchronously before starting server async def connect_bridge(): try: await hue_client.connect() logger.info("Connected to Hue Bridge successfully") except Exception as e: logger.error(f"Failed to connect to Hue Bridge: {e}", exc_info=True) sys.exit(1) asyncio.run(connect_bridge()) # Get the Starlette app with Streamable HTTP transport app = mcp.streamable_http_app() # Add health check endpoint async def health_check(request): """ Health check endpoint for monitoring and load balancers. Returns: JSON response with health status and bridge connection state """ from starlette.responses import JSONResponse if hue_client and hue_client.is_connected(): return JSONResponse( {"status": "healthy", "bridge_connected": True, "service": "hue-mcp-server"}, status_code=200, ) else: return JSONResponse( { "status": "unhealthy", "bridge_connected": False, "service": "hue-mcp-server", "error": "Hue Bridge not connected", }, status_code=503, ) # Register the health check route from starlette.routing import Route app.routes.append(Route("/health", health_check, methods=["GET"])) # Add shutdown handler @app.on_event("shutdown") async def shutdown(): logger.info("Shutting down Hue MCP HTTP Server...") if hue_client: try: await hue_client.disconnect() logger.info("Disconnected from Hue Bridge") except Exception as e: logger.error(f"Error disconnecting: {e}", exc_info=True) # Log server information logger.info(f"Hue MCP HTTP Server starting on http://{host}:{port}") logger.info(f"MCP endpoint: http://{host}:{port}/mcp") logger.info(f"Other computers can connect to: http://<this-machine-ip>:{port}/mcp") # Run the HTTP server uvicorn.run(app, host=host, port=port, log_level="info") if __name__ == "__main__": main()