EventBridge Pipes for MCP Servers

Pipes vs EventBridge Rules — when to use each

EventBridge Rules and EventBridge Pipes both move events between AWS services but at different layers of the architecture.

EventBridge Rules sit on an event bus and match events that were published to that bus via PutEvents. They're the right choice when: multiple independent consumers subscribe to the same event type (fan-out), the event source already puts events on a bus (AWS service integrations like S3/EC2/ECS all emit to the default bus), or you need content-based routing to different targets based on event fields.

EventBridge Pipes poll a source directly and don't need an event bus at all. They're the right choice when: your source is already a stream (Kinesis) or queue (SQS) and putting events on a bus first is an extra hop; you want built-in batching from the source; you need one enrichment Lambda called for every batch before the target receives events; or you want to avoid the double-serialization of wrapping source payloads in an EventBridge envelope.

Pipe lifecycle states

A pipe transitions through states that you must understand to debug creation failures and operational issues. Unlike Lambda, a pipe doesn't "fail silently" on a bad invocation — it stops the source poller entirely when errors accumulate.

• CREATING — pipe resource is being provisioned; source polling has not started.
• RUNNING — pipe is polling the source and processing batches normally.
• UPDATING — a configuration change is being applied; source polling is paused during the update.
• STOPPING / STOPPED — you called StopPipe; source polling is paused. Use to pause a pipe without deleting it (useful for maintenance windows).
• CREATE_FAILED — the pipe could not be created. Most common cause: IAM execution role missing permissions for source, enrichment, or target. Check DescribePipe → StateReason field for the specific denied action.
• UPDATE_FAILED — a configuration update failed. The pipe continues running with its previous configuration.
• RUNNING_FAILED — the pipe was running but has encountered repeated enrichment or target failures that exceed the retry policy. Source polling stops. You must resolve the underlying issue and then call StartPipe to resume.

// CDK — check pipe state and state reason after creation
import { aws_pipes as pipes } from "aws-cdk-lib";

// CfnPipe is the L1 construct — no L2 as of 2026
const pipe = new pipes.CfnPipe(this, "McpEventPipe", {
  name: "mcp-tool-call-router",
  roleArn: pipeRole.roleArn,
  source: sqsQueue.queueArn,
  sourceParameters: {
    sqsQueueParameters: {
      batchSize: 10,
      maximumBatchingWindowInSeconds: 5,
    },
  },
  // Optional filter (see filtering guide)
  // Optional enrichment (see enrichment guide)
  target: targetLambda.functionArn,
  targetParameters: {
    lambdaFunctionParameters: {
      invocationType: "FIRE_AND_FORGET",  // or "REQUEST_RESPONSE"
    },
  },
});

// Post-deploy: check state via CLI
// aws pipes describe-pipe --name mcp-tool-call-router \
//   --query '{State: State, StateReason: StateReason}'

IAM execution role — grant permissions at every stage

A single IAM role must cover all three stages of the pipe: source read, enrichment invoke (if configured), and target write. Missing any one permission causes CREATE_FAILED and the StateReason message names the specific denied API.

import {
  Role, ServicePrincipal, PolicyStatement, Effect
} from "aws-cdk-lib/aws-iam";
import { Queue } from "aws-cdk-lib/aws-sqs";
import { Function } from "aws-cdk-lib/aws-lambda";
import { Table } from "aws-cdk-lib/aws-dynamodb";

// Pipe execution role — one role for all three stages
const pipeRole = new Role(this, "McpPipeRole", {
  assumedBy: new ServicePrincipal("pipes.amazonaws.com"),
  // Scope trust to this account/region to prevent confused deputy
  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)
pipeRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [
    "sqs:ReceiveMessage",
    "sqs:DeleteMessage",
    "sqs:GetQueueAttributes",
  ],
  resources: [sqsQueue.queueArn],
}));

// Stage 2 — enrichment invoke (Lambda)
pipeRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: ["lambda:InvokeFunction"],
  resources: [enrichmentLambda.functionArn],
}));

// Stage 3 — target write (DynamoDB)
pipeRole.addToPolicy(new PolicyStatement({
  effect: Effect.ALLOW,
  actions: [
    "dynamodb:PutItem",
    "dynamodb:UpdateItem",
  ],
  resources: [targetTable.tableArn],
}));

// If target is Lambda (invocationType: REQUEST_RESPONSE), use:
// pipeRole.addToPolicy(new PolicyStatement({
//   actions: ["lambda:InvokeFunction"],
//   resources: [targetLambda.functionArn],
// }));

Note the aws:SourceArn condition key in the trust policy. EventBridge Pipes uses a service role, which means the service assumes the role on your behalf. Without the source ARN condition, any pipe in your account — including one created by a compromised actor — could use this role. Scoping to a specific pipe ARN prevents that confused deputy attack.

Concurrent executions and throughput model

EventBridge Pipes does not spawn multiple concurrent invocations of the same pipe for the same source partition. The processing model is:

• SQS source: Pipes polls the queue with long polling. One batch is in-flight at a time per pipe. There is no parallelization factor setting (unlike Lambda ESM for SQS). To process multiple batches in parallel you need multiple pipes or Lambda ESM directly.
• Kinesis source: Pipes processes one batch per shard per pipe. The parallelizationFactor setting (1–10) controls how many concurrent batches a single shard can have in-flight. Setting it to 1 (default) means strictly serial within each shard.
• DynamoDB Streams source: Same as Kinesis — one batch per shard per pipe, with optional parallelizationFactor.

For SQS-sourced pipes where you need higher throughput, consider using Lambda ESM directly (which does scale horizontally across queue partitions) and reserving Pipes for patterns where the enrichment + target abstraction outweighs the throughput limitation.

Payload size limits

EventBridge Pipes enforces a 6 MB payload limit at each stage boundary:

• Source records → enrichment Lambda: the combined batch payload must be under 6 MB.
• Enrichment Lambda response → target: the enrichment output that Pipes passes to the target must be under 6 MB.

For MCP servers that log full tool call request/response bodies (which can include large base64-encoded images or file contents), this limit matters. The standard pattern is to store the raw payload in S3 at ingestion time and put only a reference object (S3 bucket + key + metadata) into SQS/Kinesis. The enrichment Lambda then fetches from S3 if it needs the full body, adds tenant metadata to the reference object, and passes the enriched reference to the target. The target writes the reference to DynamoDB; downstream consumers pull the full payload from S3 only when needed.

Full CDK pipe with SQS source, Lambda enrichment, and DynamoDB target

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, PolicyStatement } from "aws-cdk-lib/aws-iam";

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

    // Source queue
    const toolCallQueue = new Queue(this, "ToolCallQueue", {
      queueName: "mcp-tool-calls",
      visibilityTimeout: cdk.Duration.seconds(120),  // >= enrichment + target time
      deadLetterQueue: {
        queue: new Queue(this, "ToolCallDlq", { queueName: "mcp-tool-calls-dlq" }),
        maxReceiveCount: 3,
      },
    });

    // Enrichment Lambda
    const enrichFn = new Function(this, "EnrichFn", {
      runtime: Runtime.NODEJS_22_X,
      handler: "index.handler",
      code: Code.fromInline(`
        exports.handler = async (events) => {
          // events is an array of SQS messages (batch)
          // Must return array of same length — each element replaces the source record
          return events.map(evt => ({
            ...JSON.parse(evt.body),
            tenant: evt.messageAttributes?.tenantId?.stringValue ?? "default",
            enrichedAt: new Date().toISOString(),
          }));
        };
      `),
      timeout: cdk.Duration.seconds(25),  // Pipes hard cap is 29s
    });

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

    // Pipe IAM role
    const pipeRole = new Role(this, "PipeRole", {
      assumedBy: new ServicePrincipal("pipes.amazonaws.com"),
    });
    toolCallQueue.grantConsumeMessages(pipeRole);
    enrichFn.grantInvoke(pipeRole);
    eventTable.grantWriteData(pipeRole);

    // The pipe
    new pipes.CfnPipe(this, "McpEventPipe", {
      name: "mcp-tool-call-router",
      roleArn: pipeRole.roleArn,
      source: toolCallQueue.queueArn,
      sourceParameters: {
        sqsQueueParameters: {
          batchSize: 10,
          maximumBatchingWindowInSeconds: 5,
        },
        filterCriteria: {
          filters: [
            {
              // Only route events from MCP servers with eventType = tool_call
              pattern: JSON.stringify({
                body: { eventType: ["tool_call"] }
              }),
            },
          ],
        },
      },
      enrichment: enrichFn.functionArn,
      enrichmentParameters: {
        inputTemplate: "$.body",  // Pass only the SQS message body to enrichment
      },
      target: eventTable.tableArn,
      targetParameters: {
        dynamoDbParameters: {
          operation: "PUT_ITEM",
          // Input transformer maps enrichment output fields to DynamoDB attribute map
        },
      },
    });
  }
}

Common creation failures