AWS CloudTrail for MCP Servers: Three Patterns for Audit Logging

Why CloudTrail for MCP is different: the audit surface table

An MCP server's CloudTrail audit surface is dominated by two things that conventional web applications rarely encounter at the same scale: high-frequency data-plane calls on DynamoDB (session state, tool call records, context storage) and a rich set of management-plane API calls on IAM, S3, and STS that the server's role should never make. Both require different CloudTrail configurations, and conflating them is the root cause of most CloudTrail problems in MCP deployments.

Thread 1: The data events billing trap — and three layers of cost control

The billing trap is predictable but catches nearly every team. A developer reads the documentation that says "add data event selectors to your trail" and adds a blanket selector for their DynamoDB tables without specifying readOnly: false. The trail starts capturing every GetItem (session lookup at tool call start), every Query (context retrieval), every Scan (admin tooling), plus all the writes. For a modestly busy MCP server at 50 tool calls per minute with 5 reads and 10 writes each, the read volume alone is 21.6 million events per month — $21.60/month just from reads that add no compliance value beyond what the writes already provide.

The cost model: management events on the first trail in each region are free. Data events cost $0.10 per 100,000 events on every trail, always. With write-only selectors, that same 50 tool-calls/minute server generates 10 writes per call = 500 writes/minute × 60 × 24 × 30 = 21.6 million write events/month = $21.60/month. Add reads and it doubles. Add S3 object operations and it climbs further. The lever that prevents the trap is a single field selector value.

Layer 1 — Write-only selector

CloudTrail sets readOnly: true for non-mutating operations (GetItem, Query, Scan, GetObject, HeadObject) and readOnly: false for mutations (PutItem, UpdateItem, DeleteItem, BatchWriteItem, TransactWriteItems, PutObject, DeleteObject). Adding {"Field": "readOnly", "Equals": ["false"]} to every data event selector halves or quarters the bill immediately.

# CLI: write-only DynamoDB data events for specific tables
aws cloudtrail put-event-selectors \
  --trail-name mcp-audit-trail \
  --advanced-event-selectors '[
    {
      "Name": "DynamoDB-WriteOnly-SpecificTables",
      "FieldSelectors": [
        {"Field": "eventCategory", "Equals": ["Data"]},
        {"Field": "resources.type", "Equals": ["AWS::DynamoDB::Table"]},
        {"Field": "resources.ARN", "StartsWith": [
          "arn:aws:dynamodb:us-east-1:123456789012:table/mcp-sessions",
          "arn:aws:dynamodb:us-east-1:123456789012:table/mcp-toolcalls"
        ]},
        {"Field": "readOnly", "Equals": ["false"]}
      ]
    }
  ]'

Layer 2 — ARN prefix scoping

A selector without resources.ARN constraints captures data events for all tables in the account — including tables you didn't intend to monitor. For organizations running many MCP tenants with separate DynamoDB tables, use an ARN prefix (StartsWith: ["arn:aws:dynamodb:us-east-1:123456789012:table/mcp-"]) rather than enumerating individual tables. You can have at most 500 advanced event selectors per trail, each with at most 500 field selector values, so a wildcard prefix scales cleanly.

Layer 3 — Storage tier selection via lifecycle rules

Data events produce far more log volume than management events — potentially 50-500 GB/month for an active MCP server. Without lifecycle rules, S3 costs compound. The key insight for audit log lifecycle design is that Glacier Instant Retrieval is transparent to Athena — Athena can query objects in GIR without a restore step, making it the optimal storage tier for 1-3 year old logs that are rarely queried but must remain queryable. Do not use Glacier Flexible Retrieval or Deep Archive for logs that need ad-hoc querying; those require multi-hour restore jobs.

// CDK: complete data event trail with write-only selectors and lifecycle management
import * as cdk from "aws-cdk-lib";
import * as s3 from "aws-cdk-lib/aws-s3";
import * as cloudtrail from "aws-cdk-lib/aws-cloudtrail";
import * as iam from "aws-cdk-lib/aws-iam";
import { aws_cloudtrail as cfnTrail } from "aws-cdk-lib";

export class McpAuditTrailStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string) {
    super(scope, id);

    const auditBucket = new s3.Bucket(this, "AuditLogs", {
      bucketName: `mcp-cloudtrail-audit-${this.account}`,
      encryption: s3.BucketEncryption.S3_MANAGED,
      versioned: true,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      lifecycleRules: [
        {
          id: "log-retention",
          prefix: "AWSLogs/",
          enabled: true,
          transitions: [
            // S3-IA at 90 days — 60% cost reduction, Athena queries unaffected
            { storageClass: s3.StorageClass.INFREQUENT_ACCESS, transitionAfter: cdk.Duration.days(90) },
            // Glacier Instant Retrieval at 1yr — 68% vs S3-IA, Athena still works without restore
            { storageClass: s3.StorageClass.GLACIER_INSTANT_RETRIEVAL, transitionAfter: cdk.Duration.days(365) },
          ],
          expiration: cdk.Duration.days(2555), // 7yr compliance retention
          // Versioned bucket requires these to avoid zombie delete markers
          noncurrentVersionExpiration: cdk.Duration.days(30),
          expiredObjectDeleteMarker: true,
        },
      ],
    });

    // Both statements required — missing GetBucketAcl causes silent delivery failure
    auditBucket.addToResourcePolicy(new iam.PolicyStatement({
      principals: [new iam.ServicePrincipal("cloudtrail.amazonaws.com")],
      actions: ["s3:GetBucketAcl"],
      resources: [auditBucket.bucketArn],
      conditions: {
        StringEquals: { "AWS:SourceArn": `arn:aws:cloudtrail:${this.region}:${this.account}:trail/mcp-audit-trail` },
      },
    }));
    auditBucket.addToResourcePolicy(new iam.PolicyStatement({
      principals: [new iam.ServicePrincipal("cloudtrail.amazonaws.com")],
      actions: ["s3:PutObject"],
      resources: [`${auditBucket.bucketArn}/AWSLogs/${this.account}/*`],
      conditions: {
        StringEquals: {
          "s3:x-amz-acl": "bucket-owner-full-control",
          "AWS:SourceArn": `arn:aws:cloudtrail:${this.region}:${this.account}:trail/mcp-audit-trail`,
        },
      },
    }));

    const trail = new cloudtrail.Trail(this, "McpAuditTrail", {
      trailName: "mcp-audit-trail",
      bucket: auditBucket,
      isMultiRegionTrail: true,
      includeGlobalServiceEvents: true,
      enableFileValidation: true,
      managementEvents: cloudtrail.ReadWriteType.WRITE_ONLY,
    });

    // DynamoDB data events — CDK Trail L2 does not expose a DynamoDB selector method
    // Must use L1 escape hatch after trail construction
    const cfn = trail.node.defaultChild as cfnTrail.CfnTrail;
    cfn.addPropertyOverride("AdvancedEventSelectors", [
      {
        Name: "DynamoDB-WriteOnly",
        FieldSelectors: [
          { Field: "eventCategory", Equals: ["Data"] },
          { Field: "resources.type", Equals: ["AWS::DynamoDB::Table"] },
          { Field: "resources.ARN", StartsWith: [
            `arn:aws:dynamodb:${this.region}:${this.account}:table/mcp-sessions`,
            `arn:aws:dynamodb:${this.region}:${this.account}:table/mcp-toolcalls`,
          ]},
          { Field: "readOnly", Equals: ["false"] },
        ],
      },
      {
        Name: "S3-WriteOnly",
        FieldSelectors: [
          { Field: "eventCategory", Equals: ["Data"] },
          { Field: "resources.type", Equals: ["AWS::S3::Object"] },
          { Field: "resources.ARN", StartsWith: [
            `arn:aws:s3:::mcp-tool-outputs/`,
          ]},
          { Field: "readOnly", Equals: ["false"] },
        ],
      },
    ]);
  }
}

Correlating data events to specific tool calls

CloudTrail data events capture the IAM principal and source IP but not application-level context like the MCP session ID or tool call ID. Two approaches work for correlation without STS overhead. The cheaper approach: embed the tool call ID in the DynamoDB partition key prefix so CloudTrail's logged requestParameters.key reveals the correlation directly.

// Table schema: pk = "toolcall#", sk = "output#"
// CloudTrail logs requestParameters.key.pk = "toolcall#tc_abc123"
// No STS AssumeRole needed — the key itself is the correlation handle.

import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
import { marshall } from "@aws-sdk/util-dynamodb";

const ddb = new DynamoDBClient({});

export async function recordToolCallOutput(
  toolCallId: string,
  sessionId: string,
  toolName: string,
  outputSummary: string
) {
  await ddb.send(new PutItemCommand({
    TableName: "mcp-toolcalls",
    Item: marshall({
      pk: `toolcall#${toolCallId}`,          // surfaced in CloudTrail key log
      sk: `output#${new Date().toISOString()}`,
      session_id: sessionId,
      tool_name: toolName,
      output_summary: outputSummary,         // NOT logged — item values are redacted
      ttl: Math.floor(Date.now() / 1000) + 86400 * 90,
    }),
    ConditionExpression: "attribute_not_exists(pk)",
  }));
}

Important: CloudTrail logs DynamoDB key attributes only — attribute values in requestParameters.item are omitted for privacy. The key (pk, sk) appears in the record; the item body does not. For full item-level audit with attribute values, use DynamoDB Streams with a Lambda writing to CloudWatch Logs or Kinesis — CloudTrail alone cannot provide it.

Lake vs S3+Athena: the storage and query tradeoff

Both CloudTrail Lake and S3+Athena support SQL queries over event history. The choice matters for total cost and operational overhead:

Decision rule: if you are starting a new deployment and want zero infrastructure, use Lake. If you need to JOIN CloudTrail events with application tables, need retention beyond 7 years, or already have an S3 trail for other purposes, use S3+Athena. For new deployments that specifically need data event anomaly detection via scheduled SQL, Lake is the better choice — the scheduled query capability is native and the SQL syntax is simpler than configuring Athena.

Thread 2: The EventBridge deny-list pattern for zero-configuration security automation

CloudTrail management events flow to the EventBridge default event bus in each region automatically, at no extra cost, without any trail configuration change. This integration exists regardless of whether you have a trail — the trail only adds persistence (S3) and Insights; for real-time EventBridge consumption you do not need a trail at all. The correct mental model: every IAM, STS, S3, EC2, and Lambda management API call your MCP server makes shows up in EventBridge within seconds.

The deny-list philosophy: instead of maintaining an allow-list of permitted API calls (which becomes stale as the codebase evolves), enumerate the API calls your MCP server role should never make and alert on any occurrence. An MCP server that does DynamoDB reads and writes for session state and S3 object writes for tool output should never call AttachRolePolicy, CreateAccessKey, PutBucketPolicy, AssumeRole to arbitrary targets, or log in to the console. Any occurrence of these calls from the MCP role's principal indicates either a security incident or a serious misconfiguration.

The envelope structure: why every first implementation is wrong

The most common failure is writing event patterns that reference top-level CloudTrail fields. The EventBridge event is an envelope that wraps the CloudTrail record under a detail key:

{
  "source": "aws.cloudtrail",
  "detail-type": "AWS API Call via CloudTrail",
  // detail is the full CloudTrail record — NOT top-level
  "detail": {
    "eventName": "AttachRolePolicy",          // detail.eventName — correct path
    "eventSource": "iam.amazonaws.com",        // detail.eventSource
    "userIdentity": {
      "principalId": "AROAEXAMPLEID:mcp-server-task",
      "arn": "arn:aws:sts::123456789012:assumed-role/mcp-server-task-role/mcp-server-task"
    },
    "requestParameters": {
      "roleName": "mcp-server-task-role",
      "policyArn": "arn:aws:iam::aws:policy/AdministratorAccess"
    }
  }
}

// WRONG pattern — matches nothing
{ "eventName": ["AttachRolePolicy"] }

// CORRECT pattern — must use detail. prefix for all CloudTrail fields
{
  "source": ["aws.cloudtrail"],
  "detail-type": ["AWS API Call via CloudTrail"],
  "detail": {
    "eventName": ["AttachRolePolicy"],
    "userIdentity": {
      "sessionContext": {
        "sessionIssuer": { "arn": [{ "suffix": "mcp-server-task-role" }] }
      }
    }
  }
}

CDK deny-list rules

Build four EventBridge rules covering the deny-list categories most likely to indicate MCP server compromise or misconfiguration:

import * as events from "aws-cdk-lib/aws-events";
import * as targets from "aws-cdk-lib/aws-events-targets";
import * as sns from "aws-cdk-lib/aws-sns";

const securityAlertTopic = new sns.Topic(this, "SecurityAlerts", {
  topicName: "mcp-security-alerts",
});

// Rule 1: IAM privilege escalation — MCP server should never modify IAM
new events.Rule(this, "IamEscalationRule", {
  ruleName: "mcp-iam-escalation-detect",
  eventPattern: {
    source: ["aws.cloudtrail"],
    detailType: ["AWS API Call via CloudTrail"],
    detail: {
      eventSource: ["iam.amazonaws.com"],
      eventName: [
        "AttachRolePolicy", "DetachRolePolicy", "PutRolePolicy", "DeleteRolePolicy",
        "CreateRole", "DeleteRole", "CreateAccessKey", "DeleteAccessKey",
        "UpdateAccessKey", "CreatePolicy", "CreatePolicyVersion", "SetDefaultPolicyVersion",
      ],
      // Filter to successful calls only — AccessDenied spikes are separate alert
      errorCode: [{ "anything-but": ["AccessDenied", "AccessDeniedException", "UnauthorizedOperation"] }],
      userIdentity: {
        sessionContext: {
          sessionIssuer: { arn: [{ suffix: "mcp-server-task-role" }] },
        },
      },
    },
  },
  targets: [new targets.SnsTopic(securityAlertTopic)],
});

// Rule 2: S3 bucket policy modification
new events.Rule(this, "S3PolicyModRule", {
  ruleName: "mcp-s3-policy-modification",
  eventPattern: {
    source: ["aws.cloudtrail"],
    detailType: ["AWS API Call via CloudTrail"],
    detail: {
      eventSource: ["s3.amazonaws.com"],
      eventName: [
        "PutBucketPolicy", "DeleteBucketPolicy",
        "PutBucketAcl", "PutBucketPublicAccessBlock", "DeletePublicAccessBlock",
      ],
      userIdentity: {
        sessionContext: {
          sessionIssuer: { arn: [{ suffix: "mcp-server-task-role" }] },
        },
      },
    },
  },
  targets: [new targets.SnsTopic(securityAlertTopic)],
});

// Rule 3: AssumeRole to any target (MCP server should not be vending credentials)
new events.Rule(this, "AssumeRoleAuditRule", {
  ruleName: "mcp-assume-role-audit",
  eventPattern: {
    source: ["aws.cloudtrail"],
    detailType: ["AWS API Call via CloudTrail"],
    detail: {
      eventSource: ["sts.amazonaws.com"],
      eventName: ["AssumeRole"],
      userIdentity: {
        sessionContext: {
          sessionIssuer: { arn: [{ suffix: "mcp-server-task-role" }] },
        },
      },
    },
  },
  targets: [new targets.SnsTopic(securityAlertTopic)],
});

// Rule 4: ConsoleLogin — task roles should never sign in to the console
new events.Rule(this, "ConsoleLoginRule", {
  ruleName: "mcp-console-login-detect",
  eventPattern: {
    source: ["aws.cloudtrail"],
    detailType: ["AWS Console Sign In via CloudTrail"],
    detail: {
      userIdentity: { arn: [{ suffix: "mcp-server-task-role" }] },
    },
  },
  targets: [new targets.SnsTopic(securityAlertTopic)],
});

Lambda enrichment handler

Raw CloudTrail events are not immediately actionable in a Slack alert. A Lambda handler fetches the current IAM policy state of the role at alert time — so you can see whether AttachRolePolicy already succeeded, which matters for incident triage priority:

import { IAMClient, ListAttachedRolePoliciesCommand } from "@aws-sdk/client-iam";

const iam = new IAMClient({});

export async function handler(event: {
  detail: {
    eventName: string; eventSource: string; eventTime: string;
    userIdentity: { principalId: string; arn: string };
    requestParameters: Record<string, string>;
    sourceIPAddress: string; errorCode?: string;
  };
}) {
  const detail = event.detail;
  const roleName = detail.userIdentity.arn.match(/assumed-role\/([^/]+)\//)?.[1];

  let currentPolicies: string[] = [];
  if (detail.eventSource === "iam.amazonaws.com" && roleName) {
    try {
      const resp = await iam.send(new ListAttachedRolePoliciesCommand({ RoleName: roleName }));
      currentPolicies = (resp.AttachedPolicies ?? []).map(p => p.PolicyArn ?? "");
    } catch {
      currentPolicies = ["(failed to retrieve — check IAM permissions)"];
    }
  }

  const isFailed = !!detail.errorCode;
  await fetch(process.env.SLACK_WEBHOOK_URL!, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      text: `🚨 *Security event: ${detail.eventName}* ${isFailed ? "(FAILED)" : "(SUCCEEDED)"}`,
      attachments: [{
        color: isFailed ? "warning" : "danger",
        fields: [
          { title: "API call", value: `${detail.eventSource} / ${detail.eventName}`, short: true },
          { title: "Status", value: detail.errorCode ?? "Success", short: true },
          { title: "Principal", value: detail.userIdentity.principalId, short: false },
          { title: "Source IP", value: detail.sourceIPAddress, short: true },
          { title: "Time", value: detail.eventTime, short: true },
          { title: "Request params", value: JSON.stringify(detail.requestParameters), short: false },
          ...(currentPolicies.length ? [{
            title: "Current attached policies on role",
            value: currentPolicies.join("\n"),
            short: false,
          }] : []),
        ],
      }],
    }),
  });
}

MCP server self-monitoring: startup IAM integrity check

The EventBridge pattern detects external modifications. For MCP servers that may have tool handlers capable of executing arbitrary API calls (e.g., a tools-as-code pattern), add a startup integrity check that verifies the role's attached policies match an expected set. Wire the failure to a CloudWatch Logs metric filter so any CRITICAL log line triggers an alarm before the server begins accepting connections:

import { IAMClient, ListAttachedRolePoliciesCommand } from "@aws-sdk/client-iam";
import { STSClient, GetCallerIdentityCommand } from "@aws-sdk/client-sts";

const EXPECTED_POLICIES = new Set([
  "arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess",
  "arn:aws:iam::123456789012:policy/mcp-s3-output-write",
]);

export async function checkIamIntegrity(): Promise<void> {
  const sts = new STSClient({});
  const iam = new IAMClient({});

  const { Arn } = await sts.send(new GetCallerIdentityCommand({}));
  const roleName = Arn?.match(/assumed-role\/([^/]+)\//)?.[1];
  if (!roleName) return;

  const { AttachedPolicies } = await iam.send(
    new ListAttachedRolePoliciesCommand({ RoleName: roleName })
  );

  const attached = new Set(AttachedPolicies?.map(p => p.PolicyArn!) ?? []);
  const unexpected = [...attached].filter(p => !EXPECTED_POLICIES.has(p));

  if (unexpected.length > 0) {
    // CloudWatch Logs metric filter catches level=CRITICAL → alarm → SNS
    console.error(JSON.stringify({
      level: "CRITICAL",
      event: "unexpected_iam_policy",
      role: roleName,
      unexpected_policies: unexpected,
    }));
    throw new Error(`Unexpected IAM policies detected: ${unexpected.join(", ")}`);
  }
}

// Call at startup, before accepting MCP connections
await checkIamIntegrity();

Wire the corresponding CloudWatch Logs metric filter on the trail's CWL delivery (or directly on the application log group) to increment a custom metric on level = "CRITICAL", then alarm with a threshold of 1 — any single critical log line triggers the alarm. This gives you sub-minute detection for policy modifications caught at the next server restart, complementing the EventBridge rule's real-time detection during runtime.

Thread 3: Matching anomaly class to the right detection mechanism

CloudTrail provides three distinct anomaly detection mechanisms, each suited to a different class of MCP server failure. Using the wrong mechanism for an anomaly class either produces no signal (Insights won't catch a DynamoDB data event spike) or requires manual baseline tuning that Insights would handle automatically. The key axis of differentiation: which event category does the anomaly manifest in?

CloudTrail Insights: what it does and the 7-day gap

Insights uses a machine-learning model that observes the last 7 days of management write events for each API and establishes an expected hourly rate. When the observed rate in a 10-minute window exceeds the baseline by a statistically significant margin, Insights fires a START event to EventBridge. When the rate normalizes, it fires an END event — the START/END pair enables automatic incident resolution in systems that support it.

The 7-day gap is a hard constraint: Insights fires nothing during the learning period. The practical implication for new deployments is that you need a day-0 fallback. CloudWatch metric filters on the CloudTrail → CloudWatch Logs delivery are the right fallback: configurable threshold, immediate coverage, works for any API in the management event stream.

// CDK: enable Insights on trail
const trail = new cloudtrail.Trail(this, "McpAuditTrail", {
  trailName: "mcp-audit-trail",
  bucket: auditBucket,
  isMultiRegionTrail: true,
  enableFileValidation: true,
  managementEvents: cloudtrail.ReadWriteType.WRITE_ONLY,
  insightTypes: [
    cloudtrail.InsightType.API_CALL_RATE,
    cloudtrail.InsightType.API_ERROR_RATE,
  ],
});

// Day-0 fallback: CloudWatch Logs delivery + metric filter
const trailLogGroup = new logs.LogGroup(this, "TrailLogs", {
  logGroupName: "/aws/cloudtrail/mcp-audit",
  retention: logs.RetentionDays.THREE_MONTHS,
});
trail.logAllCloudWatchLogEvents(trailLogGroup);

// Metric filter: AssumeRole rate spike — day-0 coverage before Insights baseline
new logs.MetricFilter(this, "AssumeRoleFilter", {
  logGroup: trailLogGroup,
  filterPattern: logs.FilterPattern.literal(
    '{ $.eventName = "AssumeRole" && $.errorCode NOT EXISTS }'
  ),
  metricNamespace: "MCP/CloudTrail",
  metricName: "AssumeRoleSuccessCount",
  metricValue: "1",
  defaultValue: 0,
});

// Alarm: more than 50 AssumeRole calls in 60 seconds = potential credential vending
new cloudwatch.Alarm(this, "AssumeRoleAlarm", {
  metric: new cloudwatch.Metric({
    namespace: "MCP/CloudTrail",
    metricName: "AssumeRoleSuccessCount",
    statistic: "Sum",
    period: cdk.Duration.seconds(60),
  }),
  threshold: 50,
  evaluationPeriods: 1,
  alarmName: "mcp-AssumeRole-spike",
  alarmDescription: "Possible credential vending — unusually high AssumeRole rate from MCP role",
});

Reading Insights events: triage by anomaly magnitude

The Insights event includes both the baseline average and the observed average, allowing severity triage without a fixed threshold. A 2× spike during a known deployment window is likely benign; a 27× spike at 3 AM from the MCP server role is not. Route Insights START events through a Lambda handler that computes the magnitude ratio and assigns a severity before paging on-call:

export async function insightsHandler(event: {
  detail: {
    eventName: string; eventTime: string;
    userIdentity: { principalId: string };
    insightDetails: {
      state: "Start" | "End";
      insightType: string;
      insightContext: {
        statistics: {
          baseline: { average: number };
          insight: { average: number };
        };
      };
    };
  };
}) {
  const details = event.detail.insightDetails;
  if (details.state !== "Start") return; // END events routed separately for auto-resolve

  const baseline = details.insightContext.statistics.baseline.average;
  const observed = details.insightContext.statistics.insight.average;
  const ratio = observed / (baseline || 1);

  // HIGH ≥ 10×: potential incident; MEDIUM 3-10×: investigate; LOW < 3×: monitor
  const severity = ratio >= 10 ? "HIGH" : ratio >= 3 ? "MEDIUM" : "LOW";

  console.log(JSON.stringify({
    level: severity === "HIGH" ? "ERROR" : severity === "MEDIUM" ? "WARN" : "INFO",
    event: "cloudtrail_insights_anomaly",
    api: `${event.detail.eventName}`,
    insight_type: details.insightType,
    baseline_rate: baseline.toFixed(1),
    observed_rate: observed.toFixed(1),
    ratio: ratio.toFixed(1),
    severity,
    principal: event.detail.userIdentity?.principalId,
  }));

  // Only page on HIGH or MEDIUM — LOW goes to a monitoring dashboard
  if (severity === "LOW") return;

  await fetch(process.env.SLACK_WEBHOOK_URL!, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      text: `${severity === "HIGH" ? "🚨" : "⚠️"} *CloudTrail Insights — ${severity}*`,
      attachments: [{
        color: severity === "HIGH" ? "danger" : "warning",
        fields: [
          { title: "API", value: event.detail.eventName, short: true },
          { title: "Insight type", value: details.insightType, short: true },
          { title: "Rate", value: `${baseline.toFixed(1)}/min → ${observed.toFixed(1)}/min (${ratio.toFixed(1)}×)`, short: false },
          { title: "Principal", value: event.detail.userIdentity?.principalId, short: false },
        ],
      }],
    }),
  });
}

CloudTrail Lake for data-plane anomaly detection

Data events (PutItem, DeleteItem, GetObject) are invisible to both Insights and EventBridge. For data-plane anomaly detection, CloudTrail Lake scheduled queries are the right mechanism. A Lambda on an EventBridge scheduler runs hourly, queries Lake for anomalous patterns, and alerts when thresholds are exceeded:

import {
  CloudTrailClient, StartQueryCommand,
  GetQueryCommand, GetQueryResultsCommand,
} from "@aws-sdk/client-cloudtrail";

const cloudtrail = new CloudTrailClient({ region: "us-east-1" });
const EDS_ARN = process.env.EVENT_DATA_STORE_ARN!;

// Scheduled query: detect DeleteItem/DeleteObject spikes in last hour
// Run hourly via EventBridge scheduler
export async function detectDataEventAnomalies() {
  const { QueryId } = await cloudtrail.send(new StartQueryCommand({
    QueryStatement: `
      SELECT
        date_trunc('minute', eventTime) AS minute,
        eventName,
        count(*) AS count
      FROM ${EDS_ARN}
      WHERE eventTime > date_add('hour', -1, NOW())
        AND eventName IN ('DeleteItem', 'DeleteObject', 'BatchWriteItem')
        AND userIdentity.arn LIKE '%mcp-server-task-role%'
      GROUP BY 1, 2
      ORDER BY 1 DESC, 3 DESC
      LIMIT 60
    `,
  }));

  // Poll for completion
  while (true) {
    const { QueryStatus } = await cloudtrail.send(new GetQueryCommand({
      EventDataStore: EDS_ARN, QueryId,
    }));
    if (QueryStatus === "FINISHED") break;
    if (QueryStatus === "FAILED" || QueryStatus === "CANCELLED")
      throw new Error(`Anomaly detection query failed: ${QueryId}`);
    await new Promise(r => setTimeout(r, 3000));
  }

  const { QueryResultRows } = await cloudtrail.send(new GetQueryResultsCommand({
    EventDataStore: EDS_ARN, QueryId,
  }));

  for (const row of QueryResultRows ?? []) {
    const cols = Object.fromEntries(row.map(c => [c.key!, c.value]));
    const count = parseInt(cols.count ?? "0", 10);
    // Alert if more than 1,000 delete operations in any single minute
    if (count > 1000) {
      console.error(JSON.stringify({
        level: "CRITICAL",
        event: "data_event_anomaly",
        event_name: cols.eventName,
        minute: cols.minute,
        count,
      }));
    }
  }
}

The key SQL note: requestParameters in CloudTrail Lake is stored as an escaped JSON string. To extract nested fields, use double extraction: json_extract_scalar(json_parse(requestParameters), '$.tableName'). The outer json_parse deserializes the escaped string; the inner json_extract_scalar navigates the parsed object. Attempting json_extract_scalar(requestParameters, '$.tableName') directly returns NULL on escaped JSON fields.

S3 delivery: the two failure modes that break everything silently

CloudTrail S3 delivery has two silent failure modes worth understanding before you write the bucket policy and create the Athena table.

Silent delivery failure: The bucket policy requires both s3:GetBucketAcl (on the bucket ARN) and s3:PutObject (on the bucket ARN + /AWSLogs/* prefix). CloudTrail checks the ACL permission before attempting delivery; if that check fails, delivery stops silently. The trail health indicator in the console shows "Healthy" — it only reflects whether the trail configuration is valid, not whether delivery is succeeding. The only way to detect the failure is checking that log files are actually appearing in S3 every 15 minutes.

Athena prefix error: CloudTrail delivers log files to AWSLogs/<account>/CloudTrail/ and digest files to AWSLogs/<account>/CloudTrail-Digest/. If Athena's LOCATION or storage.location.template points at AWSLogs/<account>/ (the parent), the CloudTrailSerde attempts to parse digest files and returns empty results or serialization errors. The template must end at the CloudTrail/ subdirectory level.

-- Correct Athena partition projection template — ends at CloudTrail/ subdirectory
TBLPROPERTIES (
  "projection.enabled" = "true",
  "projection.account.type" = "enum",
  "projection.account.values" = "123456789012",
  "projection.region.type" = "enum",
  "projection.region.values" = "us-east-1,us-west-2,eu-west-1",
  "projection.year.type" = "integer",
  "projection.year.range" = "2026,2030",
  "projection.month.type" = "integer",
  "projection.month.range" = "1,12",
  "projection.month.digits" = "2",
  "projection.day.type" = "integer",
  "projection.day.range" = "1,31",
  "projection.day.digits" = "2",
  -- Note: template ends at CloudTrail/${region}/... — excludes CloudTrail-Digest/
  "storage.location.template" =
    "s3://mcp-cloudtrail-audit-123456789012/AWSLogs/${account}/CloudTrail/${region}/${year}/${month}/${day}"
);

Log file validation provides a tamper-evident chain through SHA-256 digest files. Enable it at trail construction (enableFileValidation: true) and run monthly validation via the CLI for compliance audits. One gotcha: enabling S3 Object Lock in Compliance mode (not Governance) prevents lifecycle rule transitions and expirations — objects are immutable until the Object Lock retention period expires, regardless of lifecycle rules. Use Governance mode for audit buckets where you want both immutability and lifecycle management.

Consolidated failure modes reference

Implementation checklist

1. Create the S3 audit bucket first with versioning enabled, Block Public Access, and DenyDeleteByAnyone + DenyNonTLS bucket policy statements alongside the CloudTrail service principal statements.
2. Add both s3:GetBucketAcl (on bucket ARN) and s3:PutObject (on AWSLogs/<account>/*) statements for cloudtrail.amazonaws.com with AWS:SourceArn condition scoped to the trail ARN.
3. Create a multi-region trail with enableFileValidation: true and managementEvents: WRITE_ONLY. Include Insights types for ApiCallRateInsight and ApiErrorRateInsight.
4. Add data event selectors using the CDK L1 escape hatch (CfnTrail.addPropertyOverride("AdvancedEventSelectors", [...])) — include readOnly: false and specific table ARN prefixes on every selector.
5. Add lifecycle rules to the audit bucket: S3-IA at 90 days, Glacier Instant Retrieval at 1 year. Add noncurrentVersionExpiration and expiredObjectDeleteMarker for the versioned bucket.
6. Create an Athena table with CloudTrailSerde and partition projection. Set the storage template to end at the CloudTrail/ subdirectory — not the account prefix or bucket root.
7. Enable CloudWatch Logs delivery on the trail (separate IAM role with logs:CreateLogStream + logs:PutLogEvents). Add metric filters as day-0 coverage before Insights baseline establishes.
8. Deploy EventBridge deny-list rules for the APIs your MCP server should never call: IAM modification, S3 bucket policy mutation, AssumeRole to arbitrary targets, ConsoleLogin. Use detail.* field paths — not top-level paths.
9. Add the startup IAM integrity check to the MCP server initialization routine. Wire it to a CloudWatch Logs metric filter and alarm on any level: CRITICAL log line.
10. If querying across multiple accounts, create a CloudTrail Lake event data store with organizationEnabled: true. For data-plane anomaly detection, deploy a Lambda on an hourly EventBridge schedule running Lake SQL queries for DeleteItem/DeleteObject spikes.

Further reading

• MCP Server CloudTrail Data Events — write-only advanced event selectors for S3 and DynamoDB, CDK trail construction with L1 escape hatch for DynamoDB selectors, tool call correlation via partition key embedding, Athena query patterns for data event audit
• MCP Server CloudTrail → EventBridge — automatic management event integration, deny-list EventBridge rule patterns, Lambda enrichment handler fetching current IAM policy state, MCP server IAM self-check, cross-region aggregation via CloudWatch Logs
• MCP Server CloudTrail Insights — ApiCallRateInsight and ApiErrorRateInsight types, 7-day baseline learning model, START/END event pair for auto-resolve, Lambda severity triage by anomaly magnitude ratio, day-0 CloudWatch metric filter fallback
• MCP Server CloudTrail Lake — Lake vs S3+Athena comparison matrix, StartQuery → poll GetQueryCommand → paginate GetQueryResults pattern, CloudTrail Lake SQL subset limitations, cross-account aggregation via Organizations channel, double json_extract_scalar for escaped requestParameters fields
• MCP Server CloudTrail S3 Delivery — delivery prefix structure (CloudTrail/ vs CloudTrail-Digest/), complete bucket policy with both GetBucketAcl and PutObject statements, SHA-256 digest chain validation, partition projection Athena setup, Glacier Instant Retrieval transparency to Athena