Coverage for gco_mcp/tools/costs.py: 90.65%
87 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""Cost tracking MCP tools."""
3import cli_runner
4from audit import audit_logged
5from server import mcp
8@mcp.tool(tags={"safe", "costs"})
9@audit_logged
10def cost_summary(days: int = 30) -> str:
11 """Get total GCO spend broken down by AWS service.
13 Args:
14 days: Number of days to look back.
15 """
16 return cli_runner._run_cli("costs", "summary", "--days", str(days))
19@mcp.tool(tags={"safe", "costs"})
20@audit_logged
21def cost_by_region(days: int = 30) -> str:
22 """Get cost breakdown by AWS region.
24 Args:
25 days: Number of days to look back.
26 """
27 return cli_runner._run_cli("costs", "regions", "--days", str(days))
30@mcp.tool(tags={"safe", "costs"})
31@audit_logged
32def cost_trend(days: int = 14) -> str:
33 """Get daily cost trend.
35 Args:
36 days: Number of days to show.
37 """
38 return cli_runner._run_cli("costs", "trend", "--days", str(days))
41@mcp.tool(tags={"safe", "costs"})
42@audit_logged
43def cost_workloads(region: str | None = None) -> str:
44 """Estimate accumulated and hourly cost for running workloads.
46 Args:
47 region: Region to inspect, or omit to inspect every deployment region.
48 """
49 args = ["costs", "workloads"]
50 if region: 50 ↛ 52line 50 didn't jump to line 52 because the condition on line 50 was always true
51 args += ["-r", region]
52 return cli_runner._run_cli(*args)
55@mcp.tool(tags={"safe", "costs"})
56@audit_logged
57def cost_forecast(days_ahead: int = 30) -> str:
58 """Forecast GCO costs for the next N days.
60 Args:
61 days_ahead: Days to forecast ahead.
62 """
63 return cli_runner._run_cli("costs", "forecast", "--days", str(days_ahead))
66@mcp.tool(tags={"safe", "costs"})
67@audit_logged
68def cost_allocation_status(extra_tags: list[str] | None = None) -> str:
69 """Show cost allocation tag activation status for GCO's billing keys.
71 Reports whether the Project tag (which every cost query filters on) and
72 the AWS-generated aws:eks:cluster-name tag (per-cluster attribution for
73 EKS Auto Mode compute) are Active, Inactive, or not yet discovered by
74 Billing, plus backfill history and split-cost-allocation-data guidance.
76 Args:
77 extra_tags: Additional tag keys to check.
78 """
79 args = ["costs", "allocation", "status"]
80 for tag in extra_tags or []:
81 args += ["-t", tag]
82 return cli_runner._run_cli(*args)
85@mcp.tool(tags={"low-risk", "costs"})
86@audit_logged
87def cost_allocation_activate(
88 extra_tags: list[str] | None = None,
89 backfill_from: str | None = None,
90) -> str:
91 """Activate GCO's cost allocation tag keys in the billing account.
93 Activates the Project and aws:eks:cluster-name tag keys (plus any
94 extras) so Cost Explorer can attribute spend to them. Reversible in
95 the Billing console; in an AWS Organization this requires the
96 management (payer) account. Only affects billing data from activation
97 onward unless a backfill date is given.
99 Args:
100 extra_tags: Additional tag keys to activate.
101 backfill_from: Optionally re-tag historical usage from this date
102 (YYYY-MM-DD, up to 12 months back).
103 """
104 args = ["costs", "allocation", "activate", "-y"]
105 for tag in extra_tags or []: 105 ↛ 106line 105 didn't jump to line 106 because the loop on line 105 never started
106 args += ["-t", tag]
107 if backfill_from:
108 args += ["--backfill-from", backfill_from]
109 return cli_runner._run_cli(*args)
112@mcp.tool(tags={"safe", "costs"})
113@audit_logged
114def cost_k8s_namespaces(days: int = 7, region: str | None = None) -> str:
115 """Show Kubernetes cost by namespace (Athena over OpenCost reports).
117 Args:
118 days: Days to look back.
119 region: Restrict to one deployment region, or omit for all.
120 """
121 args = ["costs", "k8s", "namespaces", "--days", str(days)]
122 if region: 122 ↛ 124line 122 didn't jump to line 124 because the condition on line 122 was always true
123 args += ["-r", region]
124 return cli_runner._run_cli(*args)
127@mcp.tool(tags={"safe", "costs"})
128@audit_logged
129def cost_k8s_regions(days: int = 7) -> str:
130 """Show Kubernetes allocation cost by deployment region.
132 Args:
133 days: Days to look back.
134 """
135 return cli_runner._run_cli("costs", "k8s", "regions", "--days", str(days))
138@mcp.tool(tags={"safe", "costs"})
139@audit_logged
140def cost_k8s_trend(
141 days: int = 14,
142 granularity: str = "daily",
143 namespace: str | None = None,
144) -> str:
145 """Show Kubernetes cost over time.
147 Args:
148 days: Days to look back.
149 granularity: Trend bucket size ("daily" or "hourly").
150 namespace: Restrict to one namespace, or omit for all.
151 """
152 args = ["costs", "k8s", "trend", "--days", str(days), "--granularity", granularity]
153 if namespace: 153 ↛ 155line 153 didn't jump to line 155 because the condition on line 153 was always true
154 args += ["-n", namespace]
155 return cli_runner._run_cli(*args)
158@mcp.tool(tags={"safe", "costs"})
159@audit_logged
160def cost_k8s_top(limit: int = 10, by: str = "namespace", days: int = 7) -> str:
161 """Show the top-N Kubernetes spenders by namespace, region, or cluster.
163 Args:
164 limit: Number of results.
165 by: Grouping dimension ("namespace", "region", or "cluster").
166 days: Days to look back.
167 """
168 return cli_runner._run_cli(
169 "costs", "k8s", "top", "-n", str(limit), "--by", by, "--days", str(days)
170 )
173@mcp.tool(tags={"safe", "costs"})
174@audit_logged
175def cost_report_status(region: str | None = None) -> str:
176 """Show cost monitoring health, including OpenCost status.
178 Args:
179 region: Region whose cost monitor to check, or omit for the
180 nearest healthy region via the global API.
181 """
182 args = ["costs", "report", "status"]
183 if region: 183 ↛ 185line 183 didn't jump to line 185 because the condition on line 183 was always true
184 args += ["-r", region]
185 return cli_runner._run_cli(*args)
188@mcp.tool(tags={"safe", "costs"})
189@audit_logged
190def cost_report_list(region: str | None = None, adhoc: bool = False, limit: int = 20) -> str:
191 """List recent OpenCost allocation report objects in S3.
193 Args:
194 region: Region whose reports to list.
195 adhoc: List ad-hoc instead of scheduled reports.
196 limit: Maximum results.
197 """
198 args = ["costs", "report", "list", "-l", str(limit)]
199 if region: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 args += ["-r", region]
201 if adhoc: 201 ↛ 203line 201 didn't jump to line 203 because the condition on line 201 was always true
202 args += ["--adhoc"]
203 return cli_runner._run_cli(*args)
206@mcp.tool(tags={"low-risk", "costs"})
207@audit_logged
208def cost_report_generate(region: str | None = None, window_hours: int = 24) -> str:
209 """Generate an ad-hoc OpenCost allocation report now (written to S3).
211 Args:
212 region: Region whose cost monitor generates the report.
213 window_hours: Trailing window the report covers (1-168).
214 """
215 args = ["costs", "report", "generate", "--window-hours", str(window_hours)]
216 if region: 216 ↛ 218line 216 didn't jump to line 218 because the condition on line 216 was always true
217 args += ["-r", region]
218 return cli_runner._run_cli(*args)