SmartThings MCP Server

from __future__ import annotations import logging import os from typing import Any, Dict, List, Optional import httpx # Configure logging logger = logging.getLogger(__name__) ST_BASE_DEFAULT = "https://api.smartthings.com/v1" class SmartThingsError(RuntimeError): """Custom exception for SmartThings API errors.""" pass class SmartThingsClient: def __init__(self, token: str | None = None, base_url: str | None = None, timeout: float = 15.0): self.token = token or os.environ.get("SMARTTHINGS_PAT") if not self.token: raise SmartThingsError("SMARTTHINGS_PAT is required (env var)") self.base_url = (base_url or os.environ.get("SMARTTHINGS_BASE_URL") or ST_BASE_DEFAULT).rstrip("/") logger.info(f"Initializing SmartThings client with base URL: {self.base_url}") self._client = httpx.AsyncClient(timeout=timeout, headers={ "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "Accept": "application/json", }) async def close(self): """Close the HTTP client and clean up resources.""" logger.info("Closing SmartThings HTTP client") await self._client.aclose() async def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Make a GET request to the SmartThings API.""" url = f"{self.base_url}{path}" logger.debug(f"GET {url} with params: {params}") r = await self._client.get(url, params=params) if r.status_code >= 400: error_msg = f"GET {url} -> {r.status_code}: {r.text}" logger.error(error_msg) raise SmartThingsError(error_msg) logger.debug(f"GET {url} -> {r.status_code}") return r.json() async def _post(self, path: str, json: Dict[str, Any]) -> Dict[str, Any]: """Make a POST request to the SmartThings API.""" url = f"{self.base_url}{path}" logger.debug(f"POST {url} with body: {json}") r = await self._client.post(url, json=json) if r.status_code >= 400: error_msg = f"POST {url} -> {r.status_code}: {r.text}" logger.error(error_msg) raise SmartThingsError(error_msg) logger.debug(f"POST {url} -> {r.status_code}") return r.json() async def locations(self) -> List[Dict[str, Any]]: """Get all locations for this account.""" logger.info("Fetching locations") data = await self._get("/locations") return data.get("items", []) async def rooms(self, location_id: str) -> List[Dict[str, Any]]: """Get all rooms in a location.""" logger.info(f"Fetching rooms for location: {location_id}") data = await self._get(f"/locations/{location_id}/rooms") return data.get("items", []) async def devices(self, *, location_id: Optional[str] = None) -> List[Dict[str, Any]]: """Get all devices, optionally filtered by location.""" logger.info(f"Fetching devices (location_id={location_id})") params = {"locationId": location_id} if location_id else None data = await self._get("/devices", params=params) return data.get("items", []) async def get_device(self, device_id: str) -> Dict[str, Any]: """Get a single device by ID.""" logger.info(f"Fetching device: {device_id}") return await self._get(f"/devices/{device_id}") async def device_status(self, device_id: str) -> Dict[str, Any]: """Get the current status of a device.""" logger.info(f"Fetching status for device: {device_id}") return await self._get(f"/devices/{device_id}/status") async def device_health(self, device_id: str) -> Dict[str, Any]: """Get the health status of a device.""" logger.info(f"Fetching health for device: {device_id}") try: return await self._get(f"/devices/{device_id}/health") except SmartThingsError as e: logger.warning(f"Could not fetch health for device {device_id}: {e}") return {"state": "unknown"} async def send_commands(self, device_id: str, commands: List[Dict[str, Any]]) -> Dict[str, Any]: """Send commands to a device.""" logger.info(f"Sending commands to device {device_id}: {commands}") body = {"commands": commands} return await self._post(f"/devices/{device_id}/commands", json=body) def flatten_status(status: Dict[str, Any]) -> Dict[str, Any]: """ Flatten nested status dictionary into a simple key-value structure. Converts: {"main": {"switch": {"switch": {"value": "on"}}}} To: {"main.switch.switch": "on"} """ flat: Dict[str, Any] = {} for component, caps in (status or {}).items(): for cap, attrs in (caps or {}).items(): for attr, val in (attrs or {}).items(): if isinstance(val, dict) and "value" in val: flat[f"{component}.{cap}.{attr}"] = val["value"] else: flat[f"{component}.{cap}.{attr}"] = val return flat async def find_device_by_name(client: SmartThingsClient, name: str, *, location_id: Optional[str] = None): """ Search for a device by name (case-insensitive substring match). Args: client: The SmartThings client instance. name: The name or label to search for. location_id: Optional location ID to narrow the search. Returns: The first matching device or None. """ q = (name or "").lower() logger.info(f"Searching for device with name containing: '{q}'") devices = await client.devices(location_id=location_id) for d in devices: label = (d.get("label") or d.get("name") or "").lower() if q == label or q in label: logger.info(f"Found matching device: {d.get('label')} ({d.get('deviceId')})") return d logger.warning(f"No device found matching name: '{name}'") return None