vmware-nsx

Instructions

Implementation Reference

• mcp_server/server.py:670-701 (handler)

  This is the MCP tool registration and entry-point function that handles the 'create_ip_pool' tool request, delegating to the operational implementation.

  @mcp.tool()
  def create_ip_pool(
      pool_id: str,
      display_name: str,
      start_ip: str,
      end_ip: str,
      cidr: str,
      gateway_ip: str | None = None,
      target: str | None = None,
  ) -> dict:
      """Create a new IP address pool with a static subnet allocation range.
  
      Args:
          pool_id: Unique ID for the IP pool.
          display_name: Human-readable name.
          start_ip: Start IP address of the allocation range.
          end_ip: End IP address of the allocation range.
          cidr: Subnet CIDR (e.g. "192.168.1.0/24").
          gateway_ip: Gateway IP for the subnet (optional).
          target: Optional NSX Manager target name from config. Uses default if omitted.
      """
      from vmware_nsx.ops.ip_pool_mgmt import create_ip_pool as _create
  
      client = _get_connection(target)
      return _create(
          client, pool_id,
          display_name=display_name,
          start_ip=start_ip,
          end_ip=end_ip,
          cidr=cidr,
          gateway_ip=gateway_ip,
      )

• vmware_nsx/ops/nat_route_mgmt.py:246-314 (handler)

  The actual operational implementation that performs the NSX Policy API calls to create an IP pool and its subnets. Note: The import in server.py (vmware_nsx.ops.ip_pool_mgmt) seems to point here (nat_route_mgmt.py), likely due to an aliased package structure or a file rename not reflected in the import.

  def create_ip_pool(
      client: NsxClient,
      pool_id: str,
      display_name: str,
      subnets: list[dict[str, Any]],
  ) -> dict:
      """Create an IP pool via Policy API (PUT).
  
      Args:
          client: Authenticated NSX API client.
          pool_id: Unique IP pool identifier.
          display_name: Human-readable name.
          subnets: List of subnet dicts, each containing:
              - allocation_ranges (list[dict]): Each with "start" and "end" IPs.
              - cidr (str): Subnet CIDR (e.g., "192.168.1.0/24").
              - gateway_ip (str, optional): Gateway IP for the subnet.
  
      Returns:
          Created IP pool dict from NSX API.
      """
      _validate_id(pool_id)
  
      if not subnets:
          raise ValueError("At least one subnet is required.")
  
      # Validate subnet structure
      for sub in subnets:
          if "allocation_ranges" not in sub or "cidr" not in sub:
              raise ValueError(
                  "Each subnet must have 'allocation_ranges' and 'cidr'. "
                  "allocation_ranges: [{start: ip, end: ip}], cidr: x.x.x.x/y"
              )
  
      body: dict[str, Any] = {
          "display_name": _sanitize(display_name),
      }
  
      path = f"/policy/api/v1/infra/ip-pools/{pool_id}"
      result = client.put(path, body)
  
      # Create IP subnets as sub-resources (IpAddressPoolStaticSubnet)
      for idx, sub in enumerate(subnets):
          subnet_id = f"{pool_id}-subnet-{idx}"
          subnet_body: dict[str, Any] = {
              "resource_type": "IpAddressPoolStaticSubnet",
              "display_name": f"{display_name} subnet {idx}",
              "cidr": sub["cidr"],
              "allocation_ranges": [
                  {"start": r["start"], "end": r["end"]}
                  for r in sub["allocation_ranges"]
              ],
          }
          if "gateway_ip" in sub:
              subnet_body["gateway_ip"] = sub["gateway_ip"]
  
          subnet_path = (
              f"/policy/api/v1/infra/ip-pools/{pool_id}"
              f"/ip-subnets/{subnet_id}"
          )
          client.put(subnet_path, subnet_body)
  
      _log.info(
          "Created IP pool %s (%s) with %d subnet(s)",
          pool_id,
          display_name,
          len(subnets),
      )
      return result