get_connectivity_report

Instructions

Implementation Reference

• src/domoticz_mcp/server.py:1129-1164 (handler)

  The main handler for the 'get_connectivity_report' tool. It is registered as an MCP tool via @mcp.tool() decorator. Fetches all devices, filters those whose LastUpdate is older than the specified hours threshold, and returns a sorted JSON list of unresponsive devices.

  @mcp.tool()
  async def get_connectivity_report(hours: int = 24) -> str:
      """Get a list of devices that haven't checked in/updated within the specified timeframe.
      
      Args:
          hours: Number of hours threshold for considering a device 'unresponsive' (default: 24).
      """
      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")
          now = datetime.now()
          threshold_time = now - timedelta(hours=hours)
          results = []
          
          for dev in devices:
              last_update_str = dev.get("LastUpdate")
              if not last_update_str:
                  continue
                  
              try:
                  # Domoticz format: YYYY-MM-DD HH:MM:SS
                  last_update = datetime.strptime(last_update_str, "%Y-%m-%d %H:%M:%S")
                  if last_update < threshold_time:
                      results.append({
                          "idx": dev.get("idx"),
                          "Name": dev.get("Name"),
                          "LastUpdate": last_update_str,
                          "HardwareName": dev.get("HardwareName"),
                          "Type": dev.get("Type"),
                          "Data": dev.get("Data")
                      })
              except ValueError:
                  continue
                  
          # Sort by oldest update first
          results.sort(key=lambda x: x["LastUpdate"])
          return json.dumps({"status": "OK", "result": results})

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

  Registration of the tool via the @mcp.tool() decorator on line 1129, which registers it with the FastMCP instance named 'Domoticz'.

  @mcp.tool()
  async def get_connectivity_report(hours: int = 24) -> str:

• src/domoticz_mcp/server.py:1130-1134 (schema)

  The function signature and docstring define the schema: takes an optional 'hours' parameter (int, default 24) and returns a JSON string.

  async def get_connectivity_report(hours: int = 24) -> str:
      """Get a list of devices that haven't checked in/updated within the specified timeframe.
      
      Args:
          hours: Number of hours threshold for considering a device 'unresponsive' (default: 24).

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

  The _get_cached_data helper function used internally to fetch and cache device data 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"]

• tests/test_server.py:223-245 (helper)

  Test for get_connectivity_report, verifying it correctly identifies devices not updated within 24 hours.

  @pytest.mark.asyncio
  @respx.mock
  async def test_get_connectivity_report():
      from domoticz_mcp.server import get_connectivity_report
      from datetime import datetime, timedelta
      
      old_time = (datetime.now() - timedelta(hours=48)).strftime("%Y-%m-%d %H:%M:%S")
      new_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
      
      mock_data = {
          "result": [
              {"idx": "1", "Name": "Dead Sensor", "LastUpdate": old_time},
              {"idx": "2", "Name": "Live Sensor", "LastUpdate": new_time}
          ]
      }
      respx.get(f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true").mock(
          return_value=Response(200, json=mock_data)
      )
      
      response = await get_connectivity_report(hours=24)
      data = json.loads(response)
      assert len(data["result"]) == 1
      assert data["result"][0]["Name"] == "Dead Sensor"