Add tools for static network configuration

server.py

@@ -15,10 +15,13 @@
 from assisted_service_client import models
 from mcp.server.fastmcp import FastMCP
 
-
+from metrics import metrics, track_tool_usage, initiate_metrics
 from service_client import InventoryClient
 from service_client.logger import log
-from metrics import metrics, track_tool_usage, initiate_metrics
+from service_client.static_net import (
+    add_or_update_static_host_config,
+    remove_static_host_config,
+)
 
 
 transport_type = os.environ.get("TRANSPORT", "sse").lower()
@@ -386,6 +389,92 @@ async def create_cluster(  # pylint: disable=too-many-arguments,too-many-positio
     return cluster.id
 
 
+@mcp.tool()
+@track_tool_usage()
+async def update_static_network_config_for_host(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    cluster_id: str,
+    dns_server: str,
+    mac_address: str,
+    ip_address: str,
+    subnet_prefix_len: int,
+    gateway_address: str,
+) -> str:
+    """
+    Add or update static networking configuration for a single host.
+
+    Args:
+        cluster_id (str): The unique identifier of the cluster
+        dns_server (str): The DNS server to use for host resolution
+        mac_address (str): The MAC address of the main interface on the host
+        ip_address (str): The IP address to assign to the host with the given MAC address
+        subnet_prefix_len (int): The length of the subnet prefix, between 0 and 32
+        gateway_address (str): The IP address of the gateway for the default network route
+    """
+    client = InventoryClient(get_access_token())
+    infra_env_id = await _get_cluster_infra_env_id(client, cluster_id)
+    infra_env = await client.get_infra_env(infra_env_id)
+
+    static_network_config = add_or_update_static_host_config(
+        existing_static_network_config=infra_env.static_network_config,
+        dns_server=dns_server,
+        mac_address=mac_address,
+        ip_address=ip_address,
+        gateway_address=gateway_address,
+        subnet_prefix_len=subnet_prefix_len,
+    )
+
+    result = await client.update_infra_env(
+        infra_env_id, static_network_config=static_network_config
+    )
+    return result.to_str()
+
+
+@mcp.tool()
+@track_tool_usage()
+async def delete_static_network_config_for_host(
+    cluster_id: str, mac_address: str
+) -> str:
+    """
+    Delete static networking configuration for a single host by MAC address.
+
+    Args:
+        cluster_id (str): The unique identifier of the cluster
+        mac_address (str): The MAC address associated with this host
+    """
+    client = InventoryClient(get_access_token())
+    infra_env_id = await _get_cluster_infra_env_id(client, cluster_id)
+    infra_env = await client.get_infra_env(infra_env_id)
+
+    static_network_config = remove_static_host_config(
+        existing_static_network_config=infra_env.static_network_config,
+        mac_address=mac_address,
+    )
+    result = await client.update_infra_env(
+        infra_env_id, static_network_config=static_network_config
+    )
+    return result.to_str()
+
+
+@mcp.tool()
+@track_tool_usage()
+async def list_infra_envs(cluster_id: str) -> str:
+    """
+    List all of the infra_envs associated with the given cluster_id.
+
+    Args:
+        cluster_id (str): The unique identifier of the cluster to configure.
+
+    Returns:
+        str: A JSON-formatted list of infra environments, including static network
+        config
+    """
+    client = InventoryClient(get_access_token())
+    infra_envs = await client.list_infra_envs(cluster_id)
+    log.info("Found %d InfraEnvs for cluster %s", len(infra_envs), cluster_id)
+
+    return json.dumps(infra_envs)
+
+
 @mcp.tool()
 @track_tool_usage()
 async def set_cluster_vips(cluster_id: str, api_vip: str, ingress_vip: str) -> str:

service_client/static_net.py

@@ -0,0 +1,127 @@
+"""Static networking related functionality."""
+
+import json
+from typing import TypedDict, cast
+
+
+class MacInterfaceMap(TypedDict):
+    """Maps a NIC to a MAC Address."""
+
+    logical_nic_name: str
+    mac_address: str
+
+
+class HostStaticNetworkConfig(TypedDict):
+    """Matches the structure in the Assisted Installer API."""
+
+    mac_interface_map: list[MacInterfaceMap]
+    network_yaml: str
+
+
+def add_or_update_static_host_config(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    existing_static_network_config: str | None,
+    dns_server: str,
+    mac_address: str,
+    ip_address: str,
+    subnet_prefix_len: int,
+    gateway_address: str,
+) -> list[HostStaticNetworkConfig]:
+    """Generate and append or replace the config for a given mac address."""
+    config: list[HostStaticNetworkConfig] = []
+    if existing_static_network_config:
+        config = [
+            cast(HostStaticNetworkConfig, d)
+            for d in json.loads(existing_static_network_config)
+        ]
+
+    i = find_host_config_index_for_mac(mac_address, config)
+    if i is None:
+        host_conf = HostStaticNetworkConfig(
+            {
+                "mac_interface_map": [
+                    {
+                        "logical_nic_name": "eth0",
+                        "mac_address": mac_address,
+                    }
+                ],
+                "network_yaml": "",
+            }
+        )
+        config.append(host_conf)
+        i = len(config) - 1
+
+    config[i]["network_yaml"] = generate_basic_nmstate_yaml(
+        dns_server,
+        ip_address,
+        subnet_prefix_len,
+        gateway_address,
+    )
+
+    return config
+
+
+def find_host_config_index_for_mac(
+    mac_address: str, host_configs: list[HostStaticNetworkConfig]
+) -> int | None:
+    """Find the index of the host config for the given mac_address.
+
+    Do a simple linear search since the list should be fairly small. The returned config refers to
+    the original object in the list.
+    """
+    for i, c in enumerate(host_configs):
+        for m in c["mac_interface_map"]:
+            if m["mac_address"].lower() == mac_address.lower():
+                return i
+    return None
+
+
+def generate_basic_nmstate_yaml(
+    dns_server: str,
+    ip_address: str,
+    subnet_prefix_len: int,
+    gateway_address: str,
+) -> str:
+    """Generate a basic NMState config with the given parameters."""
+    return f"""
+interfaces:
+  - name: eth0
+    type: ethernet
+    state: up
+    ipv4:
+      address:
+        - ip: {ip_address}
+          prefix-length: {subnet_prefix_len}
+      enabled: true
+      dhcp: false
+dns-resolver:
+  config:
+    server:
+      - "{dns_server}"
+routes:
+  config:
+    - destination: 0.0.0.0/0
+      next-hop-address: {gateway_address}
+      next-hop-interface: eth0
+      table-id: 254
+"""
+
+
+def remove_static_host_config(
+    existing_static_network_config: str | None,
+    mac_address: str,
+) -> list[HostStaticNetworkConfig]:
+    """Remove the static config for the given MAC."""
+    config: list[HostStaticNetworkConfig] = []
+    if existing_static_network_config:
+        config = [
+            cast(HostStaticNetworkConfig, d)
+            for d in json.loads(existing_static_network_config)
+        ]
+
+    i = find_host_config_index_for_mac(mac_address, config)
+    if i is not None:
+        del config[i]
+    else:
+        raise ValueError(f"host config for mac address {mac_address} does not exist")
+
+    return config