AWS EventBridge Pipes for MCP Servers: Three Pipeline Patterns for Decision Logic, Enrichment Contracts, and Source Reliability

The core mental model: Pipes is a pipeline, not a bus

Before the three patterns, it helps to be precise about where EventBridge Pipes sits relative to the other EventBridge primitives and Lambda event source mappings:

The key distinction: if you need fan-out (one event → multiple independent consumers), use EventBridge Rules on a bus. If you need a single target and want enrichment handled without writing a wrapper Lambda, use Pipes. The enrichment step is the unique value — it invokes a Lambda function for every batch of filtered source records, and the enrichment's response completely replaces the source payload before the target receives anything. Your target Lambda or DynamoDB table never sees the raw SQS envelope or Kinesis base64 blob; it sees the clean, enriched shape your enrichment function returned.

Pattern 1 — The Pipes-vs-Rules decision and pipe lifecycle failure modes

When Pipes is the right choice

The Pipes-vs-Rules decision reduces to three questions:

1. Is the source already a stream or queue? If your MCP server emits tool call events to SQS, DynamoDB, or Kinesis, Pipes can consume that source directly. Rules require events to be on a bus — adding a bus hop adds serialization overhead and requires you to wrap your SQS message in an EventBridge envelope.
2. Do you need enrichment before the target? The most common MCP server use case: raw tool call events land in SQS, the target is DynamoDB, but you need to add tenant plan tier, region, and computed keys before the write. With Rules you'd write a Lambda that receives the event, enriches it, and then writes to DynamoDB — the enrichment is invisible in your architecture. With Pipes the enrichment step is first-class: you configure it explicitly, it has its own IAM permissions, timeout, and failure handling.
3. Is one target sufficient? Pipes routes each batch to exactly one target. If you need to fan the same tool call event out to DynamoDB AND to an analytics Kinesis stream AND to an alerting SNS topic, Pipes is not the right tool — use an EventBridge bus with Rules and three separate targets.

For the most common MCP server event pipeline pattern — SQS queue → enrich with tenant data → write to DynamoDB — Pipes is the clean choice. The alternative (Lambda ESM + Lambda that calls DynamoDB) has the same runtime behavior but puts enrichment logic inside your processing Lambda rather than as a declared pipeline stage.

IAM execution role — three permissions required

The single most common cause of CREATE_FAILED is a pipe role that's missing permissions at one of the three stages. The pipe needs one role, and that role must cover all three:

// CDK: pipe execution role — all three stages on one role
import { Role, ServicePrincipal, PolicyStatement } from "aws-cdk-lib/aws-iam";

const pipeRole = new Role(this, "McpPipeRole", {
  assumedBy: new ServicePrincipal("pipes.amazonaws.com"),
  // Confused deputy mitigation — scope trust to this specific pipe
  conditions: {
    StringEquals: {
      "aws:SourceAccount": this.account,
      "aws:SourceArn": `arn:aws:pipes:${this.region}:${this.account}:pipe/mcp-tool-call-router`,
    },
  },
});

// Stage 1: source read (SQS)
toolCallQueue.grantConsumeMessages(pipeRole);
// Equivalent to: sqs:ReceiveMessage, sqs:DeleteMessage, sqs:GetQueueAttributes

// Stage 2: enrichment invoke (Lambda)
enrichmentFn.grantInvoke(pipeRole);
// Equivalent to: lambda:InvokeFunction

// Stage 3: target write (DynamoDB)
eventTable.grantWriteData(pipeRole);
// Equivalent to: dynamodb:PutItem, dynamodb:UpdateItem, dynamodb:BatchWriteItem

// If source queue is SSE-KMS encrypted, also add:
// pipeRole.addToPolicy(new PolicyStatement({
//   actions: ["kms:Decrypt", "kms:GenerateDataKey"],
//   resources: [queueKmsKey.keyArn],
// }));

The aws:SourceArn condition in the trust policy prevents confused deputy attacks: without it, any pipe in your account (including one created by a compromised principal) could assume this role. Scoping to the specific pipe ARN is a one-line addition that eliminates the risk.

Pipe lifecycle states — what CREATE_FAILED and RUNNING_FAILED actually mean

A pipe transitions through well-defined states, and the two failure states require different responses:

The critical distinction between CREATE_FAILED and RUNNING_FAILED: CREATE_FAILED means the pipe never started (fix the IAM role, the pipe itself may need to be deleted and recreated); RUNNING_FAILED means the pipe was working and then stopped due to repeated failures (fix the enrichment or target, then call start-pipe to resume from where it left off).

# Diagnose any pipe failure state:
aws pipes describe-pipe \
  --name mcp-tool-call-router \
  --query '{State: State, StateReason: StateReason}'

# Example CREATE_FAILED StateReason:
# "Access denied for action lambda:InvokeFunction on resource arn:aws:lambda:..."

# After fixing the IAM role, update the pipe to force re-evaluation:
aws pipes update-pipe \
  --name mcp-tool-call-router \
  --role-arn arn:aws:iam::123456789012:role/updated-pipe-role

# After fixing RUNNING_FAILED root cause, resume:
aws pipes start-pipe --name mcp-tool-call-router

Concurrent execution — Pipes is serial per source partition

One architectural property that surprises teams coming from Lambda ESM: EventBridge Pipes processes one batch at a time per pipe. For SQS sources, there is no parallelization factor — only one batch is in-flight at a time, regardless of queue depth. For Kinesis and DynamoDB Streams sources, there is a parallelizationFactor setting (1–10) that controls concurrent batches per shard, but within a single shard at factor 1, processing is strictly serial.

This matters for MCP server pipelines with bursty traffic: an enrichment Lambda that takes 5 seconds per batch and a batchSize of 10 can only process 2 batches per second from an SQS queue. If your MCP server generates 1,000 tool call events per second, a single SQS-sourced pipe is not the right architecture — use Lambda ESM directly, which does scale horizontally across concurrent pollers.

Pattern 2 — The filter-enrichment pipeline contract

Filter expressions: AND/OR logic and the JSON.stringify trap

Filter expressions evaluate source records before enrichment, dropping records that don't match. For MCP servers routing all tool call events through one SQS queue, a filter is the zero-cost way to route only tool_call events to a DynamoDB pipeline while silently discarding heartbeat and metrics events — no Lambda code required.

The logic rules for Pipes filter expressions follow the same convention as EventBridge event patterns on a bus, but teams consistently get them backwards:

• Multiple conditions within a single pattern object → AND. Both conditions must match. {"body": {"eventType": ["tool_call"], "priority": ["high"]}} only passes records where eventType is both tool_call AND priority is high.
• Multiple values in a field's array → OR for that field. {"body": {"eventType": ["tool_call", "tool_result"]}} passes records where eventType is tool_call OR tool_result.
• Multiple pattern objects in the filters list → OR across patterns. A record passes if it matches any pattern in the list.

// CDK: AND/OR filter logic — the correct pattern
sourceParameters: {
  sqsQueueParameters: { batchSize: 10 },
  filterCriteria: {
    filters: [
      // Pattern 1: (eventType=tool_call AND priority=high)
      {
        pattern: JSON.stringify({  // ← JSON.stringify is required — see below
          body: {
            eventType: ["tool_call"],
            priority: ["high"],
          },
        }),
      },
      // Pattern 2: eventType=tool_result (separate pattern = OR with Pattern 1)
      {
        pattern: JSON.stringify({
          body: { eventType: ["tool_result"] },
        }),
      },
    ],
    // Net effect: pass if (tool_call AND high-priority) OR (tool_result)
  },
},

The JSON.stringify requirement — the silent catch-all bug

In CDK and CloudFormation, the filter pattern must be passed as a JSON-encoded string — a string of JSON, not a JSON object. If you pass a plain JavaScript object as the pattern property, CDK serializes it as an empty object, and an empty pattern matches all records. Every event passes the filter. There is no error, no warning, and no CloudWatch alarm — the pipe processes all records, enrichment runs on all of them, and you have no idea filtering isn't working until you add a log line and check.

// WRONG: plain object — silently passes all records
filters: [{ pattern: { body: { eventType: ["tool_call"] } } }]

// CORRECT: JSON-encoded string
filters: [{ pattern: JSON.stringify({ body: { eventType: ["tool_call"] } }) }]

This is the most common filter bug in Pipes deployments. The fix is a one-word addition, but discovering it requires either a code review that checks the CDK output or noticing that event counts upstream and downstream of the filter are identical.

Nested JSON fields — object structure, not dot-notation

For tool call events with nested JSON bodies, filter patterns use nested object structure — not dot-notation string keys. The filter engine does not interpret "body.event.type" as a path expression:

// WRONG: dot-notation key — matches nothing
{ "body.event.type": ["tool_call"] }

// CORRECT: nested objects
{
  body: {
    event: {
      type: ["tool_call"],
      context: { tenant: { plan: ["team", "enterprise"] } },
    },
  },
}

SQS message attributes — set by producers using MessageAttributes in the SendMessage call — are accessible at a different path than the message body. Use messageAttributes.KEY.stringValue, not body.KEY. A multi-tenant MCP server that sets tenantId as a message attribute for routing can filter on it without parsing the body at all:

// Filter on SQS message attribute (not body field)
{
  pattern: JSON.stringify({
    messageAttributes: {
      tenantId: { stringValue: ["t-enterprise-01", "t-enterprise-02"] },
    },
  }),
}

Enrichment Lambda — the 29-second timeout wall

The enrichment step inserts a processing Lambda between filter and target. The Lambda receives the entire batch as an array and must return an array. Three constraints determine whether your enrichment design will hold under production load:

The 29-second Pipes-enforced timeout. EventBridge Pipes hard-caps enrichment Lambda invocations at 29 seconds, independent of your Lambda function's configured timeout. If your Lambda is configured for 5 minutes, Pipes still abandons the invocation at 29 seconds and retries the batch. The Lambda itself continues running in the background — this is the dangerous part. If your enrichment Lambda does a DynamoDB write as a side effect (not just enriching data but also writing to a table), that write may complete after Pipes has already retried the batch with a new Lambda invocation, leaving you with duplicate writes. Enrichment Lambda writes must be idempotent.

// Fast enrichment pattern — parallel batch lookup with self-imposed timeout
exports.handler = async (events) => {
  // 22s budget — well under the 29s Pipes-enforced limit
  const DEADLINE_MS = 22_000;
  const startAt = Date.now();

  // Batch DynamoDB lookup for all unique tenant IDs in this batch
  const tenantIds = [...new Set(events.map(e => JSON.parse(e.body).tenantId))];
  const tenantMap = await batchGetTenants(tenantIds);  // BatchGetItem

  return events.map((evt) => {
    if (Date.now() - startAt > DEADLINE_MS) {
      // Mark as timeout — don't throw; let Pipes retry would just repeat the issue
      return { ...JSON.parse(evt.body), enrichmentStatus: "timeout" };
    }
    const body = JSON.parse(evt.body);
    const tenant = tenantMap[body.tenantId] ?? { plan: "free" };
    return {
      pk: `TENANT#${body.tenantId}`,
      sk: `TOOL_CALL#${evt.messageId}`,
      toolName: body.toolName,
      sessionId: body.sessionId,
      durationMs: body.durationMs,
      tenantPlan: tenant.plan,
      enrichedAt: new Date().toISOString(),
    };
  });
};

// Anti-pattern: sequential N+1 DynamoDB calls — at 3ms/call, fails at ~100 events
// for (const evt of events) { const t = await dynamodb.getItem(...); }

The response-length contract. Pipes takes the enrichment Lambda's return value as the batch payload it sends to the target. If you return fewer items than you received, the missing records are silently dropped — they do not retry, they do not go to a DLQ, they simply disappear. This is the intended behavior when you want to use the enrichment Lambda as a dynamic filter (return an empty array to drop all records from this batch, or filter out individual records by returning a shorter array). But it means a bug in your enrichment Lambda — an unhandled exception on some records that causes you to return a shorter array — will cause silent data loss with no error signal.

// Enrichment as dynamic filter — intentional use of shorter return array
exports.handler = async (events) => {
  const tenantMap = await batchGetTenants(events);

  // Return only events for paid tenants — free-tier events are dropped silently
  // (not sent to DLQ — this is intentional filtering, not a failure)
  return events
    .map(evt => {
      const body = JSON.parse(evt.body);
      const plan = tenantMap[body.tenantId]?.plan;
      if (plan === "free") return null;  // drop this record
      return { ...body, tenantPlan: plan };
    })
    .filter(Boolean);

  // If all events are from free-tier tenants, this returns []
  // Pipes treats an empty array as "zero records for target" — no error
};

Enrichment failure is all-or-nothing per batch. Unlike Lambda ESM, which supports batchItemFailures for per-record failure reporting, Pipes enrichment has no per-record failure granularity. If enrichment throws, the entire batch fails. If enrichment times out, the entire batch retries. Design enrichment to be internally robust: handle individual record parse errors without throwing, and only throw when the entire batch is unrecoverable.

Input and output transformation — JSONPath unwrapping

Before passing source records to enrichment, and before passing enrichment output to the target, Pipes can apply a JSONPath-based input transformation. The most useful application for SQS-sourced MCP pipelines is unwrapping the SQS envelope so that enrichment receives the message body directly rather than the full SQS record:

// CDK: inputTemplate to unwrap SQS envelope before enrichment
enrichmentParameters: {
  // Without inputTemplate: enrichment receives the full SQS record object
  // {messageId, receiptHandle, body, attributes, messageAttributes, ...}

  // With inputTemplate "$.body": enrichment receives only the parsed body
  // {toolName, tenantId, sessionId, durationMs, ...}
  inputTemplate: "$.body",

  // Compose a custom object including fields from the envelope:
  // inputTemplate: JSON.stringify({
  //   "messageId": "<$.messageId>",
  //   "tenantId": "<$.messageAttributes.tenantId.stringValue>",
  //   "body": "<$.body>",
  // }),
  // Note: angle-bracket JSONPath syntax is for embedding values inside templates
},

The angle-bracket syntax ("<$.field>") extracts individual field values into a JSON template. A bare "$.body" without angle brackets extracts the entire body object as the enrichment input. These two syntaxes serve different purposes and mixing them up produces unexpected results.

Pattern 3 — Source-specific gotchas and the two-DLQ-layer reliability model

SQS source — visibility timeout sizing and FIFO limitation

The SQS source has one critical timing constraint and one permanent limitation.

Visibility timeout sizing. The SQS queue's visibilityTimeout must be greater than the total time Pipes needs to process a batch from that queue. If the visibility timeout expires before Pipes finishes, the message becomes visible again, re-enters the poll loop, and Pipes processes it a second time — creating a duplicate event downstream. The formula:

// Visibility timeout sizing formula:
// visibilityTimeout > batchWindow + enrichmentTimeout + targetTimeout + safetyBuffer

// Example MCP server pipeline:
// batchWindow = 5s (maximumBatchingWindowInSeconds)
// enrichmentTimeout = 25s (our enrichment Lambda safe budget)
// targetTimeout = 10s (DynamoDB batch write, generous estimate)
// safetyBuffer = 30s
// Total = 5 + 25 + 10 + 30 = 70s → set visibilityTimeout to 120s

const toolCallQueue = new Queue(this, "ToolCallQueue", {
  visibilityTimeout: cdk.Duration.seconds(120),
  deadLetterQueue: {
    queue: sourceDlq,
    maxReceiveCount: 3,
  },
});

FIFO limitation. SQS FIFO queues are not supported as Pipes sources. If your MCP server requires strict per-tenant ordering, the alternatives are: Lambda ESM with SQS FIFO (supports MessageGroupId ordering); standard SQS + Kinesis (Kinesis preserves strict shard-level ordering, use tenant ID as the partition key, then use Kinesis as the Pipes source); or standard SQS with idempotent DynamoDB conditional writes that handle occasional duplicates without requiring strict ordering.

DynamoDB Streams source — no AT_TIMESTAMP and unmarshall requirement

DynamoDB Streams as a Pipes source is the canonical change-data-capture pattern for MCP server state changes. When a session record in DynamoDB is modified, the stream record flows through Pipes to enrichment and then to a target — without polling, without Lambda trigger wiring, and with built-in batching.

Two things catch teams new to DynamoDB Streams sources:

No AT_TIMESTAMP starting position. Unlike Kinesis, DynamoDB Streams sources only support LATEST and TRIM_HORIZON. If you need to process changes from a specific historical time, you cannot set an ISO 8601 timestamp as the starting position — you can only start from now (LATEST) or from the oldest available change in the stream (TRIM_HORIZON, up to 24 hours back).

DynamoDB typed JSON requires unmarshalling. The NewImage and OldImage fields in DynamoDB stream records use DynamoDB's typed JSON format: {"S": "string-value"}, {"N": "123"}, {"BOOL": true}. Your enrichment Lambda must call unmarshall from @aws-sdk/util-dynamodb to get plain JavaScript objects — otherwise you're comparing {"S": "active"} to strings, which never matches.

// Enrichment Lambda for DynamoDB Streams source
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
    // Keys are always present; Images depend on the stream's StreamViewType
    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 (requires NEW_AND_OLD_IMAGES stream type)
    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,
    };
  });
};

Kinesis source — parallelization factor and ordering tradeoff

Kinesis is the highest-throughput source type for Pipes. The parallelizationFactor setting controls how many concurrent batches a single shard can have in-flight within this pipe:

For MCP server tool call events where each event is independent (audit logs, analytics ingestion), parallelizationFactor: 3 triples the per-shard throughput without correctness concerns. For events that represent state transitions on a session (session opened → tool called → session closed), you need strict ordering — leave parallelizationFactor: 1 and scale by increasing shard count instead.

The AT_TIMESTAMP starting position, available for Kinesis but not DynamoDB Streams, requires an ISO 8601 UTC timestamp string:

// CDK: Kinesis source with AT_TIMESTAMP starting position
sourceParameters: {
  kinesisStreamParameters: {
    startingPosition: "AT_TIMESTAMP",
    startingPositionTimestamp: "2026-09-01T00:00:00Z",  // ISO 8601 UTC — quotes required
    batchSize: 100,
    maximumBatchingWindowInSeconds: 5,
    maximumRetryAttempts: 3,
    maximumRecordAgeInSeconds: 3600,
    parallelizationFactor: 1,
    onPartialBatchItemFailure: "AUTOMATIC_BISECT",
    destinationConfig: {
      onFailure: { destination: pipeDlq.queueArn },
    },
  },
},

The two-DLQ-layer model

The error handling model for EventBridge Pipes involves two independent DLQ layers that catch different failure modes. Confusing them is the second most common operational mistake after incorrect IAM roles.

Layer 1 — SQS source queue DLQ catches failures at the queue level: messages that fail to be polled, messages that exceed maxReceiveCount due to visibility timeout cycling. This DLQ is configured on the SQS queue itself, independent of Pipes.

Layer 2 — Pipe DLQ catches failures during enrichment and target invocation: Lambda enrichment throws, enrichment times out, DynamoDB target write fails, retries exhausted. This DLQ is configured inside the pipe's source parameters (destinationConfig.onFailure) — and is only available for Kinesis and DynamoDB Streams sources, not SQS sources.

The architectural implication: for SQS-sourced pipes, the only dead-letter path for enrichment/target failures is the SQS queue's own DLQ via visibility timeout cycling. If visibility timeout is too short (see the sizing formula above), messages cycle back to the queue and get reprocessed before eventually hitting maxReceiveCount and landing in the SQS DLQ. If maxReceiveCount is too high, poison-pill messages spend a long time in the retry loop before dead-lettering. The right settings for SQS-sourced Pipes: visibility timeout comfortably above total processing time (so successful processing always completes before timeout), and maxReceiveCount of 3 (so poison pills dead-letter quickly).

// CDK: complete two-layer DLQ configuration for SQS-sourced pipe

// Layer 1: SQS source DLQ
const sourceDlq = new Queue(this, "SourceDlq", {
  retentionPeriod: cdk.Duration.days(14),
});
const toolCallQueue = new Queue(this, "ToolCallQueue", {
  visibilityTimeout: cdk.Duration.seconds(120),  // > batchWindow + enrichment + target + buffer
  deadLetterQueue: { queue: sourceDlq, maxReceiveCount: 3 },
});

// Layer 2: Pipe DLQ — for Kinesis/DynamoDB Streams sources only
// For SQS sources, enrichment/target failures cycle through visibility timeout → sourceDlq
const pipeDlq = new Queue(this, "PipeDlq", {
  retentionPeriod: cdk.Duration.days(14),
});

// Kinesis-sourced pipe with both retry config and pipe DLQ
const kinesisPipe = new pipes.CfnPipe(this, "KinesisPipe", {
  source: eventStream.streamArn,
  sourceParameters: {
    kinesisStreamParameters: {
      startingPosition: "LATEST",
      batchSize: 100,
      maximumRetryAttempts: 3,          // finite — don't leave at -1
      maximumRecordAgeInSeconds: 3600,   // drop stale records after 1h
      onPartialBatchItemFailure: "AUTOMATIC_BISECT",
      destinationConfig: {
        onFailure: { destination: pipeDlq.queueArn },  // pipe DLQ — layer 2
      },
    },
  },
  // ...
});

AUTOMATIC_BISECT — isolating poison-pill records

For Kinesis and DynamoDB Streams sources, onPartialBatchItemFailure: "AUTOMATIC_BISECT" is the mechanism for isolating individual failing records without blocking the entire shard. When a batch fails (enrichment throws, or the target throws for the entire batch), Pipes splits the batch in half and retries each half independently. This bisection continues until the failing record is alone in a batch of size 1, at which point it's sent to the pipe DLQ.

AUTOMATIC_BISECT is not available for SQS sources — for SQS, use Lambda ESM directly with the batchItemFailures response format if you need per-record failure isolation on an SQS queue.

Self-referential pipe loops — the EventBridge Bus target trap

When a Pipes target is an EventBridge Bus and there's a Rule on that bus that routes matching events back to the same SQS queue that is the pipe's source, you have an infinite loop:

// The loop:
// SQS → Pipe → EventBridge Bus → Rule → SQS → Pipe → EventBridge Bus → ...
// Each round multiplies: 10 messages → 100 → 1,000 → 10,000

// Prevention option 1: add a "processed" marker and filter it out
// Pipe filter: only process events where processedByPipe does NOT exist
const noLoopFilter = {
  body: {
    processedByPipe: [{ "exists": false }],  // only pass unprocessed events
  },
};

// Prevention option 2: use different event detail-type on the bus
// Enrichment adds detail-type = "McpToolCallProcessed"
// Rule only matches detail-type = "McpToolCallRaw"
// → processed events don't match the rule → no re-routing

// Prevention option 3: target a different downstream queue (safest)
// SQS-source → Pipe → Bus → Rule → DIFFERENT_SQS_QUEUE (not the pipe's source)

// Detection: CloudWatch alarm on ApproximateNumberOfMessagesVisible
// An exponential spike signature is the tell-tale sign of an event loop

Monitor for loops by setting a CloudWatch alarm at 10× the expected maximum queue depth on any SQS queue that is a Pipes source. An exponential spike in ApproximateNumberOfMessagesVisible is the characteristic signature of an event loop, and catching it early (before costs compound) requires the alarm to be pre-configured.

Full CDK pipe: SQS → filter → Lambda enrichment → DynamoDB

The complete pattern combining all three structural elements — correct IAM role, filter with JSON.stringify, enrichment with timeout budget, and SQS DLQ sizing:

import * as cdk from "aws-cdk-lib";
import { aws_pipes as pipes } from "aws-cdk-lib";
import { Queue } from "aws-cdk-lib/aws-sqs";
import { Function, Runtime, Code } from "aws-cdk-lib/aws-lambda";
import { Table, BillingMode, AttributeType } from "aws-cdk-lib/aws-dynamodb";
import { Role, ServicePrincipal } from "aws-cdk-lib/aws-iam";

export class McpEventPipelineStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Source queue — visibility timeout = batchWindow(5) + enrichment(25) + target(10) + buffer(30)
    const sourceDlq = new Queue(this, "SourceDlq", {
      retentionPeriod: cdk.Duration.days(14),
    });
    const toolCallQueue = new Queue(this, "ToolCallQueue", {
      visibilityTimeout: cdk.Duration.seconds(120),
      deadLetterQueue: { queue: sourceDlq, maxReceiveCount: 3 },
    });

    // Enrichment Lambda — timeout set to 25s (pipe cap is 29s)
    const enrichFn = new Function(this, "EnrichFn", {
      runtime: Runtime.NODEJS_22_X,
      handler: "index.handler",
      code: Code.fromAsset("lambda/enrich"),
      timeout: cdk.Duration.seconds(25),
    });

    // Target table
    const eventTable = new Table(this, "EventTable", {
      partitionKey: { name: "pk", type: AttributeType.STRING },
      sortKey: { name: "sk", type: AttributeType.STRING },
      billingMode: BillingMode.PAY_PER_REQUEST,
    });

    // Pipe role — all three stages
    const pipeRole = new Role(this, "PipeRole", {
      assumedBy: new ServicePrincipal("pipes.amazonaws.com"),
      // Confused deputy mitigation (add aws:SourceArn condition after first deploy when ARN is known)
    });
    toolCallQueue.grantConsumeMessages(pipeRole);
    enrichFn.grantInvoke(pipeRole);
    eventTable.grantWriteData(pipeRole);

    new pipes.CfnPipe(this, "McpEventPipe", {
      name: "mcp-tool-call-router",
      roleArn: pipeRole.roleArn,
      source: toolCallQueue.queueArn,
      sourceParameters: {
        sqsQueueParameters: {
          batchSize: 10,
          maximumBatchingWindowInSeconds: 5,
        },
        filterCriteria: {
          filters: [
            {
              // JSON.stringify required — plain object silently passes all records
              pattern: JSON.stringify({
                body: { eventType: ["tool_call"] },
              }),
            },
          ],
        },
      },
      enrichment: enrichFn.functionArn,
      enrichmentParameters: {
        inputTemplate: "$.body",  // unwrap SQS envelope — enrichment sees body directly
      },
      target: eventTable.tableArn,
      targetParameters: {
        dynamoDbParameters: {
          operation: "PUT_ITEM",
        },
      },
    });
  }
}

Debugging checklist

When a Pipes deployment isn't behaving as expected, work through this checklist before reaching for CloudWatch:

Where AliveMCP fits in

EventBridge Pipes is the integration layer — it moves events from a source to a target with enrichment. What it doesn't do is monitor whether that pipeline is running. Pipes doesn't page you when a pipe enters RUNNING_FAILED. It doesn't alert when queue depth spikes because enrichment started timing out. It doesn't catch when a code deploy broke your enrichment Lambda and all batches are now silently returning empty arrays.

AliveMCP monitors your MCP server endpoints — the upstream source of the events flowing through your Pipes pipeline. When your MCP server starts returning errors, failing health checks, or changing its tool schema without notice, AliveMCP catches it within 60 seconds. That's the signal your Pipes pipeline should be designed to handle: an upstream MCP server that's behaving unexpectedly is what generates the malformed events, the oversized payloads, and the enrichment exceptions that put your pipe in RUNNING_FAILED.

Knowing your MCP server is healthy before an event pipeline incident is cheaper than debugging a RUNNING_FAILED pipe at 2am. Join the waitlist to monitor your MCP endpoints.