analyze_network_topology

Instructions

Implementation Reference

• src/homelab_mcp/tool_handlers/network_handlers.py:34-39 (handler)

  Tool handler function that executes analyze_network_topology - creates a NetworkSiteMap instance, calls analyze_network_topology() method, and returns the result as JSON

  async def handle_analyze_network_topology(arguments: dict[str, Any]) -> dict[str, Any]:
      """Handle analyze_network_topology tool."""
      sitemap = NetworkSiteMap()
      analysis = sitemap.analyze_network_topology()
      result = json.dumps({"status": "success", "analysis": analysis}, indent=2)
      return {"content": [{"type": "text", "text": result}]}

• src/homelab_mcp/sitemap.py:132-210 (handler)

  Core implementation of analyze_network_topology in NetworkSiteMap class - analyzes all discovered devices, provides insights on OS distribution, CPU architectures, network segments, and resource utilization (identifies high disk/memory usage and low-resource devices)

  def analyze_network_topology(self) -> dict[str, Any]:
      """Analyze the network topology and provide insights."""
      devices = self.get_all_devices()
  
      analysis = {
          "total_devices": len(devices),
          "online_devices": len([d for d in devices if d["status"] == "success"]),
          "offline_devices": len([d for d in devices if d["status"] == "error"]),
          "operating_systems": {},
          "cpu_architectures": {},
          "network_segments": {},
          "resource_utilization": {
              "high_memory_usage": [],
              "high_disk_usage": [],
              "low_resources": [],
          },
      }
  
      for device in devices:
          if device["status"] != "success":
              continue
  
          # OS distribution
          os_info = device.get("os_info", "Unknown")
          if isinstance(analysis["operating_systems"], dict):
              analysis["operating_systems"][os_info] = analysis["operating_systems"].get(os_info, 0) + 1
  
          # CPU models
          cpu_model = device.get("cpu_model", "Unknown")
          if isinstance(analysis["cpu_architectures"], dict):
              analysis["cpu_architectures"][cpu_model] = analysis["cpu_architectures"].get(cpu_model, 0) + 1
  
          # Network segments (by IP prefix)
          connection_ip = device.get("connection_ip", "")
          if "." in connection_ip:
              network_prefix = ".".join(connection_ip.split(".")[:3]) + ".0/24"
              if isinstance(analysis["network_segments"], dict):
                  analysis["network_segments"][network_prefix] = (
                      analysis["network_segments"].get(network_prefix, 0) + 1
                  )
  
          # Resource utilization analysis
          if device.get("disk_use_percent"):
              try:
                  disk_usage = int(device["disk_use_percent"].rstrip("%"))
                  if disk_usage > 80:
                      if (
                          isinstance(analysis["resource_utilization"], dict)
                          and "high_disk_usage" in analysis["resource_utilization"]
                      ):
                          analysis["resource_utilization"]["high_disk_usage"].append(
                              {
                                  "hostname": device["hostname"],
                                  "usage": device["disk_use_percent"],
                              }
                          )
              except (ValueError, AttributeError):
                  pass
  
          # Identify resource-constrained devices
          cpu_cores = device.get("cpu_cores")
          if cpu_cores is not None and cpu_cores <= 2:
              memory_total = device.get("memory_total")
              if memory_total:
                  memory_gb = self._parse_memory_gb(str(memory_total))
                  if (
                      memory_gb <= 2
                      and isinstance(analysis["resource_utilization"], dict)
                      and "low_resources" in analysis["resource_utilization"]
                  ):
                      analysis["resource_utilization"]["low_resources"].append(
                          {
                              "hostname": device["hostname"],
                              "cpu_cores": device["cpu_cores"],
                              "memory": device["memory_total"],
                          }
                      )
  
      return analysis

• src/homelab_mcp/tool_handlers/__init__.py:86-86 (registration)

  Tool registration in TOOL_HANDLERS dictionary - maps 'analyze_network_topology' tool name to handle_analyze_network_topology function

  "analyze_network_topology": handle_analyze_network_topology,

• src/homelab_mcp/tool_schemas/network_tools_schema.py:61-64 (schema)

  Tool schema definition for analyze_network_topology - defines description and input schema (empty object with no required parameters)

  "analyze_network_topology": {
      "description": "Analyze the network topology and provide insights about the discovered devices",
      "inputSchema": {"type": "object", "properties": {}, "required": []},
  },