ako_sync_status
Read-onlyIdempotent
Check synchronization status between Kubernetes resources and AVI Controller objects, displaying counts for in-sync, pending, and error states.
Instructions
[READ] Check sync status between K8s resources and AVI Controller objects.
Shows in-sync, pending, and error counts.
Args: context: K8s context name (optional).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| context | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- vmware_avi/ops/ako_sync.py:15-43 (handler)Core handler: check_sync_status() function that compares K8s Ingress count vs AVI Virtual Services count to determine sync status.
def check_sync_status(context: str | None = None) -> None: """Check whether K8s Ingress objects are in sync with AVI Controller VS objects.""" 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() k8s_count = len(ingresses.items) mgr = AviConnectionManager(cfg) session = mgr.connect() resp = session.get("virtualservice", params={"page_size": "1000"}) vs_list = resp.json().get("results", []) avi_count = len(vs_list) console.print("\n[bold]Sync Status[/bold]") console.print(f" K8s Ingress objects: {k8s_count}") console.print(f" AVI Virtual Services: {avi_count}") if k8s_count == avi_count: console.print(" [green]Counts match (basic check OK).[/green]") else: console.print( f" [yellow]Count mismatch: {k8s_count} Ingresses vs {avi_count} VS. " "Run 'vmware-avi ako sync diff' for details.[/yellow]" ) console.print() - mcp_server/server.py:24-48 (helper)Output capture helper _capture_output used by the MCP tool to capture Rich console output as plain text.
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() - mcp_server/server.py:415-427 (registration)MCP tool registration: ako_sync_status decorated with @mcp.tool and @vmware_tool, with read-only annotation and low risk level.
@mcp.tool(annotations={"readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True}) @vmware_tool(risk_level="low") def ako_sync_status(context: str | None = None) -> str: """[READ] Check sync status between K8s resources and AVI Controller objects. Shows in-sync, pending, and error counts. Args: context: K8s context name (optional). """ from vmware_avi.ops.ako_sync import check_sync_status return _capture_output(check_sync_status, context) - vmware_avi/cli.py:364-369 (handler)CLI command registration: ako_sync_status_cmd() that wraps check_sync_status() for the vmware-avi CLI.
@ako_app.command("sync-status") def ako_sync_status_cmd() -> None: """Check K8s-Controller sync status.""" from vmware_avi.ops.ako_sync import check_sync_status check_sync_status()