AWS MCP Audit

Implementation Reference

• src/aws_mcp_audit/server.py:55-127 (handler)

  The @mcp.tool decorated handler function that implements the collect_snapshot tool. It collects AWS resource data (EC2, ELBv2, RDS, S3, CloudTrail, CloudWatch) across specified or all enabled regions, computes summaries, saves the snapshot, and returns snapshot_id, regions, and summary.

  @mcp.tool
  def collect_snapshot(scope: Dict[str, Any], auth: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
      session = build_boto3_session(auth)
      sid = _snapshot_id()
  
      who = aws_whoami(auth=auth)
      requested = scope.get("regions")
      if scope.get("all") is True or requested is None:
          regions = list_enabled_regions(session)
      else:
          regions = list(requested)
  
      ec2_by_region: Dict[str, Any] = {}
      elbv2_by_region: Dict[str, Any] = {}
      rds_by_region: Dict[str, Any] = {}
      telemetry_cloudwatch: List[Dict[str, Any]] = []
  
      # collectors (MVP: sequential; add concurrency later)
      for r in regions:
          ec2_by_region[r] = collect_ec2_region(session, r)
          elbv2_by_region[r] = collect_elbv2_region(session, r)
          rds_by_region[r] = collect_rds_region(session, r)
          telemetry_cloudwatch.append(collect_cloudwatch_alarm_count(session, r))
  
      # global-ish collectors
      s3_blob = collect_s3(session)
      # CloudTrail: use caller region or us-east-1; trails are global objects but API is regional
      ct_region = session.region_name or "us-east-1"
      cloudtrail_blob = collect_cloudtrail(session, ct_region)
  
      total_alarm_count = 0
      alarm_count_known = True
      for row in telemetry_cloudwatch:
          if row.get("alarm_count") is None:
              alarm_count_known = False
          else:
              total_alarm_count += int(row.get("alarm_count"))
  
      summary = {
          "ec2_instances": sum(len(v.get("instances", [])) for v in ec2_by_region.values()),
          "security_groups": sum(len(v.get("security_groups", [])) for v in ec2_by_region.values()),
          "ebs_volumes": sum(len(v.get("volumes", [])) for v in ec2_by_region.values()),
          "elastic_ips": sum(len(v.get("eips", [])) for v in ec2_by_region.values()),
          "elbv2_load_balancers": sum(len(v.get("load_balancers", [])) for v in elbv2_by_region.values()),
          "elbv2_target_groups": sum(len(v.get("target_groups", [])) for v in elbv2_by_region.values()),
          "rds_instances": sum(len(v.get("instances", [])) for v in rds_by_region.values()),
          "rds_clusters": sum(len(v.get("clusters", [])) for v in rds_by_region.values()),
          "s3_buckets": len(s3_blob.get("buckets", [])),
      }
  
      snapshot: Dict[str, Any] = {
          "meta": {
              "snapshot_id": sid,
              "account_id": who.get("account"),
              "collected_at": now_iso_utc(),
              "regions": regions,
          },
          "summary": summary,
          "ec2_by_region": ec2_by_region,
          "elbv2_by_region": elbv2_by_region,
          "rds_by_region": rds_by_region,
          "s3": s3_blob,
          "telemetry": {
              "cloudtrail": cloudtrail_blob,
              "cloudwatch_alarms": {
                  "by_region": telemetry_cloudwatch,
                  "total_alarm_count": total_alarm_count if alarm_count_known else None,
              },
          },
      }
  
      save_snapshot(DATA_DIR, sid, snapshot)
      return {"snapshot_id": sid, "regions": regions, "summary": summary}

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

  The @mcp.tool decorator registers the collect_snapshot function as an MCP tool.

  @mcp.tool