Kinesis Consumer Patterns for MCP Servers

Lambda Event Source Mapping — configuration and failure handling

Lambda ESM is the lowest-operational-overhead way to consume a Kinesis stream. Lambda polls the stream on your behalf (using enhanced fan-out by default since 2023) and invokes your function with a batch of records from a single shard. You pay for Lambda invocation time; no consumer infrastructure to run.

The most important configuration is bisect-on-error. Without it, a single record that causes your Lambda to throw an error causes the entire batch to be retried — and since Kinesis ordering is per-shard, the batch blocks all later records on that shard. With bisect-on-error enabled, Lambda splits a failing batch in half and retries each half independently, rapidly isolating the single bad record. Combined with item-level failure reporting, you can process 499 good records and send only the 1 bad record to the DLQ.

import type { KinesisStreamEvent, KinesisStreamBatchResponse } from "aws-lambda";

export async function handler(
  event: KinesisStreamEvent
): Promise<KinesisStreamBatchResponse> {
  const failures: Array<{ itemIdentifier: string }> = [];

  for (const record of event.Records) {
    try {
      const payload = JSON.parse(
        Buffer.from(record.kinesis.data, "base64").toString("utf-8")
      );

      await processToolCallEvent(payload, record.kinesis.sequenceNumber);
    } catch (err) {
      console.error(`Failed to process record ${record.kinesis.sequenceNumber}:`, err);
      // Report this record as failed without failing the entire batch
      failures.push({ itemIdentifier: record.kinesis.sequenceNumber });
    }
  }

  // Return batchItemFailures — Lambda retries only these records
  // Records NOT in this list are considered successfully processed
  return { batchItemFailures: failures };
}

async function processToolCallEvent(
  payload: unknown,
  sequenceNumber: string
): Promise<void> {
  // Idempotency: use sequenceNumber as idempotency key
  // DynamoDB conditional write: only insert if sequenceNumber not already present
  // This handles Lambda ESM's at-least-once delivery guarantee
  // ...
}

Lambda ESM configuration best practices for Kinesis:

• StartingPosition: TRIM_HORIZON for new consumers that should process historical records; LATEST for consumers that only care about future events
• BatchSize: 100 — default and a good starting point; increase to 10,000 for high-throughput bulk processing where per-record latency doesn't matter
• BisectBatchOnFunctionError: true — always enable
• MaximumBatchingWindowInSeconds: 5 — Lambda waits up to 5 seconds to fill a batch before invoking; reduces invocations at low throughput
• ParallelizationFactor: 1–10 — number of concurrent Lambda invocations per shard; set to 1 for ordered processing, increase for throughput at the cost of ordering
• DestinationConfig.OnFailure.Destination — point to an SQS queue; records that fail even after bisection and maximum retry attempts land here

# CDK configuration for Kinesis ESM
import { EventSourceMapping, StartingPosition } from "aws-cdk-lib/aws-lambda";
import { SqsDlq } from "aws-cdk-lib/aws-lambda-event-sources";
import { KinesisEventSource } from "aws-cdk-lib/aws-lambda-event-sources";

processorFn.addEventSource(
  new KinesisEventSource(stream, {
    startingPosition: StartingPosition.TRIM_HORIZON,
    batchSize: 100,
    maxBatchingWindow: Duration.seconds(5),
    bisectBatchOnError: true,
    reportBatchItemFailures: true,
    parallelizationFactor: 1,         // increase for throughput over ordering
    retryAttempts: 3,                 // retry each bisected batch up to 3x before DLQ
    onFailure: new SqsDlq(dlqQueue),
  })
);

KCL worker model — when to use it and checkpointing semantics

The Kinesis Client Library (KCL) is a Java/Python/Node.js library that manages shard assignment, lease coordination across multiple worker instances, and checkpointing to DynamoDB. Use KCL over Lambda ESM when you need: stateful aggregation across records (windowed counts, session stitching), processing that must run on a long-lived server rather than Lambda, or multi-shard joins that require a global view of the stream.

The checkpointing semantic is the key operational concept: KCL tracks progress by writing a sequence number to DynamoDB (the lease table). When you call checkpointer.checkpoint(), KCL records the sequence number of the last record you passed. If your process crashes after checkpoint, the next KCL worker picks up from after that sequence number. If you crash before checkpoint, the next worker reprocesses everything since the last successful checkpoint — at-least-once delivery. Never checkpoint after processing each record in a loop — always checkpoint after the entire batch to minimize DynamoDB write overhead. Never checkpoint before confirming all records in the batch succeeded — partial batches with a checkpoint lose the unprocessed tail.

// KCL v3 TypeScript worker (Node.js 22)
// npm install amazon-kinesis-client

import { Checkpointer, InitializationInput, ProcessRecordsInput, ShutdownInput } from "amazon-kinesis-client";

class McpEventProcessor {
  private shardId = "";

  async initialize(input: InitializationInput): Promise<void> {
    this.shardId = input.shardId;
    console.log(`Initialized shard ${this.shardId} at ${input.extendedSequenceNumber}`);
  }

  async processRecords(input: ProcessRecordsInput): Promise<void> {
    const { records, checkpointer } = input;

    const processed: string[] = [];

    for (const record of records) {
      try {
        const payload = JSON.parse(record.data.toString("utf-8"));
        await processEvent(payload);
        processed.push(record.sequenceNumber);
      } catch (err) {
        console.error(`Error processing ${record.sequenceNumber}:`, err);
        // Don't checkpoint past the failing record
        // Break here to reprocess from last checkpoint on restart
        break;
      }
    }

    // Checkpoint only if we processed records — and only after ALL succeeded
    if (processed.length === records.length) {
      const lastSeq = records[records.length - 1].sequenceNumber;
      await checkpointer.checkpoint(lastSeq);
    }
  }

  async leaseLost(_input: unknown): Promise<void> {
    console.log(`Shard ${this.shardId} lease lost — another worker took over`);
  }

  async shardEnded(input: { checkpointer: Checkpointer }): Promise<void> {
    // Must checkpoint with SHARD_END to signal shard is fully processed
    // Required after shard split/merge so KCL advances to child shards
    await input.checkpointer.checkpoint();
    console.log(`Shard ${this.shardId} ended`);
  }

  async shutdownRequested(input: { checkpointer: Checkpointer }): Promise<void> {
    await input.checkpointer.checkpoint();
  }
}

Enhanced fan-out with SubscribeToShard

Enhanced fan-out consumers register with RegisterStreamConsumer and receive records via a long-lived HTTP/2 push connection (SubscribeToShard). Unlike standard GetRecords polling (shared 2 MB/s across all consumers), each enhanced fan-out consumer gets a dedicated 2 MB/s per shard pipe. Maximum 20 enhanced fan-out consumers per stream.

The re-subscribe loop trap: SubscribeToShard connections automatically expire after 5 minutes. The SDK raises an event when the subscription ends — you must handle this event and immediately call SubscribeToShard again. If you miss this event handler, the consumer silently stops receiving records with no error thrown on the producer side.

import {
  KinesisClient,
  SubscribeToShardCommand,
  StartingPosition,
} from "@aws-sdk/client-kinesis";

const kinesis = new KinesisClient({ region: "us-east-1" });

async function subscribeToShard(
  consumerArn: string,
  shardId: string,
  lastSequenceNumber?: string
): Promise<void> {
  const startingPosition = lastSequenceNumber
    ? {
        Type: "AFTER_SEQUENCE_NUMBER" as const,
        SequenceNumber: lastSequenceNumber,
      }
    : { Type: "LATEST" as const };

  // SubscribeToShard returns an async iterable of event batches
  const response = await kinesis.send(
    new SubscribeToShardCommand({
      ConsumerARN: consumerArn,
      ShardId: shardId,
      StartingPosition: startingPosition,
    })
  );

  let lastProcessedSeq = lastSequenceNumber;

  try {
    for await (const event of response.EventStream!) {
      if (event.SubscribeToShardEvent) {
        const { Records, ContinuationSequenceNumber } = event.SubscribeToShardEvent;

        for (const record of Records ?? []) {
          const payload = JSON.parse(
            Buffer.from(record.Data!).toString("utf-8")
          );
          await processEvent(payload);
          lastProcessedSeq = record.SequenceNumber;
        }
      }
    }
  } catch (err: unknown) {
    const errName = (err as Error).name ?? "";
    if (errName === "ResourceInUseException") {
      // Another consumer has this shard — back off and retry
      await new Promise((r) => setTimeout(r, 5000));
    }
    // All other errors: fall through to re-subscribe
  }

  // Subscription expired (5-min limit) or error — re-subscribe immediately
  await subscribeToShard(consumerArn, shardId, lastProcessedSeq);
}

Idempotent processing — handling at-least-once delivery

Kinesis guarantees that every record is delivered at least once but does not guarantee exactly once. Duplicate delivery occurs: after a Lambda ESM retry, after a KCL worker failover before checkpoint, or after a network interruption that causes a partial batch to be re-delivered. Your consumer must be idempotent — processing the same record twice must produce the same outcome as processing it once.

The cleanest idempotency mechanism for MCP server event consumers is a conditional DynamoDB write keyed on the Kinesis sequence number:

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

const dynamo = new DynamoDBClient({ region: "us-east-1" });

async function processEventIdempotently(
  payload: Record<string, unknown>,
  sequenceNumber: string
): Promise<void> {
  // Conditional write: only proceed if this sequence number hasn't been processed
  try {
    await dynamo.send(
      new PutItemCommand({
        TableName: "mcp-processed-events",
        Item: marshall({
          pk: `seq#${sequenceNumber}`,
          processedAt: new Date().toISOString(),
          ttl: Math.floor(Date.now() / 1000) + 7 * 24 * 3600, // expire after 7 days
        }),
        ConditionExpression: "attribute_not_exists(pk)",
      })
    );
  } catch (err) {
    if (err instanceof ConditionalCheckFailedException) {
      // Already processed — idempotent skip
      return;
    }
    throw err;
  }

  // Only reaches here if this is the first time processing this sequence number
  await doActualProcessing(payload);
}

Consumer pattern selection guide

Common failure modes reference