control_blinds

Instructions

Implementation Reference

• src/domoticz_mcp/server.py:638-656 (handler)

  The main handler function for the control_blinds tool. It accepts 'command' ('Open', 'Close', or 'Stop'), resolves the device by idx or name, and sends a switchlight command to the Domoticz API.

  @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

• src/domoticz_mcp/server.py:639-646 (schema)

  Docstring and parameter schema for control_blinds. Defines the 'command' parameter (Open/Close/Stop) and optional idx/name parameters.

  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.
      """

• src/domoticz_mcp/server.py:638-638 (registration)

  The @mcp.tool() decorator registers control_blinds as an MCP tool.

  @mcp.tool()

• src/domoticz_mcp/server.py:346-376 (helper)

  Helper functions _resolve_idx and _resolve_device_idx used by control_blinds to resolve a device name or idx. These cache and look up devices from the Domoticz API.

  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")