ako_ingress_map

Instructions

Implementation Reference

• mcp_server/server.py:371-382 (handler)

  MCP tool handler for ako_ingress_map - calls show_ingress_map via _capture_output helper

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def ako_ingress_map(context: str | None = None) -> str:
      """[READ] Show mapping between K8s Ingress resources and AVI Virtual Services.
  
      Use to verify which Ingresses have corresponding VS objects.
  
      Args:
          context: K8s context name (optional).
      """
      from vmware_avi.ops.ako_ingress import show_ingress_map
      return _capture_output(show_ingress_map, context)

• mcp_server/server.py:371-382 (registration)

  Tool registered as MCP tool with FastMCP using @mcp.tool decorator; marked read-only and low risk

  @mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True})
  @vmware_tool(risk_level="low")
  def ako_ingress_map(context: str | None = None) -> str:
      """[READ] Show mapping between K8s Ingress resources and AVI Virtual Services.
  
      Use to verify which Ingresses have corresponding VS objects.
  
      Args:
          context: K8s context name (optional).
      """
      from vmware_avi.ops.ako_ingress import show_ingress_map
      return _capture_output(show_ingress_map, context)

• vmware_avi/ops/ako_ingress.py:72-101 (helper)

  Core implementation - queries all k8s Ingresses and builds a table mapping Ingress resources to their hosts and IngressClass (proxy for VS mapping)

  def show_ingress_map(context: str | None = None) -> None:
      """Show Ingress to VS mapping across all namespaces."""
      cfg = load_config()
      k8s = K8sConnectionManager(cfg)
  
      from kubernetes.client import NetworkingV1Api
  
      net_v1 = NetworkingV1Api(k8s.get_client(context))
      ingresses = net_v1.list_ingress_for_all_namespaces()
  
      table = Table(title="Ingress → VS Mapping")
      table.add_column("Namespace")
      table.add_column("Ingress")
      table.add_column("Host")
      table.add_column("IngressClass")
  
      for ing in ingresses.items:
          ns = ing.metadata.namespace
          name = ing.metadata.name
          annotations = ing.metadata.annotations or {}
          ingress_class = ing.spec.ingress_class_name or annotations.get(
              "kubernetes.io/ingress.class", ""
          )
          hosts = []
          if ing.spec.rules:
              hosts = [r.host or "*" for r in ing.spec.rules]
  
          table.add_row(ns, name, ", ".join(hosts), ingress_class or "N/A")
  
      console.print(table)

• mcp_server/server.py:24-48 (helper)

  Helper that captures show_ingress_map's Rich console output into a string for MCP response

  def _capture_output(func, *args, **kwargs) -> str:
      """Run a function and capture its Rich console output as plain text."""
      import importlib  # noqa: F401 — used via sys.modules lookup
      import sys
  
      buf = StringIO()
      from rich.console import Console
      capture_console = Console(file=buf, force_terminal=False, width=120)
  
      mod_name = func.__module__
      mod = sys.modules.get(mod_name)
      original_console = getattr(mod, "console", None) if mod else None
  
      if mod and original_console is not None:
          mod.console = capture_console
  
      try:
          func(*args, **kwargs)
      except SystemExit:
          pass
      finally:
          if mod and original_console is not None:
              mod.console = original_console
  
      return buf.getvalue()

• vmware_avi/cli.py:342-347 (registration)

  CLI command registration for 'ako ingress-map' (separate from MCP tool registration)

  @ako_app.command("ingress-map")
  def ako_ingress_map_cmd() -> None:
      """Show Ingress to VS mapping."""
      from vmware_avi.ops.ako_ingress import show_ingress_map
  
      show_ingress_map()