MCP Server Lambda Event Source Mapping Filters — FilterCriteria, SQS, Kinesis, DynamoDB Streams

Filter pattern syntax

Filter patterns are JSON objects where each key is a path in the event record and each value is an array of match conditions. A record matches a pattern if all field conditions in that pattern are satisfied (AND). A record is processed if it matches any pattern (OR).

Conditions are applied after automatic JSON parsing: SQS body is parsed from a string if it is valid JSON; Kinesis data is decoded from base64 then parsed as JSON if valid. If the body is not valid JSON, only string-level conditions (exact, prefix, suffix) on the raw string apply.

SQS filter examples for MCP probe results

# Create an ESM with filter: only process messages where level == "ERROR"
# and where endpoint starts with "https://"
aws lambda create-event-source-mapping \
  --function-name ProbeAlertHandler \
  --event-source-arn arn:aws:sqs:us-east-1:123456789012:probe-results \
  --batch-size 10 \
  --filter-criteria '{
    "Filters": [
      {
        "Pattern": "{\"body\":{\"level\":[\"ERROR\"],\"endpoint\":[{\"prefix\":\"https://\"}]}}"
      },
      {
        "Pattern": "{\"body\":{\"level\":[\"DEGRADED\"],\"latencyMs\":[{\"numeric\":[\">\",5000]}]}}"
      }
    ]
  }'

This example has two filter patterns with OR semantics: Lambda is invoked if a message has level == "ERROR" AND endpoint starts with https://, OR if the message has level == "DEGRADED" AND latencyMs > 5000. An INFO-level message with any latency is discarded.

DynamoDB Streams filter: act only on INSERT or specific item types

// CDK: EventSourceMapping with DynamoDB Streams filters
import * as lambda from "aws-cdk-lib/aws-lambda";
import * as dynamodb from "aws-cdk-lib/aws-dynamodb";

const probeResultsTable = new dynamodb.Table(this, "ProbeResults", {
  partitionKey: { name: "pk", type: dynamodb.AttributeType.STRING },
  sortKey: { name: "sk", type: dynamodb.AttributeType.STRING },
  stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,  // required for NewImage filters
});

const analyticsWorker = new lambda.Function(this, "AnalyticsWorker", {
  // ...
});

// Filter: only INSERT events where item type is "probe_result"
// This skips MODIFY (status updates) and REMOVE (TTL expiry cleanup)
new lambda.EventSourceMapping(this, "ProbeResultStream", {
  target: analyticsWorker,
  eventSourceArn: probeResultsTable.tableStreamArn,
  startingPosition: lambda.StartingPosition.LATEST,
  batchSize: 50,
  bisectBatchOnError: true,
  filters: [
    lambda.FilterCriteria.filter({
      eventName: lambda.FilterRule.isEqual("INSERT"),
      dynamodb: {
        NewImage: {
          itemType: { S: lambda.FilterRule.isEqual("probe_result") },
        },
      },
    }),
  ],
});

The dynamodb.NewImage path uses DynamoDB's typed attribute format: { "S": "value" } for strings, { "N": "42" } for numbers (numbers are strings in the DynamoDB JSON wire format). A filter on NewImage requires the EventSourceMapping to use NEW_AND_OLD_IMAGES or NEW_IMAGE stream view type — KEYS_ONLY does not include image data and the filter will never match.

DynamoDB REMOVE for TTL expiry filtering

// Only process TTL-expired items (not manual deletes)
// TTL expiry events have userIdentity.type == "Service" and
// userIdentity.principalId == "dynamodb.amazonaws.com"
lambda.FilterCriteria.filter({
  eventName: lambda.FilterRule.isEqual("REMOVE"),
  userIdentity: {
    type: lambda.FilterRule.isEqual("Service"),
    principalId: lambda.FilterRule.isEqual("dynamodb.amazonaws.com"),
  },
})

Kinesis filter: only process specific event types from a shared stream

// Kinesis ESM: only invoke Lambda for records where data.eventType == "probe_failure"
// Kinesis data field is base64-encoded; Lambda decodes it and parses JSON
new lambda.EventSourceMapping(this, "ProbeLogStream", {
  target: alertHandlerFn,
  eventSourceArn: probeLogStream.streamArn,
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
  batchSize: 100,
  bisectBatchOnError: true,
  filters: [
    lambda.FilterCriteria.filter({
      data: {
        eventType: lambda.FilterRule.isEqual("probe_failure"),
      },
    }),
    lambda.FilterCriteria.filter({
      data: {
        eventType: lambda.FilterRule.isEqual("schema_drift"),
      },
    }),
  ],
});

With these two filters, the Kinesis ESM triggers the alert handler only when a record's decoded JSON contains eventType == "probe_failure" OR eventType == "schema_drift". eventType == "probe_success" records flow through the Kinesis shard without invoking Lambda — the shard's iterator advances past them as usual (filters do not pause shard processing).

Cost and throughput impact

One nuance for SQS: the filter applies per-message within a batch. If a batch of 10 messages has 2 matching and 8 non-matching, Lambda is invoked with only the 2 matching messages. The 8 non-matching messages are deleted from the queue automatically — they do not go to the DLQ. This is correct behavior (they were intentionally excluded), but may surprise teams that monitor DLQ depth to catch non-matching messages.

Common failure modes