update_user_variable
Update an existing user variable in Domoticz by specifying its name, type (integer, float, string, date, or time), and new value.
Instructions
Update an existing user variable.
vtype (Variable Type): 0: Integer 1: Float 2: String 3: Date (DD/MM/YYYY) 4: Time (HH:MM)
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| name | Yes | ||
| vtype | Yes | ||
| value | Yes |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/domoticz_mcp/server.py:739-753 (handler)The handler function for the update_user_variable tool. It is registered via @mcp.tool() decorator, accepts name (str), vtype (int, e.g. 0=Integer, 1=Float, 2=String, 3=Date, 4=Time), and value (str). It makes a GET request to the Domoticz API with param=updateuservariable, then invalidates the user variable cache.
@mcp.tool() async def update_user_variable(name: str, vtype: int, value: str) -> str: """Update an existing user variable. vtype (Variable Type): 0: Integer 1: Float 2: String 3: Date (DD/MM/YYYY) 4: Time (HH:MM) """ async with create_client() as client: response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=updateuservariable&vname={name}&vtype={vtype}&vvalue={value}") _user_variable_cache["timestamp"] = 0 # Invalidate cache return response.text - src/domoticz_mcp/server.py:739-739 (registration)The tool is registered via the @mcp.tool() decorator on line 739, which makes it available as an MCP tool named 'update_user_variable'.
@mcp.tool() - src/domoticz_mcp/server.py:269-755 (helper)The _do_request helper function used by the update_user_variable handler to perform the HTTP GET request to the Domoticz API with automatic retry on 401.
async def _do_request(client: httpx.AsyncClient, method: str, url: str, **kwargs) -> httpx.Response: """Perform a request with a single retry on 401 Unauthorized to handle expired tokens.""" global _oauth_token_cache try: resp = await client.request(method, url, **kwargs) if resp.status_code == 401: # Token might be expired. Clear cache and retry once. _oauth_token_cache = None # Re-fetch token (this will trigger OAuth flow if needed) new_token = await _fetch_oauth_token(force_refresh=True) if new_token: # Update headers for the retry if "headers" not in kwargs: kwargs["headers"] = {} kwargs["headers"]["Authorization"] = f"Bearer {new_token}" # Retry the request resp = await client.request(method, url, **kwargs) resp.raise_for_status() return resp except httpx.HTTPStatusError as e: if e.response.status_code == 401: raise Exception("Authentication failed. Please check your credentials or re-authenticate.") raise e # Custom AsyncClient wrapper that ensures the token is added class DomoticzClient: def __init__(self, own_client: bool = False): self._own_client = own_client if own_client or _global_http_client is None: self.client: httpx.AsyncClient = httpx.AsyncClient(timeout=30.0) self._owns_client = True else: self.client = _global_http_client self._owns_client = False async def __aenter__(self) -> "httpx.AsyncClient": oauth_token = None if DOMOTICZ_CLIENT_ID: oauth_token = await _fetch_oauth_token() if oauth_token: self.client.headers["Authorization"] = f"Bearer {oauth_token}" elif DOMOTICZ_USERNAME and DOMOTICZ_PASSWORD: self.client.auth = (DOMOTICZ_USERNAME, DOMOTICZ_PASSWORD) else: self.client.headers.pop("Authorization", None) self.client.auth = None return self.client async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: if self._owns_client: await self.client.aclose() def create_client(own_client: bool = False) -> DomoticzClient: """Create a DomoticzClient instance. Args: own_client: If True, creates a dedicated client that will be closed on exit. If False (default), uses a shared client for connection pooling. """ return DomoticzClient(own_client=own_client) async def close_global_client() -> None: """Close the global HTTP client. Call this on application shutdown.""" global _global_http_client if _global_http_client is not None: await _global_http_client.aclose() _global_http_client = None async def _get_cached_data(client: "httpx.AsyncClient", cache_obj: Dict[str, Any], api_url: str, key_path: str = "result") -> List[Dict[str, Any]]: now = time.time() if cache_obj["data"] is None or (now - cache_obj["timestamp"]) > CACHE_TTL: response = await _do_request(client, "GET", api_url) cache_obj["data"] = response.json().get(key_path, []) cache_obj["timestamp"] = now return cache_obj["data"] async def _resolve_idx( client: "httpx.AsyncClient", idx: Optional[int], name: Optional[str], cache: Dict[str, Any], api_url: str ) -> Optional[int]: """Resolve an entity to its idx by either using the provided idx or looking up by name.""" if idx is not None: return idx if not name: return None items = await _get_cached_data(client, cache, api_url) for item in items: if item.get("Name", "").lower() == name.lower(): return int(str(item.get("idx"))) return None async def _resolve_device_idx(client: "httpx.AsyncClient", idx: Optional[int] = None, name: Optional[str] = None) -> Optional[int]: """Resolve a device to its idx.""" return await _resolve_idx(client, idx, name, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true") async def _resolve_scene_idx(client: "httpx.AsyncClient", idx: Optional[int] = None, name: Optional[str] = None) -> Optional[int]: """Resolve a scene to its idx.""" return await _resolve_idx(client, idx, name, _scene_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getscenes") async def _resolve_user_variable_idx(client: "httpx.AsyncClient", idx: Optional[int] = None, name: Optional[str] = None) -> Optional[int]: """Resolve a user variable to its idx.""" return await _resolve_idx(client, idx, name, _user_variable_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getuservariables") def _simplify_device(dev: Dict[str, Any]) -> Dict[str, Any]: """Reduce device dictionary to essential fields to save context space.""" keys_to_keep = [ "idx", "Name", "Type", "SubType", "Data", "Status", "BatteryLevel", "Favorite", "HardwareName", "LastUpdate", "TypeImg", "Usage", "CounterToday", "Temp", "Humidity" ] return {k: dev[k] for k in keys_to_keep if k in dev} def _paginate(data: list, offset: int, limit: int) -> list: """Paginate a list of results.""" return data[offset:offset + limit] @mcp.tool() async def get_overview(detail_level: str = "minimal") -> str: """Get a high-level overview of the Domoticz system. Args: detail_level: 'minimal' (default) for counts and summary, 'standard' for including a sample of devices. """ async with create_client() as client: # Get system info resp = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=getversion") sys_info = resp.json() # Get counts from various caches devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true") scenes = await _get_cached_data(client, _scene_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getscenes") vars = await _get_cached_data(client, _user_variable_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getuservariables") plans = await _get_cached_data(client, _plans_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getplans&order=name&used=true") # Hardware count hw_resp = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=gethardware") hardware = hw_resp.json().get("result", []) overview: Dict[str, Any] = { "system": { "version": sys_info.get("version"), "build_time": sys_info.get("build_time"), "domoticz_url": DOMOTICZ_BASE_URL }, "counts": { "devices": len(devices), "scenes_and_groups": len(scenes), "user_variables": len(vars), "rooms_plans": len(plans), "hardware_gateways": len(hardware) } } if detail_level != "minimal": # Add a sample of favorite/active devices favorites = [d for d in devices if d.get("Favorite") == 1][:10] overview["favorite_devices"] = [_simplify_device(d) for d in favorites] return json.dumps({"status": "OK", "result": overview}) @mcp.tool() async def get_system_health() -> str: """Check the health of the Domoticz system and hardware gateways.""" async with create_client() as client: hw_resp = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=gethardware") hardware = hw_resp.json().get("result", []) health_report = [] for hw in hardware: status = "Online" if hw.get("Enabled") == "true" else "Disabled" health_report.append({ "Name": hw.get("Name"), "Type": hw.get("Type"), "Status": status, "Address": hw.get("Address"), "Port": hw.get("Port") }) # Check for unresponsive devices (last update > 24h) devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true") now = datetime.now() unresponsive_count = 0 for dev in devices: last_update_str = dev.get("LastUpdate") if last_update_str: try: last_update = datetime.strptime(last_update_str, "%Y-%m-%d %H:%M:%S") if now - last_update > timedelta(hours=24): unresponsive_count += 1 except ValueError: continue return json.dumps({ "status": "OK", "result": { "hardware_health": health_report, "unresponsive_devices_count": unresponsive_count, "recommendation": "Use `get_connectivity_report` for a detailed list of unresponsive devices." if unresponsive_count > 0 else "System looks healthy." } }) @mcp.tool() async def search_scripts(query: str) -> str: """Search for a specific string inside event scripts (Lua, dzVents, Python, etc.).""" async with create_client() as client: # 1. Get the list of scripts list_resp = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=events&evparam=list") scripts = list_resp.json().get("result", []) matches = [] query_lower = query.lower() # 2. For each script, load its source and search for script in scripts: script_id = script.get("idx") load_resp = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=events&evparam=load&event={script_id}") script_data = load_resp.json().get("result", [{}])[0] source_code = script_data.get("xml", "") or script_data.get("source", "") if query_lower in source_code.lower(): matches.append({ "idx": script_id, "Name": script.get("name"), "Interpreter": script.get("interpreter"), "Type": script.get("eventtype"), "Status": "Enabled" if script.get("eventstatus") == "1" else "Disabled" }) return json.dumps({"status": "OK", "result": matches, "count": len(scripts)}) @mcp.resource("domoticz://logs/error") async def get_error_logs_resource() -> str: """Read only the 'Error' level entries from the Domoticz system log.""" async with create_client() as client: # loglevel 4 is ERR response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=getlog&lastlogtime=0&loglevel=4") return response.text @mcp.prompt() def agent_guidance() -> str: """Provides the AI agent with critical knowledge about Domoticz-specific logic and best practices.""" return """ You are an expert assistant controlling a Domoticz Home Automation system. To be effective, follow these GUIDELINES: 1. ORIENTATION: Start a new session with `get_overview` to understand the home's scale and available hardware. 2. RESOLVING NAMES: Domoticz tools prefer `idx` (index). If you only have a name, use `search_devices` first to find the correct `idx`. 3. BATTERY LEVELS: A `BatteryLevel` of 255 is a special value meaning the device is mains-powered or doesn't report battery. Ignore these when auditing health. 4. RANGES: - Dimmers and Dimmer levels: Always 0 to 100. (0=Off, 100=Full). - Color Temperature (Kelvin): 0 to 100 (Where 0 is warmest/coldest depending on hardware, usually 0=Warm, 100=Cold). 5. TROUBLESHOOTING: - If a device is "Timed Out" or "Unresponsive", use `get_system_health` and check `domoticz://logs/error`. - To find which automation controls a device, use `search_scripts` with the device's name or idx. 6. USER VARIABLES: Use `get_user_variables` to read state that isn't attached to a physical device. """ @mcp.tool() async def get_all_devices(offset: int = 0, limit: int = 50) -> str: """Get all devices and their current states from Domoticz.""" async with create_client() as client: devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true&order=Name") paginated = _paginate(devices, offset, limit) return json.dumps({"status": "OK", "result": paginated, "total_count": len(devices), "offset": offset, "limit": limit}) @mcp.tool() async def search_devices(query: str, offset: int = 0, limit: int = 50) -> str: """Search for devices by name or data (status). Returns a list of matching devices.""" async with create_client() as client: devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true") query_lower = query.lower() results = [] for dev in devices: if query_lower in dev.get("Name", "").lower() or query_lower in dev.get("Data", "").lower(): results.append(dev) paginated = _paginate(results, offset, limit) return json.dumps({"status": "OK", "result": paginated, "total_count": len(results), "offset": offset, "limit": limit}) @mcp.tool() async def get_device(idx: int | None = None, name: str | None = None) -> str: """Get a specific device state by IDX or Name from Domoticz.""" if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&rid={resolved_idx}") return response.text @mcp.tool() async def toggle_switch(idx: int | None = None, name: str | None = None) -> str: """Toggle a switch or light by IDX or Name. Prefer using IDX for precision.""" if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=switchlight&idx={resolved_idx}&switchcmd=Toggle") return response.text @mcp.tool() async def set_switch_state(state: str, idx: int | None = None, name: str | None = None) -> str: """Set a switch or light to On or Off. Args: state: Must be 'On' or 'Off'. idx: Device index. name: Device name (case-insensitive). """ if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' if state.lower() not in ['on', 'off']: return '{"status": "error", "message": "state must be \'On\' or \'Off\'"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=switchlight&idx={resolved_idx}&switchcmd={state.capitalize()}") return response.text @mcp.tool() async def set_dimmer_level(level: int, idx: int | None = None, name: str | None = None) -> str: """Set the brightness level of a dimmer switch. Args: level: Integer from 0 to 100. Note: 0 is Off, 100 is Full Brightness. idx: Device index. name: Device name. """ if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' if not (0 <= level <= 100): return '{"status": "error", "message": "level must be between 0 and 100"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=switchlight&idx={resolved_idx}&switchcmd=Set%20Level&level={level}") return response.text @mcp.tool() async def set_temperature_setpoint(setpoint: float, idx: int | None = None, name: str | None = None) -> str: """Set the temperature setpoint for a thermostat. Args: setpoint: Target temperature in Celsius (e.g., 21.5). idx: Device index. name: Device name. """ if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=setsetpoint&idx={resolved_idx}&setpoint={setpoint}") return response.text @mcp.tool() async def control_blinds(command: str, idx: int | None = None, name: str | None = None) -> str: """Control blinds or covers. Args: command: Must be 'Open', 'Close', or 'Stop'. idx: Device index. name: Device name. """ if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' if command.capitalize() not in ['Open', 'Close', 'Stop']: return '{"status": "error", "message": "command must be \'Open\', \'Close\', or \'Stop\'"}' async with create_client() as client: resolved_idx = await _resolve_device_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Device not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=switchlight&idx={resolved_idx}&switchcmd={command.capitalize()}") return response.text @mcp.tool() async def get_scenes() -> str: """Get all scenes and groups from Domoticz.""" async with create_client() as client: scenes = await _get_cached_data(client, _scene_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getscenes") return json.dumps({"status": "OK", "result": scenes}) @mcp.tool() async def switch_scene(command: str, idx: int | None = None, name: str | None = None) -> str: """Turn a scene or group On or Off by IDX or Name in Domoticz. command must be 'On', 'Off', or 'Toggle'. Scenes can only be turned 'On'.""" if idx is None and name is None: return '{"status": "error", "message": "Must provide either idx or name"}' if command.capitalize() not in ['On', 'Off', 'Toggle']: return '{"status": "error", "message": "command must be \'On\', \'Off\', or \'Toggle\'"}' async with create_client() as client: resolved_idx = await _resolve_scene_idx(client, idx, name) if resolved_idx is None: return '{"status": "error", "message": "Scene not found"}' response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=switchscene&idx={resolved_idx}&switchcmd={command.capitalize()}") return response.text @mcp.tool() async def get_rooms() -> str: """Get all rooms (Room Plans) from Domoticz.""" async with create_client() as client: plans = await _get_cached_data(client, _plans_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getplans&order=name&used=true") return json.dumps({"status": "OK", "result": plans}) @mcp.tool() async def get_room_devices(idx: int | None = None, room_name: str | None = None) -> str: """Get all devices and their current states in a specific room. Provide either idx or room_name.""" if idx is None and room_name is None: return '{"status": "error", "message": "Must provide either idx or room_name"}' async with create_client() as client: if idx is None: plans = await _get_cached_data(client, _plans_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getplans&order=name&used=true") for plan in plans: if plan.get("Name", "").lower() == str(room_name).lower(): idx = plan.get("idx") break if idx is None: return f'{{"status": "error", "message": "Room \'{room_name}\' not found"}}' # Using plan=idx returns the full status of all devices in the room, rather than just their IDs response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&plan={idx}") data = response.json() if "result" in data: data["result"] = [_simplify_device(d) for d in data["result"]] return json.dumps(data) @mcp.tool() async def get_user_variables() -> str: """Get all user variables.""" async with create_client() as client: vars = await _get_cached_data(client, _user_variable_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getuservariables") return json.dumps({"status": "OK", "result": vars}) @mcp.tool() async def add_user_variable(name: str, vtype: int, value: str) -> str: """Add a new user variable. vtype (Variable Type): 0: Integer 1: Float 2: String 3: Date (DD/MM/YYYY) 4: Time (HH:MM) """ async with create_client() as client: response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=adduservariable&vname={name}&vtype={vtype}&vvalue={value}") _user_variable_cache["timestamp"] = 0 # Invalidate cache return response.text @mcp.tool() async def update_user_variable(name: str, vtype: int, value: str) -> str: """Update an existing user variable. vtype (Variable Type): 0: Integer 1: Float 2: String 3: Date (DD/MM/YYYY) 4: Time (HH:MM) """ async with create_client() as client: response = await _do_request(client, "GET", f"{DOMOTICZ_API_URL}?type=command¶m=updateuservariable&vname={name}&vtype={vtype}&vvalue={value}") _user_variable_cache["timestamp"] = 0 # Invalidate cache return response.text @mcp.tool() - src/domoticz_mcp/server.py:383-384 (helper)Helper that resolves a user variable name to its idx by querying the API and using the cache.
async def _resolve_user_variable_idx(client: "httpx.AsyncClient", idx: Optional[int] = None, name: Optional[str] = None) -> Optional[int]: """Resolve a user variable to its idx.""" - src/domoticz_mcp/server.py:743-748 (schema)Docstring describing the input schema for vtype parameter (integer codes for variable types).
vtype (Variable Type): 0: Integer 1: Float 2: String 3: Date (DD/MM/YYYY) 4: Time (HH:MM)