In this article, we use the AWS CDK for Python to build a high-performance architecture powered by AWS AppConfig, API Gateway, and Lambda. You will learn how to leverage these AWS services to safely update advanced application configurations and operational parameters live in production, with zero cold starts, and without ever redeploying code.

Prerequisites

• AWS account
• AWS CLI setup and configured with appropriate credentials
• AWS CDK set up.
• Python3

Case Study

Imagine managing a high-volume GenAI orchestrator that stores operational parameters like token limits and active model strings as static Lambda environment variables. During a viral traffic spike, a sudden surge of massive prompts can skyrocket costs and trigger upstream 429 rate-limit errors. Fixing this by modifying variables or waiting over 10 minutes for a CI/CD pipeline to redeploy code causes severe downtime, while basic boolean feature flags lack the data complexity needed to tweak these parameters on the fly.

Your proposed solution is:

To gain real-time control without code deployments, you move all dynamic metrics into AWS AppConfig Free-form configurations, which act as a serverless switchboard delivering structured JSON payloads.

You propose attaching the AWS AppConfig Agent Lambda Extension via the Python CDK to the Lambda function; this way, the function gains an isolated local server on port 2772 that caches configurations automatically. Lambda fetches active rules instantly through a localhost HTTP request, bypassing slow, expensive AWS SDK calls entirely.

When a cost or rate-limiting emergency occurs, you simply update the remote AppConfig JSON profile to reduce token caps and swap to cheaper fallback models. The background sidecar pulls these updates automatically, allowing warm Lambda functions to inherit strict new guardrails in under two seconds. This creates a live feedback loop that lets your application layer safely self-correct mid-execution with zero cold starts, zero downtime, and complete structural isolation.

Proposed Architecture

We can see that:

1. The User invokes the Lambda function via the API Gateway
2. The Lambda function uses the AWS AppConfig Agent, which is attached as a Lambda layer, to locally pull the latest cache from AWS AppConfig. We don't need to worry about the internals, as that is fully handled by the layer.

Why the need for this Lambda Layer Sidecar approach?

Storing configurations inside Lambda code requires redeployments, while querying remote sources like DynamoDB or directly calling AppConfig introduces high network latency.

The sidecar layer acts as an automated local cache. When your code requests an endpoint, the agent instantly serves the data with zero network overhead. If the configuration expires, the agent asynchronously fetches updates from AppConfig in the background. Because calls are handled locally, you can poll the cache as often as needed without triggering external API limits or slowing down your execution thread.

What is AWS AppConfig?

AWS AppConfig is a capability of AWS Systems Manager that enables developers to safely modify application behavior in production without redeploying code. By utilizing feature flags and dynamic configurations, you can roll out features gradually, monitor performance, and instantly update parameters like rate limits or logging levels to mitigate live operational issues.

To manage this workflow, AppConfig uses structural containers. An Application serves as your central namespace, housing Environments (e.g., Dev, Production) and Configuration Profiles (defining feature flags or free-form JSON sourced from S3, Secrets Manager, or the AppConfig-hosted store).

Deployments are governed by a Deployment Strategy, which controls the rollout pace and bake time. Integrated Monitors watch Amazon CloudWatch alarms, triggering automatic rollbacks if issues arise. To eliminate runtime latency, the AWS AppConfig Agent runs alongside your application, asynchronously fetching and caching configuration data locally.

Feature Flags vs Freeform Configuration

There are two configuration options to choose from when using AppConfig:

We would go with the Freeform configuration approach and use the AWS AppConfig hosted configuration store, so we don't need to create multiple feature flags and instead have a single file/config that contains the features we want to track.

AWS CDK

The AWS Cloud Development Kit (AWS CDK) is an open-source software development framework for defining cloud resources as code.

Follow the prompt to complete the setup process.

1. Install AWS CDK CLI:

npm install -g aws-cdk

2. Set up environment:

cd infrastructure
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

3. Deploy infrastructure:

cdk bootstrap # First time only

# Deploy stack
cdk deploy AppStack --require-approval never

Project Overview

Stack:

cdk-app-config/
├── app.py                  # CDK Main App Entrypoint
├── app_stack.py            # Main Stack
├── cdk_constructs/
│   ├── __init__.py
│   ├── auth_construct.py
│   └── appconfig_construct.py
├── lambda/
│   └── index.py            # Standalone Lambda source folder
└── config/
    └── config.json   # AppConfig JSON configuration file

The folder structure is designed as follows:

1. The app.py acts as the entry point.
2. The app_stack.py is the main stack; it utilizes constructs to create and manage resources.
3. The cdk_constructs folder contains the constructs, both auth_construct (which contains the API Gateway and Lambda) and the appconfig_construct (which contains the App Config resources), which the application needs.
4. The config/config.py is the config file that would be uploaded as an AppConfig FreeForm configuration.
5. The lambda/index.py is the source code of the Lambda function; it contains the business logic and interaction with AWS AppConfig.

We will review some parts of the codebase, not all, however.

App Config Construct

import os
from aws_cdk import (
    aws_appconfig as appconfig
)
from constructs import Construct

class AppConfigConstruct(Construct):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        self.application = appconfig.CfnApplication(
            self, "AiGatewayApp",
            name="AiGateway"
        )

        self.environment = appconfig.CfnEnvironment(
            self, "ProdEnvironment",
            application_id=self.application.ref,
            name="Prod"
        )

        self.configuration_profile = appconfig.CfnConfigurationProfile(
            self, "ModelRouterProfile",
            application_id=self.application.ref,
            name="ModelRouter",
            location_uri="hosted"
        )

        base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        config_path = os.path.join(base_dir, "config", "config.json")
        
        with open(config_path, "r", encoding="utf-8") as file:
            config_content = file.read()

        hosted_version = appconfig.CfnHostedConfigurationVersion(
            self, "InitialConfigVersion",
            application_id=self.application.ref,
            configuration_profile_id=self.configuration_profile.ref,
            content_type="application/json",
            content=config_content
        )

        appconfig.CfnDeployment(
            self, "InitialDeployment",
            application_id=self.application.ref,
            environment_id=self.environment.ref,
            configuration_profile_id=self.configuration_profile.ref,
            configuration_version=hosted_version.ref,
            deployment_strategy_id="AppConfig.AllAtOnce"  # Immediate rollout for testing
        )

From the above, we set up the App Config application (AiGateway) and an environment (Prod), and also the configuration profile (ModelRouter) as a Freeform type, which uses the AppConfig hosted configuration store as its store.

Next, we upload the content of config/config.json and lastly create a deployment with a deployment strategy of AppConfig.AllAtOnce.

Auth Construct

import os
from aws_cdk import (
    Duration,
    CfnOutput,
    Stack,
    aws_lambda as _lambda,
    aws_apigateway as apigateway,
    aws_iam as iam,
    aws_appconfig as appconfig
)
from constructs import Construct

class AuthConstruct(Construct):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        current_region = Stack.of(self).region

        app_config_layer_arn = appconfig.Application.get_lambda_layer_version_arn(
            region=current_region,
            platform=appconfig.Platform.X86_64
        )

        app_config_extension_layer = _lambda.LayerVersion.from_layer_version_arn(
            self, "AppConfigExtension",
            layer_version_arn=app_config_layer_arn
        )

        base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        lambda_path = os.path.join(base_dir, "lambda")

        gateway_lambda = _lambda.Function(
            self, "AiGatewayFunction",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="index.lambda_handler",
            code=_lambda.Code.from_asset(lambda_path),
            layers=[app_config_extension_layer],
            memory_size=256,
            timeout=Duration.seconds(10)
        )

        gateway_lambda.add_to_role_policy(
            iam.PolicyStatement(
                actions=[
                    "appconfig:GetLatestConfiguration",
                    "appconfig:StartConfigurationSession"
                ],
                resources=["*"]
            )
        )

        api = apigateway.RestApi(
            self, "AiRouterApi",
            rest_api_name="Smart AI Gateway Service",
            description="Dynamically manages LLM endpoints utilizing AppConfig presets.",
            default_cors_preflight_options=apigateway.CorsOptions(
                allow_origins=apigateway.Cors.ALL_ORIGINS,
                allow_methods=apigateway.Cors.ALL_METHODS
            )
        )

        prompt_resource = api.root.add_resource("prompt")
        prompt_resource.add_method(
            "POST", 
            apigateway.LambdaIntegration(gateway_lambda),
            api_key_required=True
        )

        key = api.add_api_key(
            "DevApiKey",
            api_key_name="ai-gateway-dev-key"
        )
        plan = api.add_usage_plan(
            "UsagePlan",
            name="AiGatewayBasicPlan"
        )
        plan.add_api_stage(stage=api.deployment_stage)
        plan.add_api_key(key)

        CfnOutput(self, "ApiUrl", value=f"{api.url}prompt")

This sets up the Lambda with its source code and layer(uses the AWS AppConfig Agent as its layer). The Lambda layer sets up the AWS agent, which acts as a sidecar and caches the app config data that the Lambda would need.

Since this runs as a sidecar alongside the runtime code, it can then be accessed locally from within the function’s execution environment by querying it via http://localhost:2772/...

Since the Agent acts on behalf of the function, the function’s IAM execution role must be explicitly granted permissions for appconfig:StartConfigurationSession and appconfig:GetLatestConfiguration.

Finally, it sets up the prompt route on API Gateway with an API Key that points to the Lambda function and outputs the endpoint.

Config.json

{
    "active_llm_model": "gpt-4o",
    "enable_fallback_to_cheap_model": true,
    "max_input_tokens_allowed": 20,
    "system_prompt_version": "v2.1"
}

This contains the JSON payload that the Lambda function would be expecting. During testing, we would update this code and verify that the lambda function reacts differently and responds accordingly.

Lambda Function

import json
import urllib.request

def lambda_handler(event, context):
    url = "http://localhost:2772/applications/AiGateway/environments/Prod/configurations/ModelRouter"
    
    try:
        response = urllib.request.urlopen(url)
        config = json.loads(response.read().decode('utf-8'))
    except Exception as e:
        return {"statusCode": 500, "body": json.dumps({"error": f"Failed to fetch AppConfig: {str(e)}"})}

    body = json.loads(event.get("body", "{}")) if event.get("body") else {}
    user_prompt = body.get("prompt", "")

    max_tokens = config.get("max_input_tokens_allowed", 1000)
    if len(user_prompt.split()) > max_tokens:
        return {
            "statusCode": 400, 
            "body": json.dumps({"error": f"Prompt exceeds safe limits of {max_tokens} words."})
        }

    selected_model = config.get("active_llm_model", "gpt-4o-mini")
    prompt_version = config.get("system_prompt_version", "v1.0")

    return {
        "statusCode": 200,
        "headers": { "Content-Type": "application/json" },
        "body": json.dumps({
            "status": "success",
            "model_used": selected_model,
            "prompt_version": prompt_version,
            "ai_response": f"[Mocked LLM Response] Processed using model: {selected_model}"
        })
    }

From the above, we pull the current config via the local sidecar running and decide whether to allow the user-provided prompt, based on the token limit in the latest config or not.

Testing the App

After the stack has been successfully deployed, we now have to verify if the proposed solution is working as expected.

Get the output URL from the terminal after deploy, it should look something like below:

https://unique-id.execute-api.us-east-1.amazonaws.com/prod/prompt

Next, head over to AWS API Gateway to retrieve the API Key created, so as to attach it when making a POST request to the endpoint.

Next, we invoke the API. Remember to add the X-Api-Key header when making this request, with its value being the API Key you just retrieved.

We can see we have the mocked response, which means this worked as expected, since we currently have this in place in the config: “max_input_tokens_allowed”: 20 .

Next, we would update the prompt to go beyond the 20 token/words allowed:

Now we also see that we get a 400 response for a token count more than the allowed limit.

Assuming there is a new requirement to change the model and the token limit to a smaller limit. We would update the config.json and trigger a new deployment to test if we can get changes reflected as proposed in this article:

{
  "active_llm_model": "gpt-4o-mini",
  "enable_fallback_to_cheap_model": true,
  "max_input_tokens_allowed": 10,
  "system_prompt_version": "v3.0-emergency-fallback"
}

To test, we run the deploy:

 cdk deploy AppStack --require-approval never

Since this is an article, I just wanted to deploy to test, in prod, we would utilize CI/CD tools to perform this deployment, in that way we could have specific deployment for configs, so it doesnt get to deploy the full stack when a minor config change is done, to keep one of the wins of using AWS AppConfig in the first place, which is faster deployemnt.

One approach is to track file changes in the config folder and, rather than do a CDK deploy, run an AWS CLI AppConfig deployment command, so it simply updates and deploys. This way, you have a separate workflow for configs that's faster and independent.

Here is a sample below:

aws appconfig create-hosted-configuration-version \
  --application-id <app-id> \
  --configuration-profile-id <profile-id> \
  --content fileb://config/config.json \
  --content-type application/json

aws appconfig start-deployment \
  --application-id <app-id> \
  --environment-id <env-id> \
  --configuration-profile-id <profile-id> \
  --configuration-version <new-version> \
  --deployment-strategy-id AppConfig.Linear50PercentEvery30Seconds

That out of the way, head over to the AWS console.

We see the environment is updating and deploying:

Moving in/clicking:

Also, looking at the ModelRouter, we see the latest version(2) now exists.

Now, head over to test with a prompt that has a word count of 12:

Now we see it fails for this token limit, just as we expected.

Lastly, we would update the prompt to be less than 10 words:

Perfect, it picked up the updated config changes without needing to perform any code changes on Lambda.

Great!

Conclusion

This has been an interesting ride; we learnt that to truly be agile, serverless systems require us to rethink where our configuration data lives.

We utilized AWS AppConfig to remove hardcoded configurations from within our source code and utilized a sidecar approach for faster updates and fetching of configurations, keeping our architecture rigid and highly scalable.

You can find the complete source code here.

Excited to see what you build.

Beyond Boolean Toggles: Data-Rich Feature Flags with AWS AppConfig and Python CDK was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we will develop a production-ready serverless Document management system on AWS, using Cognito for authentication, Amazon Verified Permissions (Cedar) for fine-grained RBAC/ABAC, Lambda and API Gateway for APIs, and S3 and DynamoDB for storage and metadata sharing.

Already excited? Let’s get started.

What would we cover

This article came from my need to deeply understand how to build a scalable authentication and authorization layer on AWS for a real serverless product.

It tackles the common pain points you hit as requirements grow, such as authorization logic scattered across Lambdas, endless if role == checks mixing RBAC and ABAC in inconsistent ways, and patterns that become hard to scale, hard to audit, and easy to get wrong.

We would get to achieve the following:

• Build an app using AWS CDK as IaC
• Create Cedar Policies in Amazon Verified Permission (AVP)
• Authentication via Amazon Cognito
• API layer: Lambda + API Gateway endpoints and a Lambda authorizer that validates JWTs and forwards identity context.
• Presigned URL workflow: upload/download with S3 presigned URLs, including pending → complete metadata updates (upload_complete flow).
• Implement RBAC and ABAC via AVP, DynamoDB, and Cognito
• Document Management using S3 and DynamoDB
• UI: Streamlit app hosted on ECS Fargate (behind ALB) for the front-end/admin experience.
• Observability & Auditability: CloudWatch Logs + Logs Insights and AWS X-Ray Tracing

Prerequisites

• AWS account
• AWS CLI setup and configured with appropriate credentials
• AWS CDK set up.
• Python3
• Docker
• Node.js and npm (for AWS CDK CLI)

Case Study: DocNext

Building DocNext’s serverless document management platform started simple: a few Lambda handlers with inline checks like if role == “Admin”. But the access model quickly became really complicated in the real world. The team now needed to support document owners, departmental boundaries, cross-user sharing, and multiple roles (Admin, Manager, Editor, Viewer), while also maintaining an auditable trail of who accessed what, when, and why it was allowed or denied.

As the number of endpoints grew (upload, list, download, delete, share, audit), duplicating authorization logic across Lambdas became risky and hard to review. A small policy change required code edits, deployments, and regressions across multiple handlers. DocNext needed a way to separate authorization from business logic, keep authorization rules version-controlled, and make changes easy to review via pull requests.

You, as an engineer in the team, proposed the following:

Using Amazon Verified Permissions (AVP) with Cedar policies as the centralized authorization layer. Amazon Cognito provides identity and user attributes (role and department), and DynamoDB stores document attributes (owner, department, and shared_with). At runtime, the API uses a Lambda authorizer to validate Cognito JWTs and pass user context downstream, while each business Lambda calls AVP is_authorized using the user and document attributes to make a clear ALLOW/DENY decision.

To keep the system operable, observability would be built early on: Lambdas emit structured JSON logs to CloudWatch Logs, audit and error investigations run through CloudWatch Logs Insights, and AWS X-Ray traces capture performance and dependency behavior across calls to Cognito, AVP, DynamoDB, and S3.

You envision this will produce a system where authorization is consistent, auditable, and easy to evolve, without scattering security rules throughout the codebase. This gets approved by Management, now it's time to build and bring it to life!

Proposed Architecture

Document Upload using Pre-Signed URL approach

The client(Streamlit/User) logs in and selects an image from their system, which sends a request to an API requesting a Presigned URL. Once the user is verified, the document details are stored in DynamoDB with a pending status, and the URL is sent back to the client.

Next, the user clicks the upload button after filling in other necessary information like the department the document belongs to, tags & description (optional). This uploads the document securely to S3.

An EventBridge rule is attached to the bucket that gets triggered when an object is created. What this means is that after the document is added to S3, this event is triggered, and in turn, it calls a lambda function that updates the DynamoDB table with a status of complete and other metadata

API Security (Authentication and Authorization)

Clients authenticate with Amazon Cognito and receive a JWT. That token is sent on every API request as Authorization: Bearer <token>.

API Gateway uses a Lambda Authorizer to validate the JWT (Cognito JWKS), then attaches identity context (e.g., userId/sub, role, department, email) to the request and returns an IAM policy to allow or deny the call. If allowed, API Gateway forwards the request to the target Lambda with event.requestContext.authorizer populated.

Inside the target Lambda, the handler calls AVP is_authorized with the action (read/write/delete/share) and resource attributes (document owner, department, shared_with). Cedar policies evaluate this data and return ALLOW/DENY. Only when authorization is ALLOW does the Lambda proceed; otherwise, it returns an authorization failure.

The Lambda then returns the response through API Gateway to the client.

Observability in Action

All Lambdas write structured JSON logs using a shared logger, so CloudWatch logs stay consistent and easy to query (event type, request id, user/document id, role, outcome). CloudWatch Logs Insights powers the Audit page and makes it straightforward to trace “who did what, when”.

For tracing, AWS X-Ray captures end-to-end request traces, and AWS SDK calls (Cognito, AVP, DynamoDB, S3) to spot latency and failures quickly. A CloudWatch Dashboard pulls the key signals together (errors, latency, throttles, uploads).

Amazon Verified Permission

So what exactly is Amazon Verified Permission?

AVP (Amazon Verified Permissions) is a scalable service for fine-grained authorization. It helps teams externalize access rules into centrally managed policies, so applications can enforce security consistently while moving faster.

It uses Cedar Permission, which is an open-source language for writing authorization policies and making authorization decisions based on those policies.

This video at AWS re:Inforce 2023 explains this well.

RBAC and ABAC

We implement RBAC (Role-Based Access Control) by assigning each user a role in a Cognito custom attribute and enforcing those roles centrally in Cedar policies via Amazon Verified Permissions.

In the Cedar schema, that becomes principal.role, and policies check principal.role == “Admin”, “Manager”, “Editor”, “Viewer”

We implement ABAC (Attribute-Based Access Control) by using user and resource attributes captured in Amazon Cognito (e.g., principal.department, principal.userId) and document attributes stored as the authorization resource in DynamoDB (e.g., resource.department, resource.owner, resource.sharedWith).

In the Cedar schema, these attributes appear as:

• principal.department (user attribute from Cognito)
• principal.userId (Cognito sub as the user identifier)
• resource.department, resource.owner, resource.sharedWith (document attributes)

Policies then enforce ABAC rules, such as:

• Managers can access documents only when resource.department == principal.department
• Owners can access their own documents when resource.owner == principal.userId
• Shared document access is evaluated using resource.sharedWith.contains(principal.userId)

Currently, we have 5 Cedar Policies. Let's bring them down and explain:

1. Admin Full Access

// Cedar Policy: Admin Full Access
// Admin users have full access to all documents for all operations

permit(
    principal,
    action in [
        DocumentManagement::Action::"read",
        DocumentManagement::Action::"write",
        DocumentManagement::Action::"delete",
        DocumentManagement::Action::"share"
    ],
    resource
)
when {
    principal has role &&
    principal.role == "Admin"
};

This grants the admin full access to all documents

2. Document Owner Access

// Cedar Policy: Document Owner Permissions
// Document owners can always manage their own documents regardless of role

permit(
    principal,
    action in [
        DocumentManagement::Action::"read",
        DocumentManagement::Action::"write",
        DocumentManagement::Action::"delete",
        DocumentManagement::Action::"share"
    ],
    resource
)
when {
    principal has userId &&
    resource has owner &&
    resource.owner == principal.userId
};

This grants the document owner( this means the user who uploaded the document on the platform) full access to the document. This is irrespective of whatever role they have in the platform, since they “own” this document.

3. Editor Access

// Cedar Policy: Editor Owned/Shared Document Access
// Editor users can read and write documents they own or have been shared with

permit(
    principal,
    action in [
        DocumentManagement::Action::"read",
        DocumentManagement::Action::"write"
    ],
    resource
)
when {
    principal has role &&
    principal.role == "Editor" &&
    principal has userId &&
    resource has owner &&
    (
        resource.owner == principal.userId ||
        (resource has sharedWith && resource.sharedWith.contains(principal.userId))
    )
};

This grants the Editor role read(list and download functionality via the web app) to resources, and also documents shared with them by other users.

Also, remember that if this Editor user uploads a document, they would also have full access, like delete, share, not included in this particular cedar. This is because when evaluating, AVP would also look at that Cedar policy and approve those actions. So in theory, the two Cedar policy files(editor and owner) would be used.

4. Departmental Manager Access

// Cedar Policy: Manager Departmental Access
// Manager users can manage documents in their department

permit(
    principal,
    action in [
        DocumentManagement::Action::"read",
        DocumentManagement::Action::"write",
        DocumentManagement::Action::"delete",
        DocumentManagement::Action::"share"
    ],
    resource
)
when {
    principal has role &&
    principal.role == "Manager" &&
    principal has department &&
    resource has department &&
    principal.department == resource.department
};

This grants the Manager role full access to documents created in their department.

So if we had 2 departments, Hr and IT. The Hr manager would only see documents uploaded for the Hr department and not those of the IT department.

5. Viewer Read-only Access

// Cedar Policy: Viewer Read-Only Access
// Viewer users have read-only access to documents shared with them

permit(
    principal,
    action == DocumentManagement::Action::"read",
    resource
)
when {
    principal has role &&
    principal.role == "Viewer" &&
    principal has userId &&
    resource has sharedWith &&
    resource.sharedWith.contains(principal.userId)
};

This grants the viewer read-only access to documents shared with them.

Amazon Cognito and Amazon Verified Permissions (AVP)

Amazon Cognito and Amazon Verified Permissions (AVP) work together to provide a comprehensive authentication and fine-grained authorization solution for the application.

Cognito handles the authentication while AVP handles authorization.

AWS CDK

The AWS Cloud Development Kit (AWS CDK) is an open-source software development framework for defining cloud resources as code.

Follow the prompt to complete the setup process.

1. Install AWS CDK CLI:

npm install -g aws-cdk

2. Set up environment:

cd infrastructure
python3 -m venv .venv
source .venv/bin/activate

3. Deploy infrastructure:

N:B: The compute stack uses Docker to build the necessary packages used by Lambda; you would need to have that up and running locally.

cd infrastructure # Move into cdk directory

cdk bootstrap # First time only

# Deploy all (except ECS)
cdk deploy DocumentManagement-Networking DocumentManagement-Storage DocumentManagement-Auth DocumentManagement-Authorization DocumentManagement-Compute DocumentManagement-Observability  --require-approval never

# Build & Push docker streamlit image to ECR
cd ../ && chmod +x build_and_push_streamlit.sh && ./build_and_push_streamlit.sh

# Deploy ECS stack
cd infrastructure && cdk deploy DocumentManagement-ECS --require-approval never

4. Create Admin User:

Run the following command to create the first user (Admin)

python3 create_admin_user.py admin@example.com SecurePassword123!

5. Visit app:

Next, access the Streamlit application via the ALB URL. Copy the DNS name and visit the URL.

The app should be up and show this:

Now the app is set up and ready for Testing.

Project Overview

Stack:

infrastructure/stacks/
├── networking_stack.py           # VPC, subnets, NAT gateway
├── storage_stack.py              # S3, DynamoDB tables
├── auth_stack.py                 # Cognito User Pool
├── authorization_stack.py        # Verified Permissions
├── compute_stack.py              # Lambda, API Gateway
├── observability_stack.py        # CloudWatch, X-Ray
└── ecs_stack.py                  # ECS

Streamlit app:

app/
├── app.py                 # Main application entry point
├── config.py              # Configuration management
├── auth.py                # Cognito authentication
├── api_client.py          # API Gateway client
├── session_manager.py     # Optional DynamoDB-backed Streamlit session
├── pages/                 # Application pages
│   ├── documents.py       # Document list page
│   ├── upload.py          # Document upload page
│   ├── share.py           # Document sharing page
│   ├── audit.py           # Audit logs page
│   └── admin.py           # Admin panel page
├── requirements.txt       # Python dependencies
├── Dockerfile             # Docker configuration
└── README.md              # This file

Testing

Time to see the app in action.

User Creation

The Admin user creates 3 users in the same department Hr.

Two users are viewers, while one is a Manager of the Hr department:

Document Creation

When we log in as User 1 and upload a document, we can see it in the documents section:

As the owner of this document, the user can perform all actions like read, share, delete, and download on this document.

Next, we test the sharing functionality.

Sharing Documents

The user shares this document with User 2:

How sharing works

Sharing does not call AVP directly. Instead, the Share Lambda updates the document’s shared_with attribute in DynamoDB.

When another user later requests access, the receiving Lambda calls AVP is_authorized. Cedar policies read shared_with from the document resource attributes to decide ALLOW or DENY.

Now, when User 2 logs in, they should be able to see this document, even when they didn’t create it:

As can be seen, User 2 can view the document User 1 shared with them. They can download, but can't delete, since Amazon Verified Permission only allows the read action.

Manager Oversight

User 3 is the manager of the Hr department, and according to the Cedar policies we created, we should be able to see all documents created in the department.

This document wasn't shared with the manager, nor does it belong to the manager(User 3). But the role of Manager allows the user to view all documents in their department.

They can also perform the delete and share action on the document, as their role permits these actions.

Audit Logs

The audit logs show a structured log entry:

Admin Oversight

The Admin can view all documents and perform all actions on the document:

and also share and view all shared documents:

Observability

Traces: Looking at some request traces.

We start with the list document flow:

Viewing the first trace, we see:

Scrolling down, we see the breakdown of the traces shown in timeline format:

Dashboard: Viewing the CloudWatch Dashboard:

Looking at the Dashboard, we see more information about the app:

Lessons learnt

1. Separating Authorization: Using AVP and Cedar policy from business logic makes scalability easier; we can simply update or add a new policy document without touching business logic.
2. Cost Implications: Billed per authorization evaluation. In our implementation, this means roughly one AVP call per document we return in each API. This means if we list 50 documents and call AVP per document, that’s 50 AVP evaluations for one API call. This is correct from a security standpoint, but something to take note of in terms of cost.
3. RBAC alone isn’t enough: Real systems need ABAC (department, owner, shared_with) to express scope cleanly.
4. Observability matters: Structured “authorization decision” logs + X-Ray traces reduce time-to-debug when policies or attributes don’t match expectations. Use logging and Insights early to validate decisions (who got DENY and why).

Conclusion

This has been an interesting ride. We got to explore the Authentication and Authorization part of AWS, implemented various best practices, and designed a production-ready serverless document management system powered by Amazon Verified Permission.

You can find the complete source code here.

Excited to see what you build!

Secure Serverless Document Management system with Amazon Verified Permission, CDK & Python was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we will explore how to create AI agents that use built-in tools in Amazon Bedrock AgentCore, Code Interpreter, and Browser.

Let's get started.

Prerequisites

• AWS account
• AWS CLI setup locally
• Python 3
• Previous article

What would we cover

• We cover what the Amazon Bedrock AgentCore Browser is and its role in the application.
• We cover what the Amazon Bedrock Code Interpreter is and its role in the application.
• We walk through the Python code.
• We perform some tests to see it in action.

Getting started

To get started, run the command below. This sets up the project and installs the required packages.

mkdir strands-browser && cd strands-browser
python3 -m venv .venv
source .venv/bin/activate
pip install bedrock-agentcore strands-agents strands-agents-tools playwright nest-asyncio

AgentCore Browser

AgentCore Browser tool enables AI agents interact with web pages in a secure and managed environment, particularly useful for integrating search into an AI agent.

It uses a self-managed Chrome browser (Chromium) for this browser automation.

We would utilize this tool to enable browser automation within our Strands agent.

Before going forward, we need to add the required permissions needed to work with the AgentCore Browser.

Attach this policy to your IAM user:

{  
    "Version":"2012-10-17",                                    
    "Statement": [  
        {  
            "Sid": "BedrockAgentCoreBrowserFullAccess",  
            "Effect": "Allow",  
            "Action": [  
                "bedrock-agentcore:CreateBrowser",  
                "bedrock-agentcore:ListBrowsers",  
                "bedrock-agentcore:GetBrowser",  
                "bedrock-agentcore:DeleteBrowser",  
                "bedrock-agentcore:StartBrowserSession",  
                "bedrock-agentcore:ListBrowserSessions",  
                "bedrock-agentcore:GetBrowserSession",  
                "bedrock-agentcore:StopBrowserSession",  
                "bedrock-agentcore:UpdateBrowserStream",  
                "bedrock-agentcore:ConnectBrowserAutomationStream",  
                "bedrock-agentcore:ConnectBrowserLiveViewStream"  
            ],  
            "Resource": "arn:aws:bedrock-agentcore:<Region>:<account_id>:browser/*"  
        },  
        {  
            "Sid": "BedrockModelAccess",  
            "Effect": "Allow",  
            "Action": [  
                "bedrock:InvokeModel",  
                "bedrock:InvokeModelWithResponseStream"  
            ],  
            "Resource": [  
                "*"  
            ]  
        }  
    ]  
}

Note: Replace <Region> with your chosen region and <account_id> with your account id

Next, create a new file called browser_agent.py in the root of the project (strands-browser/browser_agent.py):

from strands import Agent
from strands_tools.browser import AgentCoreBrowser

# Initialize the Browser tool
browser_identifier = "browser_identifier"
browser_tool = AgentCoreBrowser(region="us-east-1", identifier=browser_identifier)

agent = Agent(tools=[browser_tool.browser])

prompt = "What can AgentCore Browser do, least 5 benefits it brings to an AI agent? Use this website link if needed: https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/browser-tool.html"print(f"\\n\\nPrompt: {prompt}\\n\\n")
response = agent(prompt)
print("\\n\\nAgent Response:")
print(response.message["content"][0]["text"])

Notice, we have a variable browser_identifier.

Head over to your AWS Console and into Amazon Bedrock AgentCore. Next, under the built-in tool, click on the Browser tool. Create a new Browser tool.

Create a new S3 bucket; we would need this, as recordings would be stored there. Select the bucket after creating.

For Permission, we would need a new IAM role with the needed permissions for S3 and CloudWatch. This would enable this Browser tool to save recordings and logs.

Create an IAM policy with these permissions.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3Permissions",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListMultipartUploadParts",
                "s3:AbortMultipartUpload"
            ],
            "Resource": [
                "arn:aws:s3:::s3-bucket",
                "arn:aws:s3:::s3-bucket/*"
            ]
        },
        {
            "Sid": "CloudWatchLogsPermissions",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "logs:DescribeLogStreams"
            ],
            "Resource": "*"
        }
    ]
}

Attach this policy to an IAM role, and select this role in the Permissions section. Finally, click the Create button.

Copy the Browser tool ID, which is the actual value of browser_identifier we have in the code. Replace that variable content with the value you copied.

Now run the app.

python browser_agent.py

Checking the AWS Console, we see we have an active session:

The output in the terminal:

Observability

AgentCore also provides observability into what occurs when we use this tool. We can see how the agent interacted with the browser, view live sessions, recordings, and more.

We can also see the live session by clicking the View live session button

Once we terminate the session, we can see the recording as well:

Code Interpreter

The Amazon Bedrock AgentCore Code Interpreter enables AI agents to write and execute code securely in sandbox environments, reducing hallucination and increasing their accuracy.

Attach this policy to your IAM user:

{
    "Version":"2012-10-17",
    "Statement": [
        {
            "Sid": "BedrockAgentCoreCodeInterpreterFullAccess",
            "Effect": "Allow",
            "Action": [
                "bedrock-agentcore:CreateCodeInterpreter",
                "bedrock-agentcore:StartCodeInterpreterSession",
                "bedrock-agentcore:InvokeCodeInterpreter",
                "bedrock-agentcore:StopCodeInterpreterSession",
                "bedrock-agentcore:DeleteCodeInterpreter",
                "bedrock-agentcore:ListCodeInterpreters",
                "bedrock-agentcore:GetCodeInterpreter",
                "bedrock-agentcore:GetCodeInterpreterSession",
                "bedrock-agentcore:ListCodeInterpreterSessions"
            ],
            "Resource": "arn:aws:bedrock-agentcore:<region>:<account_id>:code-interpreter/*"
        }
    ]
}

Note: Replace <region> with your chosen region and <account_id> with your account id

Next, create a new file called code_interpreter_agent.py in the root of the project (strands-browser/code_interpreter_agent.py):

from strands import Agent
from strands_tools.code_interpreter import AgentCoreCodeInterpreter

# Initialize the Code Interpreter tool
code_interpreter_tool = AgentCoreCodeInterpreter(region="us-east-1")

SYSTEM_PROMPT = """You are an AI assistant that validates answers through code execution.
When asked about code, algorithms, or calculations, write Python code to verify your answers."""

agent = Agent(
    tools=[code_interpreter_tool.code_interpreter],
    system_prompt=SYSTEM_PROMPT
)

prompt = "What is 10 factorial, then add 12 to the result"
print(f"\n\nPrompt: {prompt}\n\n")

response = agent(prompt)
print(response.message["content"][0]["text"])

Now run the app.

python code_interpreter_agent.py

This output in the terminal:

As can be seen, it uses the code_interpreter tool(Tool #1) and gives an accurate error, preventing Hallucination.

Conclusion

Building on the previous Bedrock AgentCore article, we explored the built-in tools in the article. We also experimented with Browser and Code Interpreter tools using the Strands Agent.

Go and experiment, and can't wait to see what you build!

Autonomous AI Agents: Web Browsing and Code Execution with Strands on Amazon Bedrock AgentCore was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we combine LangGraph with Amazon Bedrock AgentCore to build and deploy AI agents with persistent memory capabilities. We explore how to create agents that maintain conversation context across sessions and server restarts, leveraging the powerful memory management features of Amazon Bedrock AgentCore alongside the flexible agent orchestration provided by the open-source LangGraph framework.

Let’s get right into it!

Prerequisites

• AWS account
• AWS CLI setup locally
• AWS CDK set up.
• Python 3
• Docker

What would we cover

• We cover what Amazon Bedrock AgentCore is
• We cover what Langraph is and its role in the application
• We cover the core services of Amazon Bedrock AgentCore
• We walk through the Python code.
• We perform some tests to see it in action.

What is Amazon Bedrock AgentCore

Amazon Bedrock Agentcore is an agent-centric platform for building, deploying, and operating effective agents securely at scale, eliminating the need to manage any infrastructure.

It consists of the following core services broken down into 3 main stages: Build, Deploy & Operate:

The Build stage:

1. AgentCore Gateway: Acts as an intermediary between your agent and tools, and could assist in turning lambda functions or api integrations into tools.
2. AgentCore Memory: Allows the integration of both short-term and long-term memory into your AI application, that persist across sessions. We would see more of this later on in this article.
3. AgentCore Policy: Provides the capability to define boundaries for your agents that suit business rules.
4. AgentCore Identity: This brings a secure and scalable agent identity, access, and authentication management service.
5. AgentCore CodeIntepreter(Inbuilt tool): Provides an isolated sandbox environment for agents to execute code.
6. AgentCore Browser(Inbuilt tool): Provides a cloud-browser-based runtime environment that enables AI agents to interact with web applications, while in a fully managed environment.

The Deploy stage:

1. AgentCore Runtime: This is a secure, serverless runtime environment purpose-built for deploying and scaling AI agents and tools, with benefits like fast cold starts, built-in identity, and more.

The Operate stage:

1. AgentCore Observability: Provides a centralised and detailed monitor and debug agent performance in production.
2. AgentCore Evaluations: Provides measurable quality signals that enable teams optimize performance with the structured insights provided.

What is LangGraph

LangGraph is the library that uses a graph-based approach to build and manage complex stateful AI agent workflows.

For the app, we would build a simple agent that takes users' input and outputs a response.

Getting started

Following this guide, we run the following command. This uses the AgentCore starter toolkit.

pip install bedrock-agentcore-starter-toolkit
agentcore create 

Follow all the prompts to create a new app. For the article, I used/choose the following:

Where should we create your new agent?
./pythonlanggraph     
How would you like to start?
A production-ready agent defined with Terraform or CDK
What agent framework should we use?
LangChain + LangGraph
Which model provider will power your agent?
Amazon Bedrock
Which IaC proivder will define your AgentCore resources?
CDK
Run agentcore configure first? (Further define configuration and reference exisiting re
Yes
 
Configuring Bedrock AgentCore…
🏷️ Inferred agent name: create_agent
Press Enter to use this name, or type a different one (alphanumeric without '-')
Agent name [create_agent]:
✓ Using agent name: create_agent
🚀 Deployment Configuration
Create mode only uses the container deployment type.
✓ Using: Container
🔐 Authorization Configuration
By default, Bedrock AgentCore uses IAM authorization.
Configure OAuth authorizer instead? (yes/no) [no]: no
✓ Using default IAM authorization
Configure request header allowlist? (yes/no) [no]: no
✓ Using default request header configuration

N:B: At the time of writing this article, AWS CDK for Amazon Bedrock Agentcore is still experimental. More info here.

Testing Locally

Replace the code src/main.py with:

import os
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from bedrock_agentcore import BedrockAgentCoreApp
from .mcp_client.client import get_streamable_http_mcp_client as deployed_get_tools
from model.load import load_model

if os.getenv("LOCAL_DEV") == "1":
    # In local dev, instantiate dummy MCP client so the code runs without deploying
    async def get_tools():
        return []
else:
    get_tools = deployed_get_tools

# Instantiate model
llm = load_model()


# Import AgentCore Gateway as Streamable HTTP MCP Client

# Integrate with Bedrock AgentCore
app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke(payload):
    # assume payload input is structured as { "prompt": "<user input>" }


    # Define the agent
    graph = create_agent(llm)

    # Process the user prompt
    prompt = payload.get("prompt", "What is Agentic AI?")

    # Run the agent
    result = await graph.ainvoke({"messages": [HumanMessage(content=prompt)]})

    # Return result
    return {
        "result": result["messages"][-1].content
    }

if __name__ == "__main__":
    app.run()

Next, we run the agent locally and test. Run the code below in the open terminal

cd pythonlanggraph
agentcore dev

In another terminal, run this to test. This ensures everything works well locally.

agentcore invoke --dev '{"prompt": "who are you?"}'

You could also make a POST request to the localhost:8080/invocations endpoint with the payload body as {“prompt”: “Who are you?”}.

CURL -X POST http://localhost:8080/invocations \
-H "Content-Type": "application/json" \
-d '{"prompt": "Who are you?"}'

This should return this as the outcome in the terminal:

✓ Response from dev server:
{
    'response': '{"result": "I\'m Claude, an AI assistant made
by Anthropic. I\'m here to help answer questions, have 
conversations, assist with tasks, and provide information on a
wide variety of topics. In this particular context, I have 
access to a simple math function that can add two numbers 
together, though I can help with much more than just 
arithmetic. \\n\\nIs there something specific I can help you 
with today?"}'
}

Adding Memory

Next, we update the code to create a LangGraph state graph with a node that calls an LLM. The code below uses an in-built memory.

import os
from typing import TypedDict, List, Annotated
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from bedrock_agentcore import BedrockAgentCoreApp
from .mcp_client.client import get_streamable_http_mcp_client as deployed_get_tools
from .model.load import load_model

# Environment-based tool selection
if os.getenv("LOCAL_DEV") == "1":
    # In local dev, instantiate dummy MCP client so the code runs without deploying
    async def get_tools():
        return []
else:
    get_tools = deployed_get_tools

# Initialize components once at module level
llm = load_model()

# Initialize memory saver for LangGraph
memory_saver = MemorySaver()

# Integrate with Bedrock AgentCore
app = BedrockAgentCoreApp()

# Define the state structure with proper message accumulation
class GraphState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# Define the LLM node
async def llm_node(state: GraphState) -> GraphState:
    """Node that processes messages through the LLM."""
    messages = state["messages"]
    
    # Process all messages (including conversation history) through the LLM
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

# Create the StateGraph
workflow = StateGraph(GraphState)

# Add nodes
workflow.add_node("llm", llm_node)

# Set entry point
workflow.set_entry_point("llm")

# Add edge from llm to END
workflow.add_edge("llm", END)

# Compile the graph with memory
graph = workflow.compile(checkpointer=memory_saver)

@app.entrypoint
async def invoke(payload):
    """Process user prompt through the LangGraph with memory and return response."""
    # Extract prompt and session_id from payload
    prompt = payload.get("prompt", "What is Agentic AI?")
    session_id = payload.get("session_id", "default_session")
    
    # Create human message
    human_message = HumanMessage(content=prompt)
    
    # Configure with session for memory persistence
    config = {"configurable": {"thread_id": session_id}}
    
    # Invoke the graph with the new message
    result = await graph.ainvoke(
        {"messages": [human_message]},
        config=config
    )
    
    # Get the last AI message
    last_message = result["messages"][-1]
    
    # Return the response
    return {
        "result": last_message.content,
        "session_id": session_id
    }
if __name__ == "__main__":
    app.run()

To test run:

agentcore invoke --dev '{"prompt": "who are you?", "session_id": "10"}'

Output:

Running again:

agentcore invoke --dev '{"prompt": "what did I ask last?", "session_id": "10"}'

Output:

Running one more time:

agentcore invoke --dev '{"prompt": "what is the capital of Nigeria?", "session_id": "10"}'

Output:

The problem with this is when we restart the server, the session id become invalid. Restart the server and invoke the agent one more time:

agentcore invoke --dev '{"prompt": "what did I ask last?", "session_id": "10"}'

Output:

It doesn’t retain the conversation after we restart. We can fix this issue with Agentcore memory, as we would store the conversation in AgentCore, making it persistent even after server restarts.

Integrating Amazon Bedrock AgentCore Memory

Now, we integrate AgentCore Memory to see how it supports.

Update the codebase with the following:

import os
import asyncio
import uuid
from typing import TypedDict, List, Annotated
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, END, add_messages
from langgraph.checkpoint.memory import MemorySaver
from bedrock_agentcore import BedrockAgentCoreApp, BedrockAgentCoreContext
from bedrock_agentcore_starter_toolkit.operations.memory.manager import MemoryManager
from bedrock_agentcore.memory.session import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole
from bedrock_agentcore_starter_toolkit.operations.memory.models.strategies import SummaryStrategy
from .mcp_client.client import get_streamable_http_mcp_client as deployed_get_tools
from .model.load import load_model

# Environment-based tool selection
if os.getenv("LOCAL_DEV") == "1":
    # In local dev, instantiate dummy MCP client so the code runs without deploying
    async def get_tools():
        return []
else:
    get_tools = deployed_get_tools

# Initialize components once at module level
llm = load_model()

# Integrate with Bedrock AgentCore
app = BedrockAgentCoreApp()

# Initialize Bedrock AgentCore Memory Manager
memory_manager = MemoryManager(region_name=os.getenv("AWS_REGION", "us-east-1"))

# Single user constant
SINGLE_USER_ID = "single_user"

# Create or get memory resource with strategies for single user
memory_resource = memory_manager.get_or_create_memory(
    name="LangGraphSingleUserMemory",
    description="Memory for single-user LangGraph agent conversations.",
    strategies=[
        SummaryStrategy(
            name="ConversationSummarizer",
            namespaces=[f"/summaries/{SINGLE_USER_ID}/{{sessionId}}"]
        )
    ]
)

memory_id = memory_resource.get('id')
memory_session_manager = MemorySessionManager(memory_id=memory_id)

# Initialize LangGraph memory for immediate conversation history
langgraph_memory = MemorySaver()

# Define the state structure with proper message accumulation
class GraphState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# Define the LLM node
async def llm_node(state: GraphState) -> GraphState:
    """Node that processes messages through the LLM."""
    messages = state["messages"]
    
    # Process all messages (including conversation history) through the LLM
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

# Create the StateGraph
workflow = StateGraph(GraphState)

# Add nodes
workflow.add_node("llm", llm_node)

# Set entry point
workflow.set_entry_point("llm")

# Add edge from llm to END
workflow.add_edge("llm", END)

# Compile the graph with LangGraph memory for immediate history
graph = workflow.compile(checkpointer=langgraph_memory)

@app.entrypoint
async def invoke(payload, context: BedrockAgentCoreContext):
    """Process user prompt through LangGraph with persistent memory."""
    prompt = payload.get("prompt", "What is Agentic AI?")
    
    # Handle session_id - can come from context (CLI -s) or payload
    session_id = context.session_id or payload.get("session_id")
    
    if session_id is not None:
        # session_id provided - validate it
        if len(session_id) < 33:
            return {
                "error": "session_id must be at least 33 characters long",
                "provided_length": len(session_id),
                "required_length": 33
            }
    else:
        # No session_id provided - generate one
        session_id = f"user-single-conversation-{str(uuid.uuid4())}"
    
    human_message = HumanMessage(content=prompt)
    config = {"configurable": {"thread_id": session_id}}
    
    # Check if LangGraph has existing conversation history
    try:
        current_state = await graph.aget_state(config)
        existing_messages = current_state.values.get("messages", []) if current_state and current_state.values else []
        has_existing_history = len(existing_messages) > 0
    except:
        has_existing_history = False
    
    # Get or create Bedrock memory session
    memory_session = None
    try:
        memory_session = memory_session_manager.create_memory_session(
            actor_id=SINGLE_USER_ID,
            session_id=session_id
        )
    except:
        pass
    
    # If no LangGraph history, try to restore context from Bedrock memory
    if not has_existing_history and memory_session:
        try:
            await asyncio.sleep(1)  # Allow Bedrock processing time
            
            # Try to get session-specific memory records
            session_namespace = f"/summaries/{SINGLE_USER_ID}/{session_id}"
            memory_records = None
            
            if hasattr(memory_session, 'list_long_term_memory_records'):
                memory_records = memory_session.list_long_term_memory_records(
                    namespace_prefix=session_namespace
                )
            
            # Extract context from memory records
            if memory_records and len(memory_records) > 0:
                first_record = memory_records[0]
                summary_text = ""
                
                # Handle MemoryRecord objects
                if hasattr(first_record, 'content'):
                    content_obj = first_record.content
                    if isinstance(content_obj, dict):
                        summary_text = content_obj.get('text', '').strip()
                    elif hasattr(content_obj, 'text'):
                        summary_text = content_obj.text.strip()
                
                # Enhance message with context if found
                if summary_text:
                    enhanced_content = f"Previous conversation context from this session:\n{summary_text}\n\nCurrent question: {prompt}"
                    human_message = HumanMessage(content=enhanced_content)
                    
        except:
            pass  # Continue without context if memory retrieval fails
    
    # Invoke LangGraph
    result = await graph.ainvoke({"messages": [human_message]}, config=config)
    last_message = result["messages"][-1]
    ai_response = last_message.content
    
    # Store conversation in Bedrock memory
    if memory_session and hasattr(memory_session, 'add_turns'):
        try:
            user_msg = ConversationalMessage(prompt, MessageRole.USER)
            memory_session.add_turns(messages=[user_msg])
            
            assistant_msg = ConversationalMessage(ai_response, MessageRole.ASSISTANT)
            memory_session.add_turns(messages=[assistant_msg])
        except:
            pass  # Continue if memory storage fails
    
    return {
        "result": ai_response,
        "session_id": session_id
    }
if __name__ == "__main__":
    app.run()

Next, run : pip install bedrock_agentcore_starter_toolkit in your terminal, add the package.

Then rerun the server. This would create a new AgentCore Memory, which would take some time:

On the console, we see it been active:

Now, perform a similar test we did above. Restart the server and ask the last series of questions. We use a new session id of user-123456-conversation-12345679 as AgentCore runtime needs a unique session ID for each user or conversation with at least 33 characters. More info on that here.

agentcore invoke --dev '{"prompt": "what is the capital of Ghana?", "session_id": "user-123456-conversation-12345679"}'      

Output:

Now, we restart our server and ask

agentcore invoke --dev '{"prompt": "what did I ask last?", "session_id": "user-123456-conversation-12345679"}'

Output:

This time it does remember our conversation!

N:B: From experience, it takes some seconds for the last conversation to reflect in AgentCore Memory and be ready for retrieval.

Deploy and Test

For deployment, we are going with AWS CDK:

cd cdk && npm install && npm run cdk synth && npm run cdk:deploy 

Once deployed, head over to Bedrock AgentCore on the AWS Console:

Clicking on the available runtime:

To test, run:

agentcore invoke '{"prompt": "Who are you?"}' --session-id 'user-123456-conversation-87654321'           

This time, we pass the session id (user-123456-conversation-87654321) via the --session-id flag, as that's the way via agentcore invoke in prod to pass a unique session id.

The response:

When we ask again with a different session id user-123456-conversation-22345678:

agentcore invoke '{"prompt": "what did I ask last?"}' --session-id 'user-123456-conversation-22345678'

The response:

This shows the change is session id, created a new session.

Now, we ask with the session id user-123456-conversation-87654321 which we used to ask the question “Who are you?”:

agentcore invoke '{"prompt": "what did I ask last?"}' --session-id 'user-123456-conversation-87654321'

And now get this response:

This time it remembers!

Remember to add the needed permissions, like AgentCore Memory, to the AgentCore Runtime role via CDK.

See below:

new iam.PolicyStatement({
                    sid: 'BedrockAgentCoreMemoryAccess',
                    effect: iam.Effect.ALLOW,
                    actions: [
                        'bedrock-agentcore:ListMemories',
                        'bedrock-agentcore:GetMemory',
                        'bedrock-agentcore:CreateMemory',
                        'bedrock-agentcore:UpdateMemory',
                        'bedrock-agentcore:DeleteMemory',
                        'bedrock-agentcore:CreateEvent',
                        'bedrock-agentcore:CreateMemorySession',
                        'bedrock-agentcore:GetMemorySession',
                        'bedrock-agentcore:UpdateMemorySession',
                        'bedrock-agentcore:DeleteMemorySession',
                        'bedrock-agentcore:ListMemorySessions',
                        'bedrock-agentcore:PutMemoryEvent',
                        'bedrock-agentcore:GetMemoryEvent',
                        'bedrock-agentcore:ListMemoryEvents',
                        'bedrock-agentcore:DeleteMemoryEvent',
                        'bedrock-agentcore:ListLongTermMemoryRecords',
                        'bedrock-agentcore:SearchLongTermMemories'
                    ],
                    resources: [
                        `arn:aws:bedrock-agentcore:${region}:${accountId}:memory/*`,
                        `arn:aws:bedrock-agentcore:${region}:${accountId}:memory-session/*`
                    ],
                }),

Observability

LangGraph, in combination with Langsmith, provides a great way to monitor your LangGraph Agents. When combined with CloudWatch Logs and Bedrock AgentCore observability, this provides complete observability, monitoring both application and infrastructure.

The image gives insight into Memory usage while building this Agent.

Cleaning Up

Remember to destroy all resources when done with testing the app.

cdk destroy --all

This command deletes the Bedrock AgentCore Agent, ECR image, the IAM roles used by the agent, and any other resource CDK created to launch the project on AWS.

To delete the memory, run the command below

aws bedrock-agentcore-control delete-memory --memory-id <MEMORY_ID>

Alternatively, head over to the console and navigate to the Amazon Bedrock AgentCore memory, then delete the created memory.

Conclusion

I had fun exploring AgentCore as a first-time user, and hopefully you did as well!

We learnt about this awesome AWS service, and saw how it helps build production-ready AI agents. We also experimented with Memory, one of its core services, and how it changes the game in building AI agents that persist across sessions, retain context, and learn.

Excited to see what you build!

Building Persistent Memory Agents with LangGraph and Amazon Bedrock AgentCore was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

Recently, I’ve been exploring the Langchain ecosystem and how this powerful framework powers a lot of Generative AI applications. This article, my first with Langchain, is my way of consolidating my findings, insights and sharing them with you.

In this article, we will build a knowledge management system using Langchain, Chroma, and Bedrock. Users upload documents (PDFs, Markdown), and chat with an AI assistant that answers questions based on the uploaded documents.

Already excited? Let’s get started

Prerequisites

• AWS account
• AWS CLI setup locally
• Python 3

What would we cover

• We cover what Langchain is
• We cover what Chroma is, and its role in the application
• We cover the role AWS Bedrock plays in the application
• We walk through the Python code.
• We perform some tests to see it in action.

What is Langchain

Simply put, it is an open-source orchestration framework that simplifies the development of applications using large language models (LLMs).

In our application, it acts as the orchestrator, integrating with both AWS Bedrock and Chroma to give an accurate response based on users query. It achieves this using Retrieval-Augmented Generation (RAG) workflow, where it combines the power of LLMs (AWS Bedorck) with external source(Documents embeddings in Chroma).

Learn more about it here

What is Chroma

Chroma, is an open-source vector store used for storing and retrieving vector embeddings.

In our application, we use it to save embeddings along with metadata to be used later by the LLM.

It achieves this by storing chunked documents plus metadata in a local persistent vector store (chroma_db/ on disk), then uses the AWS Bedrock embedding vectors (generated at ingest time) as its index.

Given a user question, it runs a similarity search over that index and returns the top‑k Document objects, preserving metadata like source file, chunk index, etc.

What is AWS Bedrock

AWS Bedrock, is a fully managed, serverless service that provides a simple way to build and deploy Gen AI applications. It gives developers access to a variety of industry-leading foundation models (FMs), which can be customized, all while integrating with other AWS services.

We would focus on the foundational model it provides us with in this article. Acting as the brain power within the application, it plays two key roles:

• Embeddings model (via, amazon.titan-embed-text-v1): transforms each chunk into a vector during ingestion so Chroma can index/search it later
• Chat model (via, anthropic.claude-3-sonnet-20240229-v1:0): receives the retrieved context plus the user’s question and produces the natural-language answer

Project Setup & Overview

The folder structure is designed as follows:

1. The src contains the main files of the app. Here, we define the AWS Bedrock Embedding model, the document processing system, the vector store module, and how we retrieve response using the query system.
2. The chroma_db is used by Chroma as a persistent storage.
3. The app.py file is the entry point of the application and returns a GUI powered by Streamlit.

How it works

User Uploads Documents > Document Processor (PDF/Markdown) > Text Chunking > Bedrock Embeddings > ChromaDB Vector Store (Persistent)

Next, when a user queries the AI assistant, it follows this approach.

Question > LangChain RAG Chain (adds Prompt and retrieves relevant context from Chroma) > AWS Bedrock LLM > Returns Answer to user

We will review some parts of the codebase, not all of it, however.

Creating the Document Processor

"""Document processing module for PDF and Markdown files."""
import os
from typing import List
from pathlib import Path
from pypdf import PdfReader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

class DocumentProcessor:
    """Process PDF and Markdown documents for ingestion."""
    
    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        """Initialize the document processor.
        
        Args:
            chunk_size: Size of text chunks
            chunk_overlap: Overlap between chunks
        """
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
        )
    
    def process_pdf(self, file_path: str) -> List[Document]:
        """Extract text from a PDF file and split into chunks.
        
        Args:
            file_path: Path to the PDF file
            
        Returns:
            List of Document objects
        """
        documents = []
        try:
            with open(file_path, "rb") as file:
                pdf_reader = PdfReader(file)
                text = ""
                for page in pdf_reader.pages:
                    text += page.extract_text() + "\n"
                
                # Split text into chunks
                chunks = self.text_splitter.split_text(text)
                
                # Create Document objects with metadata
                file_name = Path(file_path).name
                for i, chunk in enumerate(chunks):
                    documents.append(
                        Document(
                            page_content=chunk,
                            metadata={
                                "source": file_name,
                                "file_path": file_path,
                                "chunk_index": i,
                                "file_type": "pdf",
                            }
                        )
                    )
        except Exception as e:
            raise Exception(f"Error processing PDF {file_path}: {str(e)}")
        
        return documents
    
    def process_markdown(self, file_path: str) -> List[Document]:
        """Extract text from a Markdown file and split into chunks.
        
        Args:
            file_path: Path to the Markdown file
            
        Returns:
            List of Document objects
        """
        documents = []
        try:
            with open(file_path, "r", encoding="utf-8") as file:
                text = file.read()
            
            # Split text into chunks
            chunks = self.text_splitter.split_text(text)
            
            # Create Document objects with metadata
            file_name = Path(file_path).name
            for i, chunk in enumerate(chunks):
                documents.append(
                    Document(
                        page_content=chunk,
                        metadata={
                            "source": file_name,
                            "file_path": file_path,
                            "chunk_index": i,
                            "file_type": "markdown",
                        }
                    )
                )
        except Exception as e:
            raise Exception(f"Error processing Markdown {file_path}: {str(e)}")
        
        return documents
    
    def process_file(self, file_path: str) -> List[Document]:
        """Process a file based on its extension.
        
        Args:
            file_path: Path to the file
            
        Returns:
            List of Document objects
        """
        file_ext = Path(file_path).suffix.lower()
        
        if file_ext == ".pdf":
            return self.process_pdf(file_path)
        elif file_ext in [".md", ".markdown"]:
            return self.process_markdown(file_path)
        else:
            raise ValueError(f"Unsupported file type: {file_ext}")

From the above, we set some methods that process the uploaded file, breaks it into text chunks and finally returns this chunks alongside some metadata.

Creating Embeddings with AWS Bedrock

"""Bedrock embeddings integration module."""
from typing import List
from langchain_aws import BedrockEmbeddings
import config


class BedrockEmbeddingService:
    """Service for generating embeddings using AWS Bedrock."""
    
    def __init__(
        self,
        model_id: str = None,
        region_name: str = None,
        aws_access_key_id: str = None,
        aws_secret_access_key: str = None,
    ):
        """Initialize the Bedrock embedding service.
        
        Args:
            model_id: Bedrock embedding model ID
            region_name: AWS region
            aws_access_key_id: AWS access key ID
            aws_secret_access_key: AWS secret access key
        """
        self.model_id = model_id or config.BEDROCK_EMBEDDING_MODEL_ID
        self.region_name = region_name or config.AWS_REGION
        self.aws_access_key_id = aws_access_key_id or config.AWS_ACCESS_KEY_ID
        self.aws_secret_access_key = aws_secret_access_key or config.AWS_SECRET_ACCESS_KEY

        # Initialize Bedrock embeddings
        credentials = {}
        if self.aws_access_key_id:
            credentials["aws_access_key_id"] = self.aws_access_key_id
        if self.aws_secret_access_key:
            credentials["aws_secret_access_key"] = self.aws_secret_access_key
      
        self.embeddings = BedrockEmbeddings(
            model_id=self.model_id,
            region_name=self.region_name,
            **credentials
        )
        print(f"AWS Access Key ID: {self.aws_access_key_id}")
        print(f"AWS Secret Access Key: {self.aws_secret_access_key}")
        print(f"AWS Region: {self.region_name}")
        print(f"AWS Model ID: {self.model_id}")

    
    def get_embeddings_model(self):
        """Get the LangChain embeddings model instance.
        
        Returns:
            BedrockEmbeddings instance
        """
        return self.embeddings

This sets up AWS Bedrock as the embedding model which would later be used by Chroma.

Creating the Vector store

"""ChromaDB vector store integration module."""
from typing import List
import chromadb
from chromadb.config import Settings
from langchain_chroma import Chroma
from langchain_core.documents import Document
import config
from src.bedrock_embeddings import BedrockEmbeddingService


class VectorStore:
    """ChromaDB vector store for document embeddings."""
    
    def __init__(self, collection_name: str = None, reset: bool = False):
        """Initialize the vector store.
        
        Args:
            collection_name: Name of the ChromaDB collection
            reset: Whether to reset the collection if it exists
        """
        self.collection_name = collection_name or config.COLLECTION_NAME
        self.persist_directory = config.CHROMA_PERSIST_DIRECTORY
        
        # Initialize embedding service
        self.embedding_service = BedrockEmbeddingService()
        
        # Initialize ChromaDB client
        self.client = chromadb.PersistentClient(
            path=self.persist_directory,
            settings=Settings(
                anonymized_telemetry=False,
                allow_reset=True
            )
        )
        
        # Initialize or get collection
        if reset:
            try:
                self.client.delete_collection(name=self.collection_name)
            except:
                pass
        
        # Create LangChain Chroma vector store
        self.vector_store = Chroma(
            collection_name=self.collection_name,
            embedding_function=self.embedding_service.get_embeddings_model(),
            persist_directory=self.persist_directory,
            client=self.client,
        )
    
    def add_documents(self, documents: List[Document]) -> List[str]:
        """Add documents to the vector store.
        
        Args:
            documents: List of Document objects to add
            
        Returns:
            List of document IDs
        """
        return self.vector_store.add_documents(documents)
    
    def get_collection_info(self) -> dict:
        """Get information about the collection.
        
        Returns:
            Dictionary with collection information
        """
        try:
            collection = self.client.get_collection(name=self.collection_name)
            count = collection.count()
            return {
                "name": self.collection_name,
                "document_count": count,
                "status": "active"
            }
        except Exception as e:
            return {
                "name": self.collection_name,
                "document_count": 0,
                "status": "not_found",
                "error": str(e)
            }

This initializes the store, and provides a method for adding documents(chunks) to the vector store.

Retrieving results with the Query system

"""LangChain query and chat module."""
from typing import List
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_aws import ChatBedrock
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
import config
from src.vector_store import VectorStore


class QueryChain:
    """LangChain-based query and chat system."""

    def __init__(self, vector_store: VectorStore):
        """Initialize the query chain.

        Args:
            vector_store: VectorStore instance for retrieval
        """
        self.vector_store = vector_store

        # Initialize Bedrock chat model
        credentials = {}
        if config.AWS_ACCESS_KEY_ID:
            credentials["aws_access_key_id"] = config.AWS_ACCESS_KEY_ID
        if config.AWS_SECRET_ACCESS_KEY:
            credentials["aws_secret_access_key"] = config.AWS_SECRET_ACCESS_KEY

        self.llm = ChatBedrock(
            model_id=config.BEDROCK_CHAT_MODEL_ID,
            region_name=config.AWS_REGION,
            model_kwargs={"temperature": 0.7, "max_tokens": 2000},
            **credentials
        )

        # Create custom prompt template
        self.prompt_template = PromptTemplate(
            template="""You are a helpful AI assistant that answers questions based on the provided context from company documents.

Context from documents:
{context}

Question: {question}

Instructions:
- Answer the question based only on the provided context
- If the context doesn't contain enough information, say so
- Be concise and accurate
- Use bullet points or numbered lists when appropriate
- If asked about something not in the context, politely explain that you don't have that information

Answer:""",
            input_variables=["context", "question"]
        )

        # Create retriever
        self.retriever = self.vector_store.vector_store.as_retriever(
            search_kwargs={"k": 4}
        )
        
        # Create RAG chain using RunnablePassthrough pattern
        def format_docs(docs):
            return "\n\n".join(doc.page_content for doc in docs)
        
        self.qa_chain = (
            {
                "context": self.retriever | format_docs,
                "question": RunnablePassthrough(),
            }
            | self.prompt_template
            | self.llm
            | StrOutputParser()
        )

    def query(self, question: str) -> dict:
        """Query the knowledge base with a question.

        Args:
            question: User's question

        Returns:
            Dictionary with answer and source documents
        """
        # Get answer from chain
        answer = self.qa_chain.invoke(question)
        
        # Get source documents separately
        source_documents = self.retriever.invoke(question)

        return {
            "answer": answer,
            "source_documents": source_documents,
        }

Lastly, this combines the users query, Prompt template, retieved context, and LLM to provide a valid answer to the user.

Getting Started

The source code can be found here.

Create a virtual environment, and install dependencies:

cd python-langchain

python3 -m venv venv
source venv/bin/activate


pip install -r requirements.txt

Set up environment variables:

cp .env.example .env

Edit .env and add your AWS credentials:

AWS_REGION=us-east-1 
AWS_ACCESS_KEY_ID=your_access_key_here 
AWS_SECRET_ACCESS_KEY=your_secret_key_here 
BEDROCK_MODEL_ID=amazon.titan-embed-text-v1 
BEDROCK_CHAT_MODEL_ID=anthropic.claude-3-sonnet-20240229-v1:0
CHROMA_PERSIST_DIRECTORY=./chroma_db
CHUNK_SIZE=1000
CHUNK_OVERLAP=200

Then launch the app:

streamlit run app.py

Testing the app

After running the streamlit application, the image below should display as the landing page.

Click the “Initialize/Reset Vector Store” at the top left corner to proceed.

Currently there are no active documents in the store, lets fix that by testing the application.

To test the app, upload the sample_document.pdf file, and click the Load Documents button to begin the indexing process.

Now that is done, time to test the application!

Query 1

User: What is the contact info of the Engineering and HR team?

Assistant: According to the provided context from the company documents, the contact information for the Engineering and HR teams is:

• Engineering: engineering@company.com
• HR: hr@company.com

As can be seen, it also provided the Source Document/Chunks where it retrived this information from.

Looking at the source material, we can confirm we got back an accurate result.

Query 2

User: What are the team’s core working hours, and when should PTO requests be submitted?

Assistant: Based on the provided context, here are the relevant details:

• Core working hours:
• Core hours are 10 AM — 3 PM EST for team collaboration
• All employees must be available during these core hours
• PTO requests:
• PTO requests should be submitted at least 2 weeks in advance

Looking at the source material, we can confirm we got back an accurate result.

What’s Next?

In my next article, we will learn how to deploy this application for production using Amazon Bedrock AgentCore, and also tweak some parts of the project to experiment with other tools and approaches.

This would enable end users to test the app from anywhere in the world. Anticipate!

You can find the complete source code here

Build an AI Assistant with LangChain, Chroma & AWS Bedrock was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we build a production-ready, cloud-native observability stack on AWS ECS Fargate using CDK, OpenTelemetry, ADOT, and Grafana. This article is an upgrade to the previous article here, where we solely focused on logging and visualization using Loki and Grafana.

Already excited? Let’s get started

Prerequisites

• AWS account
• AWS CLI setup locally
• AWS CDK set up.
• Python3
• Docker
• GitHub Actions
• Previous article here

Case Study: Trimbal

From the previous article, we saw Trimbal implement a cloud-native logging and visualization solution using Loki and Grafana.

Management has seen the great impact it had, and wants a complete implementation of full observability using the right tools.

After research, you learnt about OpenTelemetry and decided to hop on the train and utilize it in this app. You’ve done the research, validated the approach, and now it’s time to design the architecture and start building.

Your proposed solution is:

• Logs: Loki for log aggregation
• Metrics: Amazon Managed Service for Prometheus(AMP) for Prometheus metrics
• Traces: AWS X-Ray for distributed tracing
• Visualization: Grafana dashboards

What is OpenTelemetry?

OTel, for short, is an open-source vendor-neutral framework of tools, SDKs, APIS for the generation, collection, and exportation of telemetry data like logs, metrics, and traces. This standard(under CNCF) provides a unified way to instrument applications for observability, without being tied down to a specific vendor.

One standard for metrics, logs, and traces means less complexity, more visibility.

We would be making use of OTel to add full observability to the current implementation. Even AWS is moving away from using X-Ray SDK for collecting traces, and focusing on open-source — OTel. More on that here.

What is AWS Distro for OpenTelemetry?

ADOT, for short, is a secure, production-ready, AWS-supported distribution of the OpenTelemetry project.

From the official page, we can see that:

With AWS Distro for OpenTelemetry, you can instrument your applications just once to send correlated metrics and traces to multiple AWS and Partner monitoring solutions. Use auto-instrumentation agents to collect traces without changing your code. AWS Distro for OpenTelemetry also collects metadata from your AWS resources and managed services, so you can correlate application performance data with underlying infrastructure data, reducing the mean time to problem resolution

We would be making use of ADOT to instrument the current application running on Amazon Elastic Container Service (ECS).

This would be possible by implementing a sidecar pattern, where the ADOT collector container would forward metrics and traces to Amazon Managed Service for Prometheus(AMP) and AWS X-Ray ,respectively, similar to what with did with Firelens & Loki to forward logs to Loki.

Current Architecture

Proposed Architecture

We can notice some new additions:

1. Introduced a new sidecar — The ADOT sidecar collector for forwarding metrics and traces to AWS.
2. Grafana + Loki + AMP + X-Ray integration: Gives us full observability in our app.
3. Added AMP workspace, SQS queues, and DynamoDB table.

Project Overview

We will review some parts of the codebase, not all, however.

ECS Stack

....   
# Add Add AMP permissions, CloudWatch permissions and  X-Ray metrics)
        task_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("AmazonPrometheusFullAccess"))
        execution_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("AmazonPrometheusFullAccess"))
        task_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("CloudWatchFullAccess"))
        task_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("AWSXRayDaemonWriteAccess"))
        task_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("AWSXrayReadOnlyAccess"))


# Logger app container
        logger_container=logger_task_def.add_container(
            "LoggerAppContainer",
            image=ecs.ContainerImage.from_ecr_repository(ecr_logger),
            essential=True,
            logging=ecs.LogDrivers.firelens(
                options={
                    "Name": "grafana-loki",
                    "Url": "http://loki.internal.com/loki/api/v1/push",
                    #"Labels": "job=firelens-logs,environment=production",
                    "LabelKeys": "container_name,ecs_task_definition,source,ecs_cluster",
                    "LineFormat": "key_value",
                    "RemoveKeys": "container_id,ecs_task_arn"
                }
            ),
            environment={
                "OTEL_SERVICE_NAME": "logger-app",
                "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:4317",
                "OTEL_TRACES_SAMPLER": "parentbased_always_on",
                "AWS_REGION": "us-east-1",
                "SQS_MESSAGE_QUEUE_URL": sqs_stack.message_queue.queue_url,
                "DYNAMODB_APP_TABLE": dynamodb_stack.app_table.table_name,
                
            }
        )



    adot_collector_container = logger_task_def.add_container(
            "AdotCollector",
            image=ecs.ContainerImage.from_registry("public.ecr.aws/aws-observability/aws-otel-collector:latest"),
            essential=False,
            logging=ecs.LogDriver.aws_logs(stream_prefix="adot-collector"),
            environment={
                "AWS_REGION": "us-east-1",
                "AWS_XRAY_TRACING_NAME": "logger-app",
                "OTEL_RESOURCE_ATTRIBUTES": "service.name=logger-app,service.version=1.0.0",
                "AOT_CONFIG_CONTENT": f"""receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  awsecscontainermetrics:
    collection_interval: 10s
processors:
  batch:
    timeout: 1s
    send_batch_size: 1024
  filter:
    metrics:
      include:
        match_type: strict
        metric_names:
          - ecs.task.memory.utilized
          - ecs.task.memory.reserved
          - ecs.task.cpu.utilized
          - ecs.task.cpu.reserved
          - ecs.task.network.rate.rx
          - ecs.task.network.rate.tx
          - ecs.task.storage.read_bytes
          - ecs.task.storage.write_bytes
exporters:
  awsxray:
    region: us-east-1
  prometheusremotewrite:
    endpoint: {amp_workspace.attr_prometheus_endpoint}api/v1/remote_write
    auth:
      authenticator: sigv4auth
extensions:
  health_check:
  pprof:
    endpoint: :1888
  zpages:
    endpoint: :55679
  sigv4auth:
    region: us-east-1
    service: aps
service:
  extensions: [pprof, zpages, health_check, sigv4auth]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [awsxray]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [prometheusremotewrite]
    metrics/ecs:
      receivers: [awsecscontainermetrics]
      processors: [filter]
      exporters: [prometheusremotewrite]"""
            }
        )

        adot_collector_container.add_port_mappings(
            ecs.PortMapping(container_port=4317, protocol=ecs.Protocol.TCP)
        )
        adot_collector_container.add_port_mappings(
            ecs.PortMapping(container_port=4318, protocol=ecs.Protocol.TCP)
        )
        adot_collector_container.add_port_mappings(
            ecs.PortMapping(container_port=1888, protocol=ecs.Protocol.TCP)
        )
        adot_collector_container.add_port_mappings(
            ecs.PortMapping(container_port=55679, protocol=ecs.Protocol.TCP)
        )
.....

We can notice some new additions to the current ECS stack, some of which are not captured above.

However, let's highlight the following additions:

• Added necessary permissions to the Task and Execution role.
• Added required environment variables: OTEL_EXPORTER_OTLP_ENDPOINT, and others to the logger container.
• Added the ADOT collector sidecar container as a sidecar to the logger app. As can be noticed, the collector configuration receives some variables, AOT_CONFIG_CONTENT the most important, which defines most of the configuration the container requires. Configurations like where to write the Prometheus metrics to, send the X-Ray traces to, the authentication method to use, and more. This is where you could define third-party vendors as receivers, as defined by OTel.
• Exposed required ports that the ADOT collector container uses.

AMP Stack

# AMP (Amazon Managed Prometheus) Stack module
from aws_cdk import (
    aws_aps as amp,
    aws_iam as iam,
    Stack
)
from constructs import Construct

class AmpStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        # Create AMP Workspace
        self.workspace = amp.CfnWorkspace(
            self, "PrometheusWorkspace",
            alias="logger-app-prometheus"
        )

Creates the Amazon Managed Service for Prometheus workspace, which would be used to receive metrics from the application.

SQS Stack

# SQS Stack module
from aws_cdk import (
    aws_sqs as sqs,
    aws_iam as iam,
    Stack,
    Duration
)
from constructs import Construct

class SqsStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        # Create DLQ separately so we can reference its ARN
        self.message_dlq = sqs.Queue(
            self, "LoggerAppMessageQueueDLQ",
            queue_name="logger-app-messages-dlq",
            retention_period=Duration.days(14)
        )

        # Create SQS Queue for message processing
        self.message_queue = sqs.Queue(
            self, "LoggerAppMessageQueue",
            queue_name="logger-app-messages",
            visibility_timeout=Duration.seconds(30),
            retention_period=Duration.days(14),
            receive_message_wait_time=Duration.seconds(20),  # Long polling
            dead_letter_queue=sqs.DeadLetterQueue(
                max_receive_count=3,
                queue=self.message_dlq
            )
        )

Creates the SQS queues used in the logger app to receive messages. This was introduced so we could see traces as the application interacts with SQS, mocking production use cases.

DynamoDB Stack

# DynamoDB Stack module
from aws_cdk import (
    aws_dynamodb as dynamodb,
    aws_iam as iam,
    Stack,
    RemovalPolicy
)
from constructs import Construct

class DynamoDbStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        # Create DynamoDB table for application data
        self.app_table = dynamodb.Table(
            self, "LoggerAppTable",
            table_name="logger-app-data",
            partition_key=dynamodb.Attribute(
                name="id",
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(
                name="timestamp",
                type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY,
            stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES
        )

        # Add Global Secondary Index for querying by user
        self.app_table.add_global_secondary_index(
            index_name="user-index",
            partition_key=dynamodb.Attribute(
                name="user_id",
                type=dynamodb.AttributeType.STRING
            ),
            sort_key=dynamodb.Attribute(
                name="created_at",
                type=dynamodb.AttributeType.STRING
            )
        )

Creates the DynamoDB table and indexes used in the logger app to save data. This was introduced so we could see traces as the application interacts with dynamodb, mocking production use cases.

Logger Web App

A lot of changes were made to the logger app, like addition of new endpoints, integration with Amazon SQS and Amazon DynamoDB, AWS SDK auto-instrumentation, Application instrumentation, implementation of metrics and traces collection and exportation via the OTLP(OTel Protocol).

This enabled the simulation of a real-world scenario where an application would interact with multiple AWS services, as well as generate custom metrics and traces

Deploy Stack

Remember to rebuild the Docker images and redeploy to ECR, as we made some new changes.

Run the following:

./bootstrap-ecr-images.sh \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-logger-repo \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-grafana-repo \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-loki-repo

Since we added 3 new stacks (AmpStack, DynamoDbStack, SqsStack), we need to add them to the last deploy step we have from the previous article:

cdk deploy EcsStack Route53Stack AmpStack DynamoDbStack SqsStack -c vpn_server_cert_arn=$SERVER_CERT_ARN -c vpn_client_cert_arn=$CLIENT_CERT_ARN

Replace the $SERVER_CERT_ARN and $CLIENT_CERT_ARN with their respective values.

Testing the app

To test the app, head over to EC2 and click on the public load balancer that the CDK stack created. You should see the public DNS name.

Copy this DNS name, and enter it into a browser or send a GET request. This should show the app landing page running and returning logs!

Next, connect to the Client VPN using OpenVPN to access the private VPC and access the private Route53 Grafana and Loki hosted zone.

N:B: Follow the guide for this setup if you haven't already, from the previous article

Here test Grafana and see the 3 layers, logs, metrics, and traces.

Visiting the base route /

After visiting the logger app via its ALB on the route /, we then proceed to see the generated logs on Loki via the Grafana dashboard:

For Metrics, we see the system-specific metric as well:

Visiting the route /send-message

Next, we make a POST request to the /send-message route to send a message to the SQS queue.

This returns this response:

For Metrics, we see:

When we hit the POST request a second time, we see the update:

and via DrillDown:

For the Traces, we would check both Grafana and AWS X-Ray.

For Grafana, we see:

Scrolling down, each span contains relevant information that could be further explored by expanding it.

For X-Ray/AWS Application Signals, we see:

Looking more into a selected trace, we see more information about the request:

We can also see more details for a selected trace, as well as any metadata it has.

A point to note is that custom metadata can be added via the logger app as well.

Visiting the route /receive-messages

Next, we make a GET request to the /receive-message route to simulate sending a message to the SQS queue.

This returns this response:

Checking the logs:

Checking the Metrics after hitting the /send-message route a few times:

Checking the Traces on Grafana:

Expanding:

Visiting the route /save-data

Next, we make a POST request to the /save-data route to save some data into AWS DynamoDB.

We copy the corresponding id it returns, as we would need it when hitting the get-data route next:

We would also review the metrics and traces in the next section when we attempt to retrieve the data.

Visiting the route /get-data/{data_id}

Next, we make a GET request to the /get-data/{data-id} route to fetch the saved data from AWS DynamoDB.

Use the id we got from saving the data as the {data-id} to retrieve the data we stored in the previous section:

As can be seen, this returns a 404, even though the data exists; this is because we simulated a 404 error, as during querying Dynamodb, we search for timestamp:latest, instead of the right timestamp.

Now, let's see the metrics and traces for this simulated 404 response:

Metrics:

Traces:

Expanding, we can see the color change in the Node Graph, as well as the red error on the get_dynamodb_data span, indicating the trace has an issue, just as we expected!

Visiting the route /workflow

Lastly, we make a GET request to the /workflow route to simulate a complete flow involving SQS and DynamoDB.

Checking the Metrics:

Checking the Traces:

This shows all green, indicating we have successfully simulated the full workflow:

Expanding, we see more details:

And that's a wrap for testing the application.

We were able to simulate a production use case involving interaction with multiple AWS services, a failure and success scenario, and get to see the full observability in action!

Testing GitHub workflow

We added the Trivy workflow file (.github/workflows/trivy-scan.yml) to the existing GitHub workflow to enable security scanning.

Once we push to GitHub, we see the pipeline passes and deploys to AWS.

Conclusion

This journey has been both challenging and rewarding. We successfully implemented OpenTelemetry(OTel) to achieve comprehensive observability across our application, enabling real-time anomaly detection and maintaining robust security standards. Along the way, we’ve streamlined developer workflows and applied industry best practices to create a production-ready serverless observability solution.

We successfully combined AWS CDK, OpenTelemetry, ADOT, and Grafana to provide a solid foundation for monitoring modern containerized applications on AWS ECS. By following the patterns and practices outlined in this article, you can also build similar observability solutions that scale with your infrastructure needs.

You can find the complete source code here

Excited to see what you build!

AWS ECS Observability with CDK, OpenTelemetry, ADOT, and Grafana was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we build a production-ready, cloud-native logging and visualization stack on AWS ECS Fargate with Loki, Grafana, Firelens, GitHub(CI/CD), and CDK for automation.

Already excited? Let’s get started!

What would we cover

This article stemmed from my own need to get hands-on in multiple AWS core services like Advanced Networking, Security, Storage, Containers/CloudNative, Serverless, and industry best practices like IaC, CI/CD using GitHub, OIDC for third-party authentication, and enabling secure, encrypted connectivity to private VPC networks and AWS resources using AWS Client VPN.

This article aims to eliminate multiple pain points usually encountered when building apps, like manual configuration, lack of scalability, and insecure patterns, among others.

We would get to achieve the following:

• Build an app using AWS CDK as IaC
• Implement security architecture by using VPC, IAM, EFS access point, Security groups, OIDC, AWS Client VPC, OpenSSL, ACM, Private Hosted Zone, and Internal Load Balancer.
• Implement FireLens for log routing using a sidecar pattern.
• Use Loki + Grafana for Logging and Visualization.
• Implement Persistent Storage using EFS and S3.
• Cloud-native solution using Docker, ECR, and ECS.
• Integrate Continuous Integration & Continuous Deployment(CI/CD) using GitHub Actions.

Prerequisites

• AWS account
• AWS CLI setup locally
• AWS CDK set up.
• Python3
• Docker
• GitHub Actions

Case Study: Trimbal

Our imaginary company, Trimbal, is preparing to deploy its cloud-native application. The team has chosen AWS ECS for container orchestration and initially relies on the default logging solution provided by CloudWatch. Everything works fine at first, but as traffic grows, the company realizes it has limited visibility into its logs. Issues start to surface:

• No real-time log streaming
• Limited search capabilities
• A clunky UI with little room for customization makes it difficult for internal teams to gain actionable insights into customer data.

Management begins exploring alternatives and hears a lot about Grafana and Loki as powerful solutions for log visualization and centralized logging. The goal is to implement these tools securely while keeping infrastructure costs under control.

After reviewing multiple success stories from other companies using Grafana and Loki, the decision is clear — Trimbal will adopt this stack. You’ve done the research, validated the approach, and now it’s time to design the architecture and start building.

Proposed Architecture

We can see that:

1. Public users visit the Logger Web app via the public-facing load balancer.

• Logger ECS service forwards logs to Loki running in private subnet via Route53 private zone record (loki.internal.com)
• Private Zone points to the internal Load Balancer and routes to the Loki ECS service
• Loki uses EFS for persistent storage, utilizing the file system and access point, and S3 for log archiving

2. Authenticated user connects to VPC using any VPN client via the AWS Client VPN credentials

• Establishing a connection with VPC
• Visits grafana.internal.com
• Gets routed to the Internal Load Balancer via the Route 53 private zone record (grafana.internal.com)
• Internal Load Balancer points to the Grafana ECS service running in the private subnet
• Grafana uses EFS for persistent storage, utilizing the file system and access point

3. GitHub Actions for CI/CD

• Authenticated using OIDC
• Once changes are pushed to the main branch(ignoring some files), a new workflow is triggered
• It builds the Docker images for Logger, Loki, and Grafana
• It runs CDK deploy to deploy any new stack changes
• It updates the ECS task definition for Logger, Loki, and Grafana to use the latest build Docker images, and subsequently updates each service to use the latest task definition.

Why this architecture?

Why EFS?

• Used for persistent data storage for both Grafana and Loki.
• Ensures that Grafana dashboards, users, sessions, and settings are preserved even after ECS service redeployments.
• Stores Loki WAL, index, and cache, which are crucial for fault-tolerance and high availability.
• Automatically replicated across AZs.

Why S3?

• Loki writes compressed log chunks and indexes to S3
• Cost-effective and scalable storage option vs local ECS disk.
• Enables long-term log retention and durability.

FireLens Sidecar Pattern

What is it?

• FireLens is an AWS integration with Fluent Bit that enables flexible log routing from ECS containers.

Why use it?

• Offers a sidecar container that forwards logs directly to Loki.
• Supports log transformation, filtering, and dynamic labeling (e.g., environment, container name).

AWS CDK

The AWS Cloud Development Kit (AWS CDK) is an open-source software development framework for defining cloud resources as code.

Run this command to set up a new CDK application.

sudo npm install -g aws-cdk && cdk init app - language python

Follow the prompt to complete the setup process.

Project Overview

The folder structure is designed as follows:

1. The .github/workflows contains the GitHub action workflow file called logger-deploy.yml.
2. The app/config contains the Docker files and config for both Loki and Grafana.
3. The app/modules contains all the stacks used in the project; that's where most logic for CDK exists.
4. The app/scripts contains the script (generate_and_import_vpn_certs.sh) that generate the certificates that are later used when creating the Client VPN Endpoint. It generates them using OpenSSL and uploads to ACM.
5. The sample_logger.py contains the logic for the logger app.
6. The architecture contains the architecture_diagram.png file that shows the UI architecture of the project

We will review some parts of the codebase, not all, however.

Logger Web App

import logging
import random
import time
from datetime import datetime
from fastapi import FastAPI, Request
import threading
import uvicorn

# Configure standard logger
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s %(message)s',
    datefmt='%Y-%m-%dT%H:%M:%S%z'
)
logger = logging.getLogger("sample_logger")

LOG_LEVELS = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL]
LOG_LEVEL_NAMES = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
SAMPLE_MESSAGES = [
    "User login succeeded",
    "User login failed",
    "File uploaded successfully",
    "File upload failed",
    "Database connection established",
    "Database connection lost",
    "Cache miss",
    "Cache hit",
    "API request received",
    "API request failed",
    "Background job started",
    "Background job completed",
    "Unexpected exception occurred"
]

app = FastAPI()

def random_log():
    while True:
        level = random.choice(LOG_LEVELS)
        level_name = LOG_LEVEL_NAMES[LOG_LEVELS.index(level)]
        msg = random.choice(SAMPLE_MESSAGES)
        logger.log(level, f"[{level_name}] {msg}")
        time.sleep(30)

@app.get("/")
async def index(request: Request):
    now = datetime.utcnow().isoformat()
    client_host = request.client.host
    logger.info(f"[INFO] Web endpoint visited at {now} from {client_host}. Deployed with Github workflow")
    return {"message": f"Logger app running. Visit time: {now}. Deployed with Github workflow"}

if __name__ == "__main__":
    t = threading.Thread(target=random_log, daemon=True)
    t.start()
    uvicorn.run(app, host="0.0.0.0", port=8080)

From the above, we spin a server that outputs various levels of logs randomly, and also when a user visits the web app.

Route53 Stack

from aws_cdk import (
    aws_route53 as route53,
    aws_route53_targets as targets,
    Stack
)
from constructs import Construct

class Route53Stack(Stack):
    def __init__(self, scope: Construct, id: str, vpc, internal_alb, **kwargs):
        super().__init__(scope, id, **kwargs)

        # Private hosted zone
        self.internal_zone = route53.PrivateHostedZone(
            self, "InternalHostedZone",
            zone_name="internal.com",
            vpc=vpc
        )
        # A records for loki and grafana
        self.loki_record = route53.ARecord(
            self, "LokiInternalRecord",
            zone=self.internal_zone,
            record_name="loki.internal.com",
            target=route53.RecordTarget.from_alias(targets.LoadBalancerTarget(internal_alb))
        )
        self.grafana_record = route53.ARecord(
            self, "GrafanaInternalRecord",
            zone=self.internal_zone,
            record_name="grafana.internal.com",
            target=route53.RecordTarget.from_alias(targets.LoadBalancerTarget(internal_alb))
        )

Creates the Private Zone, as well as A records within this zone that point to the internal Load balancer for both Loki and Grafana.

ClientVPN Stack

from aws_cdk import (
    aws_ec2 as ec2,
    Stack,
    CfnOutput
)
from constructs import Construct

class ClientVpnStack(Stack):
    def __init__(self, scope: Construct, id: str, vpc: ec2.Vpc, client_vpn_sg,  **kwargs):
        super().__init__(scope, id, **kwargs)

        # Context values for certificates
        server_cert_arn = self.node.try_get_context("vpn_server_cert_arn") or "arn:aws:acm:REGION:ACCOUNT:certificate/SERVER_CERT_ID"
        client_cert_arn = self.node.try_get_context("vpn_client_cert_arn") or "arn:aws:acm:REGION:ACCOUNT:certificate/CLIENT_CERT_ID"

        # ✅ Client VPN Endpoint
        self.client_vpn = ec2.CfnClientVpnEndpoint(
            self, "ClientVpnEndpoint",
            authentication_options=[{
                "type": "certificate-authentication",
                "mutualAuthentication": {
                    "clientRootCertificateChainArn": client_cert_arn
                }
            }],
            client_cidr_block="10.100.0.0/16",
            connection_log_options={"enabled": False},
            server_certificate_arn=server_cert_arn,
            vpc_id=vpc.vpc_id,
            security_group_ids=[client_vpn_sg.security_group_id],
            split_tunnel=True,
            # Uses VPC DNS so internal Route53 works
            dns_servers=["10.0.0.2"],
            description="Client VPN for secure VPC access"
        )

        # ✅ Associate VPN with private subnets
        for i, subnet in enumerate(vpc.select_subnets(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS).subnets):
            assoc = ec2.CfnClientVpnTargetNetworkAssociation(
                self, f"ClientVpnAssoc{i}",
                client_vpn_endpoint_id=self.client_vpn.ref,
                subnet_id=subnet.subnet_id
            )
        
        # ✅ Authorization rule: allow all users (for now)
        ec2.CfnClientVpnAuthorizationRule(
            self, "ClientVpnAuthRule",
            client_vpn_endpoint_id=self.client_vpn.ref,
            target_network_cidr=vpc.vpc_cidr_block,
            authorize_all_groups=True
        )

        CfnOutput(self, "ClientVpnEndpointId", value=self.client_vpn.ref, description="Client VPN Endpoint ID")

Creates the Client VPN Endpoint, associates it with the existing private subnet in the VPC, and creates an Authorization rule.

Pre-Deploy

We need to run a script to create a certificate that would be imported to ACM and used by our Client VPN to connect to the VPC.

We would be using the Mutual Authentication approach, where the AWS Client VPN uses certificates to perform authentication between the client and the server.

Create a file generate_and_import_vpn_certs.sh, then make it executable in the app/scripts folder.

To do that, run this command:

mkdir app/scripts && cd app/scripts && touch generate_and_import_vpn_certs.sh
chmod +x generate_and_import_vpn_certs.sh

Add this to the file:

#!/bin/bash
set -e

# ---- CONFIG ----
AWS_REGION="us-east-1"
DOMAIN="serverlessvpn"

# ---- GENERATE CERTIFICATES ----
mkdir -p vpn-certs && cd vpn-certs

echo "Generating Root CA..."
openssl genrsa -out clientvpn-rootCA.key 2048
openssl req -x509 -new -nodes -key clientvpn-rootCA.key -sha256 -days 3650 -out clientvpn-rootCA.crt -subj "/CN=ClientVPN-RootCA"

echo "Generating Server Certificate..."
openssl genrsa -out clientvpn-server.key 2048
openssl req -new -key clientvpn-server.key -out clientvpn-server.csr -subj "/CN=server.${DOMAIN}"
openssl x509 -req -in clientvpn-server.csr -CA clientvpn-rootCA.crt -CAkey clientvpn-rootCA.key -CAcreateserial -out clientvpn-server.crt -days 3650 -sha256

echo "Generating Client Certificate..."
openssl genrsa -out clientvpn-client.key 2048
openssl req -new -key clientvpn-client.key -out clientvpn-client.csr -subj "/CN=client.${DOMAIN}"
openssl x509 -req -in clientvpn-client.csr -CA clientvpn-rootCA.crt -CAkey clientvpn-rootCA.key -CAcreateserial -out clientvpn-client.crt -days 3650 -sha256

# ---- IMPORT TO ACM ----
echo "Importing Server Certificate to ACM..."
SERVER_CERT_ARN=$(aws acm import-certificate \
  --certificate fileb://clientvpn-server.crt \
  --private-key fileb://clientvpn-server.key \
  --certificate-chain fileb://clientvpn-rootCA.crt \
  --region $AWS_REGION \
  --query CertificateArn --output text)

echo "Importing Client Root CA to ACM..."
CLIENT_CERT_ARN=$(aws acm import-certificate \
  --certificate fileb://clientvpn-rootCA.crt \
  --private-key fileb://clientvpn-rootCA.key \
  --region $AWS_REGION \
  --query CertificateArn --output text)

echo "----"
echo "Server Certificate ARN: $SERVER_CERT_ARN"
echo "Client Root CA ARN:     $CLIENT_CERT_ARN"
echo "----"
echo "Use these ARNs in your CDK deployment:"
echo "  cdk deploy -c vpn_server_cert_arn=$SERVER_CERT_ARN -c vpn_client_cert_arn=$CLIENT_CERT_ARN"
echo "----"
echo "Distribute these files to VPN users:"
echo "  clientvpn-client.crt, clientvpn-client.key, clientvpn-rootCA.crt"

Then run this script.

This would import the Server Certificate and Client Root CA to ACM.

It also outputs certificate files to the app/scripts/vpn-certs folder. We would copy the contents of some files from this folder when we need to connect to the VPC using the Client VPN.

Deploy the Stack

We would break the deployment into 3 stages.

For the first stage, run the command below to deploy some initial stacks:

cdk deploy VpcStack ClientVpnStack SgStack S3Stack EcrStack AlbStack -c vpn_server_cert_arn=$SERVER_CERT_ARN -c vpn_client_cert_arn=$CLIENT_CERT_ARN

Replace the $SERVER_CERT_ARN and $CLIENT_CERT_ARN with the right value from the Pre-Deploy step above.

Before running the next stack deploy, we need to build images for the logger, Grafana, and Loki services. This would ensure the images are in ECR when ECS tries to start up.

We do this manually for the first time; subsequently, GitHub Actions will be used to build new images and update the task definition and update each service, respectively.

Run the following:

./bootstrap-ecr-images.sh \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-logger-repo \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-grafana-repo \
  <aws-account-id>.dkr.ecr.us-east-1.amazonaws.com/ecrstack-loki-repo

Replace the <aws-account-id> with your AWS account ID.

For the second stage, run the next deploy:

cdk deploy EfsStack EfsAccessPointsStack -c vpn_server_cert_arn=$SERVER_CERT_ARN -c vpn_client_cert_arn=$CLIENT_CERT_ARN

Before running ECS, we need to update the EFS file system permission to enable IAM authentication.

Due to CDK limitations, we can't do this directly in CDK as it would create a circular dependency. Hence, we would do it via CLI.

Run this code to delete the default policy created, so it can use the IAM permissions we have set up.

aws efs delete-file-system-policy --file-system-id fs-<id>

Replace fs-<id> with the valid file system ID

Next, deploy the Route53Stack and the EcsStack.

cdk deploy EcsStack Route53Stack -c vpn_server_cert_arn=$SERVER_CERT_ARN -c vpn_client_cert_arn=$CLIENT_CERT_ARN

Testing the app

To test the app, head over to EC2 and click on the public load balancer that the CDK stack created. You should see the public DNS name.

Copy this DNS name, and enter it into a browser. This should show the app running and returning logs!

Connecting to the VPC via the AWS Client VPN

To confirm that logs are being forwarded securely to Loki, we need to see them. We use Grafana for visibility, which is only accessible from within the VPC.

Visiting the grafana.internal.com URL would return the image below. This is because we are not currently within the VPC, so it looks for that URL on the public internet:

Let's fix this!

We created a ClientVPNStack for this purpose. We will be connecting from our machine using the VPN we created, so as to access the app within the VPC.

We use OpenVPN as the software for connecting; it's quick to download and set up.

After setup, head over to AWS to download the needed configuration. In the Vpc -> Client VPN endpoints, select the client endpoint CDK created, and click the “Download client configuration” button.

This would download a .ovpn file, which is what we would use in OpenVPN to authenticate.

Open the file in a file editor to make the necessary updates to it, and add the following just below the <ca>… </ca>:

<cert>
-----BEGIN CERTIFICATE-----
clientvpn-client.crt contents from app/scripts/vpn-certs

-----END CERTIFICATE-----
</cert>

<key>
-----BEGIN PRIVATE KEY-----
clientvpn-client.key contents from app/scripts/vpn-certs
-----END PRIVATE KEY-----
</key>

Next, save this and upload it to OpenVPN Connect. If everything goes fine, you should see:

Now time to test this out.

Head over to your browser and enter http://grafana.internal.com.

We now see the Grafana login page:

Log in with the default credentials of “admin” for username and “admin” for password.

After logging in, we see that the “Add your first data source” has already been completed. This is because in the Grafana Docker image built, we configured this.

The Docker file included a command that pushed a prebuilt config file to the granafa image.

The Dockfile for Grafana:

FROM grafana/grafana-oss

COPY app/config/grafana/grafana-config.yaml /etc/grafana/provisioning/datasources/datasources.yaml

The grafana-config.yaml file

apiVersion: 1

datasources:
  - name: Loki
    type: loki
    access: proxy
    orgId: 1
    url: http://loki.internal.com:80
    version: 1
    editable: true

Let's explore the logs the logger app generated:

Clicking on the “Show logs” button, we see the logs:

And just like that, we have the full power of Grafana, and we can explore!

We also see different labels:

And once we turn off the connection with OpenVPN, we can no longer access Grafana:

Testing GitHub workflow

The last step is integrating the GitHub workflow. This would fully automate the app to enable redeployment for each ECS service application once there is a change, as well as update the stack once there is a new change.

Create a new file in .github/workflow/ folder called logger-deploy.yml with the following content:

name: Build and Deploy Logger to ECS

on:
  push:
    branches: [ main ]
    paths:
      - 'app/sample_logger.py'
      - 'Dockerfile'
      - '.github/workflows/logger-deploy.yml'
      - 'requirements.txt'
      - 'app.py'
      - 'app/**'
      - 'cdk.json'

env:
  ECR_REPO: ecrstack-logger-repo
  ECR_GRAFANA_REPO: ecrstack-grafana-repo
  ECR_LOKI_REPO: ecrstack-loki-repo
  ECS_CLUSTER: ecsstack-cluster
  LOGGER_SERVICE: logger-service
  GRAFANA_SERVICE: grafana-service
  LOKI_SERVICE: loki-service
  AWS_REGION: us-east-1
  IMAGE_TAG: ${{ github.sha }}
  SERVER_CERT_ARN:  ${{ secrets.SERVER_CERT_ARN }}
  CLIENT_CERT_ARN:  ${{ secrets.CLIENT_CERT_ARN }}

permissions:
      id-token: write
      contents: read

jobs:
  build-and-push:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v2
        with:
          role-to-assume: ${{ secrets.ASSUME_ROLE_ARN }}
          role-session-name: GitHub_to_AWS_via_FederatedOIDC
          aws-region: ${{ env.AWS_REGION }}

      - name: Login to Amazon ECR
        id: login-ecr
        uses: aws-actions/amazon-ecr-login@v2

      - name: Build Logger Docker image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker build -t logger-app:${IMAGE_TAG} .
          docker tag logger-app:${IMAGE_TAG} $ECR_REGISTRY/$ECR_REPO:${IMAGE_TAG}

      - name: Push Logger Docker image to ECR
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker push $ECR_REGISTRY/$ECR_REPO:${IMAGE_TAG}

      - name: Build Grafana Docker image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker build -f app/config/grafana/Dockerfile -t grafana-app:${IMAGE_TAG} .
          docker tag grafana-app:${IMAGE_TAG} $ECR_REGISTRY/$ECR_GRAFANA_REPO:${IMAGE_TAG}

      - name: Push Grafana Docker image to ECR
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker push $ECR_REGISTRY/$ECR_GRAFANA_REPO:${IMAGE_TAG}

      - name: Build Loki Docker image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker build -f app/config/loki/Dockerfile -t loki-app:${IMAGE_TAG} .
          docker tag loki-app:${IMAGE_TAG} $ECR_REGISTRY/$ECR_LOKI_REPO:${IMAGE_TAG}

      - name: Push Loki Docker image to ECR
        env:
          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
        run: |
          docker push $ECR_REGISTRY/$ECR_LOKI_REPO:${IMAGE_TAG}

  deploy:
    needs: build-and-push
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v2
        with:
          role-to-assume: ${{ secrets.ASSUME_ROLE_ARN }}
          role-session-name: GitHub_to_AWS_via_FederatedOIDC
          aws-region: ${{ env.AWS_REGION }}

      - name: Login to Amazon ECR
        id: login-ecr-update
        uses: aws-actions/amazon-ecr-login@v2

      - name: Install CDK
        run: npm install -g aws-cdk

      - name: Install Python dependencies
        run: pip install -r requirements.txt

      - name: Deploy CDK stack
        run: |
          cdk deploy --all \
          -c vpn_server_cert_arn="$SERVER_CERT_ARN" \
          -c vpn_client_cert_arn="$CLIENT_CERT_ARN" \
          --require-approval never
      
      - name: Update Logger ECS service with new image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr-update.outputs.registry }}
        run: |
          TASK_DEF_ARN=$(aws ecs describe-services --cluster $ECS_CLUSTER --services $LOGGER_SERVICE --region $AWS_REGION --query "services[0].taskDefinition" --output text)
          aws ecs describe-task-definition --task-definition $TASK_DEF_ARN --region $AWS_REGION > logger-task-def.json
          cat logger-task-def.json | jq '.taskDefinition | {family, executionRoleArn, taskRoleArn, networkMode, containerDefinitions, requiresCompatibilities, cpu, memory, volumes}' > logger-task-def-min.json
          IMAGE_URI="$ECR_REGISTRY/$ECR_REPO:$IMAGE_TAG"
          cat logger-task-def-min.json | jq --arg IMAGE_URI "$IMAGE_URI" ' .containerDefinitions[0].image = $IMAGE_URI ' > logger-task-def-updated.json
          NEW_TASK_DEF_ARN=$(aws ecs register-task-definition --cli-input-json file://logger-task-def-updated.json --region $AWS_REGION --query 'taskDefinition.taskDefinitionArn' --output text)
          aws ecs update-service --cluster $ECS_CLUSTER --service $LOGGER_SERVICE --task-definition $NEW_TASK_DEF_ARN --region $AWS_REGION

      - name: Update Grafana ECS service with new image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr-update.outputs.registry }}
        run: |
          TASK_DEF_ARN=$(aws ecs describe-services --cluster $ECS_CLUSTER --services $GRAFANA_SERVICE --region $AWS_REGION --query "services[0].taskDefinition" --output text)
          aws ecs describe-task-definition --task-definition $TASK_DEF_ARN --region $AWS_REGION > grafana-task-def.json
          cat grafana-task-def.json | jq '.taskDefinition | {family, executionRoleArn, taskRoleArn, networkMode, containerDefinitions, requiresCompatibilities, cpu, memory, volumes}' > grafana-task-def-min.json
          IMAGE_URI="$ECR_REGISTRY/$ECR_GRAFANA_REPO:$IMAGE_TAG"
          cat grafana-task-def-min.json | jq --arg IMAGE_URI "$IMAGE_URI" ' .containerDefinitions[0].image = $IMAGE_URI ' > grafana-task-def-updated.json
          NEW_TASK_DEF_ARN=$(aws ecs register-task-definition --cli-input-json file://grafana-task-def-updated.json --region $AWS_REGION --query 'taskDefinition.taskDefinitionArn' --output text)
          aws ecs update-service --cluster $ECS_CLUSTER --service $GRAFANA_SERVICE --task-definition $NEW_TASK_DEF_ARN --region $AWS_REGION

      - name: Update Loki ECS service with new image
        env:
          ECR_REGISTRY: ${{ steps.login-ecr-update.outputs.registry }}
        run: |
          TASK_DEF_ARN=$(aws ecs describe-services --cluster $ECS_CLUSTER --services $LOKI_SERVICE --region $AWS_REGION --query "services[0].taskDefinition" --output text)
          aws ecs describe-task-definition --task-definition $TASK_DEF_ARN --region $AWS_REGION > loki-task-def.json
          cat loki-task-def.json | jq '.taskDefinition | {family, executionRoleArn, taskRoleArn, networkMode, containerDefinitions, requiresCompatibilities, cpu, memory, volumes}' > loki-task-def-min.json
          IMAGE_URI="$ECR_REGISTRY/$ECR_LOKI_REPO:$IMAGE_TAG"
          cat loki-task-def-min.json | jq --arg IMAGE_URI "$IMAGE_URI" ' .containerDefinitions[0].image = $IMAGE_URI ' > loki-task-def-updated.json
          NEW_TASK_DEF_ARN=$(aws ecs register-task-definition --cli-input-json file://loki-task-def-updated.json --region $AWS_REGION --query 'taskDefinition.taskDefinitionArn' --output text)
          aws ecs update-service --cluster $ECS_CLUSTER --service $LOKI_SERVICE --task-definition $NEW_TASK_DEF_ARN --region $AWS_REGION

We will use this guide to set up an IAM OIDC identity provider for GitHub.

Create an inline policy with this content when creating the role:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ECRPermissions",
            "Effect": "Allow",
            "Action": [
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:InitiateLayerUpload",
                "ecr:UploadLayerPart",
                "ecr:CompleteLayerUpload",
                "ecr:PutImage"
            ],
            "Resource": "*"
        },
        {
            "Sid": "ECSPermissions",
            "Effect": "Allow",
            "Action": [
                "ecs:DescribeServices",
                "ecs:DescribeTaskDefinition",
                "ecs:RegisterTaskDefinition",
                "ecs:UpdateService"
            ],
            "Resource": "*"
        },
        {
            "Sid": "IAMPassRole",
            "Effect": "Allow",
            "Action": "iam:PassRole",
            "Resource": [
                "*"
            ]
        },
        {
            "Sid": "CDKDeployment",
            "Effect": "Allow",
            "Action": [
                "cloudformation:*",
                "s3:*",
                "ec2:*",
                "elasticloadbalancing:*",
                "route53:*",
                "logs:*",
                "acm:*",
                "iam:GetRole",
                "iam:CreateRole",
                "iam:AttachRolePolicy",
                "iam:DetachRolePolicy",
                "iam:DeleteRole",
                "iam:PutRolePolicy",
                "iam:DeleteRolePolicy"
            ],
            "Resource": "*"
        },
        {
            "Sid": "STSGetCallerIdentity",
            "Effect": "Allow",
            "Action": "sts:GetCallerIdentity",
            "Resource": "*"
        },
        {
            "Sid": "SSMGetPermission",
            "Effect": "Allow",
            "Action": "ssm:GetParameter",
            "Resource": "*"
        }
    ]
}

Once that is done, and the role is created, your trust policy would look something like this:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Federated": "arn:aws:iam::<account-id>:oidc-provider/token.actions.githubusercontent.com"
            },
            "Action": "sts:AssumeRoleWithWebIdentity",
            "Condition": {
                "StringEquals": {
                    "token.actions.githubusercontent.com:aud": "sts.amazonaws.com"
                },
                "StringLike": {
                    "token.actions.githubusercontent.com:sub": "repo:<github-username>/<repository-name>:*"
                }
            }
        }
    ]
}

Next, add the required secrets in GitHub:

Before pushing to GitHub to trigger the workflow, we make a minor change to the logger app.

Updating the sample_logger.py file from the below:

async def index(request: Request):
    now = datetime.utcnow().isoformat()
    client_host = request.client.host
    logger.info(f"[INFO] Web endpoint visited at {now} from {client_host}")
    return {"message": f"Logger app running. Visit time: {now}"}

to:

async def index(request: Request):
    now = datetime.utcnow().isoformat()
    client_host = request.client.host
    logger.info(f"[INFO] Web endpoint visited at {now} from {client_host}. Deployed with Github workflow")
    return {"message": f"Logger app running. Visit time: {now}. Deployed with Github workflow"}

Next, commit your changes and push to GitHub on the main branch. This should trigger an action run.

The running workflow:

We now see the ECS service is updating

Once deployment is completed, we visit and see the public app to see if the changes are reflected.

Visiting Grafana, we also see the new updated log:

Conclusion

This has been an interesting ride. We got to explore multiple core areas in AWS, implemented various best practices, and designed a production-ready serverless cloud-native logging and visualization application from the ground up.

You can find the complete source code here. In the next article here, we implement full observability in the application.

Excited to see what you build!

Serverless Cloud-Native Logging and Visualization on AWS ECS Using CDK was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

In this article, we’ll explore how I built a real-time serverless data pipeline using AWS services such as Data Catalog, DataBrew, DynamoDB, and seamlessly deployed it to AWS with the help of Terraform.

Whether you are new to the world of data engineering or a seasoned expert, this guide has you covered!

Prerequisites

• AWS account
• Python 3
• Terraform

Why Terraform

• Open source
• Large developer community
• Immutable Infrastructure
• IaC
• Cloud agnostic
• Looking to improve my knowledge of a new technology

Project Structure

Overview

I am not a Data Engineer, nor do I actively build data pipelines; however, I recently became exposed to some data services. Combined with my general knowledge of AWS, I decided to get my hands dirty with some services.

This article stems from my experience building a data pipeline as a fun project, and I found the whole experience quite interesting and rewarding.

Key Services used

1. Amazon S3

• Acts as the data storage layer.
• Stored the JSON files used in the project

2. Amazon DynamoDB

• Serves as the database layer.
• Export JSON data from DynamoDB to S3 using DynamoDB streams and AWS Lambda

3. AWS Glue

• Handles Data Extraction, Transformation, and Loading (ETL).
• Use Glue Data Catalog in managing metadata for datasets.
• Also supports crawling using a crawler that automatically discovers schema and creates tables; however, in this project, we manually define the schema, so we didn't require this.

4. Amazon DataBrew

• Used to transform the data stored in S3 by removing duplicate entries.
• Runs as a triggered job(from lambda) once items are placed in the S3 bucket on the path /data
• Points to Glue Data Catalog as an input dataset.

5. Amazon Athena

• Query transformed data stored in Glue Catalog using standard SQL.
• Fully serverless and integrated with the Glue Data Catalog.

Resulting Architecture

Architecture Explained

First Flow:

• A .json file is added to S3 on the path /data
• First Glue Catalog Table reads from the /data path in S3
• Upload to /data triggers a lambda function which starts the DataBrew transformation job that cleans the data in the First Glue Catalog Table(input dataset) by removing any duplicate rows using the email column
• DataBrew job outputs the transformed data into a new path in S3 under the /cleaned path, overriding other items in that path to avoid duplicates in the output path.
• Second Glue Catalog Table reads from the /cleaned path in S3
• Athena workgroup reads from the Second Glue Catalog Table and runs queries against it. This, in turn, stores the query result in a new output location in S3 /athena-results/

Second flow:

• An item is added to DynamoDB.
• DynamoDB stream is triggered as a new item is added
• Lambda function(in Python) connected to the DynamoDB Stream is invoked, which converts the new items into a .json file
• All steps from the First flow

In summary, the goal of real-time data processing is achieved using DynamoDB Streams and S3 Bucket Notification integration with Lambda.

Code Definition

main.tf

Provider: The first step is usually to define the provider. Here we define the cloud provider as aws.

Also, we include the various modules we would be using in our project and passing all required variables

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
  required_version = ">= 1.3.0"
}

provider "aws" {
  region = "us-east-1"
}

module "s3_bucket" {
  source = "./modules/s3"
  bucket_name = "upload-bucket-data-pipeline-234"
}

module "dynamodb_table" {
  source = "./modules/dynamodb"
  table_name = "dynamodb-table"
}

module "lambda_function" {
  source = "./modules/lambda"
  lambda_name        = "dynamodb_to_s3"
  handler_trigger     = "dynamodb_to_s3_trigger.lambda_handler"
  s3_bucket   = module.s3_bucket.bucket_name
  table_name  = module.dynamodb_table.table_name
  dynamodb_stream_arn = module.dynamodb_table.stream_arn
}

module "glue_catalog" {
  source         = "./modules/glue/glue_raw"
  database_name  = "catalog_db"
  table_name     = "catalog_json_table"
  s3_location    = "s3://${module.s3_bucket.bucket_name}/data/"
}

module "databrew" {
  source       = "./modules/databrew"
  glue_table   = module.glue_catalog.table_name
  glue_db      = module.glue_catalog.database_name
  s3_bucket_name    = module.s3_bucket.bucket_name
  data_zip     = module.lambda_function.data_zip
}

module "glue_catalog_cleaned" {
  source = "./modules/glue/glue_cleaned"
  database_name = "catalog_cleaned_db"
  table_name    = "catalog_cleaned_json_table"
  s3_location   = "s3://${module.s3_bucket.bucket_name}/cleaned/"
}

module "athena" {
  source = "./modules/athena"
  result_output_location = "s3://${module.s3_bucket.bucket_name}/athena-results/"
}

backend.tf

First, create the S3 bucket for storing the Terraform state file remotely, which fosters collaboration.

After creation, add the following content to the backend.tf file:

terraform {
  backend "s3" {
    bucket = "serverless-data-pipeline-backend-bucket"
    key    = "serverless-pipeline/dev/terraform.tfstate"
    region = "us-east-1"
  }
}

Once we run the terraform init and terraform apply to deploy the project, we should see the state stored in the bucket as shown below:

[module]/main.tf

Each service’s infrastructure and configuration are placed in its respective main.tf file

[module]/output.tf

Each service’s output detail is placed in its respective output.tf file

[module]/variable.tf

Each service's desired exported variables are placed in its respective variable.tf file

Deploy the application

Now we are all set to deploy our application. Run the following command to deploy.

Terraform init: This initializes the project, pulling all required packages needed for deployment.

terraform init

Terraform plan: This visualizes the proposed changes, quite useful in catching any unintended changes before deployment.

terraform plan

Terraform apply: This deploys the project to AWS

terraform apply

Testing the application

The S3 bucket is initially loaded with a sample.json file, which has the following content:

[{
  "id": "1",
  "name": "John Doe",
  "email": "john@example.com",
  "timestamp": "2025-04-19T12:00:00Z"
},
{
  "id": "2",
  "name": "Mary Doe",
  "email": "mary@example.com",
  "timestamp": "2025-06-20T12:00:00Z"
},
{
  "id": "3",
  "name": "Jane Doe",
  "email": "john@example.com",
  "timestamp": "2025-06-22T12:00:00Z"
}
]

N:B: As can be seen, we have a duplicate email of john@example.com, the ideal outcome would be to remove the duplicate entry.

This is uploaded immediately after creating the bucket using Terraform, hence it doesn’t trigger the DataBrew job since that trigger setup happens later in the project.

To test the Real-time feature and the full flow of the project, we can either upload new objects to S3 or add new items to DynamoDB. We will be adding a new item to DynamoDB to test the full flow:

Head over to DynamoDB via the AWS console and add a new item:

Adding this would trigger the DynamoDB stream, which uses the associated lambda function to insert a new object into S3.

Heading over to S3, we see the new item 7.json added:

Next, heading over to Data Catalog, we see the Databases and Tables

Heading over to DataBrew, we see the project.

Once the lambda function is triggered by a new addition to the /data path, it starts the DataBrew transformation job:

Once completed, it looks like this:

DataBrew Data lineage allows us to see the graphical representation of the flow with DataBrew.

Idle Job:

Running Job:

Lastly, head over to Athena and select the my-athena-workgroup workgroup to run queries in.

Before the DataBrew Job runs, if we run a query with Athena, there won't be any results, as the /cleaned path won't have any items yet:

However, after the Databrew Job runs successfully, we click the Run again button to re-run the query in Athena:

The result below:

If you observe, it can be seen that it removed the duplicate entry with an email john@example.com found in the sample.json file which uploaded to S3 during the project creation.

It also has the new DynamoDB item with an email mathew@gmail.com added.

This means our pipeline works as expected!!

Adding a duplicate item via DynamoDB

Let's test further by adding a new item to DynamoDB, but this time it would have an email matching an existing record we have.

We added a new item with a conflicting email mathew@gmail.com

This triggers a new DataBrew Job:

Once it succeeds, we see:

Next, head over to Athena and run the query again, and this time it doesn’t have the duplicate record!

Conclusion

I created this project and article to improve my skills in Terraform as well and get my hands dirty in the data engineering world while sharpening my Python skills.

You can find the complete source code here

I hope this article has helped you feel more confident about building a data engineering project in AWS. I’m excited to see what you create!

Building a Serverless Real-Time Data Pipeline with Terraform, AWS, and Python was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

We saw how users could seamlessly interact with it to get information about menus and deals and create and delete reservations.

This article will take it further by adding more agents to the mix. These agents are designed to perform specific tasks for Steakhouse and would be managed by a Supervisor agent. We will see multi-agent collaboration in AWS and how to build a robust system that can handle multiple use cases.

Let’s get right into it!

Prerequisites

• AWS account
• AWS CLI setup locally
• AWS CDK set up.
• Python3
• Model access is enabled for the required models. Specifically, Amazon Bedrock has access to the Amazon Titan Embeddings V2 model and Anthropic Claude 3 Haiku.
• Previous article

What would we cover

• Current architecture
• We review the new Steakhouse architecture.
• We walk through the Python and AWS CDK code.
• We perform some tests to see it in action.

Case Study: Steakhouse

Our imaginary company Steakhouse has seen a lot of success from deploying the Bedrock agent we discussed in our last article, this convinced management to invest more into agents for various use cases.

They have also expanded and are no longer solely focused on being a steakhouse, but now have short lets they manage for customers, as well as an HR department for internal staff. They also added a ticket support team and would love to have an agent help in this department.

So Steakhouse now has 4 sub-agents: A Reservation Agent(would keep managing Steakhouse meal orders for customers), an HR Agent(would manage all staff HR issues), a Shortlet(would manage the new short let for customers), and ticket support(would manage ticket support for customers).

Then a Supervisor Agent made possible by AWS Bedrock multi-agent collaboration, would be directly connected to the client to receive all requests and coordinate with the sub-agents to give back an optimal response.

Current Steakhouse Architecture

New Steakhouse Architecture

We can see that:

1. The user visits the Streamlit Web app to interact with the Agent by visiting the URL exposed by the load balancer.
2. The Streamlit Web app interacts with the Supervisor agent.
3. The supervisor agent then decides which agent(s) to forward the request to, aggregates the response, and outputs the final relevant response to the user.

Project Overview

The folder structure is designed as follows:

1. The agents_python folder contains the 2 stacks for the app, the Agent stack, and the Streamlit stack.
2. The streamlit folder contains the files for setting up the streamlit app via the app.py file and interacting with the bedrock agent via the agent.py helper file.
3. The dataset contains the files that would be added to the knowledge base and used by each agent.
4. The lambdas/actiongroup includes the action group lambda functions of the 4 sub-agents.

We will review some parts of the newly added code.

Creating the Supervisor Agent

  supervisor_agent_role = iam.Role(
            scope=self,
            id='SupervisorAgentRole',
            role_name='AmazonBedrockExecutionRoleForSupervisorAgent-Steakhouse',
            assumed_by=iam.ServicePrincipal('bedrock.amazonaws.com'))

supervisor_agent_role.add_to_policy(iam.PolicyStatement(
            sid='SupervisorInvokeBedrockLambda',
            effect=iam.Effect.ALLOW,
            resources=[
                agent_model_arn],
            actions=['bedrock:InvokeModel', 'lambda:InvokeFunction']))
        
supervisor_agent_role.add_to_policy(iam.PolicyStatement(
            sid='SupervisorAgentInvoke',
            effect=iam.Effect.ALLOW,
            resources=["*"],
            actions=['bedrock:*']))

  # Create supervisor agent
        cfn_supervisor_agent = bedrock.CfnAgent(self, "SupervisorAgent",
                                                agent_name=supervisor_agent_name,
                                                agent_resource_role_arn=supervisor_agent_role.role_arn,
                                                auto_prepare=True,
                                                description=supervisor_agent_description,
                                                foundation_model=agent_model_id,
                                                instruction=supervisor_agent_instruction,
                                                idle_session_ttl_in_seconds=1800,
                                                )
        cfn_supervisor_agent_alias = bedrock.CfnAgentAlias(
            self, "SupervisorAgentAlias",
            agent_alias_name=supervisor_agent_alias_name,
            agent_id=cfn_supervisor_agent.attr_agent_id)

        self.bedrock_supervisor_agent_id = cfn_supervisor_agent.attr_agent_id
        self.bedrock_supervisor_agent_alias_id = cfn_supervisor_agent_alias.attr_agent_alias_id

From the above, we set up the supervisor agent with a role and required permissions.

We also create an Alias for the supervisor agent and set it up for production use. Lastly, we export the bedrock_supervisor_agent_id and bedrock_supervisor_agent_alias_id to be used by the Streamlit Stack.

Updating app.py

app = cdk.App()
agent_stack = AgentStack(app, "AgentStack")
streamlit_stack = StreamlitStack(app, "StreamlitStack",
                                 bedrock_agent_id=agent_stack.bedrock_supervisor_agent_id,
                                 bedrock_agent_alias_id=agent_stack.bedrock_supervisor_agent_alias_id
                                 )

streamlit_stack.add_dependency(agent_stack)

We update the app.py to use the supervisor agent and agent alias in the streamlit stack. This way the app would interact with the supervisor agent rather than a specific sub-agent.

A point to note: We created 4 sub-agents and assigned each an action group, with corresponding lambda function and connected them to a single Knowledge Base. However, due to the code snippet size we won't be seeing them in this article, but you can always see the full project on the repo here.

Time to deploy 🔥 🔥

To deploy the app:

cdk deploy --all

This would deploy the 2 stacks, first the Agent Python stack and then the Streamlit stack.

Enabling Multi-agent Collaboration

Update: This section is no longer needed as the source code has been updated with the latest CDK version that supports multi-agent collaboration

Before we test the app, we need to set up the multi-agent collaboration manually using the AWS Console.

This is because AWS CDK doesn't currently support this feature, so we would carry it out via the console.

Click on the bedrock agents in the console, then select the Supervisor Agent and follow these steps:

Click the Edit in Agent Builder button:

Next, scroll down and click the Edit button in the Multi-agent collaboration feature section:

Next, enable the Multi-agent collaboration feature.

Scroll down and add the 4 sub-agent collaborators

1. HR Agent Collaborator:

2. Reservation Agent Collaborator:

3. Shortlet Agent Collaborator:

4. Ticket Agent Collaborator:

Next click the “Save and exit” button at the top. This would save the changes and take us to the agent details.

Now we have 4 collaborators:

Click the “Save and exit” button on this page, so it saves the changes.

Next, proceed to prepare the agent so the changes we made would take effect on the supervisor agent.

Click on the “Prepare” button

Next, click on the alias and edit so we can update it to use the latest changes:

In the Edit alias modal, choose “Create a new version and associate it to this alias”. Then click the save button.

This would create a new version and update the alias to point to this new version.

Now the changes made are in the supervisor alias that the streamlit app is connected to.

Testing the app

We would test the app via the Streamlit app and the AWS console. The reason for this is the AWS console allows us to see traces, this way we can see how the supervisor agent interacts with the sub-agents when answering user queries, something we won't be able to see directly when we test using the streamlit app.

Testing the app via AWS Console

Query 1

Customer: Whats the best steak I can get as well as best dessert

Agent: For the best steak option, I would recommend the Wagyu Ribeye. At $60 for a 12oz cut, it is a premium and flavorful steak with unmatched marbling and richness. The Tomahawk Steak is also an excellent choice for sharing.

For dessert, the Chocolate Lava Cake served with vanilla ice cream sounds like an indulgent and delicious option. The Tiramisu is another great choice for those who prefer a coffee-flavored dessert.

You may also want to consider taking advantage of the weekly steak specials like the Filet Mignon Feast on Thursdays or the discounts on T-Bones on Tuesdays.

Note: As can be seen, the Supervisor agent calls only the Reservation sub-agents which uses the Knowledge Base for context.

Query 2

Customer: Steakhouse Menu for Tuesday and available short lets

Agent: According to the information gathered: The Steakhouse menu for Tuesday features the “T-Bone Tuesday” special, where T-Bone steaks are $5 off and served with any two sides. The available short lets at the Steakhouse this week include: — Standard Room: $50 per night — Deluxe Room: $80 per night — Suite: $120 per night — Penthouse: $250 per night All short let rooms come with amenities such as free WiFi, air conditioning, 24/7 security, housekeeping services, and access to a shared lounge. Additional amenities like kitchenettes and private pools are available in select room types.

Note: As can be seen, the Supervisor agent calls both sub-agents which each use the Knowledge Base for context

Query 3

Customer: What is the sick leave policy for me as a staff, also could you create a ticket for John with the following details: incident date: 22/03/2022 and reason: Shortlet room was untidy

Agent: Based on the information provided:

1. The Steakhouse sick leave policy states that all employees, including part-time staff, are eligible for sick leave. Employees can use sick leave for personal illness, medical appointments, or caring for an immediate family member. Employees must notify their manager as soon as possible if they need to take sick leave, and for absences longer than three consecutive days, a medical certificate is required.

2. I have successfully created a new ticket for John with the following details: — Incident date: 22/03/2022 — Reason: Shortlet room was untidy The new ticket has been assigned the booking ID ‘59175490’.

Note: As can be seen, the Supervisor agent calls both sub-agents. It uses the Hr sub-agent Knowledge Base for context and uses the Ticket sub-agent Action group to make an api call that created the ticket in DynamoDB.

Query 4

Customer: Create shortlet reservations for John for 4 days on 28/02/2025 for 2 guests, for shortlet type use standard. Then recommend 1 good dessert I could take.

Agent: Based on the information provided, I have completed the following actions:

1. Created a new shortlet booking for John with the following details: — Booking ID: 679b0fce — Date: 28/02/2025 — Shortlet Type: standard — Number of Guests: 2 — Number of Days: 4

2. Recommended the Tiramisu or the Crème Brûlée as the best dessert options to pair with the steakhouse shortlet booking: — Tiramisu is a “perfect” pairing for Prosecco wine, which is recommended for “lighter desserts”. This would complement the indulgence of a shortlet stay. — Crème Brûlée is an elegant, classic dessert option that would pair well with a steakhouse meal. Its rich, creamy texture would make for a sophisticated end to a shortlet experience. So in summary, I would recommend either the Tiramisu or the Crème Brûlée as the best dessert option to pair with the steakhouse shortlet booking.

Note: As can be seen, the Supervisor agent calls the shortlet and reservation sub-agents. It uses the Shortlet sub-agent Action group to make an api call that created the shortlet reservation in DynamoDB and the Reservation sub-agent Knowledge Base for context for Dessert.

Testing the app via Streamlit App

Query 5

Customer:

Create HR booking for sick leave for Janet from 03/04/2025 to 010/04/2025, the main reason is cough, detailed comment is severe cold and doctor recommended rest. Return Shortlet details for Shortlet id: 679b0fce

Agent:

My Observation

1. I was faced with Request per minute(RPM) rate throttling when I was actively testing the agent. This wasn’t a pleasant experience. I had to update my model to use Haiku 3, which gave me higher RPM, as I couldn't get the Bedrock team to increase my Claude Sonnet 3 RPM on time(I still have an open ticket for over a week now!). You may need to request an increase in service quota for Claude's quota, more info here.
   Update: I was able to use the Cross-region Inference, more details here, to fix this issue
2. Also, the Supervisor Agent could act as a security agent where you could apply checks using guardrails.

Closing

This has been an interesting ride. This brings an end to an exciting use case of Steakhouse, where we got to build an AI Agent using Bedrock, and extended it to support multi-agent collaboration.

You can find the complete source code here.

Excited to see what you build!

Multi-Agent Collaboration: Serverless AI Agents using AWS CDK, Bedrock, and Python was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.

Update: This framework has been renamed to the AWS Agent Squad framework!

Prerequisites

• AWS account
• AWS CLI setup locally
• Python3
• Model access is enabled for the required models.

What would we cover

• We cover what Amazon Multi-Agent Orchestrator is.
• We walk through the multiple approaches to building an Agent.
• We perform some tests to see it in action.
• Some observations

Introduction

This article will utilize the AWS Multi-agent to build Multi-Agents. The goal is to get an introduction to the framework by seeing what is possible and trying out some use cases.

Let's get started!

What is the AWS Multi-Agent Orchestrator?

The AWS Multi-Agent Orchestrator is an open-source framework for building AI agents and multi-agents. It has inbuilt features such as intelligent routing, intent classification, context management, agent overlap analysis, and many more.

The Multi-Agent Orchestrator uses a Classifier, that does the selection of the right agent to be used for every user request.

Main Components of the Orchestrator

The main components are:

Orchestrator:

The central coordinator for all other components and handles the flow of information between all other components such as Classifiers, Agents, Storage, and Retrievers.

Handles error scenarios internally and has a great fallback mechanism.

Classifier

Examines user input alongside the previous chat history and agent description to identify the most appropriate agent for each request. Custom classifiers could also be used.

Agents

From Prebuilt Agents to custom agents, this is where you create and tailor functionality for agents. You could also extend prebuilt agents as we use later in the article.

Conversation Storage

Used to save conversation history. Has in-built memory and DynamoDb support and can also use custom storage solutions.

Retrievers

Used to enhance the agent's performance by providing context and relevant information. We could attach a Knowledge Base here to augment the agent data by providing more context.

Project Setup

Before we start creating the agents, we need to set up our environment. head over to the terminal and run the following:

mkdir aws-orchestrator
cd aws-orchestrator && touch app.py

python -m venv venv
source venv/bin/activate 

This would set up a virtual environment for the Python app.

Next, install the multi-agent-orchestrator package:

pip install "multi-agent-orchestrator[all]"

Using the Inbuilt Bedrock Agent with classifier approach

First, we explore how to build an inline agent using the inbuilt Bedrock agent. We would customize it to support our use case.

Head over to app.py and input the following:

import uuid
import asyncio
import sys

from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
from multi_agent_orchestrator.agents import (BedrockLLMAgent,
 BedrockLLMAgentOptions,
 AgentResponse,
 AgentCallbacks)

orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
  LOG_AGENT_CHAT=True,
  LOG_CLASSIFIER_CHAT=True,
  LOG_CLASSIFIER_RAW_OUTPUT=True,
  LOG_CLASSIFIER_OUTPUT=True,
  LOG_EXECUTION_TIMES=True,
  MAX_RETRIES=3,
  MAX_MESSAGE_PAIRS_PER_AGENT=10,
  USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True
))

class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        print(token, end='', flush=True)

football_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Football Insights Agent",
    description="Expert in football analysis, covering team strategies, player stats, match predictions, and historical comparisons.",
    callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(football_agent)

life_hack_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Life Hacks & Motivation Agent",
    description="Provides life hacks for efficiency, productivity tips, motivational insights, and goal-setting strategies for self-improvement.",
    callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(life_hack_agent)


async def handle_request(_orchestrator: MultiAgentOrchestrator, _user_input: str, _user_id: str, _session_id: str):
    response: AgentResponse = await _orchestrator.route_request(_user_input, _user_id, _session_id)
    
    print("\nMetadata:")
    print(f"Selected Agent: {response.metadata.agent_name}")
    if response.streaming:
        print('Response:', response.output.content[0]['text'])
    else:
        print('Response:', response.output.content[0]['text'])

if __name__ == "__main__":
    USER_ID = "user123"
    SESSION_ID = str(uuid.uuid4())
    print("Welcome to the interactive Multi-Agent system. Type 'quit' to exit.")
    while True:
        # Get user input
        user_input = input("\nYou: ").strip()
        if user_input.lower() == 'quit':
            print("Exiting the program. Goodbye!")
            sys.exit()
        # Run the async function
        asyncio.run(handle_request(orchestrator, user_input, USER_ID, SESSION_ID))

This creates 2 agents. A football agent and secondly a life hack agent.

Run the app: python3 app.py

Input Prompt 1:

Who is cristiano ronaldo

Resulting output:

As can be seen, the Classifier routed the request to the right agent.

Input Prompt 2:

I need motivation to succeed in goals

Resulting output:

The Classifier also correctly routed the request to the right agent.

Adding a Default Agent

As can be seen, we have no Default agent and hence when we provide an input with no agent coverage it doesn't give a graceful response.

Input Prompt:

I need motivation to succeed in goals

Resulting output:

Let's fix this by updating the code to include a Default agent. Update the orchestrator in app.py

orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
  LOG_AGENT_CHAT=True,
  LOG_CLASSIFIER_CHAT=True,
  LOG_CLASSIFIER_RAW_OUTPUT=True,
  LOG_CLASSIFIER_OUTPUT=True,
  LOG_EXECUTION_TIMES=True,
  MAX_RETRIES=3,
  MAX_MESSAGE_PAIRS_PER_AGENT=10,
  USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True
),
  default_agent=BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Default Agent",
    streaming=False,
    description="This is the default agent that handles general queries and tasks.",
  ))
)

Rerun the app and enter the same user prompt. This time it uses the default agent to attempt the question:

Chaining Agents

Suppose we want an ordered flow of agent collaboration, rather than the Classifier choosing the single appropriate agent.

We want some agents to work together sequentially to give an output. This is where ChainAgent comes in.

It does this by using the output of one agent as input for the next. This creates an agent chain

Create a new file called chained.py and input the following:

from multi_agent_orchestrator.agents import (BedrockLLMAgent,
                                             BedrockLLMAgentOptions,
                                             AgentCallbacks, ChainAgent, ChainAgentOptions)

class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        print(token, end='', flush=True)


research_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Research Agent",
    description="Analyzes and validates the given content, expanding on relevant topics and ensuring accuracy.",
    callbacks=BedrockLLMAgentCallbacks(),
    streaming=True
))

analysis_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Analysis Agent",
    description="Extracts key insights, trends, and relevant points from the content, identifying core themes.",
    callbacks=BedrockLLMAgentCallbacks(),
    streaming=True
))

report_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="Report Agent",
    description="Creates a structured report summarizing the research findings and key insights into a coherent format.",
    callbacks=BedrockLLMAgentCallbacks(),
    streaming=True
))


options = ChainAgentOptions(
    name='ChainAnalysisAgent',
    description='A research, analayis and report generation chain of agents that takes user research input and gives a final report after proper research & analysis',
    agents=[research_agent, analysis_agent, report_agent],
    default_output='The chain processing encountered an issue.',
    save_chat=True
)

chain_agent = ChainAgent(options)

We have 3 agents chained together. The chained agent retrieves content on any topic from the user and carries out research & analysis and returns a final report.

Using Chained Agent with Orchestrator

Let's put this chained agent to the test. Modify the app.py file to import and add this new agent to the orchestrator:

from chained import chain_agent

# previous code

#Add Chained agent
orchestrator.add_agent(chain_agent)

# previous code

Now rerun the app.py file:

Input Prompt:

Help research on LLM. I believe its large language model and its used in Gen AI. Give me a proper research, analysis and report on this

Resulting output:

It is carrying out the needed analysis to produce the report. As can be seen, it uses the information from the research agent to generate insights and reports.

Using a supervisor agent

The SupervisorAgent is an advanced orchestration component that enables sophisticated multi-agent coordination, also known as multi-agent collaboration. Its unique agent-as-tools architecture exposes team members to a supervisor agent as invocable tools.

Rather than sequentially, it enables parallel processing, allowing the lead agent to switch context between multiple team agents as it seems fit.

This means it could call Agent 1 first and then Agent 2 for one input, while also calling Agent 2 first and then Agent 1 for another input.

It can be used directly or with the classifier. In the example below, we use it with the orchestrator, just like we did with the chained agent.

Create a new file called surpervisor.py and input the following:

from multi_agent_orchestrator.agents import (BedrockLLMAgent,
                                             BedrockLLMAgentOptions,
                                             AgentCallbacks,
                                             SupervisorAgent,
                                             SupervisorAgentOptions)


class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        print(token, end='', flush=True)


ticket_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
            name="Ticket Agent",
            description="Creates a ticket about customer issues",
            callbacks=BedrockLLMAgentCallbacks(),
            streaming=True
        ))
    
credit_card_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
            name="Credit Card Agent",
            description="Handles card issues, asks for customer card details when needed",
            callbacks=BedrockLLMAgentCallbacks(),
            streaming=True
        ))

supervisor_agent = SupervisorAgent(SupervisorAgentOptions(
    lead_agent=BedrockLLMAgent(BedrockLLMAgentOptions(
        name="Support Team Lead",
        description="Coordinates support inquiries"
    )),
    team=[credit_card_agent, ticket_agent]
))

We have three agents: two team members and a lead agent. The lead agent determines which agent to call or whether to use both team agents for a single user request.

Using Supervisor Agent with Orchestrator

Let’s put this Supervisor agent to the test. Modify the app.py file to import and add this new agent to the orchestrator:

from supervisor import supervisor_agent

.....

#Add Supervisor agent
orchestrator.add_agent(supervisor_agent)

Now rerun the app.py file:

Input Prompt:

My credit card has issues, create a ticket for me, and I need help resolving it

Resulting output:

The lead supervisor agent could switch between its team agents based on the user's request. It combined both agents to give a useful response.

My Observation

1. I haven't seen any USP(Unique Selling Point) with this framework from CrewAI. What comes to my mind is the ease of integration with other AWS services since it is an AWS framework
2. This is a good initiative for building agents quickly, trying out other components like custom storage, and using existing AWS services like lambda, Bedrock agent, and others.

Closing

This has been an interesting ride. We not only got to build an AI Agent using the AWS orchestrator, but we also got to try out its multiple features/components to build AI agents.

You can find the complete source code here.

Excited to see what you build!

Getting started with the AWS Multi-Agent Orchestrator Framework was originally published in Towards AWS on Medium, where people are continuing the conversation by highlighting and responding to this story.