Static Networking Config

Dockerfile

@@ -13,6 +13,7 @@ RUN uv sync
 
 COPY server.py .
 COPY service_client ./service_client/
+COPY static_net ./static_net/
 COPY metrics ./metrics/
 
 RUN chown -R 1001:0 ${APP_HOME}

pyproject.toml

@@ -12,6 +12,9 @@ dependencies = [
     "retry>=0.9.2",
     "types-requests>=2.32.4.20250611",
     "prometheus_client>=0.22.1",
+    "pyyaml>=6",
+    "jinja2>=3.1",
+    "pydantic>=2",
 ]
 
 [dependency-groups]
@@ -22,6 +25,7 @@ dev = [
     "pylint>=3.3.7",
     "pyright>=1.1.402",
     "ruff>=0.12.1",
+    "types-pyyaml>=6",
 ]
 test = [
     "pytest>=8.0.0",
@@ -62,3 +66,8 @@ markers = [
     "integration: marks tests as integration tests",
 ]
 asyncio_mode = "auto"
+
+[tool.pydocstyle]
+add-ignore=[
+  "D400"  # Stop requiring periods after non-sentences.
+]

server.py

@@ -10,16 +10,23 @@
 import asyncio
 from typing import Any, Annotated
 
+from jinja2 import TemplateError
 import requests
 import uvicorn
 from pydantic import Field
 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 static_net import (
+    NMStateTemplateParams,
+    add_or_replace_static_host_config_yaml,
+    generate_nmstate_from_template,
+    remove_static_host_config_by_index,
+    validate_and_parse_nmstate,
+)
 
 
 transport_type = os.environ.get("TRANSPORT", "sse").lower()
@@ -441,6 +448,120 @@ async def create_cluster(  # pylint: disable=too-many-arguments,too-many-positio
     return cluster.id
 
 
+@mcp.tool()
+@track_tool_usage()
+async def validate_nmstate_yaml(nmstate_yaml: str) -> str:
+    """
+    Validate an nmstate yaml document.
+
+    The yaml should always be validated before submitted to the cluster.
+    """
+    validate_and_parse_nmstate(nmstate_yaml)
+    return "YAML is valid"
+
+
+@mcp.tool(
+    description=f"""
+    Generate an initial nmstate yaml.
+
+    You should call this after gathering information from the user to generate the initial nmstate
+    yaml. Then you can tweak it as needed. Do not generate nmstate yaml from scratch without calling
+    this tool.
+
+    Returns: the generated nmstate yaml
+
+    Input param schema:
+    {NMStateTemplateParams.model_json_schema()}
+"""
+)
+@track_tool_usage()
+async def generate_nmstate_yaml(params: NMStateTemplateParams) -> str:
+    """
+    Generate an initial nmstate yaml.
+
+    See the mcp.tool description for more details (we have to use it since we dynamically generate
+    the input schema for the params).
+    """
+    log.info("Generate nmstate yaml with params: %s", params.model_dump_json(indent=2))
+    try:
+        generated = generate_nmstate_from_template(params)
+        log.debug("Generated yaml: %s", generated)
+        return generated
+    except TemplateError as e:
+        log.error("Failed to render nmstate template", exc_info=e)
+        return "ERROR: Failed to generate nmstate yaml"
+    except Exception as e:
+        log.error("Exception generating nmstate yaml", exc_info=e)
+        return "ERROR: Unknown error"
+
+
+@mcp.tool()
+@track_tool_usage()
+async def alter_static_network_config_nmstate_for_host(
+    cluster_id: str, index: int | None, new_nmstate_yaml: str | None
+) -> str:
+    """
+    Add, replace, or delete the nmstate yaml for a single host.
+
+    Args:
+        cluster_id (str): The unique identifier of the cluster
+        index (int | None): The index of the host in the existing static config to replace, or None to append a new host to the end of the config.
+        new_nmstate_yaml (str): The new nmstate YAML for a host. Leave this None to delete the config at the given index.
+
+    Returns: the updated infra env with the new static network config
+    """
+    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)
+
+    if new_nmstate_yaml is None:
+        if index is None:
+            raise ValueError("index cannot be null when removing a host yaml")
+        if not infra_env.static_network_config:
+            raise ValueError(
+                "cannot remove host yaml with empty existing static network config"
+            )
+        static_network_config = remove_static_host_config_by_index(
+            existing_static_network_config=infra_env.static_network_config, index=index
+        )
+    else:
+        static_network_config = add_or_replace_static_host_config_yaml(
+            existing_static_network_config=infra_env.static_network_config,
+            index=index,
+            new_nmstate_yaml=new_nmstate_yaml,
+        )
+
+    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_static_network_config(cluster_id: str) -> str:
+    """
+    List all of the host static network config 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 static network config or an error string
+    """
+    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)
+
+    if len(infra_envs) != 1:
+        log.warning(
+            "cluster %s has %d infra_envs, expected 1", cluster_id, len(infra_envs)
+        )
+        return "ERROR: this cluster doesn't have exactly 1 infra env, cannot manage static network config"
+
+    return json.dumps(infra_envs[0].get("static_network_config", []))
+
+
 @mcp.tool()
 @track_tool_usage()
 async def set_cluster_vips(

static_net/__init__.py

@@ -0,0 +1,16 @@
+"""Static networking related functionality."""
+
+from .config import (
+    add_or_replace_static_host_config_yaml,
+    remove_static_host_config_by_index,
+    validate_and_parse_nmstate,
+)
+from .template import NMStateTemplateParams, generate_nmstate_from_template
+
+__all__ = [
+    "NMStateTemplateParams",
+    "add_or_replace_static_host_config_yaml",
+    "generate_nmstate_from_template",
+    "remove_static_host_config_by_index",
+    "validate_and_parse_nmstate",
+]

static_net/config.py

@@ -0,0 +1,106 @@
+"""infra env static_network_config field handling"""
+
+import json
+from typing import Any, TypedDict, cast
+
+import yaml
+
+
+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 remove_static_host_config_by_index(
+    existing_static_network_config: str, index: int
+) -> list[HostStaticNetworkConfig]:
+    """Remove a single host's config by index position."""
+    config = [
+        cast(HostStaticNetworkConfig, d)
+        for d in json.loads(existing_static_network_config)
+    ]
+    if index < 0:
+        raise IndexError("negative indexes are not allowed")
+    if index >= len(config):
+        raise ValueError(
+            f"static network config only has {len(config)} elements, cannot delete index {index}"
+        )
+    del config[index]
+    return config
+
+
+def add_or_replace_static_host_config_yaml(
+    existing_static_network_config: str | None,
+    index: int | None,
+    new_nmstate_yaml: str,
+) -> list[HostStaticNetworkConfig]:
+    """Add/update a single host's config by index.
+
+    Raises:
+     - IndexError: if the index is out of range for the existing config
+    """
+    config: list[HostStaticNetworkConfig] = []
+    if existing_static_network_config:
+        config = [
+            cast(HostStaticNetworkConfig, d)
+            for d in json.loads(existing_static_network_config)
+        ]
+
+    host_config = _generate_host_static_config(new_nmstate_yaml)
+    if index is None:
+        config.append(host_config)
+    else:
+        if index < 0:
+            raise IndexError("negative indexes are not allowed")
+        if index >= len(config):
+            raise IndexError(
+                f"static network config only has {len(config)} elements, cannot replace index {index}"
+            )
+        config[index] = host_config
+
+    return config
+
+
+def _generate_host_static_config(nmstate_yaml: str) -> HostStaticNetworkConfig:
+    nmstate = validate_and_parse_nmstate(nmstate_yaml)
+    interfaces = nmstate.get("interfaces")
+    name_and_mac_list: list[MacInterfaceMap] = [
+        {
+            "mac_address": i.get("mac-address"),
+            "logical_nic_name": i.get("name"),
+        }
+        for i in interfaces
+        if i.get("mac-address")
+    ]
+    if not name_and_mac_list:
+        raise ValueError("At least one interface must be associated to a MAC Address")
+
+    new_host: HostStaticNetworkConfig = {
+        "mac_interface_map": name_and_mac_list,
+        "network_yaml": nmstate_yaml,
+    }
+    return new_host
+
+
+def validate_and_parse_nmstate(nmstate_yaml: str) -> Any:
+    """Validate nmstate yaml and return it parsed.
+
+    Raises:
+     - ValueError: If the yaml is invalid is some way
+    """
+    # Eventually when nmstate 2.2.51 is released we should be able to validate the nmstate by doing:
+    # libnmstate.validate(nmstate_yaml)
+    # For now just make sure it is valid yaml
+    try:
+        return yaml.safe_load(nmstate_yaml)
+    except yaml.YAMLError as e:
+        raise ValueError("Invalid YAML") from e