vmware-nsx

Instructions

Implementation Reference

• vmware_nsx/ops/networking.py:67-164 (handler)

  The actual implementation of get_bgp_neighbors, which queries NSX Policy API for locale-services, BGP configuration, and status.

  def get_bgp_neighbors(client: NsxClient, tier0_id: str) -> dict:
      """Get BGP neighbor status for a Tier-0 gateway.
  
      Uses the Policy API to discover locale-services, then queries the
      Management API for realized BGP neighbor status.
  
      Args:
          client: Authenticated NSX API client.
          tier0_id: Tier-0 gateway identifier.
  
      Returns:
          Dict with locale_service info and list of BGP neighbors.
      """
      # Step 1: Get locale-services for this Tier-0
      ls_path = f"/policy/api/v1/infra/tier-0s/{tier0_id}/locale-services"
      locale_services = client.get_all(ls_path)
  
      if not locale_services:
          return {
              "tier0_id": tier0_id,
              "locale_services": [],
              "bgp_neighbors": [],
              "hint": "No locale-services found on this Tier-0 gateway.",
          }
  
      first_ls = locale_services[0]
      ls_id = first_ls.get("id", "")
  
      # Step 2: Get BGP config from Policy API
      bgp_path = (
          f"/policy/api/v1/infra/tier-0s/{tier0_id}"
          f"/locale-services/{ls_id}/bgp"
      )
      try:
          bgp_config = client.get(bgp_path)
      except Exception:
          bgp_config = {}
  
      # Step 3: Get BGP neighbors from Policy API
      neighbors_path = (
          f"/policy/api/v1/infra/tier-0s/{tier0_id}"
          f"/locale-services/{ls_id}/bgp/neighbors"
      )
      try:
          neighbors = client.get_all(neighbors_path)
      except Exception:
          neighbors = []
  
      # Step 4: Try to get realized BGP neighbor status via Management API
      bgp_status: list[dict] = []
      try:
          # Get the realized logical router ID for this Tier-0
          status_path = (
              f"/policy/api/v1/infra/tier-0s/{tier0_id}"
              f"/locale-services/{ls_id}/bgp/neighbors/status"
          )
          status_data = client.get(status_path)
          bgp_status = status_data.get("results", [])
      except Exception:
          _log.debug(
              "Could not retrieve BGP neighbor status for tier0=%s",
              tier0_id,
          )
  
      return {
          "tier0_id": tier0_id,
          "locale_service_id": _sanitize(ls_id),
          "local_as_num": bgp_config.get("local_as_num", ""),
          "enabled": bgp_config.get("enabled", False),
          "graceful_restart": bgp_config.get("graceful_restart_config", {}),
          "neighbors": [
              {
                  "id": _sanitize(n.get("id", "")),
                  "display_name": _sanitize(n.get("display_name", "")),
                  "neighbor_address": _sanitize(n.get("neighbor_address", "")),
                  "remote_as_num": n.get("remote_as_num", ""),
                  "source_addresses": n.get("source_addresses", []),
                  "hold_down_timer": n.get("hold_down_timer", 180),
                  "keep_alive_timer": n.get("keep_alive_timer", 60),
              }
              for n in neighbors
          ],
          "neighbor_status": [
              {
                  "neighbor_address": _sanitize(
                      s.get("neighbor_address", "")
                  ),
                  "remote_as_num": s.get("remote_as_number", ""),
                  "connection_state": s.get("connection_state", ""),
                  "time_since_established": s.get(
                      "time_since_established", 0
                  ),
                  "messages_received": s.get("total_in_prefix_count", 0),
                  "messages_sent": s.get("total_out_prefix_count", 0),
              }
              for s in bgp_status
          ],
      }

• mcp_server/server.py:227-238 (registration)

  The MCP tool registration for get_bgp_neighbors, which delegates to the ops/networking.py module.

  @mcp.tool()
  def get_bgp_neighbors(tier0_id: str, target: str | None = None) -> list[dict]:
      """Get BGP neighbors for a Tier-0 gateway with connection state and ASN.
  
      Args:
          tier0_id: The Tier-0 gateway ID.
          target: Optional NSX Manager target name from config. Uses default if omitted.
      """
      from vmware_nsx.ops.networking import get_bgp_neighbors as _get_bgp
  
      client = _get_connection(target)
      return _get_bgp(client, tier0_id)