eventbridge-pipes-bedrock-enrichment-cdk: AI enrichment via EventBridge Pipes

eventbridge-pipes-bedrock-enrichment-cdk/README.md

@@ -0,0 +1,99 @@
+# Amazon EventBridge Pipes with Amazon Bedrock AI Enrichment
+
+This pattern deploys an Amazon EventBridge Pipe that enriches messages in-flight using Amazon Bedrock before delivering them to the target. Messages from Amazon SQS are passed through an AWS Lambda enrichment function that calls Amazon Bedrock to classify sentiment, extract entities, and generate summaries — then writes the enriched data to Amazon DynamoDB.
+
+Important: This is the first serverless pattern combining Amazon EventBridge Pipes with Amazon Bedrock. While 48+ Pipes patterns exist in this repo, none use AI/ML enrichment. This pattern demonstrates how to add real-time AI processing to any event pipeline without changing source or target configurations.
+
+Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/eventbridge-pipes-bedrock-enrichment-cdk
+
+## Architecture
+
+```
+┌──────────────┐     ┌─────────────────────────────────────────────┐     ┌──────────────────┐
+│ Amazon SQS   │────▶│ Amazon EventBridge Pipe                      │────▶│ Amazon DynamoDB  │
+│ (Source)     │     │                                             │     │ (Enriched Data)  │
+└──────────────┘     │  ┌─────────────────────────────────────┐   │     └──────────────────┘
+                     │  │ AWS Lambda (Enrichment)              │   │
+                     │  │  → Amazon Bedrock (Claude Sonnet 4.6)│   │
+                     │  │  → Classify sentiment                │   │
+                     │  │  → Extract entities                  │   │
+                     │  │  → Generate summary                  │   │
+                     │  └─────────────────────────────────────┘   │
+                     └─────────────────────────────────────────────┘
+```
+
+**How it works:**
+
+1. Messages arrive in the Amazon SQS source queue (any format — customer feedback, support tickets, log entries)
+2. Amazon EventBridge Pipes reads the message and invokes the AWS Lambda enrichment function
+3. The enrichment function calls Amazon Bedrock (Claude Sonnet 4.6) to classify sentiment, extract named entities, and generate a one-line summary
+4. The enriched message (original + sentiment + entities + summary) is written to Amazon DynamoDB
+
+**Use cases:** Real-time sentiment analysis on customer feedback, automated ticket classification, log enrichment with AI context, content moderation pipelines.
+
+## Requirements
+
+- [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) installed and configured
+- [Node.js 20+](https://nodejs.org/) with npm
+- AWS account [bootstrapped for CDK](https://docs.aws.amazon.com/cdk/v2/guide/bootstrapping.html)
+- Amazon Bedrock model access enabled for Claude Sonnet 4.6
+- Python 3.12 (for AWS Lambda functions)
+
+## Deployment
+
+```bash
+cd eventbridge-pipes-bedrock-enrichment-cdk/cdk
+npm install
+npx cdk deploy
+```
+
+## Testing
+
+### Send a test message to Amazon SQS
+
+```bash
+QUEUE_URL=$(aws cloudformation describe-stacks \
+  --stack-name EventbridgePipesBedrockEnrichmentStack \
+  --query 'Stacks[0].Outputs[?OutputKey==`SourceQueueUrl`].OutputValue' \
+  --output text)
+
+aws sqs send-message \
+  --queue-url "$QUEUE_URL" \
+  --message-body '{"message": "I absolutely love the new feature you released! The AI suggestions save me hours every week. Your team is doing amazing work."}'
+```
+
+### Check enriched results in Amazon DynamoDB
+
+```bash
+TABLE_NAME=$(aws cloudformation describe-stacks \
+  --stack-name EventbridgePipesBedrockEnrichmentStack \
+  --query 'Stacks[0].Outputs[?OutputKey==`EnrichedTableName`].OutputValue' \
+  --output text)
+
+aws dynamodb scan --table-name "$TABLE_NAME" --query 'Items[0]'
+```
+
+Expected output includes: `sentiment: POSITIVE`, `entities: ["AI"]`, `summary: "Customer praising new AI feature..."`.
+
+## Cleanup
+
+> **Warning:** This will delete the Amazon DynamoDB table and all enriched data.
+
+```bash
+cd eventbridge-pipes-bedrock-enrichment-cdk/cdk
+npx cdk destroy
+```
+
+## Services Used
+
+| Service | Role |
+|---------|------|
+| Amazon SQS | Source queue — receives raw messages |
+| Amazon EventBridge Pipes | Orchestrates source → enrichment → target flow |
+| AWS Lambda | Enrichment step — calls Amazon Bedrock |
+| Amazon Bedrock | AI classification, entity extraction, summarization |
+| Amazon DynamoDB | Target — stores enriched messages with AI metadata |
+
+----
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0

eventbridge-pipes-bedrock-enrichment-cdk/cdk/.gitignore

@@ -0,0 +1,6 @@
+node_modules
+cdk.out
+cdk.context.json
+build
+*.js
+*.d.ts

eventbridge-pipes-bedrock-enrichment-cdk/cdk/bin/app.ts

@@ -0,0 +1,11 @@
+#!/usr/bin/env node
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT-0 (2026)
+import 'source-map-support/register';
+import * as cdk from 'aws-cdk-lib';
+import { EventbridgePipesBedrockEnrichmentStack } from '../lib/eventbridge-pipes-bedrock-enrichment-stack';
+
+const app = new cdk.App();
+new EventbridgePipesBedrockEnrichmentStack(app, 'EventbridgePipesBedrockEnrichmentStack', {
+  description: 'Amazon EventBridge Pipes with Amazon Bedrock AI enrichment (uksb-1tupboc57)',
+});

eventbridge-pipes-bedrock-enrichment-cdk/cdk/cdk.json

@@ -0,0 +1 @@
+{"app":"npx ts-node --prefer-ts-exts bin/app.ts"}

eventbridge-pipes-bedrock-enrichment-cdk/cdk/lib/eventbridge-pipes-bedrock-enrichment-stack.ts

@@ -0,0 +1,174 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT-0 (2026)
+
+import * as cdk from 'aws-cdk-lib';
+import * as sqs from 'aws-cdk-lib/aws-sqs';
+import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
+import * as lambda from 'aws-cdk-lib/aws-lambda';
+import * as iam from 'aws-cdk-lib/aws-iam';
+import * as pipes from 'aws-cdk-lib/aws-pipes';
+import * as logs from 'aws-cdk-lib/aws-logs';
+import { Construct } from 'constructs';
+import * as path from 'path';
+
+export class EventbridgePipesBedrockEnrichmentStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    const region = cdk.Stack.of(this).region;
+    const account = cdk.Stack.of(this).account;
+
+    // =========================================================
+    // 1. Amazon SQS Queue (Pipe source)
+    // =========================================================
+    const sourceQueue = new sqs.Queue(this, 'SourceQueue', {
+      queueName: 'pipes-bedrock-source',
+      visibilityTimeout: cdk.Duration.seconds(120),
+      encryption: sqs.QueueEncryption.SQS_MANAGED,
+    });
+
+    // =========================================================
+    // 2. Amazon DynamoDB Table (Pipe target — enriched messages)
+    // =========================================================
+    const enrichedTable = new dynamodb.Table(this, 'EnrichedTable', {
+      partitionKey: { name: 'messageId', type: dynamodb.AttributeType.STRING },
+      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+      timeToLiveAttribute: 'ttl',
+    });
+
+    // =========================================================
+    // 3. AWS Lambda: Bedrock Enrichment Function
+    //    Calls Amazon Bedrock to classify/summarize messages in-flight
+    // =========================================================
+    const enrichFn = new lambda.Function(this, 'EnrichFunction', {
+      runtime: lambda.Runtime.PYTHON_3_12,
+      handler: 'handler.lambda_handler',
+      code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/enrich')),
+      timeout: cdk.Duration.seconds(60),
+      memorySize: 256,
+      description: 'Enriches messages via Amazon Bedrock (classify sentiment + extract entities)',
+      environment: {
+        MODEL_ID: 'us.anthropic.claude-sonnet-4-6',
+      },
+    });
+
+    // Bedrock InvokeModel permission (scoped to inference profiles + foundation models)
+    enrichFn.addToRolePolicy(new iam.PolicyStatement({
+      effect: iam.Effect.ALLOW,
+      actions: ['bedrock:InvokeModel'],
+      resources: [
+        `arn:aws:bedrock:*::foundation-model/*`,
+        `arn:aws:bedrock:*:${account}:inference-profile/*`,
+      ],
+    }));
+
+    // =========================================================
+    // 4. IAM Role for Amazon EventBridge Pipes
+    // =========================================================
+    const pipeRole = new iam.Role(this, 'PipeRole', {
+      assumedBy: new iam.ServicePrincipal('pipes.amazonaws.com'),
+      description: 'Allows Amazon EventBridge Pipes to read SQS, invoke Lambda, write DynamoDB',
+      inlinePolicies: {
+        SourcePolicy: new iam.PolicyDocument({
+          statements: [new iam.PolicyStatement({
+            actions: [
+              'sqs:ReceiveMessage',
+              'sqs:DeleteMessage',
+              'sqs:GetQueueAttributes',
+            ],
+            resources: [sourceQueue.queueArn],
+          })],
+        }),
+        EnrichmentPolicy: new iam.PolicyDocument({
+          statements: [new iam.PolicyStatement({
+            actions: ['lambda:InvokeFunction'],
+            resources: [enrichFn.functionArn],
+          })],
+        }),
+        TargetPolicy: new iam.PolicyDocument({
+          statements: [new iam.PolicyStatement({
+            actions: ['dynamodb:PutItem'],
+            resources: [enrichedTable.tableArn],
+          })],
+        }),
+      },
+    });
+
+    // =========================================================
+    // 5. Amazon EventBridge Pipe
+    //    Source: SQS → Enrichment: Lambda (Bedrock) → Target: CloudWatch Logs
+    //    (Enriched data also stored in DynamoDB by enrichment Lambda)
+    // =========================================================
+
+    // Target: CloudWatch Logs (lightweight, always-available target for enriched output)
+    const targetLogGroup = new logs.LogGroup(this, 'EnrichedLogGroup', {
+      logGroupName: '/pipes/bedrock-enriched-output',
+      retention: logs.RetentionDays.ONE_WEEK,
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    // Update enrichment Lambda to also write to DynamoDB directly
+    enrichedTable.grantWriteData(enrichFn);
+    enrichFn.addEnvironment('ENRICHED_TABLE', enrichedTable.tableName);
+
+    // Pipe role needs CloudWatch Logs PutLogEvents for the target
+    pipeRole.addToPolicy(new iam.PolicyStatement({
+      actions: ['logs:CreateLogStream', 'logs:PutLogEvents'],
+      resources: [targetLogGroup.logGroupArn, `${targetLogGroup.logGroupArn}:*`],
+    }));
+
+    const pipe = new pipes.CfnPipe(this, 'BedrockEnrichmentPipe', {
+      name: 'sqs-bedrock-enrich-dynamodb',
+      description: 'Enriches Amazon SQS messages with Amazon Bedrock AI classification before writing to Amazon DynamoDB',
+      roleArn: pipeRole.roleArn,
+      source: sourceQueue.queueArn,
+      sourceParameters: {
+        sqsQueueParameters: {
+          batchSize: 1,
+          maximumBatchingWindowInSeconds: 5,
+        },
+      },
+      enrichment: enrichFn.functionArn,
+      target: targetLogGroup.logGroupArn,
+      targetParameters: {
+        cloudWatchLogsParameters: {
+          logStreamName: 'enriched-events',
+        },
+      },
+      logConfiguration: {
+        cloudwatchLogsLogDestination: {
+          logGroupArn: new logs.LogGroup(this, 'PipeLogGroup', {
+            logGroupName: '/aws/pipes/bedrock-enrichment',
+            retention: logs.RetentionDays.ONE_WEEK,
+            removalPolicy: cdk.RemovalPolicy.DESTROY,
+          }).logGroupArn,
+        },
+        level: 'ERROR',
+      },
+    });
+
+    // =========================================================
+    // Outputs
+    // =========================================================
+    new cdk.CfnOutput(this, 'SourceQueueUrl', {
+      value: sourceQueue.queueUrl,
+      description: 'Amazon SQS source queue URL — send messages here',
+    });
+
+    new cdk.CfnOutput(this, 'EnrichedTableName', {
+      value: enrichedTable.tableName,
+      description: 'Amazon DynamoDB table with enriched messages',
+    });
+
+    new cdk.CfnOutput(this, 'PipeName', {
+      value: 'sqs-bedrock-enrich-dynamodb',
+      description: 'Amazon EventBridge Pipe name',
+    });
+
+    new cdk.CfnOutput(this, 'EnrichFunctionName', {
+      value: enrichFn.functionName,
+      description: 'AWS Lambda enrichment function name',
+    });
+  }
+}

eventbridge-pipes-bedrock-enrichment-cdk/cdk/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "eventbridge-pipes-bedrock-enrichment-cdk",
+  "version": "1.0.0",
+  "bin": { "app": "bin/app.ts" },
+  "scripts": { "build": "tsc", "synth": "cdk synth" },
+  "dependencies": {
+    "aws-cdk-lib": "2.185.0",
+    "constructs": "^10.0.0",
+    "source-map-support": "^0.5.21"
+  },
+  "devDependencies": { "typescript": "~5.4.0", "ts-node": "^10.9.0" }
+}

eventbridge-pipes-bedrock-enrichment-cdk/cdk/tsconfig.json

@@ -0,0 +1 @@
+{"compilerOptions":{"target":"ES2020","module":"commonjs","lib":["es2020"],"declaration":true,"strict":true,"noImplicitAny":true,"strictNullChecks":true,"noImplicitReturns":true,"inlineSourceMap":true,"inlineSources":true,"experimentalDecorators":true,"strictPropertyInitialization":false,"outDir":"./build","rootDir":"."},"exclude":["node_modules","build"]}

eventbridge-pipes-bedrock-enrichment-cdk/example-pattern.json

@@ -0,0 +1,69 @@
+{
+  "title": "Amazon EventBridge Pipes with Amazon Bedrock AI Enrichment",
+  "description": "Enrich messages in-flight with AI classification and entity extraction using Amazon EventBridge Pipes and Amazon Bedrock",
+  "language": "TypeScript",
+  "level": "300",
+  "framework": "AWS CDK",
+  "introBox": {
+    "headline": "How it works",
+    "text": [
+      "Messages arrive in an Amazon SQS queue and are read by Amazon EventBridge Pipes.",
+      "The Pipe invokes an AWS Lambda enrichment function that calls Amazon Bedrock (Claude Sonnet 4.6) to classify sentiment, extract entities, and summarize the message.",
+      "The enriched message is written to Amazon DynamoDB with AI-generated metadata for downstream analytics."
+    ]
+  },
+  "gitHub": {
+    "template": {
+      "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/eventbridge-pipes-bedrock-enrichment-cdk",
+      "templateURL": "serverless-patterns/eventbridge-pipes-bedrock-enrichment-cdk",
+      "projectFolder": "eventbridge-pipes-bedrock-enrichment-cdk",
+      "templateFile": "cdk/lib/eventbridge-pipes-bedrock-enrichment-stack.ts"
+    }
+  },
+  "resources": {
+    "bullets": [
+      {
+        "text": "Amazon EventBridge Pipes enrichment",
+        "link": "https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-enrichment.html"
+      },
+      {
+        "text": "Amazon Bedrock InvokeModel API",
+        "link": "https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModel.html"
+      }
+    ]
+  },
+  "deploy": {
+    "text": [
+      "<code>cd eventbridge-pipes-bedrock-enrichment-cdk/cdk</code>",
+      "<code>npm install</code>",
+      "<code>npx cdk deploy</code>"
+    ]
+  },
+  "testing": {
+    "text": [
+      "Send a message to the Amazon SQS source queue.",
+      "Wait a few seconds for the Pipe to process.",
+      "Query the Amazon DynamoDB table to see the enriched result with sentiment, entities, and summary."
+    ]
+  },
+  "cleanup": {
+    "text": ["<code>npx cdk destroy</code>"]
+  },
+  "authors": [
+    {
+      "name": "Nithin Chandran R",
+      "image": "",
+      "bio": "Technical Account Manager at AWS",
+      "linkedin": "nithin-chandran-r",
+      "twitter": ""
+    }
+  ],
+  "patternArch": {
+    "icon1": { "x": 10, "y": 50, "service": "sqs", "label": "Amazon SQS" },
+    "icon2": { "x": 35, "y": 50, "service": "eventbridge", "label": "Amazon EventBridge Pipes" },
+    "icon3": { "x": 55, "y": 50, "service": "lambda", "label": "AWS Lambda" },
+    "icon4": { "x": 75, "y": 50, "service": "bedrock", "label": "Amazon Bedrock" },
+    "icon5": { "x": 95, "y": 50, "service": "dynamodb", "label": "Amazon DynamoDB" }
+  },
+  "patternType": "Serverless"
+}