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