AWS MCP Audit

Implementation Reference

• src/aws_mcp_audit/server.py:222-225 (handler)

  The main handler function for the 'cost_explorer_by_region' tool, decorated with @mcp.tool for automatic registration in FastMCP. It builds a Boto3 session and calls the supporting cost_explorer_by_dimension function with dimension set to 'REGION'.
  @mcp.tool def cost_explorer_by_region(days: int = 30, top_n: int = 10, auth: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: session = build_boto3_session(auth) return cost_explorer_by_dimension(session, days=days, dimension="REGION", top_n=top_n)
• src/aws_mcp_audit/server.py:222-225 (handler)

  The main handler function for the MCP tool 'cost_explorer_by_region'. It is registered via @mcp.tool decorator, builds a boto3 session from auth, and delegates to the helper function with dimension set to 'REGION'.
  @mcp.tool def cost_explorer_by_region(days: int = 30, top_n: int = 10, auth: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: session = build_boto3_session(auth) return cost_explorer_by_dimension(session, days=days, dimension="REGION", top_n=top_n)
• src/aws_mcp_audit/collectors/cost_explorer.py:29-58 (helper)

  Supporting function that performs the actual AWS Cost Explorer API call (ce.get_cost_and_usage) grouped by the given dimension ('REGION' for this tool), processes results to top N by UnblendedCost, handles exceptions.
  def cost_explorer_by_dimension(session: boto3.Session, days: int, dimension: str, top_n: int = 10) -> Dict[str, Any]: ce = session.client("ce", region_name="us-east-1") rng = _dates(days) try: resp = ce.get_cost_and_usage( TimePeriod=rng, Granularity="MONTHLY" if days >= 28 else "DAILY", Metrics=["UnblendedCost"], GroupBy=[{"Type": "DIMENSION", "Key": dimension}], ) groups = [] for t in resp.get("ResultsByTime", []): for g in t.get("Groups", []): groups.append( { "keys": g.get("Keys", []), "amount": (g.get("Metrics", {}).get("UnblendedCost", {}) or {}).get("Amount"), "unit": (g.get("Metrics", {}).get("UnblendedCost", {}) or {}).get("Unit"), } ) # naive top_n by amount def amt(x): try: return float(x.get("amount") or 0.0) except Exception: return 0.0 groups.sort(key=amt, reverse=True) return {"time_period": rng, "dimension": dimension, "top": groups[:top_n]} except ClientError as e: return {"time_period": rng, "dimension": dimension, "error": str(e), "top": []}
• src/aws_mcp_audit/collectors/cost_explorer.py:29-58 (helper)

  Core helper function that performs the AWS Cost Explorer API call grouped by the specified dimension (e.g., 'REGION' for the tool), processes the response to extract top costs, and handles errors.
  def cost_explorer_by_dimension(session: boto3.Session, days: int, dimension: str, top_n: int = 10) -> Dict[str, Any]: ce = session.client("ce", region_name="us-east-1") rng = _dates(days) try: resp = ce.get_cost_and_usage( TimePeriod=rng, Granularity="MONTHLY" if days >= 28 else "DAILY", Metrics=["UnblendedCost"], GroupBy=[{"Type": "DIMENSION", "Key": dimension}], ) groups = [] for t in resp.get("ResultsByTime", []): for g in t.get("Groups", []): groups.append( { "keys": g.get("Keys", []), "amount": (g.get("Metrics", {}).get("UnblendedCost", {}) or {}).get("Amount"), "unit": (g.get("Metrics", {}).get("UnblendedCost", {}) or {}).get("Unit"), } ) # naive top_n by amount def amt(x): try: return float(x.get("amount") or 0.0) except Exception: return 0.0 groups.sort(key=amt, reverse=True) return {"time_period": rng, "dimension": dimension, "top": groups[:top_n]} except ClientError as e: return {"time_period": rng, "dimension": dimension, "error": str(e), "top": []}
• src/aws_mcp_audit/collectors/cost_explorer.py:9-12 (helper)

  Helper utility to compute Start/End dates for Cost Explorer queries based on days back from today.
  def _dates(days: int) -> Dict[str, str]: end = datetime.date.today() start = end - datetime.timedelta(days=days) return {"Start": start.isoformat(), "End": end.isoformat()}