EventBridge Pipes Enrichment

Lambda enrichment — input shape and response contract

When Pipes invokes a Lambda enrichment function, it sends an array of source records as the event payload. The exact shape depends on the source type (SQS records, Kinesis records, or DynamoDB stream records). The enrichment Lambda must return an array — its response is what EventBridge Pipes passes to the target, completely replacing the original source records.

// Lambda enrichment for SQS-sourced pipe
// Event = array of SQS records (filtered by Pipes before reaching enrichment)
exports.handler = async (events: SQSRecord[]): Promise => {
  // Batch lookup for tenant metadata — avoid N+1 DynamoDB calls
  const tenantIds = [...new Set(events.map(e => {
    const body = JSON.parse(e.body);
    return body.tenantId;
  }))];

  const tenantMap = await batchGetTenants(tenantIds);  // BatchGetItem

  const enriched = events.map(evt => {
    const body = JSON.parse(evt.body);
    const tenant = tenantMap[body.tenantId] ?? { plan: "free", region: "us-east-1" };

    // The return value for this record is what the target receives
    // Return shape must match your target's expected input schema
    return {
      // Preserve original fields
      eventId: evt.messageId,
      toolName: body.toolName,
      sessionId: body.sessionId,
      durationMs: body.durationMs,
      // Enriched fields
      tenantId: body.tenantId,
      tenantPlan: tenant.plan,
      tenantRegion: tenant.region,
      enrichedAt: new Date().toISOString(),
      // Computed fields
      pk: `TENANT#${body.tenantId}`,
      sk: `TOOL_CALL#${evt.messageId}`,
    };
  });

  return enriched;  // Must be same length as input — no per-record drops
};

// To drop specific records, return null for them and filter:
// const result = events.map(e => shouldProcess(e) ? enrich(e) : null);
// return result.filter(Boolean);
// Warning: this changes the count — downstream target receives fewer items

Input and output transformation (InputTemplate / OutputTemplate)

Before Pipes passes source records to the enrichment function, and before it passes enrichment output to the target, you can apply a JSONPath-based input transformation. This is useful for unwrapping the SQS envelope (passing only $.body to enrichment) or reshaping the enrichment output to match a specific target schema.

// CDK: use inputTemplate to unwrap SQS body before enrichment
const pipe = new pipes.CfnPipe(this, "EnrichedPipe", {
  // ...
  enrichment: enrichFn.functionArn,
  enrichmentParameters: {
    // Pass only the parsed SQS message body to enrichment Lambda
    // enrichment receives: { toolName: "search", tenantId: "t-99", ... }
    // instead of the full SQS record envelope
    inputTemplate: "$.body",

    // Alternative: compose a custom object from multiple paths
    // inputTemplate: JSON.stringify({
    //   "messageId": "<$.messageId>",
    //   "body": "<$.body>",
    //   "tenantId": "<$.messageAttributes.tenantId.stringValue>"
    // }),
  },
  // Target input transformation — reshape enrichment output for DynamoDB
  targetParameters: {
    dynamoDbParameters: {
      operation: "PUT_ITEM",
      // input transformer can map enriched fields to DynamoDB attribute format
    },
  },
});

When using inputTemplate, template variables use angle-bracket JSONPath syntax: "<$.field.path>". A bare "$.body" (without angle brackets) extracts the entire body object as the sole event passed to enrichment. The angle-bracket syntax is used when embedding field values inside a JSON template string.

Enrichment timeout — the 29-second wall

EventBridge Pipes hard-codes a 29-second timeout for enrichment Lambda invocations. This is independent of the Lambda function's own configured timeout:

• If your Lambda function is configured with a 5-minute timeout, Pipes still waits only 29 seconds and then treats the invocation as a failure.
• The Lambda function may still continue executing after Pipes gives up — this can lead to partial work with no Pipes awareness. Use idempotent writes in enrichment (conditional DynamoDB puts, deduplication keys) to handle this.
• Pipes does not send a cancellation signal to the Lambda when it times out — the Lambda invocation continues until it finishes or until its own timeout, but its response is ignored by Pipes.

The practical implication: enrichment must complete in under 25 seconds to have a safe margin. Design your enrichment Lambda for fast batch processing:

// Fast enrichment pattern: parallel batch lookup with timeout guard
exports.handler = async (events) => {
  const ENRICHMENT_TIMEOUT_MS = 20_000;  // 20s budget — well under 29s Pipes limit
  const startAt = Date.now();

  // Use Promise.all for parallel DynamoDB batch lookup instead of sequential
  const enriched = await Promise.all(events.map(async (evt) => {
    if (Date.now() - startAt > ENRICHMENT_TIMEOUT_MS) {
      // Return partial enrichment — mark as timeout rather than fail
      return { ...JSON.parse(evt.body), enrichmentStatus: "timeout" };
    }
    return await enrichSingleEvent(evt);
  }));

  return enriched;
};

// Anti-pattern: sequential N+1 DynamoDB gets — each adds 1-5ms, fails at ~200 events
const sequential = async (events) => {
  const results = [];
  for (const evt of events) {
    const tenant = await dynamodb.getItem(/* ... */).promise();  // sequential!
    results.push({ ...JSON.parse(evt.body), tenant });
  }
  return results;
};

API Gateway enrichment

Instead of a Lambda, you can configure an API Gateway REST or HTTP API endpoint as the enrichment target. Pipes sends an HTTP POST with the batch payload and uses the HTTP response body as the enriched output passed to the target.

// CDK: API Gateway enrichment
import { RestApi, LambdaIntegration } from "aws-cdk-lib/aws-apigateway";

const enrichApi = new RestApi(this, "EnrichApi");
const enrichResource = enrichApi.root.addResource("enrich");
enrichResource.addMethod("POST", new LambdaIntegration(enrichBackendFn));

const pipe = new pipes.CfnPipe(this, "ApiEnrichedPipe", {
  // ...
  enrichment: enrichResource.url,  // API Gateway URL
  enrichmentParameters: {
    httpParameters: {
      pathParameterValues: [],  // URL path variables (not needed for POST)
      headerParameters: {
        "X-Pipe-Source": "mcp-tool-calls",
        "Content-Type": "application/json",
      },
      queryStringParameters: {
        "env": "prod",
      },
    },
    inputTemplate: "$.body",  // POST body = source record body
  },
  // ...
});

API Gateway enrichment enforces a 5-second timeout (not 29 seconds like Lambda). This makes it suitable only for lightweight lookups or pass-through enrichment. The API response body must be a valid JSON array matching the enrichment response contract. Use Lambda enrichment for anything that requires more than trivial transformation time.

Step Functions enrichment — Express Workflows only

EventBridge Pipes supports Step Functions as an enrichment target, but with a critical constraint: only Express Workflows are supported, not Standard Workflows. Express Workflows run synchronously (via StartSyncExecution), which Pipes waits for. Standard Workflows are asynchronous (you start them and poll for completion) — Pipes has no mechanism to await asynchronous executions.

// CDK: Step Functions Express Workflow enrichment
import { StateMachine, StateMachineType } from "aws-cdk-lib/aws-stepfunctions";

const enrichmentSfn = new StateMachine(this, "EnrichSfn", {
  stateMachineType: StateMachineType.EXPRESS,  // REQUIRED — Standard fails silently
  definition: /* your enrichment workflow */,
  timeout: cdk.Duration.seconds(25),  // Keep under 29s Pipes limit
});

const pipe = new pipes.CfnPipe(this, "SfnEnrichedPipe", {
  // ...
  enrichment: enrichmentSfn.stateMachineArn,
  // No enrichmentParameters needed for Step Functions — uses default input
  // ...
});

// Grant pipe role execution permission
enrichmentSfn.grantStartSyncExecution(pipeRole);

Express Workflow enrichment is useful when your enrichment logic involves complex branching, parallel lookups to multiple services, or you want the execution history for debugging. The execution output (the state machine's final state output) becomes the enriched payload passed to the target. The output must be a valid JSON array with the same structure as Lambda enrichment responses.

Enrichment failure semantics

If enrichment fails (Lambda throws, returns non-2xx, times out after 29s, or returns invalid JSON), EventBridge Pipes retries the entire batch according to the pipe's retry policy. There is no partial batch failure granularity for enrichment — unlike Lambda ESM which supports batchItemFailures to isolate individual failing records.