SmartThings MCP Server

# app/main.py from __future__ import annotations import logging import os from typing import Any, Optional, Dict, List from pydantic import BaseModel, Field, field_validator from mcp.server.fastmcp import FastMCP from .smartthings import SmartThingsClient, SmartThingsError, flatten_status, find_device_by_name # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) APP_NAME = "smartthings-mcp-server" # Single client for the process lifetime _client: SmartThingsClient | None = None async def get_client() -> SmartThingsClient: """Get or create the global SmartThings client.""" global _client if _client is None: logger.info("Initializing SmartThings client") _client = SmartThingsClient() return _client async def cleanup_client(): """Clean up the global SmartThings client.""" global _client if _client is not None: logger.info("Closing SmartThings client") await _client.close() _client = None class RoomsArgs(BaseModel): locationId: Optional[str] = None class DevicesArgs(BaseModel): locationId: Optional[str] = None roomId: Optional[str] = None class DeviceSelector(BaseModel): deviceId: Optional[str] = Field(None, description="The unique device ID") name: Optional[str] = Field(None, description="The device name/label to search for") @field_validator('deviceId', 'name') @classmethod def at_least_one_identifier(cls, v, info): """Ensure at least one identifier is provided.""" # This validator runs for each field, so we check if both are None # Note: In Pydantic V2, we need a model_validator for cross-field validation return v class SwitchArgs(DeviceSelector): command: str = Field(..., description="Switch command: 'on' or 'off'") component: str = Field("main", description="Device component (default: 'main')") @field_validator('command') @classmethod def validate_command(cls, v): """Validate switch command.""" if v not in {"on", "off"}: raise ValueError("command must be 'on' or 'off'") return v # Note: Lifespan/cleanup handler would go here if FastMCP supported it # For now, the HTTP client will be cleaned up when the process exits mcp = FastMCP(APP_NAME, json_response=True) @mcp.tool() async def list_locations() -> List[Dict[str, Any]]: """ List all SmartThings locations accessible with the current PAT. Returns: A list of location objects with locationId, name, and other metadata. """ logger.info("Listing locations") client = await get_client() locations = await client.locations() logger.info(f"Found {len(locations)} location(s)") return locations @mcp.tool() async def list_rooms(args: RoomsArgs) -> List[Dict[str, Any]]: """ List all rooms in a SmartThings location. Args: locationId: The location ID. If not provided, uses SMARTTHINGS_LOCATION_ID environment variable or the first available location. Returns: A list of room objects with roomId, name, and other metadata. """ location_id = args.locationId or os.environ.get("SMARTTHINGS_LOCATION_ID") client = await get_client() if not location_id: logger.info("No locationId provided, fetching first available location") locs = await client.locations() if not locs: raise SmartThingsError("No locations found for this token") location_id = locs[0]["locationId"] logger.info(f"Using location: {location_id}") logger.info(f"Listing rooms for location: {location_id}") rooms = await client.rooms(location_id) logger.info(f"Found {len(rooms)} room(s)") return rooms @mcp.tool() async def list_devices(args: DevicesArgs) -> List[Dict[str, Any]]: """ List all SmartThings devices, optionally filtered by location and/or room. Args: locationId: Optional location ID to filter devices. roomId: Optional room ID to filter devices. Returns: A list of device objects with deviceId, label, capabilities, and other metadata. """ logger.info(f"Listing devices (locationId={args.locationId}, roomId={args.roomId})") client = await get_client() devices = await client.devices(location_id=args.locationId) if args.roomId: devices = [d for d in devices if d.get("roomId") == args.roomId] logger.info(f"Found {len(devices)} device(s) in room {args.roomId}") else: logger.info(f"Found {len(devices)} device(s)") return devices @mcp.tool() async def device_status(args: DeviceSelector) -> Dict[str, Any]: """ Get the current status of a SmartThings device. Args: deviceId: The unique device ID (preferred for efficiency). name: The device name/label to search for (slower, searches all devices). Returns: Device information with current status and flattened summary. """ client = await get_client() device = None if args.deviceId: logger.info(f"Getting device by ID: {args.deviceId}") device = await client.get_device(args.deviceId) elif args.name: logger.info(f"Searching for device by name: {args.name}") device = await find_device_by_name(client, args.name) if not device: error_msg = f"Device not found (deviceId={args.deviceId}, name={args.name})" logger.error(error_msg) raise SmartThingsError(error_msg) logger.info(f"Getting status for device: {device['deviceId']}") status = await client.device_status(device["deviceId"]) return { "device": {"id": device["deviceId"], "label": device.get("label")}, "status": status, "summary": flatten_status(status) } @mcp.tool() async def switch_device(args: SwitchArgs) -> Dict[str, Any]: """ Turn a SmartThings switch device on or off. Args: deviceId: The unique device ID (preferred for efficiency). name: The device name/label to search for (slower, searches all devices). command: The switch command - either 'on' or 'off'. component: The device component (default: 'main'). Returns: Device information and command execution result. """ client = await get_client() device = None if args.deviceId: logger.info(f"Getting device by ID: {args.deviceId}") device = await client.get_device(args.deviceId) elif args.name: logger.info(f"Searching for device by name: {args.name}") device = await find_device_by_name(client, args.name) if not device: error_msg = f"Device not found (deviceId={args.deviceId}, name={args.name})" logger.error(error_msg) raise SmartThingsError(error_msg) logger.info(f"Sending '{args.command}' command to device {device['deviceId']} (component: {args.component})") res = await client.send_commands(device["deviceId"], [{ "component": args.component, "capability": "switch", "command": args.command }]) return { "device": {"id": device["deviceId"], "label": device.get("label")}, "result": res } @mcp.tool() async def fridge_status(args: DeviceSelector) -> Dict[str, Any]: """ Get detailed status of a SmartThings refrigerator/fridge device. Args: deviceId: The unique device ID. name: The device name/label to search for. If neither is provided, automatically searches for devices with 'fridge' or 'refrigerator' in the name. Returns: Device information with refrigerator-specific status summary and raw status data. """ client = await get_client() device = None if args.deviceId: logger.info(f"Getting fridge by ID: {args.deviceId}") device = await client.get_device(args.deviceId) elif args.name: logger.info(f"Searching for fridge by name: {args.name}") device = await find_device_by_name(client, args.name) # Auto-detect fridge if no identifier provided if not device: logger.info("No device specified, searching for refrigerator/fridge devices") devices = await client.devices() for d in devices: label = (d.get("label") or d.get("name") or "").lower() if any(k in label for k in ("fridge", "refrigerator")): device = d logger.info(f"Found fridge device: {label}") break if not device: error_msg = "Fridge not found. Specify deviceId or name, or ensure a device has 'fridge' or 'refrigerator' in its name." logger.error(error_msg) raise SmartThingsError(error_msg) logger.info(f"Getting status for fridge: {device['deviceId']}") status = await client.device_status(device["deviceId"]) flat = flatten_status(status) summary = { "power": flat.get("main.switch.switch") or flat.get("refrigerator.switch.switch"), "refrigeratorTemp": flat.get("refrigeratorTemperatureMeasurement.temperature") or flat.get("main.temperatureMeasurement.temperature"), "freezerTemp": flat.get("freezerTemperatureMeasurement.temperature"), "doorOpen": flat.get("doorControl.door") or flat.get("refrigeratorDoorState.door"), "iceMaker": flat.get("icemaker.switch.switch"), "defrost": flat.get("refrigerator.defrostMode.defrostMode"), "interesting": {k: v for k, v in flat.items() if any(s in k for s in ("temperature","door","ice","humidity","energy"))}, } return { "device": {"id": device["deviceId"], "label": device.get("label")}, "summary": summary, "raw": status } @mcp.tool() async def device_health(deviceId: str) -> Dict[str, Any]: """ Get the health status of a SmartThings device. Args: deviceId: The unique device ID. Returns: Device health information including online/offline state. """ logger.info(f"Getting health for device: {deviceId}") client = await get_client() health = await client.device_health(deviceId) return health if __name__ == "__main__": transport = os.environ.get("MCP_TRANSPORT", "stdio") logger.info(f"Starting MCP server with transport: {transport}") mcp.run(transport=transport)