Coverage for gco_mcp / tools / status.py: 100.00%

1"""Fleet-wide status MCP tool (read-only).""" 

2 

3import asyncio 

4 

5import cli_runner 

6from audit import audit_logged 

7from server import mcp 

8 

9 

10@mcp.tool(tags={"safe", "status"}) 

11@audit_logged 

12async def fleet_status( 

13 region: str | None = None, 

14 with_costs: bool = False, 

15 with_nodepools: bool = False, 

16) -> str: 

17 """`gco status` — whole-fleet deployment status as one document. 

18 

19 Aggregates stacks, queue depth, job counts, capacity telemetry, and 

20 inference endpoints across every configured region. Each section 

21 carries its own status (ok/empty/partial/unavailable/error/skipped) 

22 plus a findings list of what looks wrong, so one call answers what 

23 would otherwise take eight. 

24 

25 Args: 

26 region: Restrict the gather to a single region (default: every 

27 configured deployment region). 

28 with_costs: Include the costs section. Cost Explorer bills per 

29 request, so this is off by default. 

30 with_nodepools: Include Karpenter NodePools. Requires a publicly 

31 reachable cluster API endpoint; private endpoints are reported 

32 as unavailable. 

33 """ 

34 args = ["status"] 

35 if region: 

36 args += ["-r", region] 

37 if with_costs: 

38 args.append("--with-costs") 

39 if with_nodepools: 

40 args.append("--with-nodepools") 

41 return await asyncio.to_thread(cli_runner._run_cli, *args)