EventBridge Pipes Source Types

SQS source — batch size, batch window, and visibility timeout

The SQS source is the simplest to configure but has one critical timing constraint: the visibility timeout on the queue must be greater than the total time Pipes needs to process a batch.

// CDK: SQS source with correct visibility timeout sizing
import { Queue } from "aws-cdk-lib/aws-sqs";
import { aws_pipes as pipes } from "aws-cdk-lib";

// Sizing visibility timeout:
// batchWindow = 5s (wait up to 5s to accumulate records)
// enrichmentTimeout = 25s (Lambda enrichment max)
// targetTimeout = 10s (DynamoDB batch write, generous)
// buffer = 30s
// Total: 5 + 25 + 10 + 30 = 70s → round up to next round number
const toolCallQueue = new Queue(this, "ToolCallQueue", {
  visibilityTimeout: cdk.Duration.seconds(120),  // comfortable margin above 70s
  // SQS-level DLQ: handles messages that fail to be received (pre-pipe)
  deadLetterQueue: {
    queue: new Queue(this, "ToolCallDlq"),
    maxReceiveCount: 3,  // move to DLQ after 3 receive attempts
  },
});

// Pipe source parameters
const sqsSourceParams = {
  sqsQueueParameters: {
    batchSize: 10,          // Up to 10 messages per batch (max: 10,000 for SQS)
    maximumBatchingWindowInSeconds: 5,  // Wait up to 5s to fill the batch
  },
};

// FIFO queues are NOT supported as Pipes sources.
// Use standard SQS for Pipes integration.
// For ordering requirements, use the SQS message group + your own dedup logic.

The maximumBatchingWindowInSeconds (batch window) controls how long Pipes waits to accumulate records up to batchSize. Setting it to 0 means Pipes immediately delivers whatever it polled; setting it to 60 means Pipes will wait up to 60 seconds to fill a batch of batchSize records. A longer batch window improves batching efficiency (fewer Lambda invocations) at the cost of higher latency for individual messages.

SQS source — FIFO limitation and workaround

SQS FIFO queues are not supported as EventBridge Pipes sources. If your MCP server event pipeline requires strict per-tenant ordering, the options are:

• Use Lambda ESM with SQS FIFO — Lambda's event source mapping does support FIFO queues, with one concurrent Lambda invocation per message group. Gives strict ordering per MessageGroupId.
• Use standard SQS + Kinesis — Route ordered events into a Kinesis stream (partition key = tenant ID), then use Kinesis as the Pipes source. Within a shard, Kinesis preserves ordering strictly. Pipes processes Kinesis shards serially (or with configurable parallelization per shard).
• Use standard SQS + dedup key — For most MCP server use cases, true ordering is not required; idempotent writes in the target (DynamoDB conditional puts with version numbers) handle occasional duplicates from at-least-once delivery.

DynamoDB Streams source — starting position and record format

When using a DynamoDB Stream as a Pipes source, you're consuming the change feed of a DynamoDB table — each record represents an INSERT, MODIFY, or REMOVE operation on a table item. This is useful for building audit trails, cache invalidation, or event-driven side effects on MCP server state changes.

// CDK: DynamoDB Streams source
import { Table, BillingMode, AttributeType, StreamViewType } from "aws-cdk-lib/aws-dynamodb";

// Table must have streams enabled with the right view type
const sessionTable = new Table(this, "SessionTable", {
  billingMode: BillingMode.PAY_PER_REQUEST,
  partitionKey: { name: "pk", type: AttributeType.STRING },
  sortKey: { name: "sk", type: AttributeType.STRING },
  // Stream view type determines what's available in Pipes records
  stream: StreamViewType.NEW_AND_OLD_IMAGES,  // Both before and after state
  // StreamViewType.NEW_IMAGE — only the new state (smaller payload, most common)
  // StreamViewType.OLD_IMAGE — only the deleted/pre-update state
  // StreamViewType.KEYS_ONLY — only the partition/sort key (most compact)
});

const pipe = new pipes.CfnPipe(this, "DdbStreamPipe", {
  // ...
  source: sessionTable.tableStreamArn!,  // Must use the stream ARN, not table ARN
  sourceParameters: {
    dynamoDbStreamParameters: {
      startingPosition: "LATEST",
      // LATEST: only new changes after pipe creation (production default)
      // TRIM_HORIZON: all changes available in the stream (up to 24h retention)
      // No AT_TIMESTAMP option (unlike Kinesis)
      batchSize: 100,
      maximumBatchingWindowInSeconds: 5,
      maximumRecordAgeInSeconds: 3600,    // Drop records older than 1h
      maximumRetryAttempts: 3,
      bisectBatchOnFunctionError: false,  // Not available in Pipes (unlike Lambda ESM)
      parallelizationFactor: 1,          // 1-10 concurrent batches per shard
      onPartialBatchItemFailure: "AUTOMATIC_BISECT",  // Available for DDB streams
      destinationConfig: {
        onFailure: {
          destination: dlq.queueArn,  // Pipe-level DLQ for failed batches
        },
      },
    },
  },
  // ...
});

// Grant pipe role stream read permission
pipeRole.addToPolicy(new PolicyStatement({
  actions: [
    "dynamodb:GetRecords",
    "dynamodb:GetShardIterator",
    "dynamodb:DescribeStream",
    "dynamodb:ListStreams",
  ],
  resources: [sessionTable.tableStreamArn!],
}));

The DynamoDB stream record in the Pipes event payload includes the eventName (INSERT, MODIFY, REMOVE), dynamodb.NewImage, dynamodb.OldImage (if configured), and dynamodb.Keys. Images are in DynamoDB's typed JSON format — {"S": "value"}, {"N": "123"} — not plain JSON. Your enrichment Lambda (or target) needs to unmarshall these if you want plain objects. Use the AWS SDK's unmarshall utility from @aws-sdk/util-dynamodb.

DynamoDB Streams — computing diffs in enrichment

// Enrichment Lambda: compute diff between old and new DynamoDB image
import { unmarshall } from "@aws-sdk/util-dynamodb";
import type { DynamoDBRecord } from "aws-lambda";

exports.handler = async (records: DynamoDBRecord[]) => {
  return records.map(record => {
    const eventName = record.eventName;  // INSERT | MODIFY | REMOVE
    const keys = record.dynamodb?.Keys
      ? unmarshall(record.dynamodb.Keys as any)
      : {};
    const newImage = record.dynamodb?.NewImage
      ? unmarshall(record.dynamodb.NewImage as any)
      : null;
    const oldImage = record.dynamodb?.OldImage
      ? unmarshall(record.dynamodb.OldImage as any)
      : null;

    // Compute changed fields for MODIFY events
    const changedFields = eventName === "MODIFY" && newImage && oldImage
      ? Object.keys(newImage).filter(k => JSON.stringify(newImage[k]) !== JSON.stringify(oldImage[k]))
      : [];

    return {
      eventName,
      keys,
      newImage,
      oldImage,
      changedFields,
      sequenceNumber: record.dynamodb?.SequenceNumber,
      approximateCreationDateTime: record.dynamodb?.ApproximateCreationDateTime,
    };
  });
};

Kinesis source — starting position and parallelization

Kinesis is the highest-throughput source type for Pipes. The starting position options are more flexible than DynamoDB (AT_TIMESTAMP is available) and the parallelization factor has a direct impact on throughput.

// CDK: Kinesis source
import { Stream } from "aws-cdk-lib/aws-kinesis";

const eventStream = new Stream(this, "EventStream", {
  streamName: "mcp-tool-events",
  shardCount: 4,  // 4 MB/s write, 8 MB/s read (classic)
  retentionPeriod: cdk.Duration.days(7),
});

const pipe = new pipes.CfnPipe(this, "KinesisPipe", {
  // ...
  source: eventStream.streamArn,
  sourceParameters: {
    kinesisStreamParameters: {
      // Starting position options:
      // LATEST — only new records (default for production)
      // TRIM_HORIZON — all records in the stream (for backfill)
      // AT_TIMESTAMP — records from a specific time onwards
      startingPosition: "LATEST",
      // startingPosition: "AT_TIMESTAMP",
      // startingPositionTimestamp: "2026-09-01T00:00:00Z",  // ISO 8601 UTC

      batchSize: 100,
      maximumBatchingWindowInSeconds: 5,
      maximumRecordAgeInSeconds: -1,  // -1 = no age limit (process all available)
      maximumRetryAttempts: -1,       // -1 = retry indefinitely
      parallelizationFactor: 1,       // 1 = serial per shard; 2-10 = concurrent batches
      onPartialBatchItemFailure: "AUTOMATIC_BISECT",  // Bisect on error
      destinationConfig: {
        onFailure: { destination: pipeDlq.queueArn },
      },
    },
  },
  // ...
});

// Grant pipe role Kinesis read permissions
eventStream.grantRead(pipeRole);

The parallelizationFactor setting controls how many concurrent batches a single Kinesis shard can have in-flight within this pipe. At parallelizationFactor: 1 (default), records within a shard are processed strictly in order. At parallelizationFactor: 3, up to 3 batches from the same shard are processed concurrently — this triples throughput from that shard but means records 1-100 and 101-200 from the same shard may be enriched and written to the target in parallel, without ordering guarantees between the two batches.

Source comparison table