AWS Kinesis for MCP Servers: Three Streaming Patterns for Producer Reliability, Consumer Durability, and Event Pipeline Security

The core mental model: Kinesis is a log, not a queue

Before the three patterns, it helps to be precise about what the five Kinesis primitives actually are and what they are not:

The key distinction between Kinesis Data Streams and a queue like SQS: Kinesis is a log, not a message queue. Records are not deleted after consumption — they age out after the retention window expires. Every consumer maintains its own position (iterator or checkpoint) and can re-read records independently. This makes Kinesis appropriate when you want multiple independent consumers (audit archival, real-time analytics, and a Lambda processor simultaneously) but inappropriate when you want competing consumers to share work — SQS is the right primitive for work queues.

Pattern 1: The producer reliability pair — PutRecords partial-fail and partition key selection

Most teams treat Kinesis as a write-and-forget bus. They call PutRecords, check for a non-2xx HTTP response, and assume success. This is wrong, and it loses data silently.

PutRecords HTTP 200 ≠ all records accepted

PutRecords accepts up to 500 records per call and returns HTTP 200 regardless of whether all records were accepted. The actual acceptance status for each record is in the response body's Records array. Each element has either a SequenceNumber + ShardId (success) or an ErrorCode + ErrorMessage (failure). The FailedRecordCount field at the top of the response summarizes the count, but you need the per-record array to know which records failed so you can retry them.

The most common error codes from PutRecords are ProvisionedThroughputExceededException (shard write capacity exhausted) and InternalFailure (transient service error). Both are retryable. The canonical PutRecords call pattern:

async function putRecordsWithRetry(
  client: KinesisClient,
  streamName: string,
  records: PutRecordsRequestEntry[],
  maxAttempts = 3
): Promise<void> {
  let pending = records;
  for (let attempt = 0; attempt < maxAttempts && pending.length > 0; attempt++) {
    if (attempt > 0) {
      // exponential backoff: 100ms, 200ms, 400ms
      await new Promise(r => setTimeout(r, 100 * Math.pow(2, attempt - 1)));
    }
    const response = await client.send(
      new PutRecordsCommand({ StreamName: streamName, Records: pending })
    );
    if (response.FailedRecordCount === 0) return;
    // collect only the failed records, preserving original order for retry
    pending = response.Records!
      .map((r, i) => (r.ErrorCode ? pending[i] : null))
      .filter((r): r is PutRecordsRequestEntry => r !== null);
  }
  if (pending.length > 0) {
    throw new Error(`PutRecords: ${pending.length} records failed after ${maxAttempts} attempts`);
  }
}

The index-mapping pattern on line 17 is the critical detail: response.Records[i] corresponds to pending[i] — the response array is always the same length as the input array, with each element either succeeded or failed. You do not get a separate failed-records list; you must correlate by index.

Partition key selection — the hot-spot trap

Kinesis routes records to shards by taking the MD5 hash of the partition key and mapping it into the shard's key range. If all records use the same partition key — or a low-cardinality set like MCP tool names — they all hash to the same shard range, overloading one shard while the others sit idle. A 10-shard stream with a constant partition key effectively has 1 MB/s write capacity, not 10 MB/s.

The rule: use sessionId as the partition key when downstream consumers need to process all events from one session in order. Use a random UUID or high-cardinality composite key otherwise. Never use a constant string, an enum field alone, or any value with fewer than ~50 distinct values across your event volume.

On-demand vs provisioned shards — the break-even calculation

Kinesis Data Streams has two capacity modes. Provisioned mode charges $0.015/shard-hour plus $0.014/GB ingested. On-demand mode charges $0.08/GB ingested and scales automatically. The break-even point is approximately 137 GB/month per logical shard ($0.08 × 137 = $11/shard ≈ $0.015 × 730h). If your MCP tool call volume routinely exceeds 137 GB/month per shard, on-demand costs more than a correctly provisioned stream. If your volume is bursty or unpredictable, on-demand saves you from provisioning for peak.

When planning provisioned capacity, use the formula shards = ceil(peak_MB_s / 0.85) — the 0.85 factor gives 15% headroom above the 1 MB/s per-shard limit for retry traffic and uneven partition key distribution. Monitor WriteProvisionedThroughputExceeded CloudWatch metric and alarm when the 5-minute average exceeds zero on three consecutive periods — that means retries are eating into your headroom.

Firehose: the 60-second floor and the transformation Lambda trap

Kinesis Data Firehose (now Amazon Data Firehose) solves the archival write path for MCP event logs: it buffers records and delivers them to S3, Redshift, or OpenSearch without any consumer code. The minimum buffer interval is 60 seconds — if your use case requires near-real-time delivery (<5 seconds), Firehose is not the right tool and you need a Lambda consumer on a Data Stream instead.

The transformation Lambda trap is the single most common Firehose mistake. When you enable record transformation, Firehose sends each batch to a Lambda and expects the Lambda to return the records base64-encoded in a specific structure. Two failure modes:

1. Returning the record data as a plain string instead of base64 encoding it. Firehose delivers the bytes to S3 without error, but the content is garbage — the bytes represent the raw UTF-8 string, not the original JSON. Athena queries then fail silently or return malformed data.
2. Omitting the newline delimiter (\n) at the end of each transformed record. Firehose concatenates all records in a batch into a single S3 object. Without newlines, all records run together and Athena sees the entire batch as a single malformed row.

The correct transformation Lambda return structure:

exports.handler = async (event) => {
  return {
    records: event.records.map(record => {
      // Decode the incoming base64 payload
      const payload = Buffer.from(record.data, 'base64').toString('utf8');
      const parsed = JSON.parse(payload);
      // Add derived fields, normalize schema, etc.
      const enriched = { ...parsed, processedAt: Date.now() };
      // Re-encode as base64, with \n delimiter for Athena
      const encoded = Buffer.from(JSON.stringify(enriched) + '\n').toString('base64');
      return {
        recordId: record.recordId,
        result: 'Ok',
        data: encoded  // must be base64, must include the \n inside the encoding
      };
    })
  };
};

Error records go to S3DestinationPrefix/processing-failed/ silently — Firehose never returns a delivery error to your application. You must alarm on two CloudWatch metrics: DeliveryToS3.DataFreshness (triggers when records are held longer than expected — indicates delivery stall) and create an S3 EventBridge rule on the processing-failed/ prefix to alert on transformation failures.

Pattern 2: The consumer durability stack — bisect-on-error, checkpoint discipline, and idempotency

Once records are in a Data Stream, the consumer architecture determines whether your MCP event pipeline processes every record exactly once or loses records silently on partial failures. Three failure patterns account for the majority of data loss in Kinesis consumer deployments.

Lambda ESM: bisect-on-error and batchItemFailures

Lambda event source mapping (ESM) on a Kinesis stream polls each shard and delivers batches of records to your Lambda function. The default behavior on failure is to retry the entire batch — repeatedly — until the batch expires or the retention window ends. For MCP event pipelines, a single malformed tool call event or a transient downstream API error can block all records behind it on the same shard for hours.

Two ESM configuration properties fix this:

With both BisectBatchOnFunctionError and ReportBatchItemFailures configured, your Lambda should explicitly report which records it could not process:

export const handler = async (event: KinesisStreamEvent): Promise<KinesisStreamBatchResponse> => {
  const failures: string[] = [];
  for (const record of event.Records) {
    try {
      const payload = JSON.parse(
        Buffer.from(record.kinesis.data, 'base64').toString('utf8')
      );
      await processToolCallEvent(payload);
    } catch (err) {
      // Report this specific sequence number as failed — ESM retries only this record
      failures.push(record.kinesis.sequenceNumber);
    }
  }
  return {
    batchItemFailures: failures.map(seq => ({ itemIdentifier: seq }))
  };
};

Returning an empty batchItemFailures array signals complete success. Returning records in the array signals partial failure — ESM retries only those records while advancing past the successfully processed ones. This combination means a single malformed record can be isolated, retried, and eventually routed to the DLQ without blocking any other event on the shard.

KCL checkpointing — the checkpoint-before-all-records-processed trap

For stateful consumers — typically Flink jobs or KCL workers processing large batches — checkpointing semantics determine what happens on failure. Kinesis Checkpointing marks the sequence number position in a shard, telling the consumer "I have processed all records up to this point." After a restart, the consumer replays from the last checkpoint.

The critical invariant: checkpoint only after all records in a batch have been successfully processed. Two common violations:

1. Checkpoint inside the processing loop — checkpointing after processing record 3 of 10 and then crashing means records 4–10 are permanently lost. The consumer restarts from record 4's position (which it already checkpointed past) and never sees them.
2. Checkpoint on shardEnded without advancing — when a shard is split or merged, the consumer must call checkpoint(SHARD_END) on the now-closed shard before it begins reading from the child shards. Missing this call means the consumer never discovers the child shards and the stream goes silent from its perspective.

The correct pattern: process all records in the batch, accumulate them into a local buffer or sink batch, commit the sink, then checkpoint. If the Lambda exits between committing the sink and checkpointing, the records get replayed — which is why you also need idempotency (covered next).

SubscribeToShard — the silent 5-minute expiry

Enhanced fan-out consumers use SubscribeToShard to establish a dedicated HTTP/2 push connection that gives each consumer 2 MB/s per shard independent of other consumers. The critical operational detail: the SubscribeToShard connection expires after exactly 5 minutes. When it expires, no error is emitted on the producer side — your stream keeps accepting records normally. The consumer simply stops receiving them.

The consumer must implement an explicit re-subscribe loop:

async function consumeShard(
  client: KinesisClient,
  streamArn: string,
  consumerArn: string,
  shardId: string
): Promise<void> {
  let startingPosition: StartingPosition = {
    Type: ShardIteratorType.TRIM_HORIZON
  };
  while (true) {
    try {
      const response = await client.send(new SubscribeToShardCommand({
        ConsumerARN: consumerArn,
        ShardId: shardId,
        StartingPosition: startingPosition
      }));
      for await (const event of response.EventStream!) {
        if (event.SubscribeToShardEvent) {
          const records = event.SubscribeToShardEvent.Records ?? [];
          if (records.length > 0) {
            await processRecords(records);
            // Resume from just after the last processed record on re-subscribe
            const lastSeq = records[records.length - 1].SequenceNumber!;
            startingPosition = {
              Type: ShardIteratorType.AFTER_SEQUENCE_NUMBER,
              SequenceNumber: lastSeq
            };
          }
        }
      }
    } catch (err: any) {
      if (err.name !== 'ResourceNotFoundException') {
        // Shard closed — advance to child shards
        break;
      }
      // Other errors or normal 5-min expiry: re-subscribe immediately
      await new Promise(r => setTimeout(r, 100)); // brief backoff
    }
  }
}

When the for-await loop exits cleanly (no error), it means the 5-minute window elapsed — re-subscribe immediately from the AFTER_SEQUENCE_NUMBER position of the last processed record. Do not add a long sleep between re-subscribes; the gap between expiry and re-subscription is a window where new records queue up without being consumed, increasing consumer lag.

Idempotent processing — DynamoDB sequence number guards

Kinesis provides at-least-once delivery guarantees. Any consumer using Lambda ESM retries, KCL failover, or re-subscriptions may re-deliver records that were already processed. For MCP event pipelines that write to downstream databases or invoke APIs, duplicate processing has real consequences (duplicate audit log entries, double-charged API calls, duplicate notifications).

The standard pattern is a DynamoDB conditional write using the Kinesis sequence number as the idempotency key:

async function processRecordIdempotent(
  record: Record,
  dynamodb: DynamoDBDocumentClient
): Promise<void> {
  const sequenceNumber = record.SequenceNumber!;
  const payload = JSON.parse(Buffer.from(record.Data as Uint8Array).toString('utf8'));
  try {
    // Write the processing result only if this sequence number hasn't been seen
    await dynamodb.send(new PutCommand({
      TableName: 'mcp-event-processed',
      Item: {
        pk: sequenceNumber,
        processedAt: Date.now(),
        result: await deriveResult(payload)
      },
      ConditionExpression: 'attribute_not_exists(pk)'
    }));
  } catch (err: any) {
    if (err.name === 'ConditionalCheckFailedException') {
      // Already processed — idempotent skip
      return;
    }
    throw err;
  }
}

The DynamoDB table needs a TTL attribute set to sequence expiry (approximately the stream's retention window) to prevent unbounded growth. Set the TTL to processedAt + retentionWindowSeconds — once a sequence number ages out of the stream, no consumer can re-deliver it, so the guard entry is no longer needed.

Consumer pattern selection guide

The right consumer choice depends on three axes: processing complexity, throughput, and fan-out requirements.

Kinesis Analytics and the KPU idle cost trap

When you need windowed metrics over your MCP tool call stream — rolling p99 latency per tool, anomaly detection on error rate spikes, per-tenant throughput summaries — Kinesis Data Analytics (Managed Apache Flink) provides these without building a state management layer from scratch.

The most expensive operational mistake with Managed Flink: the application bills $0.11/KPU-hour even when the stream has no records to process. "Stopping" an application in the console sets it to READY state — it still has state stored and still incurs storage costs. The only way to stop paying is to delete the application (losing checkpointed state) or use the Managed Flink autoscaling feature to scale down to 0 KPUs during off-hours.

For MCP server monitoring use cases, the two most useful Flink SQL patterns are tumbling windows for rate metrics and Random Cut Forest (RCF) for anomaly detection:

-- Tumbling window: tool call rate per 1-minute window per tool
SELECT
  TUMBLE_END(event_time, INTERVAL '1' MINUTE) AS window_end,
  tool_name,
  COUNT(*) AS call_count,
  AVG(duration_ms) AS avg_duration_ms,
  PERCENTILE_APPROX(duration_ms, 0.99) AS p99_duration_ms
FROM tool_call_events
GROUP BY
  TUMBLE(event_time, INTERVAL '1' MINUTE),
  tool_name;

The event time vs processing time distinction is important for MCP server metrics. Processing time windows fire on wall clock — they're simple and predictable. Event time windows fire when the Flink watermark advances past the window boundary, which depends on record arrival. If MCP tool call events arrive with latency (a queued Lambda sending them, for example), event time windows may fire 30–60 seconds after the wall clock window closes. Use processing time for dashboards and alerting; use event time for accurate historical analysis where late-arriving records matter.

Pattern 3: The Kinesis security stack — encryption, network isolation, and audit logging

MCP tool call events flowing through Kinesis may include session IDs, tool inputs, partial outputs, and user identifiers. They warrant the same security posture as your API layer: encryption at rest with customer-managed keys, network isolation to your VPC, and audit logging of data plane API calls.

KMS encryption — CMK vs aws/kinesis

Kinesis Data Streams supports two server-side encryption modes. Both encrypt records before writing to disk. The difference matters when you have cross-account consumers or compliance requirements for per-operation audit logs:

Use a CMK for any stream where:

• Consumer Lambdas or services run in a different AWS account than the stream
• Compliance requires a per-decrypt audit trail (SOC 2, HIPAA, PCI DSS)
• You need the ability to revoke access by disabling the key (aws/kinesis keys cannot be disabled)

The CMK key policy for a cross-account consumer pattern:

{
  "Sid": "AllowCrossAccountConsumer",
  "Effect": "Allow",
  "Principal": {
    "AWS": "arn:aws:iam::CONSUMER_ACCOUNT_ID:role/mcp-kinesis-consumer-role"
  },
  "Action": [
    "kms:Decrypt",
    "kms:DescribeKey"
  ],
  "Resource": "*"
}

VPC interface endpoints — the private DNS trap

The AWS documentation recommends VPC interface endpoints to keep Kinesis traffic within the AWS network and off the public internet. The trap: unlike S3 and DynamoDB (which use gateway endpoints with automatic route table injection), Kinesis uses interface endpoints. An interface endpoint without private DNS enabled is effectively unused — your Lambda resolves kinesis.us-east-1.amazonaws.com to the public IP even when the endpoint exists in the VPC, and all traffic still routes over the internet.

When creating the Kinesis VPC endpoint, --private-dns-enabled is the mandatory flag. With private DNS enabled, the Kinesis regional endpoint resolves to the endpoint's private IP inside your VPC. Without it, you've paid for an endpoint that is never used:

# Correct — with private DNS enabled
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-0abc1234 \
  --service-name com.amazonaws.us-east-1.kinesis-streams \
  --vpc-endpoint-type Interface \
  --subnet-ids subnet-0def5678 \
  --security-group-ids sg-0ghi9012 \
  --private-dns-enabled   # <-- required, not the default

# The security group must allow inbound 443 from your Lambda/ECS task security group
aws ec2 authorize-security-group-ingress \
  --group-id sg-0ghi9012 \
  --protocol tcp \
  --port 443 \
  --source-group sg-lambda-sg-id

Interface endpoints cost $0.01/endpoint-hour per Availability Zone. For a three-AZ VPC, that's ~$22/month per endpoint. Verify the endpoint is being used by checking the BytesProcessed CloudWatch metric on the endpoint — if it stays at zero after deploying your Lambda, the Lambda's security group is not allowing outbound 443 to the endpoint's security group, or private DNS is not enabled.

CloudTrail data events — the opt-in cost trap

CloudTrail management events (CreateStream, DeleteStream, AddTagsToStream) are logged by default at no additional cost. Data plane operations — PutRecord, PutRecords, GetRecords, GetShardIterator — are not logged by default. You must explicitly configure a CloudTrail data event selector for Kinesis, at $0.10 per 100,000 API calls.

For MCP server audit logging use cases, the cost can be substantial. A stream processing 10,000 tool call events per minute generates 14.4 million PutRecord API calls per day — roughly $14.40/day in CloudTrail data event costs. The practical approach:

• Enable data events only on streams carrying regulated data (PII, PHI, financial data) where per-operation audit trails are required by compliance
• Use read-only data event selectors for GetRecords if you only need consumer audit trails, not producer trails
• Prefer stream-level monitoring (CloudWatch metrics + Enhanced Monitoring per-shard metrics) for operational visibility; reserve CloudTrail data events for compliance-specific audit requirements

CDK snippet for selective data event logging on a specific stream:

new cloudtrail.Trail(this, 'KinesisAuditTrail', {
  trailName: 'kinesis-regulated-streams-trail',
  bucket: auditBucket,
  s3KeyPrefix: 'kinesis',
}).addEventSelector(
  cloudtrail.DataResourceType.KINESIS_STREAM,
  [regulatedStream.streamArn],  // scope to specific streams only
  { includeManagementEvents: false }  // management events already in default trail
);

IAM least privilege — producer vs consumer separation

MCP server deployments typically have distinct producer roles (the MCP server Lambda or ECS task that emits tool call events) and consumer roles (the processing Lambda, analytics job, or Firehose delivery stream). These should be separate IAM roles with separate minimal permission sets:

Note that Kinesis Data Streams does not support resource-based policies — you cannot attach a policy to the stream itself the way you can with an SQS queue or S3 bucket. Cross-account access always goes through IAM role assumption: the cross-account consumer assumes a role in the stream's account that has GetRecords permissions on the specific stream ARN.

Security configuration reference

Common failure modes across the Kinesis stack

Putting it together: a minimal MCP event streaming stack

Assembling the three patterns gives a minimal MCP event streaming deployment:

1. Data Stream — provisioned mode with ceil(peak_MB_s / 0.85) shards, CMK encryption, 7-day retention. Producer role has only PutRecords. Consumer roles have only GetRecords/SubscribeToShard. No resource-based policies on the stream.
2. Producer Lambda — uses putRecordsWithRetry pattern: call PutRecords, check FailedRecordCount, retry failed record positions by index. Partition key is sessionId for session-level ordering or random UUID for maximum throughput. Alarm on retries exceeding 1% of records.
3. Consumer Lambda — ESM with BisectBatchOnFunctionError: true, ReportBatchItemFailures, MaximumRetryAttempts: 3, and an SQS DLQ. Idempotency via DynamoDB conditional write on sequence number, with TTL set to stream retention + 1 hour.
4. Firehose delivery stream — direct-PUT from producer Lambda for archival path. Transformation Lambda encodes output as base64 JSON lines with newline delimiters. CloudWatch alarm on DeliveryToS3.DataFreshness > 900 seconds. EventBridge S3 notification on processing-failed/ prefix to SNS.
5. VPC endpoint — Interface endpoint for Kinesis streams with --private-dns-enabled. Security group allowing inbound 443 from Lambda security group.

That stack covers the three patterns: producer retry reliability (PutRecords partial-fail + partition key cardinality), consumer durability (bisect-on-error + checkpoint discipline + idempotency + re-subscribe loop), and security (CMK + VPC endpoint with private DNS + CloudTrail data events on regulated streams only).

TL;DR

• Pattern 1 (producer reliability): PutRecords HTTP 200 does not mean all records accepted — check FailedRecordCount and retry failed positions by index. Use sessionId or UUID as partition key; never use a constant or low-cardinality field. Provision at ceil(peak_MB_s / 0.85) shards; prefer on-demand below 137 GB/month per shard. Firehose transformation Lambdas must return base64-encoded records with \n delimiter or S3 output is silently garbled.
• Pattern 2 (consumer durability): Enable BisectBatchOnFunctionError + ReportBatchItemFailures + DLQ on all Lambda ESM consumers — otherwise one bad record blocks the entire shard. Checkpoint KCL workers only after the full batch is committed, not during the loop. Wrap SubscribeToShard in a re-subscribe loop that fires immediately on stream end — the 5-minute expiry is silent. Guard downstream writes with DynamoDB conditional write on sequence number for at-least-once delivery idempotency.
• Pattern 3 (security stack): Use CMK (not aws/kinesis) for cross-account consumers or compliance audit trails. Interface VPC endpoints require --private-dns-enabled or they are never used. Enable CloudTrail data events only on regulated streams — $0.10/100K API calls adds up fast on high-volume MCP event streams. Separate IAM roles for producers and consumers; no resource-based policies on streams — cross-account access via role assumption.