Kinesis Data Firehose for MCP Servers

Firehose vs Kinesis Data Streams — choose the right service

Kinesis Data Streams and Kinesis Data Firehose are often confused because both have "Kinesis" in the name and both ingest streaming data. They solve different problems. Streams is a distributed log — you write records, multiple consumers pull them independently, and records persist for 24 hours to 365 days. Firehose is a managed delivery pipeline — you write records, Firehose buffers them and delivers them to a destination (S3, Redshift, etc.) once per buffer window. You cannot replay Firehose deliveries or have multiple independent consumers of a Firehose stream.

A common pattern for MCP server infrastructure is to combine both: Kinesis Data Streams as the event bus (multiple consumers — real-time alerting, analytics, enrichment) with a Firehose consumer on that stream for archival to S3. Firehose can use a Kinesis Data Stream as its source, consuming from the stream as one of its fan-out consumers.

Transformation Lambda — the base64 trap and newline delimiter

Firehose optionally invokes a Lambda before delivery to let you normalize, filter, or enrich records. The Lambda receives a batch of records as base64-encoded strings in the records array and must return them in the same structure with a result field (Ok, Dropped, or ProcessingFailed) and the transformed data as base64-encoded bytes.

Two silent failure modes: (1) if your Lambda returns the data as a plain string instead of base64, Firehose delivers garbage bytes to S3; (2) if you don't append \n to each record before base64-encoding, the S3 file is one long concatenated blob — SELECT * FROM athena_table returns a single row containing all records for that buffer window. Add the newline inside the Lambda before encoding.

// Lambda transformation for Kinesis Firehose
// Node.js 22.x — handler receives FirehoseTransformationEvent

import type {
  FirehoseTransformationEvent,
  FirehoseTransformationResult,
  FirehoseTransformationResultRecord,
} from "aws-lambda";

export async function handler(
  event: FirehoseTransformationEvent
): Promise<FirehoseTransformationResult> {
  const records: FirehoseTransformationResultRecord[] = event.records.map(
    (record) => {
      try {
        const raw = Buffer.from(record.data, "base64").toString("utf-8");
        const parsed = JSON.parse(raw);

        // Normalize: add ISO timestamp, tenant context, schema version
        const normalized = {
          schema_version: "1",
          tenant_id: parsed.tenantId ?? "unknown",
          tool_name: parsed.toolName ?? "unknown",
          session_id: parsed.sessionId ?? null,
          timestamp_utc: new Date(parsed.ts ?? Date.now()).toISOString(),
          duration_ms: parsed.durationMs ?? null,
          success: parsed.success ?? null,
          error_code: parsed.errorCode ?? null,
        };

        // CRITICAL: append \n before base64-encoding so Athena can parse newline-delimited JSON
        const output = JSON.stringify(normalized) + "\n";

        return {
          recordId: record.recordId,
          result: "Ok",
          data: Buffer.from(output).toString("base64"),
        };
      } catch {
        // ProcessingFailed routes record to error prefix — never silently drop
        return {
          recordId: record.recordId,
          result: "ProcessingFailed",
          data: record.data, // Return original data unchanged
        };
      }
    }
  );

  return { records };
}

Dynamic partitioning — splitting by tenant and tool at ingest

Without dynamic partitioning, all Firehose records land in a single S3 prefix — to query one tenant's events you must scan the entire dataset. Dynamic partitioning extracts a key from each record using a jq expression and routes records to separate S3 prefixes, making per-tenant or per-tool Athena queries scan only the relevant partition.

Dynamic partitioning adds a small per-GB cost ($0.02/GB) and requires each record to be valid JSON parseable by the jq expression. The jq expression must output a scalar string — nested objects or arrays fail silently and the record goes to the error prefix.

# CloudFormation / CDK L1 escape hatch — Firehose delivery stream with dynamic partitioning
# (CDK L2 construct does not expose dynamic partitioning as of 2026 — use CfnDeliveryStream)

Resources:
  McpEventDeliveryStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: mcp-tool-events
      DeliveryStreamType: KinesisStreamAsSource
      KinesisStreamSourceConfiguration:
        KinesisStreamARN: !GetAtt McpEventStream.Arn
        RoleARN: !GetAtt FirehoseRole.Arn
      ExtendedS3DestinationConfiguration:
        BucketARN: !GetAtt EventsBucket.Arn
        RoleARN: !GetAtt FirehoseRole.Arn
        # Dynamic partitioning prefix — Firehose substitutes !{partitionKeyFromQuery:...}
        Prefix: "tool-events/tenant=!{partitionKeyFromQuery:tenantId}/tool=!{partitionKeyFromQuery:toolName}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
        ErrorOutputPrefix: "tool-events-errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
        BufferingHints:
          IntervalInSeconds: 300  # 5 minutes — adjust down to 60s for near-real-time Athena
          SizeInMBs: 64           # Flush when buffer reaches 64 MB regardless of interval
        CompressionFormat: GZIP
        DynamicPartitioningConfiguration:
          Enabled: true
        ProcessingConfiguration:
          Enabled: true
          Processors:
            - Type: MetadataExtraction
              Parameters:
                - ParameterName: MetadataExtractionQuery
                  ParameterValue: "{tenantId: .tenant_id, toolName: .tool_name}"
                - ParameterName: JsonParsingEngine
                  ParameterValue: JQ-1.6
            - Type: Lambda
              Parameters:
                - ParameterName: LambdaArn
                  ParameterValue: !GetAtt TransformFunction.Arn
                - ParameterName: BufferSizeInMBs
                  ParameterValue: "3"
                - ParameterName: BufferIntervalInSeconds
                  ParameterValue: "60"

Buffer tuning — interval vs size

Firehose buffers records and flushes when either the interval or the size threshold is reached, whichever comes first. Setting a small interval reduces latency (records appear in S3 sooner) but increases S3 PUT costs because each flush is one S3 object. Setting a large interval reduces costs but delays Athena visibility.

At volumes below the size threshold (common in early-stage MCP servers), the interval always triggers first — meaning a 900-second buffer gives you 15-minute data freshness. For operational dashboards querying "what happened in the last 10 minutes," use a 60-second buffer interval. The S3 PUT cost difference between 60s and 300s at 1K records/min is roughly $0.0036/hour — not worth optimizing until you're at high scale.

Error record monitoring — the silent failure trap

Firehose never returns an error to the PutRecord or PutRecordBatch caller when transformation or delivery fails. Instead it routes failed records to the S3 error prefix you specify in ErrorOutputPrefix, and emits a CloudWatch metric. If you don't monitor this prefix, records disappear with no indication on the write side. Two error sources to monitor:

1. Transformation failures — your Lambda returned ProcessingFailed or the Lambda timed out. Records appear under ErrorOutputPrefix/processing-failed/.
2. Delivery failures — the S3 PUT failed (bucket policy, KMS key access, bucket deleted). Records appear under ErrorOutputPrefix/delivery-failed/.

# CloudWatch alarm on Firehose delivery failures (alert if any records fail delivery)
aws cloudwatch put-metric-alarm \
  --alarm-name mcp-firehose-delivery-failed \
  --namespace AWS/Firehose \
  --metric-name DeliveryToS3.DataFreshness \
  --dimensions Name=DeliveryStreamName,Value=mcp-tool-events \
  --statistic Maximum \
  --period 300 \
  --evaluation-periods 2 \
  --threshold 900 \
  --comparison-operator GreaterThanThreshold \
  --alarm-description "Firehose records are more than 15 minutes stale — likely delivery failures"

# S3 metric filter for error prefix objects (alarm if any appear)
aws s3api put-bucket-notification-configuration \
  --bucket mcp-events-bucket \
  --notification-configuration '{
    "EventBridgeConfiguration": {}
  }'
# Then EventBridge rule: source=aws.s3, detail-type=Object Created, prefix filter=tool-events-errors/

Direct PUT vs Kinesis stream source

Firehose accepts records two ways: via the Firehose PutRecord API directly, or by reading from a Kinesis Data Stream. If your MCP server already publishes to a Kinesis Data Stream for real-time consumers (alerting, dashboards), add Firehose as a stream consumer for archival — the stream fan-out is free and you avoid duplicating the PutRecord call. If you only need archival with no real-time consumers, direct PUT into Firehose is simpler and slightly cheaper (no shard provisioning overhead).

import {
  FirehoseClient,
  PutRecordBatchCommand,
} from "@aws-sdk/client-firehose";

const firehose = new FirehoseClient({ region: "us-east-1" });

// Direct PUT to Firehose (use when no real-time consumers need the event)
async function archiveToolCallBatch(
  events: Array<{ tenantId: string; toolName: string; [key: string]: unknown }>
): Promise<void> {
  // PutRecordBatch accepts up to 500 records or 4 MB per call
  const records = events.map((event) => ({
    Data: Buffer.from(JSON.stringify(event) + "\n"), // newline for S3 line parsing
  }));

  const response = await firehose.send(
    new PutRecordBatchCommand({
      DeliveryStreamName: "mcp-tool-events",
      Records: records,
    })
  );

  // Same pattern as Kinesis PutRecords — check FailedPutCount
  if ((response.FailedPutCount ?? 0) > 0) {
    const failed = response.RequestResponses?.filter((r) => r.ErrorCode);
    console.error(`${failed?.length} Firehose records failed:`, failed?.[0]?.ErrorMessage);
  }
}

Common failure modes reference