Coverage for gco_mcp / tools / nodepools.py: 100.00%
65 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Karpenter NodePool management MCP tools.
3``nodepools_list``, ``nodepools_describe`` and ``delete_nodepool`` talk to the
4cluster's EKS API endpoint, which GCO deploys private by default. These tools
5never provision a bastion (that launches a billable EC2 instance), so they
6work from inside the VPC or when the operator has a tunnel open; otherwise
7``cluster_tunnel_command`` describes the route and the ``gco nodepools``
8commands take ``--via-ssm auto`` themselves. The two generators only read
9EKS and EC2 and return a manifest.
10"""
12import asyncio
14import cli_runner
15from audit import audit_logged
16from server import mcp
19@mcp.tool(tags={"safe", "nodepools"})
20@audit_logged
21async def nodepools_list(region: str | None = None, cluster: str | None = None) -> str:
22 """`gco nodepools list` — list Karpenter NodePools in a cluster.
24 Needs the cluster's EKS API endpoint to be reachable (it is private by
25 default; see ``cluster_tunnel_command``).
27 Args:
28 region: AWS region to query.
29 cluster: EKS cluster name (defaults to ``gco-<region>``).
30 """
31 args = ["nodepools", "list"]
32 if region:
33 args += ["-r", region]
34 if cluster:
35 args += ["--cluster", cluster]
36 return await asyncio.to_thread(cli_runner._run_cli, *args)
39@mcp.tool(tags={"safe", "nodepools"})
40@audit_logged
41async def nodepools_describe(nodepool_name: str, region: str, cluster: str | None = None) -> str:
42 """`gco nodepools describe` — describe a single NodePool.
44 Args:
45 nodepool_name: NodePool name.
46 region: AWS region.
47 cluster: EKS cluster name (defaults to ``gco-<region>``).
48 """
49 args = ["nodepools", "describe", nodepool_name, "-r", region]
50 if cluster:
51 args += ["--cluster", cluster]
52 return await asyncio.to_thread(cli_runner._run_cli, *args)
55# =============================================================================
56# Mutating tools (low-risk)
57# =============================================================================
60@mcp.tool(tags={"low-risk", "nodepools"})
61@audit_logged
62async def nodepools_create_odcr(
63 name: str,
64 region: str,
65 capacity_reservation_id: str,
66 instance_type: list[str] | None = None,
67 max_nodes: int = 100,
68 fallback_on_demand: bool = False,
69 efa: bool = False,
70 cluster: str | None = None,
71) -> str:
72 """`gco nodepools create-odcr` — generate a NodePool manifest tied to an ODCR.
74 Returns an EKS Auto Mode NodeClass that binds an On-Demand Capacity
75 Reservation through ``capacityReservationSelectorTerms`` (with the
76 cluster's own node role, subnets and security group) plus a NodePool
77 that launches reserved capacity through it. Nothing is applied: apply
78 the returned YAML with kubectl, and target the pool from a pod with
79 ``nodeSelector karpenter.sh/nodepool: <name>``.
81 Args:
82 name: NodePool name.
83 region: AWS region.
84 capacity_reservation_id: EC2 Capacity Reservation ID (``cr-...``).
85 instance_type: Instance types the NodePool may provision (one per entry).
86 max_nodes: Maximum nodes in the pool.
87 fallback_on_demand: Fall back to on-demand when the ODCR is exhausted.
88 efa: Enable EFA support (adds EFA taint and labels).
89 cluster: EKS cluster name (defaults to ``gco-<region>``).
90 """
91 args = [
92 "nodepools",
93 "create-odcr",
94 "-n",
95 name,
96 "-r",
97 region,
98 "-c",
99 capacity_reservation_id,
100 "--max-nodes",
101 str(max_nodes),
102 ]
103 for it in instance_type or []:
104 args += ["-i", it]
105 if fallback_on_demand:
106 args.append("--fallback-on-demand")
107 if efa:
108 args.append("--efa")
109 if cluster:
110 args += ["--cluster", cluster]
111 return await asyncio.to_thread(cli_runner._run_cli, *args)
114@mcp.tool(tags={"low-risk", "nodepools"})
115@audit_logged
116async def nodepools_create_capacity_block(
117 name: str,
118 region: str,
119 capacity_reservation_id: str,
120 instance_type: list[str] | None = None,
121 max_nodes: int = 100,
122 fallback_on_demand: bool = False,
123 efa: bool = False,
124 cluster: str | None = None,
125) -> str:
126 """`gco nodepools create-capacity-block` — NodePool manifest for a Capacity Block.
128 The Capacity Block counterpart to ``nodepools_create_odcr``. Purchasing a
129 Capacity Block yields an EC2 Capacity Reservation id (``cr-...``), which the
130 returned EKS Auto Mode NodeClass binds through
131 ``capacityReservationSelectorTerms``. Because a block is prepaid for a
132 fixed term, the NodePool holds the capacity rather than consolidating it
133 early. Nothing is applied: apply the returned YAML with kubectl.
135 Args:
136 name: NodePool name.
137 region: AWS region.
138 capacity_reservation_id: Capacity Reservation ID (``cr-...``) of the block.
139 instance_type: Instance types the NodePool may provision (one per entry).
140 max_nodes: Maximum nodes in the pool.
141 fallback_on_demand: Fall back to on-demand when the block is exhausted/expired.
142 efa: Enable EFA support (adds EFA taint and labels).
143 cluster: EKS cluster name (defaults to ``gco-<region>``).
144 """
145 args = [
146 "nodepools",
147 "create-capacity-block",
148 "-n",
149 name,
150 "-r",
151 region,
152 "-c",
153 capacity_reservation_id,
154 "--max-nodes",
155 str(max_nodes),
156 ]
157 for it in instance_type or []:
158 args += ["-i", it]
159 if fallback_on_demand:
160 args.append("--fallback-on-demand")
161 if efa:
162 args.append("--efa")
163 if cluster:
164 args += ["--cluster", cluster]
165 return await asyncio.to_thread(cli_runner._run_cli, *args)
168# =============================================================================
169# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS
170# =============================================================================
173import contextlib # noqa: E402
175from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402
178async def _ctx_warning(message: str) -> None:
179 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context."""
180 try:
181 from fastmcp.server.dependencies import get_context
183 ctx = get_context()
184 except Exception:
185 return
186 with contextlib.suppress(Exception):
187 await ctx.warning(message)
190if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS):
192 @mcp.tool(tags={"destructive", "nodepools"})
193 @audit_logged
194 async def delete_nodepool(nodepool_name: str, region: str, cluster: str | None = None) -> str:
195 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
197 `gco nodepools delete` — delete a Karpenter NodePool.
198 Cannot be undone — the NodePool, its paired ``<name>-nodeclass``
199 NodeClass, and any nodes currently provisioned through it are removed.
201 Args:
202 nodepool_name: NodePool name.
203 region: AWS region.
204 cluster: EKS cluster name (defaults to ``gco-<region>``).
205 """
206 await _ctx_warning(
207 f"Deleting NodePool {nodepool_name!r} in {region} — this cannot be undone."
208 )
209 args = ["nodepools", "delete", nodepool_name, "-r", region, "-y"]
210 if cluster:
211 args += ["--cluster", cluster]
212 return await asyncio.to_thread(cli_runner._run_cli, *args)