AWS MCP Audit

Implementation Reference

• src/aws_mcp_audit/server.py:228-261 (handler)

  Core handler function for the 'generate_report' MCP tool. Decorated with @mcp.tool for registration. Generates a markdown or PDF report by loading snapshot data, findings, cost analysis, optionally fetching live cost explorer data, rendering via helper, and saving files.

  @mcp.tool
  def generate_report(snapshot_id: str, finding_set_id: Optional[str] = None, format: str = "md", auth: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      s_path = os.path.join(snapshot_dir(DATA_DIR, snapshot_id), "snapshot.json")
      snap = read_json(s_path)
  
      fsid = finding_set_id or snapshot_id
      f_path = os.path.join(snapshot_dir(DATA_DIR, fsid), "findings.json")
      findings = read_json(f_path) if os.path.exists(f_path) else []
  
      cost_path = os.path.join(snapshot_dir(DATA_DIR, snapshot_id), "cost.json")
      cost = read_json(cost_path) if os.path.exists(cost_path) else {}
  
      # Optionally include cost explorer in report (best-effort)
      cost_ce = {}
      try:
          if auth is not None:
              session = build_boto3_session(auth)
              cost_ce = cost_explorer_summary(session, days=30, granularity="DAILY")
      except Exception as e:
          cost_ce = {"error": str(e), "results": []}
  
      md = render_markdown(snap, findings, cost, cost_ce)
      out_dir = snapshot_dir(DATA_DIR, snapshot_id)
  
      md_path = os.path.join(out_dir, "report.md")
      with open(md_path, "w", encoding="utf-8") as f:
          f.write(md)
  
      if format.lower() == "pdf":
          pdf_path = os.path.join(out_dir, "report.pdf")
          write_pdf_from_text(pdf_path=pdf_path, title="AWS Audit Snapshot", text=md)
          return {"report_md": md_path, "report_pdf": pdf_path}
  
      return {"report_md": md_path}

• src/aws_mcp_audit/reports/render.py:10-55 (helper)

  Key helper function that formats the report data into a markdown string, including executive summary, top findings, cost snapshot, and inventory summary.

  def render_markdown(snapshot: Dict[str, Any], findings: List[Dict[str, Any]], cost: Dict[str, Any], cost_explorer: Dict[str, Any]) -> str:
      meta = snapshot.get("meta", {})
      lines: List[str] = []
  
      lines.append(f"# AWS Audit Snapshot — {meta.get('account_id', 'unknown')}")
      lines.append("")
      lines.append(f"- Snapshot ID: `{meta.get('snapshot_id')}`")
      lines.append(f"- Collected: `{meta.get('collected_at')}`")
      lines.append(f"- Regions: {', '.join(meta.get('regions', []))}")
      lines.append("")
  
      lines.append("## Executive Summary")
      lines.append(f"- Findings: **{len(findings)}**")
      lines.append("")
  
      lines.append("## Top Findings")
      if not findings:
          lines.append("- No findings generated.")
      else:
          # show first 15
          for f in findings[:15]:
              lines.append(f"- **{f.get('severity')}** — {f.get('title')} ({f.get('region') or 'global'})")
      lines.append("")
  
      lines.append("## Cost Snapshot")
      if cost:
          lines.append("### Tier 1 (inventory-derived signals)")
          lines.append(f"- Unattached EBS (GB): {cost.get('unattached_ebs_gb')}")
          lines.append(f"- Unassociated EIPs: {cost.get('unassociated_eips')}")
          lines.append("")
      if cost_explorer and cost_explorer.get("results"):
          lines.append("### Tier 2 (Cost Explorer)")
          lines.append(f"- Range: {cost_explorer.get('time_period')}")
          lines.append(f"- Points: {len(cost_explorer.get('results', []))}")
          lines.append("")
      elif cost_explorer and cost_explorer.get("error"):
          lines.append("### Tier 2 (Cost Explorer)")
          lines.append(f"- Not available: {cost_explorer.get('error')}")
          lines.append("")
  
      lines.append("## Inventory Summary")
      summary = snapshot.get("summary", {})
      for k, v in summary.items():
          lines.append(f"- {k}: {v}")
      lines.append("")
      return "\n".join(lines)

• src/aws_mcp_audit/reports/render.py:58-72 (helper)

  Supporting helper to convert markdown text to PDF using reportlab library, used when format='pdf'.

  def write_pdf_from_text(pdf_path: str, title: str, text: str) -> None:
      os.makedirs(os.path.dirname(pdf_path), exist_ok=True)
      c = canvas.Canvas(pdf_path, pagesize=LETTER)
      width, height = LETTER
      y = height - 72
      c.setTitle(title)
  
      for raw in text.splitlines():
          if y < 72:
              c.showPage()
              y = height - 72
          c.drawString(72, y, raw[:1200])
          y -= 14
  
      c.save()

• src/aws_mcp_audit/server.py:228-228 (registration)

  MCP tool registration decorator applied to the generate_report handler function.

  @mcp.tool