MCP Server DynamoDB Streams — audit trails, session event sourcing, Lambda triggers

Enabling streams and choosing StreamViewType

DynamoDB Streams is enabled per table. The StreamViewType controls what data each stream record contains:

Changing StreamViewType on an existing stream is not possible — you must disable and re-enable streams, which resets the 24-hour retention window. Set NEW_AND_OLD_IMAGES from the start.

// CDK: DynamoDB table with streams enabled
import * as dynamodb from "aws-cdk-lib/aws-dynamodb";
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as lambdaEventSources from "aws-cdk-lib/aws-lambda-event-sources";

const sessionTable = new dynamodb.Table(this, "McpSessionTable", {
  tableName: "mcp-sessions",
  partitionKey: { name: "pk", type: dynamodb.AttributeType.STRING },
  sortKey: { name: "sk", type: dynamodb.AttributeType.STRING },
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
  stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,  // required for audit
  timeToLiveAttribute: "ttl",
  removalPolicy: cdk.RemovalPolicy.RETAIN,
});

const auditFn = new lambda.Function(this, "AuditStreamHandler", {
  runtime: lambda.Runtime.NODEJS_22_X,
  handler: "index.handler",
  code: lambda.Code.fromAsset("src/audit-stream"),
  environment: {
    AUDIT_TABLE: auditTable.tableName,
  },
});

auditFn.addEventSource(
  new lambdaEventSources.DynamoEventSource(sessionTable, {
    startingPosition: lambda.StartingPosition.TRIM_HORIZON,
    batchSize: 100,
    bisectBatchOnError: true,       // isolate poison-pill records
    retryAttempts: 3,
    onFailure: new lambdaEventSources.SqsDlq(dlq),
    filters: [
      // Only process MODIFY and REMOVE events (skip INSERT noise if desired)
      // Remove this filter to capture INSERTs too
      lambda.FilterCriteria.filter({
        eventName: lambda.FilterRule.isEqual("MODIFY"),
      }),
    ],
  })
);

Processing stream records in the Lambda handler

Each Lambda invocation receives a batch of stream records. Records within a shard are delivered in strict write order for each partition key. Your handler must be idempotent — Lambda retries the batch on any unhandled exception.

import {
  DynamoDBStreamEvent,
  DynamoDBRecord,
  Context,
} from "aws-lambda";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { unmarshall } from "@aws-sdk/util-dynamodb";
import { AttributeValue } from "@aws-sdk/client-dynamodb";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export async function handler(
  event: DynamoDBStreamEvent,
  _ctx: Context
): Promise {
  for (const record of event.Records) {
    if (record.eventName === "REMOVE") {
      // REMOVE records have OldImage but no NewImage
      await writeAuditEntry(record, "session_deleted");
      continue;
    }
    if (record.eventName === "INSERT") {
      await writeAuditEntry(record, "session_created");
      continue;
    }
    if (record.eventName === "MODIFY") {
      await writeAuditEntry(record, "session_updated");
    }
  }
}

async function writeAuditEntry(
  record: DynamoDBRecord,
  eventType: string
): Promise {
  const newImage = record.dynamodb?.NewImage
    ? unmarshall(record.dynamodb.NewImage as Record)
    : null;
  const oldImage = record.dynamodb?.OldImage
    ? unmarshall(record.dynamodb.OldImage as Record)
    : null;

  // Sequence number is unique per shard — use as idempotency key
  const sequenceNumber = record.dynamodb?.SequenceNumber;

  await ddb.send(
    new PutCommand({
      TableName: process.env.AUDIT_TABLE!,
      Item: {
        pk: `audit#${record.dynamodb?.Keys?.pk.S}`,
        sk: `${Date.now()}#${sequenceNumber}`,
        eventType,
        eventSource: record.eventSourceARN,
        sequenceNumber,
        newImage,
        oldImage,
        // Compute diff for MODIFY events
        changedFields:
          eventType === "session_updated" && newImage && oldImage
            ? Object.keys(newImage).filter(
                (k) => JSON.stringify(newImage[k]) !== JSON.stringify(oldImage[k])
              )
            : null,
        recordedAt: new Date().toISOString(),
      },
      // Idempotency: skip if we already processed this sequence number
      ConditionExpression: "attribute_not_exists(sequenceNumber)",
    })
  );
}

The ConditionExpression: "attribute_not_exists(sequenceNumber)" on the audit write makes the handler idempotent — a duplicate delivery from Lambda retries produces a ConditionalCheckFailedException which should be caught and swallowed (not re-thrown), since the record was already written.

Ordering guarantees and fan-out patterns

DynamoDB Streams partitions into shards, and each shard contains records for a subset of the table's partition key space. The ordering guarantee is:

• Within a shard: records are in strict write order for each partition key. If session pk=session#abc is updated three times, the stream records appear in that order within the shard that owns the key.
• Across shards: no ordering guarantee. Two writes to different partition keys may appear in any order across shards.
• Within a Lambda batch: records from the same shard maintain order, but a batch may mix records from multiple shards (if your Lambda has multiple concurrent shard readers).

For MCP audit logs, per-session ordering is usually sufficient — you care that session abc's events are in order, not that session abc's events are interleaved correctly with session xyz's events.

Fan-out to multiple consumers: DynamoDB Streams supports up to 2 concurrent Lambda triggers per stream. For more consumers (e.g., audit Lambda + search indexer Lambda), route the stream through Kinesis Data Streams (KDS) using the Kinesis Adapter, which supports up to 20 consumers via enhanced fan-out.

Common failure modes