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

1"""Cluster connectivity MCP tools (thin wrapper over `gco cluster tunnel`). 

2 

3``cluster_tunnel_command`` is ``safe`` / read-only: it resolves how to reach a 

4cluster's (possibly private) EKS API endpoint and returns the ready-to-run 

5``aws ssm start-session`` tunnel command plus the ``kubectl`` flags to use with 

6it. There is intentionally no tool that *holds a tunnel open* — an SSM tunnel is 

7an interactive, long-running process, not a request/response operation (the same 

8reason there is no ``monitoring open`` tool). An agent calls this to learn *how* 

9to connect, then the operator (or the ``gco cluster tunnel`` CLI) runs it. 

10""" 

11 

12import asyncio 

13 

14import cli_runner 

15from audit import audit_logged 

16from server import mcp 

17 

18 

19@mcp.tool(tags={"safe", "cluster"}) 

20@audit_logged 

21async def cluster_tunnel_command( 

22 region: str | None = None, 

23 instance_id: str | None = None, 

24 local_port: int = 8443, 

25) -> str: 

26 """`gco cluster tunnel --print` — connection plan for a cluster's API endpoint. 

27 

28 Returns a JSON connection plan: the `aws ssm start-session` command that 

29 tunnels to the (possibly private) EKS API endpoint and the 

30 `kubectl --server/--tls-server-name` flags to use through it. Read-only — it 

31 resolves the endpoint and builds commands; it does not open a tunnel or 

32 launch any instance. 

33 

34 Args: 

35 region: AWS region of the target cluster (e.g. us-east-1). When omitted, 

36 resolves to the first cdk.json regional entry. 

37 instance_id: Optional SSM-managed instance id to tunnel through. It is 

38 used whenever the cluster has private endpoint access, 

39 PUBLIC_AND_PRIVATE included, and the plan fails when private 

40 access is off. When omitted, a cluster with a public endpoint 

41 gets the direct plan; a private-only one gets a command template 

42 with an <INSTANCE_ID> placeholder — use `gco cluster tunnel 

43 --via-ssm auto` in the CLI to auto-provision a self-terminating 

44 ephemeral bastion instead. 

45 local_port: Local port to bind for the API tunnel (default 8443). 

46 """ 

47 args = ["cluster", "tunnel", "--print", "--local-port", str(local_port)] 

48 if region: 

49 args += ["--region", region] 

50 if instance_id: 

51 args += ["--via-ssm", instance_id] 

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