Forem: AWS Community Builders

Add Chat AI Summary Using Amazon Bedrock and HTTP Response Streaming

Marko Djakovic — Mon, 06 Apr 2026 18:24:23 +0000

Ever since I started writing what turned out to be a series of articles on how to build real-time chat apps natively on AWS, I had a huge backlog of ideas for extending the basic chat, which included adding authentication & authorization, full-text search, threads, reactions, and many other things that make any chat system production grade and up to par with what users expect these days. Given how AI has entered every pore of the industry, extending my small chat project had to go in this direction sooner or later. It all started with creating a solution that (ab)uses IoT Core, which worked exceptionally well - maybe even too well, as AWS ended up designing AppSync Events in a similar fashion. Read more on what I mean in my blog post about Serverless Chat on AWS with AppSync Events. Shortly after the initial release, AWS pushed out the promised improvements for AppSync Events, which I explore in the following blog post about leveraging WebSockets with it.

To build on top of existing chat solution, this time we won't touch the main infrastructure that powers the chat but rather extend the group chat capabilities by adding a dedicated AI summary endpoint. This is one of the most common AI use cases, and apps like Viber or Slack already have it. Most of us can relate to the pain of having to go through numerous Slack threads with dozens of messages each - so I'd say this feature is more than only nice to have at this point. Let's see how we can do it using Amazon Bedrock's ConverseStreamCommand, and leveraging HTTP response streaming capability of API Gateway.

Architecture

As I mentioned, we won't be touching the core infrastructure of the chat solution which is AppSync and the whole WebSockets part, but rather extend the system with a dedicated endpoint to create the AI summary of chat messages. So, essentially what we're talking about here is shown in the following diagram:

New, and the most interesting addition here is Amazon Bedrock. Also, since API Gateway supports HTTP response streaming, we'll make sure to leverage that to get the fastest possible response when invoking AI summary endpoint.

If we zoom back out, the whole solution looks something like:

So just to be clear - this is what will get deployed if you decide to run this in your own account. Let's break down how the two new key pieces fit into the mix.

Amazon Bedrock

I won't waste words on this since it is quite familiar, but in short - Bedrock is a fully managed, serverless AWS service for building generative AI applications. It provides a single API access for wide variety of foundational models from leading AI companies such as Anthropic, Meta, and Amazon itself. For this particular example I will rely on one of the smallest Amazon models - amazon.nova-micro-v1:0. It is more than enough for showcasing the idea, but feel free to play with others according to your liking.

Small caveat: when specifying the model, it is not enough to put only the name of the model, but you need to include also the region prefix of where your app will run. For example in my case it was eu.amazon.nova-micro-v1:0. Worth knowing just so you don't get confused when you see it in the code.

A dedicated Lambda function for AI summary will first fetch group chat messages from the database, and then create a prompt for Bedrock to create a short summary. It uses ConverseStreamCommand by BedrockRuntimeClient to send the prompt to Bedrock and get the stream response. To make the response smoothly stream into Lambda function's response, there is a small caveat compared to how you would normally define the handler in TypeScript. One is defining the handler as:

awslambda.streamifyResponse(async (event: APIGatewayProxyEvent, responseStream: awslambda.HttpResponseStream)

and handling the streaming response:

const httpStream = awslambda.HttpResponseStream.from(responseStream, {
    statusCode: 200,
    headers: {
        'Content-Type': 'text/plain; charset=utf-8',
        'Access-Control-Allow-Origin': '*',
    },
});

...and then the httpStream will be used to write the stream response from Bedrock into it.

Note: for simplicity I have set a limit of 50 messages to fetch for the summary, however it is configurable.

HTTP Response Streaming with API Gateway

Finally, the last piece to get the streaming response to the client is required to be configured on API Gateway. This capability was introduced in November 2025 and has CDK support from version 2.227.0 via responseTransferMode property. Setting it to ResponseTransferMode.STREAM on the Lambda integration resource will enable streaming:

getSummaryResource.addMethod('GET', new LambdaIntegration(getSummaryLambda, {responseTransferMode: ResponseTransferMode.STREAM}));

Demo

To showcase how the finished feature works, I've created a short video. I prefilled the group chat with test messages to simulate a conversation between team of engineers working on a production deployment. The AI summary will summarize their agreements and key points they discussed about the deployment.

Watch the demo video

Conclusions

In the end, building an AI summary today isn’t something groundbreaking, and to be honest that might be the point. What used to feel like cutting-edge quickly became the standard what users expect in these types of applications. Amazon Bedrock doing the heavy lifting on the model side makes the overall path from idea to working solution really smooth. I encourage you to try it yourself, check out the code and follow the instructions to deploy to your own AWS account:

https://github.com/imflamboyant/serverless-aws-chat/tree/main/chat-appsync-events-websocket

💸 Note: as always, be aware that usage might incur real costs, though relatively small for this example. If you have free tier or credits, just make sure that Bedrock is covered.

Thanks for reading, and stay tuned, because next I’ll take this further into agentic chat and explore what’s possible with Strands.

Handling File Uploads in AWS Lambda with Powertools OpenAPI (From Limitation to Production Feature)

Michael Uanikehi — Mon, 06 Apr 2026 18:13:46 +0000

Introduction

Handling file uploads in serverless APIs sounds simple until you actually try to do it.

If you're building APIs with AWS Lambda Powertools and OpenAPI validation, you quickly run into a limitation:

multipart/form-data isn’t natively supported in the same way as JSON or form-encoded requests.

That gap forces teams into workarounds:

Manual multipart parsing
Base64 hacks
Disabling validation entirely

None of which are ideal in production systems.

This article walks through:

The real problem
How the feature was designed
How you can now use it in practice

The Problem: File Uploads Break the Abstraction

Before this feature, Powertools handled:

JSON payloads
Query parameters
Headers
Form data (application/x-www-form-urlencoded)

But not:

multipart/form-data (file uploads)

That meant:

@app.post("/upload")
def upload(file: bytes):
    ...

simply didn’t work with OpenAPI validation.

Instead, developers had to:

Parse raw request bodies manually
Disable validation middleware
Or redesign APIs around non-standard formats

At scale, this creates:

Inconsistent APIs
Security gaps
Poor developer experience

The Goal: Make File Uploads First-Class

The aim was simple:

Make file uploads work the same way as Query(), Header(), and Form()

That means:

Type-safe
Automatically validated
Fully reflected in OpenAPI schema
Works with Swagger UI

The Solution: `File()` Parameter Support

You can now define file inputs like this:

from typing import Annotated
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.openapi.params import File, UploadFile

app = APIGatewayRestResolver(enable_validation=True)
app.enable_swagger(path="/swagger")

@app.post("/upload")
def upload(file_data: Annotated[UploadFile, File(description="File to upload")]):
    return {
        "filename": file_data.filename,
        "content_type": file_data.content_type,
        "file_size": len(file_data),
    }

Two Ways to Work with Files

1. Raw bytes

file: Annotated[bytes, File()]

You get:

File content only

2. Rich file object

file_data: Annotated[UploadFile, File()]

You get:

Content
Filename
Content type

This is usually what you want in real systems.

Combining Files with Form Data

@app.post("/upload-csv")
def upload_csv(
    file_data: Annotated[UploadFile, File(description="CSV file")],
    separator: Annotated[str, Form(description="CSV separator")] = ",",
):
    text = file_data.content.decode("utf-8")

This unlocks:

Metadata + file uploads
Real-world API patterns

What Changed Under the Hood

Supporting this wasn’t just adding a new parameter type.

It required:

Multipart parsing logic
Boundary handling (including WebKit quirks)
Base64 decoding for Lambda event payloads
Differentiating file vs form fields
OpenAPI schema generation (format: binary)
Validation integration

There’s also a helpful runtime safeguard:

If multipart requests aren’t properly base64 encoded, a warning is emitted

This helps catch common misconfigurations early.

API Gateway Gotcha (Important)

If you're using REST API (v1), you must configure:

Globals:
  Api:
    BinaryMediaTypes:
      - "multipart~1form-data"

Without this:
File uploads won’t work correctly.

For:

HTTP API (v2)
Lambda Function URLs
ALB

It works out of the box.

Before vs After

Before

Manual parsing
No validation
Custom schemas
Inconsistent APIs

After

Native File() support
OpenAPI validation
Swagger UI integration
Cleaner, safer APIs

Why This Matters

This isn’t just about file uploads.

It’s about removing friction from real-world API design.

When basic capabilities are missing:

Engineers build workarounds
Systems become inconsistent
Reliability suffers

By making file uploads a first-class feature:

APIs become more predictable
Validation becomes reliable
Developer experience improves significantly

Open Source Insight

One interesting part of this work:

The implementation evolved through multiple iterations before reaching the final version.

Large features often:

Start broad
Get refined for maintainability
Land as cleaner, more focused implementations

That process is what makes open source powerful —
it’s not just about shipping code, but improving it collaboratively.

Final Thoughts

If you’re building serverless APIs and dealing with file uploads:

You no longer need workarounds.

This feature brings Powertools closer to frameworks like:

FastAPI
Django
Express

…but in a serverless-native way.

References

Feature PR: https://github.com/aws-powertools/powertools-lambda-python/pull/8093
Original implementation: https://github.com/aws-powertools/powertools-lambda-python/pull/7132
Feature request: https://github.com/aws-powertools/powertools-lambda-python/issues/7124

I Ditched Prisma for Raw SQL (And My Queries Got 10x Faster)

Polliog — Mon, 06 Apr 2026 13:28:19 +0000

Prisma is genuinely good software. The schema DSL is clean, the type generation works well, and for a new project it gets you to a working data layer in an hour. I used it for about a year before I started noticing things.

The first sign was a query that should have taken 5ms taking 80ms. The second was a N+1 that I'd technically solved with include but was still generating 15 SQL statements. The third was opening prisma.$queryRaw for the third time in a week because the query builder couldn't express what I needed.

At that point I stopped fighting the abstraction and started writing SQL directly.

What Prisma Actually Does to Your Queries

This is a simple query with a filter and pagination:

const logs = await prisma.logEntry.findMany({
  where: {
    organizationId: orgId,
    timestamp: {
      gte: from,
      lt: to,
    },
    level: { in: ["error", "fatal"] },
  },
  orderBy: { timestamp: "desc" },
  take: 50,
  skip: page * 50,
});

The SQL Prisma generates:

SELECT
  "public"."log_entries"."id",
  "public"."log_entries"."timestamp",
  "public"."log_entries"."service",
  "public"."log_entries"."level",
  "public"."log_entries"."message",
  "public"."log_entries"."metadata",
  "public"."log_entries"."organization_id",
  "public"."log_entries"."created_at",
  "public"."log_entries"."updated_at"
FROM "public"."log_entries"
WHERE (
  "public"."log_entries"."organization_id" = $1
  AND "public"."log_entries"."timestamp" >= $2
  AND "public"."log_entries"."timestamp" < $3
  AND "public"."log_entries"."level" IN ($4, $5)
)
ORDER BY "public"."log_entries"."timestamp" DESC
LIMIT $6 OFFSET $7

This is fine SQL. But notice: it selects every column (including created_at and updated_at that my UI doesn't need), it uses OFFSET pagination (slow on large tables), and I have no control over any of it without escaping to $queryRaw.

The equivalent raw query:

const logs = await pool.query(
  `SELECT id, timestamp, service, level, message, metadata
   FROM log_entries
   WHERE organization_id = $1
     AND timestamp >= $2
     AND timestamp < $3
     AND level = ANY($4)
     AND (timestamp, id) < ($5, $6)
   ORDER BY timestamp DESC, id DESC
   LIMIT $7`,
  [orgId, from, to, ["error", "fatal"], cursorTs, cursorId, 50]
);

Keyset pagination instead of OFFSET, only the columns I need, and the query is exactly what I want the database to run.

The N+1 Problem Prisma Doesn't Fully Solve

Prisma's include resolves N+1 queries by using IN clauses instead of per-row queries. But "no N+1" doesn't mean "one query":

const projects = await prisma.project.findMany({
  where: { organizationId: orgId },
  include: {
    members: true,
    apiKeys: { where: { active: true } },
    _count: { select: { logEntries: true } },
  },
});

Prisma executes this as 4 separate queries: one for projects, one for members, one for apiKeys, one for the count. Then it assembles the result in JavaScript.

The raw equivalent is one query. The naive approach would be chaining multiple LEFT JOIN on one-to-many tables and relying on GROUP BY - but that produces a Cartesian fan-out: if a project has 10 members, 5 API keys, and 100 log entries, the database materializes 10x5x100 = 5,000 intermediate rows per project before collapsing them. COUNT(DISTINCT ...) hides the bug in the results, but performance collapses as the tables grow.

The correct version pre-aggregates each relationship with CTEs before joining:

WITH member_stats AS (
  SELECT
    project_id,
    COUNT(user_id) AS member_count,
    jsonb_agg(jsonb_build_object('id', user_id, 'role', role)) AS members
  FROM project_members
  GROUP BY project_id
),
key_stats AS (
  SELECT project_id, COUNT(id) AS active_key_count
  FROM api_keys
  WHERE active = true
  GROUP BY project_id
),
log_stats AS (
  SELECT project_id, COUNT(id) AS log_entry_count
  FROM log_entries
  WHERE timestamp > NOW() - INTERVAL '24 hours'
  GROUP BY project_id
)
SELECT
  p.id,
  p.name,
  p.created_at,
  COALESCE(ms.member_count, 0) AS member_count,
  COALESCE(ks.active_key_count, 0) AS active_key_count,
  COALESCE(ls.log_entry_count, 0) AS log_entry_count,
  COALESCE(ms.members, '[]'::jsonb) AS members
FROM projects p
LEFT JOIN member_stats ms ON ms.project_id = p.id
LEFT JOIN key_stats ks ON ks.project_id = p.id
LEFT JOIN log_stats ls ON ls.project_id = p.id
WHERE p.organization_id = $1
ORDER BY p.created_at DESC

Each CTE scans and aggregates its table independently. The final join works on already-collapsed rows - no fan-out, no wasted intermediate rows. One round trip, and actually faster than Prisma's 4 queries at scale.

Prisma can't generate this query. $queryRaw can run it, but then you lose the type safety that was the point of using Prisma.

The Performance Numbers

Same endpoint, same data, same index configuration. 50k rows in the table.

Query type	p50	p95	p99
Prisma `findMany` with `include`	45ms	120ms	310ms
4 separate `pg` queries	18ms	40ms	95ms
Single JOIN query	6ms	14ms	28ms

The 10x headline comes from the p99 comparison. At p50 it's closer to 7x. Both are real.

The Prisma numbers aren't bad in absolute terms for most applications. They become a problem when you're doing this on every request, at scale, with connection pool pressure from concurrent requests.

Migrating Without Rewriting Everything

You don't have to replace Prisma everywhere at once. The practical path:

Step 1: Keep Prisma for writes and simple reads

Prisma is genuinely good for inserts, updates, and single-record lookups by primary key. The query generation for these is optimal and the type safety is useful.

// Keep this in Prisma - it's fine
await prisma.user.create({ data: { email, name, organizationId } });
await prisma.project.update({ where: { id }, data: { name } });
const user = await prisma.user.findUnique({ where: { id } });

Step 2: Replace list queries and anything with joins

This is where the overhead compounds. Add a pg pool alongside Prisma:

import { Pool } from "pg";
import { PrismaClient } from "@prisma/client";

export const prisma = new PrismaClient();
export const pool = new Pool({ connectionString: process.env.DATABASE_URL });

Step 3: Write a thin query layer

The thing I missed most from Prisma was typed results. TypeScript with raw SQL defaults to any. Fix it:

async function queryLogs(params: LogQuery): Promise<LogEntry[]> {
  const result = await pool.query<{
    id: string;
    timestamp: Date;
    service: string;
    level: string;
    message: string;
    metadata: Record<string, unknown>;
  }>(
    `SELECT id, timestamp, service, level, message, metadata
     FROM log_entries
     WHERE organization_id = $1
       AND timestamp >= $2
       AND timestamp < $3
     ORDER BY timestamp DESC
     LIMIT $4`,
    [params.orgId, params.from, params.to, params.limit]
  );

  return result.rows;
}

The generic parameter on pool.query<T> types the rows. It's not as ergonomic as Prisma's generated types, but it's enough to catch most mistakes at compile time.

If you want SQL-level control with Prisma-level type safety, look into Kysely or Drizzle ORM. Both let you write SQL-close queries while inferring full TypeScript types from your schema - without the ORM magic that makes query optimization hard. Kysely in particular is worth a look if the manual typing in pool.query<T> feels too brittle.

What You Actually Lose

This is important to say clearly: there are real things you give up.

Schema migrations. Prisma Migrate is good. When you drop Prisma from your query layer you still want a migration tool. I use node-pg-migrate, others use db-migrate or just raw SQL files in a migrations folder with a simple runner. None of them are as polished as Prisma Migrate.

The schema as source of truth. Prisma's schema file makes it easy to see your data model at a glance and generates types from it. With raw SQL you're maintaining types manually or generating them from the database schema with something like pgtyped or zapatos.

Prisma Studio. Minor thing but worth mentioning - having a UI to browse your data is useful during development.

Onboarding speed. New developers on a project with raw SQL need to know SQL. This is not a bad thing, but it's a real cost.

When to Keep Prisma

Prisma is the right choice when:

Your team isn't comfortable with SQL
You're building a CRUD app where the Prisma query builder covers 90%+ of your needs
You're early stage and query performance isn't a bottleneck yet
The productivity gain from the DX outweighs the performance cost

It stops being the right choice when:

Your most important queries can't be expressed through the query builder
You're regularly escaping to $queryRaw for anything beyond simple lookups
Query times are a meaningful part of your latency budget
You need fine-grained control over indexes, hints, or query plans

The answer for most production systems that have been running for more than a year is: use both. Prisma for the simple stuff, raw SQL for the queries that matter.

What ORM or query approach are you using in production? Anything that changed your mind in either direction? Comments are open.

Production Observability for Kubernetes on AWS using OpenTelemetry Operator

Mohammad Imran — Mon, 06 Apr 2026 13:13:53 +0000

Modern Kubernetes environments are highly dynamic, distributed, and complex. While this enables scalability and flexibility, it also introduces a critical challenge: observability at scale.

In production systems, simply collecting logs or metrics is not enough. You need a unified observability strategy that provides:

Metrics (system health)
Logs (events & debugging)
Traces (request flow across services)

In this blog, we’ll explore how to build a production-grade observability stack on AWS using Kubernetes and the OpenTelemetry Operator, covering architecture, implementation, and best practices.

Why Observability is Critical in Kubernetes

Kubernetes introduces several layers of abstraction:

Pods are ephemeral
Services scale dynamically
Network paths are non-linear
Failures are distributed

Without proper observability, it becomes difficult to:

Identify bottlenecks
Debug latency issues
Trace failures across services
Monitor system health

Observability Architecture Overview

End-to-end observability architecture in Kubernetes using OpenTelemetry Operator, Collector, and Grafana stack.

Architecture Flow

Applications are instrumented using OpenTelemetry
OpenTelemetry Operator injects auto-instrumentation
Telemetry is collected by OpenTelemetry Collector
Data is exported to:
- Prometheus (metrics)
- Loki (logs)
- Tempo (traces)
Grafana visualizes all signals

Key Components of the Stack

OpenTelemetry Operator

Auto-injects agents into pods
Manages collectors as CRDs
Standardizes telemetry pipelines

OpenTelemetry Collector

Receives telemetry
Processes data
Exports to backends

Prometheus (Metrics)

CPU / Memory
Request rate
Error rate
Latency

Grafana Tempo (Traces)

Distributed tracing
Service dependencies
Latency analysis

Loki (Logs)

Log aggregation
Correlation with traces

Grafana

Dashboards
Logs
Traces

Deploying on AWS (EKS-Based Architecture)

Amazon EKS → workloads
OpenTelemetry Operator → instrumentation
OpenTelemetry Collector → pipeline
S3 → storage
Grafana → visualization

Auto-Instrumentation Example

apiVersion: opentelemetry.io/v1alpha1
kind: Instrumentation
metadata:
  name: java-instrumentation
spec:
  java:
    image: ghcr.io/open-telemetry/opentelemetry-operator/autoinstrumentation-java

OpenTelemetry Collector Config

apiVersion: opentelemetry.io/v1alpha1
kind: OpenTelemetryCollector
metadata:
  name: otel-collector
spec:
  config: |
    receivers:
      otlp:
        protocols:
          grpc:
          http:

    processors:
      batch:

    exporters:
      prometheus:
        endpoint: "0.0.0.0:8889"
      tempo:
        endpoint: tempo:4317

    service:
      pipelines:
        traces:
          receivers: [otlp]
          processors: [batch]
          exporters: [tempo]

Correlation Workflow

Alert triggered
Check metrics
Inspect traces
Check logs
Identify root cause

Production Best Practices

Use sampling
Scale collectors
Separate pipelines
Monitor collectors
Secure telemetry

Common Pitfalls

No collector
Over-collection
No sampling
No correlation

Real-World Example

User → Frontend → Product → Cart → Checkout → Payment

Observability helps trace issues across services.

Production Debugging Scenario

Let’s look at a real-world scenario to understand how observability helps in production.

Scenario

Users report that the checkout service is slow in a production e-commerce application.

Step 1: Detect the Issue (Metrics)

Grafana dashboard shows:

Increased latency in checkout service
Spike in response time (P95/P99)

This indicates a performance issue but doesn’t reveal the root cause.

Step 2: Trace the Request (Traces)

Using Grafana Tempo:

Identify slow traces
Analyze request flow

Example trace:

Frontend → Cart Service → Checkout Service → Payment Service

Observation:

Checkout service is taking unusually long

Step 3: Drill Down into Spans

Within the trace:

A specific span shows high latency
Database query inside checkout service is slow

Step 4: Inspect Logs

Using Loki:

Filter logs for checkout service
Identify errors or warnings

Finding:

Database timeout errors
Slow query logs

Step 5: Root Cause Identified

Inefficient database query
Missing index

Step 6: Resolution

Optimize query
Add database index
Reduce response latency

Outcome

Latency reduced
System stabilized
Faster incident resolution

Key Insight

This workflow demonstrates the power of correlating metrics, traces, and logs:

Metrics → detect
Traces → locate
Logs → explain

This significantly reduces MTTR (Mean Time to Resolution) in production systems.

Final Thoughts

Combining:

OpenTelemetry Operator
OpenTelemetry Collector
Prometheus, Loki, Tempo
Grafana

enables a scalable, production-grade observability platform on AWS.

Observability is not optional, it is foundational.

MCP Development with Python, Gemini CLI, and Amazon AWS ECS Express

xbill — Mon, 06 Apr 2026 13:09:29 +0000

Leveraging Gemini CLI and the underlying Gemini LLM to build Model Context Protocol (MCP) AI applications with Python with a local development environment hosted on AWS ECS Express.

Aren’t There a Billion Python MCP Demos?

Yes there are.

Python has traditionally been the main coding language for ML and AI tools. The goal of this article is to provide a minimal viable basic working MCP stdio server that can be run locally without any unneeded extra code or extensions.

What Is Python?

Python is an interpreted language that allows for rapid development and testing and has deep libraries for working with ML and AI:

Welcome to Python.org

Python Version Management

One of the downsides of the wide deployment of Python has been managing the language versions across platforms and maintaining a supported version.

The pyenv tool enables deploying consistent versions of Python:

GitHub - pyenv/pyenv: Simple Python version management

As of writing — the mainstream python version is 3.13. To validate your current Python:

admin@ip-172-31-70-211:~/gemini-cli-aws/mcp-stdio-python-aws$ python --version
Python 3.13.12

Amazon ECS Express Configuration

Amazon ECS Express Mode (announced Nov 2025) is a simplified deployment feature for Amazon Elastic Container Service (ECS) designed to rapidly launch containerized applications, APIs, and web services on AWS Fargate. It automates infrastructure setup — including load balancing, networking, scaling, and HTTPS endpoints — allowing developers to deploy from container image to production in a single step.

More details are available here:

Amazon ECS Express Mode

Gemini CLI

If not pre-installed you can download the Gemini CLI to interact with the source files and provide real-time assistance:

npm install -g @google/gemini-cli

Testing the Gemini CLI Environment

Once you have all the tools and the correct Node.js version in place- you can test the startup of Gemini CLI. You will need to authenticate with a Key or your Google Account:

gemini

admin@ip-172-31-70-211:~/gemini-cli-aws/mcp-stdio-python-aws$ gemini

▝▜▄ Gemini CLI v0.33.1
    ▝▜▄
   ▗▟▀ Logged in with Google /auth
  ▝▀ Gemini Code Assist Standard /upgrade

? for shortcuts 
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
 shift+tab to accept edits 3 GEMINI.md files | 1 MCP server
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
 > Type your message or @path/to/file
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
 ~/.../mcp-stdio-python-aws (main*) no sandbox (see /docs) /model Auto (Gemini 3) | 239.8 MB

Node Version Management

Gemini CLI needs a consistent, up to date version of Node. The nvm command can be used to get a standard Node environment:

GitHub - nvm-sh/nvm: Node Version Manager - POSIX-compliant bash script to manage multiple active node.js versions

Python MCP Documentation

The official GitHub Repo provides samples and documentation for getting started:

GitHub - modelcontextprotocol/python-sdk: The official Python SDK for Model Context Protocol servers and clients

The most common MCP Python deployment path uses the FASTMCP library:

Welcome to FastMCP - FastMCP

Where do I start?

The strategy for starting MCP development is a incremental step by step approach.

First, the basic development environment is setup with the required system variables, and a working Gemini CLI configuration.

Then, a minimal Hello World Style Python MCP Server is built with stdio transport. This server is validated with Gemini CLI in the local environment.

This setup validates the connection from Gemini CLI to the local process via MCP. The MCP client (Gemini CLI) and the Python MCP server both run in the same local environment.

Next- the basic MCP server is extended with Gemini CLI to add several new tools in standard Python code.

Setup the Basic Environment

At this point you should have a working Python interpreter and a working Gemini CLI installation. The next step is to clone the GitHub samples repository with support scripts:

cd ~
git clone https://github.com/xbill9/gemini-cli-aws

Then run init.sh from the cloned directory.

The script will attempt to determine your shell environment and set the correct variables:

cd gemini-cli-aws
source init.sh

If your session times out or you need to re-authenticate- you can run the set_env.sh script to reset your environment variables:

cd gemini-cli-aws
source set_env.sh

Variables like PROJECT_ID need to be setup for use in the various build scripts- so the set_env script can be used to reset the environment if you time-out.

Hello World with HTTP Transport

One of the key features that the standard MCP libraries provide is abstracting various transport methods.

The high level MCP tool implementation is the same no matter what low level transport channel/method that the MCP Client uses to connect to a MCP Server.

The simplest transport that the SDK supports is the stdio (stdio/stdout) transport — which connects a locally running process. Both the MCP client and MCP Server must be running in the same environment.

The HTTP transport allows the MCP Client and Server to be in the same environment or distributed over the Internet.

The connection over HTTP will look similar to this:

mcp.run(
        transport="http",
        host="0.0.0.0",
        port=port,
    )

Running the Python Code

First- switch the directory with the Python MCP sample code. Then, run the release version on the local system:

cd ~/gemini-cli-aws/mcp-https-python-ecs-express
make release

If everything is running correctly — you will see the FASTMCP banner:


                 ╭──────────────────────────────────────────────────────────────────────────────╮                 
                 │ │                 
                 │ │                 
                 │ ▄▀▀ ▄▀█ █▀▀ ▀█▀ █▀▄▀█ █▀▀ █▀█ │                 
                 │ █▀ █▀█ ▄▄█ █ █ ▀ █ █▄▄ █▀▀ │                 
                 │ │                 
                 │ │                 
                 │ FastMCP 3.2.0 │                 
                 │ https://gofastmcp.com │                 
                 │ │                 
                 │ 🖥 Server: hello-world-server, 3.2.0 │                 
                 │ 🚀 Deploy free: https://horizon.prefect.io │                 
                 │ │                 
                 ╰──────────────────────────────────────────────────────────────────────────────╯                 

[04/04/26 19:10:50] INFO Starting MCP server 'hello-world-server' with transport 'http' on transport.py:299
                             http://0.0.0.0:8080/mcp                                                              
INFO: Started server process [27502]
INFO: Waiting for application startup.
{"message": "StreamableHTTP session manager started"}
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

At this point — hit Cntrl-C to exit the local HTTP server.

Deploy to ECS Express

Now that the server has been tested locally- start the remote deployment to ECS Express.

Use the deploy target:

 > make deploy
✦ I will execute the full deployment cycle, which includes building the Docker image, pushing it to ECR, and
  deploying to ECS Express Mode.

Then check the status:

 > make status
✦ I will check the status of the ECS Express Mode service.

╭────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ ✓ Shell make status [in /home/xbill/gemini-cli-aws/mcp-ecsexpress-python-aws] │
│ │
│ Checking AWS ECS service status for mcp-express-python-aws... │
│ Service: mcp-express-python-aws │
│ Status: ACTIVE │
│ Endpoint: mc-8d69e7dfa87344be98fdb3f7a8fbbbba.ecs.us-east-1.on.aws │
╰────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
✦ The ECS Express Mode service mcp-express-python-aws is ACTIVE and available at:
  mc-8d69e7dfa87344be98fdb3f7a8fbbbba.ecs.us-east-1.on.aws

And the endpoint:

  Based on the make status output earlier, the actual endpoint is:
  mc-8d69e7dfa87344be98fdb3f7a8fbbbba.ecs.us-east-1.on.aws

Gemini CLI settings.json

The default Gemini CLI settings.json has an entry for the Python source:

{
  "mcpServers": {
    "aws-ecs-express-python": {
      "httpUrl": "https://mc-8d69e7dfa87344be98fdb3f7a8fbbbba.ecs.us-east-1.on.aws/mcp"
    }
  }
}

Validation with Gemini CLI

Leave the HTTP server running and open a new terminal.

Gemini CLI is restarted and the MCP connection over HTTP to the Python Code is validated. The full Gemini CLI Session will start:


xbill@penguin:~/gemini-cli-aws/mcp-ecsexpress-python-aws$ gemini

 ▝▜▄ Gemini CLI v0.36.0
   ▝▜▄
  ▗▟▀ Signed in with Google: xbill@glitnir.com /auth
 ▝▀ Plan: Gemini Code Assist Standard /upgrade

 > /mcp list
Configured MCP servers:

🟢 aws-ecs-express-python - Ready (1 tool)
  Tools:
  - mcp_aws-ecs-express-python_greet

And you can then connect to the MCP Server over HTTP:

 > greet ECS Express!
✦ I will call the greet tool with "ECS Express!".

╭────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ ✓ greet (aws-ecs-express-python MCP Server) {"param":"ECS Express!"} │
│ │
│ ECS Express! │
╰────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
✦ The greet tool returned: ECS Express!

Validate Project Setup with Gemini CLI

Gemini CLI was used to summarize the project:

 > summarize the project deployment to Amazon ECS Express
✦ The deployment of the mcp-express-python-aws project to Amazon ECS Express Mode is a streamlined process that
  automates the container lifecycle from build to public endpoint.

  Deployment Workflow (make deploy)
   1. Containerization: Builds a Python 3.13-slim Docker image and pushes it to an Amazon ECR repository
      (mcp-server-repo).
   2. Identity & Access: Automatically provisions necessary IAM roles (ecsTaskExecutionRole and
      ecsInfrastructureRoleForExpressServices) if they do not already exist.
   3. Orchestration: Deploys the container to ECS Express Mode, which simplifies AWS Fargate deployments by
      automatically managing the Application Load Balancer, Security Groups, and Networking.

  Current Infrastructure State
   * Service Name: mcp-express-python-aws
   * Status: ACTIVE
   * Compute: AWS Fargate (Task size: 1 vCPU / 2GB RAM)
   * Scaling: Configured for 1 to 20 tasks, triggered at 60% average CPU usage.
   * Public Access: mc-8d69e7dfa87344be98fdb3f7a8fbbbba.ecs.us-east-1.on.aws (Port 8080)

  The service is currently running and reachable via its assigned ECS Express Gateway endpoint.

Summary

The strategy for using Python for MCP development with Gemini CLI and AWS was validated with a incremental step by step approach.

A minimal HTTP transport MCP Server was started from Python source code and validated with Gemini CLI running as a MCP client in the same local environment.

The HTTP server was then tested remotely from the AWS ECS Express instance.

This approach can be extended to more complex deployments using other MCP transports and Cloud based options.

25 Internal Knowledge and Productivity Agent Patterns on AWS You Can Steal Right Now

Marcelo Acosta Cavalero — Mon, 06 Apr 2026 13:09:09 +0000

Originally published on Build With AWS. Subscribe for weekly AWS builds.

An engineer spent 40 minutes last Thursday searching for the internal API rate-limiting policy. She checked Confluence, Notion, three Slack channels, and finally asked a colleague who pointed her to a Google Doc shared in a thread six months ago. The policy existed.

Finding it was the problem.

This is the second edition of a five-part series cataloging real AI architecture patterns running on AWS.

Edition 1 covered 25 customer-facing agents.

This edition shifts the lens inward: 25 patterns for employee-facing agents that handle knowledge retrieval, internal support, operational productivity, and the daily friction that slows teams down.

If you missed Edition 1, go back for the “Agent or Not?” scoring framework and the AgentCore vs Quick breakdown.

Those mental models apply here too, so this edition skips straight to the architectures and use cases.

One platform update before the cards: Edition 1 split the world into AgentCore (custom agents) and Quick (analytics).

Internal agents add a third lane. Amazon Q Business is the AWS-native default for enterprise knowledge assistants, permissions-aware search, and SaaS-connected internal help desks.

It ships with native connectors for Google Drive, Slack, Confluence, Jira, SharePoint, and dozens more, with document-level ACLs built in.

Q Business can trigger actions through plugins, but AgentCore remains the better choice when workflows require deterministic orchestration, multi-step execution, or strict policy enforcement.

AgentCore remains the right choice for custom agent backends with tool orchestration, memory, identity, and fine-grained control.

Amazon Quick stays in its lane for analytics, dashboarding, research, and workflow automation around business data.

Several patterns below use Q Business for retrieval and AgentCore for action, which turns out to be the natural split for internal workloads.

Reference Architectures for Internal Agents

Internal agents integrate with different systems than customer-facing ones. Corporate identity providers, internal wikis, HR platforms, CI/CD pipelines, and financial systems replace the CRM and e-commerce APIs from Edition 1.

The four reference architectures adapt accordingly.

Reference Architecture D - Single Agent with Internal Tool Access

Platform: AgentCore

When to use: The agent reasons about which internal tools to query, in what order, based on the employee’s role and question. One agent handles the full interaction with 3-8 internal system integrations.

Covers most IT support, HR advisory, and workflow-execution agents where the agent needs to take actions through APIs.

For pure knowledge retrieval and Q&A, see Architecture D2 below.

AgentCore Identity integrates with your corporate IdP (Okta, Azure AD) for SSO. AgentCore Policy enforces role-based access scoping - verify maturity for your target region before production rollout.

Reference Architecture E - Quick Workspace for Internal Intelligence

Platform: Quick

When to use: Teams need AI-powered analysis of internal data, operational metrics, or workforce analytics without writing code.

Covers engineering velocity dashboards, headcount planning analysis, budget tracking, and self-service reporting for managers and operations teams.

Reference Architecture F - Multi-Agent Internal Workflow

Platform: AgentCore (multi-agent)

When to use: Employee requests span IT, HR, finance, and facilities.

Each domain needs its own tools, knowledge bases, and policy constraints.

A single agent trying to handle all internal functions becomes unreliable at 15+ tools. Specialized agents behind a router keep each context window focused.

Reference Architecture G - Q Business for Enterprise Knowledge

Platform: Amazon Q Business

When to use: The primary need is permissions-aware search and Q&A across SaaS knowledge sources.

Q Business ships with native connectors for dozens of data sources and enforces document-level ACLs automatically.

No custom orchestration code required.

Covers enterprise knowledge search, policy Q&A, and any pattern where the core job is “find the right document and synthesize an answer the employee is authorized to see.”

When the same workflow also needs to take actions (create tickets, provision access, call APIs), pair Q Business for retrieval with AgentCore for execution.

The 25 Use Cases

Knowledge Management and Search

#026 - Enterprise Knowledge Search Agent

Pattern: Modernization from chatbot

Platform: Amazon Q Business (primary), AgentCore (optional action layer) Complexity: Quick Win

Reference Architecture: G

What the agent does:

Searches across internal knowledge sources - Confluence, SharePoint, Google Drive, Slack message history, Jira, and S3 - through a single conversational interface.
Understands natural language questions (”What’s our policy on vendor security reviews?”), retrieves relevant documents from multiple sources, synthesizes a direct answer with citations, and identifies when conflicting information exists across sources.
Respects document-level permissions so employees only see content they have access to. Amazon Q Business handles this natively: its built-in connectors index these sources and its ACL engine maps existing permissions without custom code.
For sources Q Business does not cover natively, Bedrock Knowledge Bases with a custom data source connector fills the gap, though note that some Bedrock connectors (such as Confluence) are in preview and do not yet support multimodal content like tables and diagrams.

AWS services: Amazon Q Business (connectors + retriever + ACL engine), Bedrock Knowledge Bases (custom RAG for unsupported sources), S3 (document store)

You need this if: Your employees regularly say “I know we documented this somewhere” and spend 20+ minutes searching across 3 or more knowledge platforms.

#027 - Policy and Compliance Q&A Agent

Pattern: New build

Platform: Amazon Q Business (primary), AgentCore (for action routing)

Complexity: Quick Win

Reference Architecture: G

What the agent does:

Answers employee questions about internal policies - travel expenses, PTO accrual, data classification, security requirements, procurement thresholds, acceptable use.
Pulls from the authoritative policy documents (not outdated wiki copies) and provides specific answers with page references.
Q Business indexes the policy corpus from S3 or SharePoint and enforces access controls so employees only see policies relevant to their role.
When policies are ambiguous or the question falls outside documented rules, an AgentCore action layer identifies the policy owner and drafts an email for the employee to send.
Tracks which policies generate the most questions, surfacing candidates for clarification.

AWS services: Amazon Q Business (retriever + ACL engine), S3 (policy document store), AgentCore Runtime (action routing), CloudWatch (query analytics)

You need this if: Your HR, legal, or compliance team answers the same policy questions repeatedly, and employees default to asking coworkers instead of reading the docs.

#028 - Institutional Knowledge Capture Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Runs structured knowledge extraction interviews with subject matter experts, particularly before role transitions, departures, or reorganizations.
Asks targeted questions about undocumented processes, tribal knowledge, key relationships, and decision context.
Transcribes and synthesizes responses into structured knowledge articles with proper metadata and cross-references.
Identifies gaps where captured knowledge contradicts or supplements existing documentation.
Generates a handoff document for successors.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Memory (interview state), Amazon Transcribe, S3 (knowledge archive), Bedrock Knowledge Bases

You need this if: Critical knowledge walks out the door when senior employees leave, and your team spends months reconstructing context that lived in someone’s head.

#029 - Technical Documentation Assistant

Pattern: Modernization from chatbot

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Helps engineers navigate internal API documentation, runbooks, architecture decision records, and system diagrams.
Answers questions like “How does the payment service authenticate with the ledger?” by pulling from code comments, README files, ADRs, and internal docs.
When documentation is stale or missing, it flags the gap and creates a draft based on the current codebase.
Understands code context so it can explain what a service does, not just repeat what the docs say.

AWS services: Bedrock (Claude), AgentCore Runtime, Bedrock Knowledge Bases (documentation + custom-ingested code artifacts), Amazon Q Developer (native repository integration)

You need this if: Your engineering team wastes hours reading outdated documentation and reverse-engineering service behavior because the docs do not match the code.

#030 - Cross-Team Decision Log Agent

Pattern: New build

Platform: Both (AgentCore backend + Quick analytics)

Complexity: Strategic Bet

Reference Architecture: D + E

What the agent does:

Captures architectural decisions, trade-off discussions, and design choices from Slack threads, meeting transcripts, and PR comments.
Structures them into searchable decision records with context, alternatives considered, rationale, and stakeholders.
When a team proposes something that contradicts or revisits a prior decision, the agent surfaces the original discussion and reasoning.
Quick dashboards show decision frequency by domain, open questions, and areas where decisions are overdue.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Memory, Amazon Quick (QuickSight + Index), Amazon Transcribe, S3 (decision archive)

You need this if: Your teams relitigate the same technical decisions every quarter because nobody remembers why the original choice was made.

IT Help Desk and Internal Support

#031 - IT Help Desk Agent

Pattern: Modernization from chatbot

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Handles common IT support requests through Slack or a web interface.
Resets passwords via the IdP API, provisions software licenses through the asset management system.
Troubleshoots VPN connectivity with diagnostic checks.
Resolves printer issues with guided walkthroughs, and manages MFA token enrollment.
For issues requiring hands-on support, it collects diagnostic information, determines priority based on impact and urgency, and creates a ticket with all relevant context pre-populated.

AWS services: Bedrock (Nova), AgentCore Runtime, AgentCore Identity (IdP integration), AgentCore Gateway (ITSM APIs), ServiceNow API, Okta/Azure AD API

You need this if: More than 50% of your IT help desk tickets are password resets, access requests, and connectivity issues that follow standard resolution procedures.

#032 - Software Access Provisioning Agent

Pattern: Migration from RPA

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Processes software access requests end-to-end. Employee asks for access to a tool (GitHub org, AWS account, Datadog, Salesforce).
The agent checks the employee’s role against the entitlement matrix, identifies whether manager approval is needed, routes the approval request, and upon approval, provisions access via the tool’s API or SCIM endpoint.
Handles license availability checks and waitlisting.
Automatically de-provisions access when employees change roles or depart based on HRIS events.

AWS services: Bedrock (Nova), AgentCore Runtime, AgentCore Policy (entitlement rules), AgentCore Identity, SCIM APIs, HRIS API (Workday/BambooHR), EventBridge (lifecycle events)

You need this if: Software access requests take 2+ business days to fulfill because they require manual approval chains and admin intervention across multiple systems.

#033 - Incident Communication Coordinator

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

During production incidents, drafts and distributes internal status updates based on real-time information from monitoring tools and the incident Slack channel.
Pulls metrics from CloudWatch and Datadog, summarizes the current state of the incident, identifies affected services and customer impact, and posts updates to the status page and stakeholder channels at configured intervals.
After resolution, compiles a timeline of events and generates a postmortem draft with contributing factors and action items pre-populated from the incident channel discussion.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Gateway (monitoring APIs), CloudWatch, EventBridge, SNS (notifications), S3 (postmortem archive)

You need this if: Your incident commanders spend more time writing status updates than resolving the incident, and postmortems take a week to produce because nobody captured the timeline in real-time.

#034 - Infrastructure Self-Service Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Lets developers request and configure cloud infrastructure through conversation instead of filing tickets.
Handles common requests: spin up a dev environment, create an S3 bucket with standard tagging, set up a new RDS instance within approved configurations, or request a temporary IAM role for cross-account access.
Validates all requests against organizational policies and guardrails (naming conventions, cost limits, security baselines) before executing via IaC templates.
Non-standard requests route to the platform team with a pre-filled request.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Policy (guardrails), AWS Service Catalog, CloudFormation/CDK, IAM, AWS Organizations

You need this if: Your platform team processes 30+ infrastructure requests per week and developers wait 1-3 days for standard environments that could be provisioned in minutes.

#035 - Security Questionnaire Response Agent

Pattern: Migration from RPA

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Completes vendor security questionnaires and customer security assessments by matching questions against a maintained library of approved responses.
Pulls from SOC 2 reports, penetration test summaries, architecture documentation, and previously approved answers.
Drafts responses for each question with confidence scores.
High-confidence answers (exact matches to prior approved responses) are auto-filled.
Low-confidence answers are flagged for security team review.
Tracks which questions appear most frequently to prioritize documentation improvements.

AWS services: Bedrock (Claude), AgentCore Runtime, Bedrock Knowledge Bases (security response library, optionally backed by OpenSearch Serverless for advanced retrieval control), S3 (compliance documents)

You need this if: Your security team spends 10+ hours per week completing repetitive security questionnaires, and the same questions appear across 80% of inbound assessments.

HR and People Operations

#036 - Employee Onboarding Navigator

Pattern: Modernization from chatbot

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Guides new hires through their first 90 days.
Sends day-one setup instructions (laptop configuration, tool access, building entry).
Answers questions about benefits enrollment deadlines, org structure, team norms, and internal processes.
Adapts the onboarding checklist based on role, department, and location.
Tracks completion of required training, compliance acknowledgments, and documentation reviews.
Nudges managers when their new hire’s onboarding milestones are stalling.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Memory (onboarding state), HRIS API (Workday/BambooHR), LMS API, SES/SNS (notifications)

You need this if: New hire ramp time exceeds 30 days, onboarding satisfaction scores are below 80%, and your HR team manually tracks checklist completion in spreadsheets.

#037 - Benefits and Leave Advisory Agent

Pattern: Modernization from chatbot

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Answers employee questions about health insurance plans, 401(k) matching, HSA/FSA eligibility, parental leave, PTO balance, and FMLA procedures.
Pulls real-time data from the HRIS and benefits platforms to give personalized answers (”You have 8.5 PTO days remaining this year”).
Walks employees through benefits enrollment during open enrollment with side-by-side plan comparisons based on their specific situation (family size, expected medical usage, contribution preferences).
Routes complex cases to HR specialists with the question and relevant context pre-attached.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Identity (employee verification), HRIS API, benefits platform API, Bedrock Guardrails (PII handling)

You need this if: Your HR inbox is dominated by benefits questions during open enrollment, and employees make suboptimal plan selections because they do not understand their options.

#038 - Internal Job Matching Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Matches employees to internal open positions based on skills, career goals, project history, and performance data.
Goes beyond keyword matching on job descriptions: analyzes the employee’s actual work (code contributions, project involvement, skills demonstrated in reviews) against what the hiring manager needs.
Surfaces opportunities employees might not have found or considered.
Provides a match explanation (”Your work on the data pipeline migration maps directly to this team’s real-time analytics build”).
Respects confidentiality so managers are not notified unless the employee applies.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Policy (confidentiality rules), HRIS API, ATS API (Greenhouse/Lever), Bedrock Knowledge Bases (job postings + employee profiles)

You need this if: Internal mobility is below 15%, employees leave for roles they could have found internally, and your job board gets low engagement because listings read like external postings.

#039 - Performance Review Preparation Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Helps managers prepare for performance reviews by compiling an employee’s contributions over the review period.
Pulls data from project management tools (Jira tickets completed, PRs merged, epics delivered), peer feedback, 1:1 notes, goal tracking systems, and prior review history.
Generates a structured draft highlighting key accomplishments, growth areas, and evidence for each.
Does not write the evaluation - it assembles the evidence so the manager spends time on assessment quality instead of data gathering.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Policy (data access controls), Jira API, GitHub API, HRIS API, 15Five/Lattice API

You need this if: Your managers spend 3+ hours per direct report gathering data for reviews, and review quality suffers because managers rely on recency bias instead of full-period evidence.

#040 - Compensation Benchmarking Agent

Pattern: New build

Platform: Both (AgentCore backend + Quick analytics)

Complexity: Foundation Build

Reference Architecture: D + E

What the agent does:

Helps HR and hiring managers make compensation decisions by pulling from internal pay bands, market survey data, and peer comparisons.
Takes a role, level, location, and candidate profile, then generates a recommended offer range with supporting data.
Flags when a proposed offer falls outside band or creates internal equity concerns.
Quick dashboards show compensation distribution by team, gender pay gap analysis, and market competitiveness by role family.
All outputs route through HR approval before reaching the hiring manager.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Policy (data access restrictions), Amazon Quick (QuickSight + Research), HRIS API, compensation survey APIs, Redshift

You need this if: Compensation decisions take a week because they require HR to manually pull market data, check internal equity, and build a justification for every offer.

Engineering and Development

#041 - Code Review Context Agent

Pattern: New build

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Enriches pull requests with context that speeds up code review.
When a PR is opened, it analyzes the changes and adds a summary: which services are affected, what architectural patterns changed, whether the change touches a critical path, and links to related PRs and design docs.
Flags potential issues: breaking API changes, missing test coverage for modified paths, configuration changes that affect other teams, and dependency updates with known vulnerabilities.
Does not approve or block - it surfaces what a reviewer should pay attention to.

AWS services: Bedrock (Claude), AgentCore Runtime, GitHub/GitLab API, Bedrock Knowledge Bases (architecture docs + ADRs), Amazon Q Developer (code review context)

You need this if: Code reviews take 2+ days because reviewers spend most of their time understanding context rather than evaluating the actual change.

#042 - Incident Postmortem Generator

Pattern: New build

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Produces structured postmortem documents from incident data.
Pulls the timeline from PagerDuty or Opsgenie, reconstructs the sequence of events from the incident Slack channel, correlates with deployment logs and monitoring data, and generates a draft postmortem following your team’s template.
Identifies contributing factors by analyzing what changed before the incident (deploys, config changes, traffic spikes).
Pre-populates action items based on patterns from previous incidents.
The on-call engineer reviews and refines instead of writing from scratch.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Gateway (PagerDuty/Opsgenie API, Slack API), CloudWatch Logs, S3 (postmortem archive)

You need this if: Postmortems take a week to produce, half of incidents never get a written postmortem, and your team keeps encountering the same failure modes.

#043 - Dependency Risk Assessment Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Continuously monitors your codebase’s dependency tree for risk signals beyond CVEs.
Analyzes maintainer activity (abandoned projects, single-maintainer risk), license compatibility, breaking change frequency in upstream releases, and supply chain indicators (typosquatting packages, unexpected maintainer changes).
When a dependency update is available, provides a risk assessment: what changed, what might break, and whether similar codebases have reported issues.
Prioritizes updates based on actual exposure, not just severity scores.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Gateway (GitHub API, package registry APIs), Amazon Inspector (vulnerability scanning + SCA), Amazon Q Developer (code-level risk context), EventBridge (scheduled scans)

You need this if: Your dependency updates are either ignored for months (creating security debt) or applied blindly (causing unexpected breakages), and Dependabot alerts alone do not give you enough context to prioritize.

#044 - On-Call Handoff Agent

Pattern: New build

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Generates end-of-rotation handoff briefs for on-call engineers.
Compiles all incidents from the rotation (alerts fired, pages received, resolutions applied), ongoing issues that need monitoring, recent deployments that might cause problems, and upcoming changes the next on-call should watch.
Pulls from PagerDuty, Slack incident channels, deployment logs, and the change calendar.
The outgoing on-call reviews and annotates the brief before it goes to the incoming engineer.

AWS services: Bedrock (Claude), AgentCore Runtime, PagerDuty API, Slack API, deployment pipeline API, SES (handoff delivery)

You need this if: On-call handoffs happen verbally (or not at all), incoming engineers start blind, and the first hour of every rotation is spent asking “what happened this week?”

#045 - Architecture Decision Record Agent

Pattern: New build

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Facilitates the creation of Architecture Decision Records from design discussions.
Monitors designated Slack channels and meeting transcripts for architectural debates.
When it detects a decision being made, it drafts an ADR: context, decision, alternatives considered, consequences, and status.
Tags the relevant teams and stakeholders for review.
Maintains a searchable index of all ADRs linked to the services they affect.
When someone proposes a change that conflicts with an existing ADR, the agent surfaces the relevant record and asks whether this is an intentional reversal.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Memory, Slack API, Amazon Transcribe, Bedrock Knowledge Bases (ADR corpus), S3

You need this if: Your team makes architectural decisions in Slack threads that nobody can find three months later, and new engineers re-propose approaches that were already evaluated and rejected.

Finance and Procurement

#046 - Expense Report Processing Agent

Pattern: Migration from RPA

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Processes expense reports by extracting data from uploaded receipts using Amazon Textract, matching expenses against the company’s travel and expense policy, flagging out-of-policy items with specific policy references, and routing compliant reports for manager approval.
Handles currency conversion for international expenses, per diem calculations by city, and mileage reimbursement.
Auto-categorizes expenses for GL coding.
Reports with flagged items go to the submitter for correction before reaching the approval queue.

AWS services: Bedrock (Nova), AgentCore Runtime, Amazon Textract, AgentCore Policy (expense rules), expense management API (Concur/Expensify), DynamoDB

You need this if: Your finance team manually reviews expense reports for policy compliance, processing takes 5+ business days, and 30% of submissions require back-and-forth corrections.

#047 - Procurement Request Agent

Pattern: Migration from RPA

Platform: AgentCore

Complexity: Strategic Bet

Reference Architecture: D

What the agent does:

Guides employees through procurement requests conversationally.
Collects requirements (what they need, why, budget, timeline), checks whether an existing contract covers the request, identifies the correct approval chain based on amount and category, and generates a purchase requisition.
For software purchases, checks the approved vendor list and existing license inventory to avoid redundant buying.
Handles the approval workflow: routes to the right approvers, sends reminders, escalates stalled approvals, and notifies the requester at each stage.

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Policy (approval rules + spend limits), ERP API (SAP/Oracle/NetSuite), contract management API, SES (notifications)

You need this if: Employees avoid the procurement process because it requires filling out forms they do not understand, and your procurement team spends hours routing requests to the right approvers.

#048 - Budget Tracking and Forecast Agent

Pattern: New build Platform: Both (AgentCore backend + Quick dashboards) Complexity: Strategic Bet

Reference Architecture: D + E

What the agent does:

Monitors department budgets against actuals in real-time.
Pulls spend data from the ERP, cloud billing (AWS Cost Explorer), and SaaS management platforms. Alerts budget owners when spending trends suggest they will exceed budget before quarter end.
Generates variance explanations by analyzing which line items are over or under plan.
Quick dashboards let managers drill into spend by category, vendor, and project.
Produces monthly budget summaries and forecast adjustments automatically.

AWS services: Bedrock (Claude), AgentCore Runtime, Amazon Quick (QuickSight + Flows), AWS Cost Explorer API, ERP API, Redshift, EventBridge (alerting triggers), SNS

You need this if: Budget reviews happen monthly from stale spreadsheets, overspend is discovered after the fact, and finance produces variance reports manually.

Meetings and Communication

#049 - Meeting Summarization and Action Tracker

Pattern: New build

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Joins meetings (via Amazon Chime SDK or calendar integration), transcribes the discussion, and produces a structured summary within minutes of the meeting ending.
Identifies decisions made, action items with owners and due dates, open questions, and topics deferred.
Posts the summary to the relevant Slack channel or project management tool.
Tracks action items across meetings and flags overdue items in the next meeting’s pre-brief.
Distinguishes between informational discussion and actionable outcomes.

AWS services: Bedrock (Claude), AgentCore Runtime, Amazon Transcribe, Amazon Chime SDK (the SDK remains supported independently of the Chime service), Slack API, Jira API (action item creation), S3 (transcript archive)

You need this if: Action items from meetings disappear into notes nobody reads, decisions get relitigated because they were not recorded, and your team spends 5+ hours per week in meetings without clear outcomes.

#050 - Status Report Generator

Pattern: New build

Platform: AgentCore

Complexity: Quick Win

Reference Architecture: D

What the agent does:

Compiles weekly or biweekly status reports by pulling from the systems where work actually happens.
Aggregates Jira ticket progress, GitHub PR activity, deployment history, incident reports, and OKR tracking data.
Produces a structured update for each team: what shipped, what is in progress, what is blocked, and key metrics.
Managers review and edit instead of writing from scratch.
Adapts format and detail level based on the audience (team standup vs executive briefing vs cross-functional update).

AWS services: Bedrock (Claude), AgentCore Runtime, AgentCore Gateway (Jira, GitHub, OKR platform APIs), S3 (report archive), SES (distribution)

You need this if: Your managers spend 2+ hours per week writing status reports by manually checking Jira, GitHub, and Slack, and the reports are outdated by the time they are sent.

What These 25 Patterns Reveal

Different dynamics emerge when agents face inward instead of outward.

Knowledge retrieval dominates the Quick Win category.

Most of them involve finding, synthesizing, or delivering information that already exists somewhere in the organization.

The hardest part of internal AI agents is not the reasoning - it is the integration with fragmented knowledge sources behind SSO, document-level permissions, and inconsistent APIs.

Amazon Q Business absorbs a significant chunk of this complexity out of the box with native connectors and built-in ACLs, which is why it appears as the default for pure retrieval patterns.

Bedrock Knowledge Bases fills in when you need a custom RAG pipeline or when Q Business lacks a connector for your source.

Permission models are the real engineering challenge.

Customer-facing agents from Edition 1 mostly deal with one customer’s data at a time.

Internal agents cross organizational boundaries constantly.

An HR agent that can see compensation data, a finance agent that reads budget forecasts, an engineering agent that accesses production logs - each needs fine-grained access controls scoped to the requester’s role.

AgentCore Identity handles IdP integration for SSO. AgentCore Policy adds rule-based access scoping - verify maturity for your target region before production rollout.

For retrieval-only patterns, Q Business’s ACL engine is the more battle-tested option today.

RPA migrations have the clearest ROI.

Expense processing, access provisioning, procurement workflows - these agents replace brittle RPA scripts that break when a UI changes.

The agentic version handles exceptions, asks clarifying questions, and adapts to edge cases instead of failing silently.

Multi-agent architectures appear less often internally (see how we did not reference architecture F).

Internal users tolerate slightly longer response times and are better at framing specific questions, which means a single well-tooled agent handles most internal scenarios effectively.

Quick fills the analytics gap. Some patterns use Quick for dashboarding and self-service analysis.

Internal teams need visibility into operational data more than they need conversational agents.

QuickSight and Quick Research provide that without custom development.

Where the Leverage Actually Is

Most of the patterns in this edition run on a single agent with tool access. That’s not a limitation of the framework, it reflects how internal work actually breaks down. Employees ask specific questions, need specific actions, and want specific answers. The architectural complexity lives in the permission model and the integration layer, not in multi-agent orchestration.

The engineer from the opening spent 40 minutes finding a rate-limiting policy. Pattern #026 solves that with Q Business, native connectors, and document-level ACLs she never has to think about.

No custom orchestration.

No agent memory.

No specialist routing.

The right document, surfaced to someone authorized to see it, in seconds. Start there.

Add AgentCore when the workflow needs to take action, not just answer questions. Add Quick when teams need dashboards, not conversations.

Every pattern in this edition follows that same decision sequence: retrieval first, action second, analytics where the data justifies it.

What Comes Next

Three more editions:

Edition 3 - Workflow automation and process agents (internal operations, no direct user interaction)
Edition 4 - Data and analytics agents (self-service BI)
Edition 5 - Compliance, security, and governance agents (high-stakes environments)

If you are building internal productivity agents, start with #026 (Enterprise Knowledge Search) or #031 (IT Help Desk).

Enterprise Knowledge Search deploys fast on Q Business with minimal custom code.

IT Help Desk needs AgentCore for the action layer but has the clearest success metrics.

Both solve a pain point every employee recognizes on day one.

I publish every week at buildwithaws.substack.com. Subscribe. It's free.

Connecting 12 MCP Servers to Amazon Q CLI: What Broke and How I Fixed It

Sarvar Nadaf — Mon, 06 Apr 2026 13:02:02 +0000

👋 Hey there, tech enthusiasts!

I'm Sarvar, a Cloud Architect with a passion for transforming complex technological challenges into elegant solutions. With extensive experience spanning Cloud Operations (AWS & Azure), Data Operations, Analytics, DevOps, and Generative AI, I've had the privilege of architecting solutions for global enterprises that drive real business impact. Through this article series, I'm excited to share practical insights, best practices, and hands-on experiences from my journey in the tech world. Whether you're a seasoned professional or just starting out, I aim to break down complex concepts into digestible pieces that you can apply in your projects.

Let's dive in and explore the fascinating world of cloud technology together! 🚀

Written from experience building AI agent integrations for AWS infrastructure management. Your mileage may vary, but the principles hold across different use cases.

Do We Still Need MCP When We Have Agent Skills?

As a cloud architect working with AI agents, I've spent the last few months exploring how to extend their capabilities. Specifically, I've been working with Amazon Q Developer Pro - AWS's AI assistant that helps with coding, infrastructure management, and cloud operations through a chat interface.

This article shares what I learned about two different approaches to extending AI agents: Model Context Protocol (MCP) and Agent Skills. While the specific implementation details are still evolving, the architectural patterns I describe here represent where the ecosystem is heading based on current capabilities and community standards.

The Problem I Was Solving

I needed Amazon Q Developer Pro to help me manage AWS infrastructure across multiple accounts. The agent needed to:

Check CloudWatch metrics
Query RDS databases
Send alerts to Slack
Generate cost reports
Review CloudFormation templates

I tried two approaches, and each had serious problems.

Approach 1: Connecting Everything via MCP

MCP is a standard protocol that lets AI agents connect to external tools. Think of it like USB ports on your computer - one standard interface that works with many devices.

I connected 12 MCP servers to Amazon Q Developer Pro:

AWS CloudWatch server
RDS query server
Slack messaging server
Cost Explorer server
GitHub server
And seven more

What Went Wrong

Before I even asked a question, 35% of the context window was already full. Each MCP server loaded all its available functions into memory. The CloudWatch server alone exposed 15 different functions with detailed parameter descriptions.

When I asked Amazon Q Developer Pro to "check if our xyz database is healthy," it had to scan through multiple function definitions to figure out which ones to use. Sometimes it picked the wrong ones.

Every function call and response consumed more context. After three or four operations, I was running out of space for the actual conversation.

Approach 2: Using Agent Skills

Agent Skills are different. Instead of connecting to external tools, you give the agent domain knowledge - a guide on how to think about a problem.

I created a skill called "database-health-check" with a simple file structure:

database-health-check/
  SKILL.md          (when to use this, what steps to follow)
  scripts/
    check_rds.py    (Python script to query RDS)

The SKILL.md file contained:

# Database Health Check Skill

## Trigger
Keywords: "database health", "RDS status", "database performance", "DB issues"

## Process
1. Check CPU utilization (threshold: 80%)
2. Check active connections (threshold: 90% of max)
3. Check replication lag (threshold: 1000ms)
4. Check storage space (threshold: 85%)

## Decision Logic
- If CPU > 80%: WARN - "High CPU usage detected"
- If connections > 90%: CRITICAL - "Connection pool nearly exhausted"
- If replication lag > 1000ms: WARN - "Replication falling behind"
- If storage > 85%: CRITICAL - "Low storage space"

## Output Format
Status: [HEALTHY|WARN|CRITICAL]
Issues: [list of problems found]
Recommendations: [suggested actions]

What Went Wrong

The skill worked perfectly on my laptop. Then my colleague tried to use it and got an error: "ModuleNotFoundError: No module named 'boto3'".

The Python script needed boto3, pandas, and psycopg2. My machine had them installed. His didn't. We had no standard way to declare or install these dependencies.

The Real Difference

After working with both, I realized they solve different problems:

MCP answers: "What can I do?"
It provides capabilities - functions the agent can call. Each MCP server is self-contained with its own dependencies and environment.

Skills answer: "How should I think?"
They provide expertise - decision logic, quality standards, and workflows. But they run in whatever environment the agent has.

The Solution: Use Both Together

Here's what actually works in real time. I'll use a real example from last week.

Example: Automated Cost Anomaly Detection

I needed Amazon Q Developer Pro to monitor our AWS costs and alert us when something unusual happens.

The MCP Layer (The Hands)

I set up two MCP servers:

aws-cost-explorer-mcp - exposes functions like get_daily_costs(), get_service_breakdown()
slack-notifications-mcp - exposes send_message(), create_incident()

Each server runs in its own Docker container with all dependencies installed. Amazon Q Developer Pro doesn't need to know about boto3 or the Slack SDK.

The Skill Layer (The Brain)

I created a cost-monitoring skill with this logic in SKILL.md:

Trigger: When user asks about cost anomalies or unusual spending

Process:
1. Get last 30 days of daily costs
2. Calculate average and standard deviation
3. Check if today's cost is more than 2 standard deviations above average
4. If yes, get service breakdown to identify which service spiked
5. If spike is over $500, send high-priority Slack alert
6. If spike is under $500, just report in conversation

Quality checks:
- Always compare against same day of week (Monday vs Monday)
- Exclude known scheduled events (monthly backups, etc.)
- Include percentage change, not just absolute numbers

How They Work Together

When I ask Amazon Q Developer Pro: "Are our AWS costs normal today?"

Amazon Q Developer Pro matches the question to the cost-monitoring skill
The skill loads its SKILL.md (only 2KB of context)
The skill requires cost data, so Amazon Q Developer Pro connects to the aws-cost-explorer-mcp server
The skill orchestrates: call get_daily_costs(), analyze the data, decide if it's anomalous
If anomalous, the skill calls the slack-notifications-mcp server
After completion, the skill unloads from memory

The MCP servers handle the technical execution. The skill handles the business logic and decision-making.

Why This Architecture Works

Context Efficiency
Only the active skill loads into memory. MCP tools load on-demand when the skill needs them. I went from 35% context consumed at startup to 5%.

Portability
The same skill works on my laptop, my colleague's Windows machine, and our CI/CD pipeline. The MCP servers can run locally, in containers, or as remote services.

Reusability
The aws-cost-explorer-mcp server is used by three different skills:

cost-monitoring (detects anomalies)
budget-planning (forecasts spending)
cost-optimization (finds savings opportunities)

Each skill brings different expertise, but they share the same data source.

Maintainability
When AWS changes their Cost Explorer API, I update one MCP server. All skills continue working. When business rules change (new alert thresholds), I update the skill. The MCP servers don't need to change.

Visual Overview

Here's how the three layers work together:

┌─────────────────────────────────────────┐
│   Agent Layer (Amazon Q Developer Pro)  │
│   Matches tasks → Loads skills          │
│   Connects to MCP servers               │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│         Skill Layer (The Brain)         │
│   cost-monitoring.SKILL.md              │
│   - When to trigger                     │
│   - Decision logic                      │
│   - Quality checks                      │
└─────────────────────────────────────────┘
                    ↓
┌─────────────────────────────────────────┐
│         MCP Layer (The Hands)           │
│   aws-cost-explorer-mcp                 │
│   slack-notifications-mcp               │
│   - Tool execution                      │
│   - Environment isolation               │
└─────────────────────────────────────────┘

A Practical Pattern

Here's the decision framework I use:

Put it in an MCP server if:

Multiple skills will use it
It has heavy dependencies (Python libraries, system tools)
It needs credentials or secrets
It's a stable, reusable capability

Put it in a skill if:

It's domain knowledge or business logic
It orchestrates multiple tools
It has conditional decision-making
It's specific to one workflow

Keep it as a simple script if:

It's a one-time operation
The entire workflow is tightly coupled
Splitting it would add unnecessary complexity

What's Still Missing

The ecosystem isn't mature yet. Here's what I wish existed:

Dependency Declaration
Skills need a standard way to say "I need these capabilities" without hardcoding specific MCP servers. Something like:

requires:
  - cloud_metrics
  - notifications

Then the runtime figures out which MCP servers provide those capabilities.

Dynamic Loading
Right now, when you connect an MCP server, all its tools load immediately. I want skills to control this: "Load only the cost analysis tools for this task."

Graceful Fallback
If an MCP server is unavailable, the skill should automatically fall back to a built-in script or tell me clearly what's missing.

Conclusion

After six months of exploring these patterns with Amazon Q Developer Pro and other AI agents, here's what I know:

MCP and Agent Skills aren't competing approaches. They're complementary layers of the same system.

MCP gives your agent reliable, isolated capabilities. Skills give your agent the expertise to use those capabilities intelligently.

You need both. MCP without skills is a toolbox with no craftsman. Skills without MCP are expertise with unreliable tools.

The architecture that works:

Skills encode what to do and when
MCP servers provide how to do it
The agent runtime connects them together

This isn't just theoretical. These patterns are being implemented in actual environments, managing real AWS infrastructure.

Getting Started

If you want to explore this architecture:

For MCP:

Check the official MCP specification at modelcontextprotocol.io
Browse existing MCP servers at github.com/modelcontextprotocol/servers
Amazon Q Developer supports MCP through the Q CLI

For Agent Skills:

Review the Agent Skills specification by Anthropic
Start with simple skills that encode your team's operational knowledge
Focus on decision logic and workflows, not heavy computation

Next Steps:

Identify one repetitive task you do with your AI agent
Ask: Does this need external tools (MCP) or decision logic (Skill)?
Build the simplest version that works
Iterate based on what you learn

The ecosystem is still maturing, but the core patterns are solid. Start small, learn from erros, and build up from there.

📌 Wrapping Up

Thank you for reading! I hope this article gave you practical insights and a clearer perspective on the topic.

Was this helpful?

❤️ Like if it added value
🦄 Unicorn if you’re applying it today
💾 Save for your next optimization session
🔄 Share with your team

Follow me for more on:

AWS architecture patterns
FinOps automation
Multi-account strategies
AI-driven DevOps

💡 What’s Next

More deep dives coming soon on cloud operations, GenAI, Agentic-AI, DevOps, and data workflows follow for weekly insights.

🌐 Portfolio & Work

You can explore my full body of work, certifications, architecture projects, and technical articles here:

👉 Visit My Website

🛠️ Services I Offer

If you're looking for hands-on guidance or collaboration, I provide:

Cloud Architecture Consulting (AWS / Azure)
DevSecOps & Automation Design
FinOps Optimization Reviews
Technical Writing (Cloud, DevOps, GenAI)
Product & Architecture Reviews
Mentorship & 1:1 Technical Guidance

🤝 Let’s Connect

I’d love to hear your thoughts drop a comment or connect with me on LinkedIn.

For collaborations, consulting, or technical discussions, feel free to reach out directly at simplynadaf@gmail.com

Happy Learning 🚀

AWS Lambda PII Handling in Production: DynamoDB Field Encryption with KMS

Matia Rašetina — Mon, 06 Apr 2026 07:00:00 +0000

Handling Personally Identifiable Information (or PII for short) in AWS Lambda-backed systems is not difficult because AWS lacks security primitives. It is difficult because the default patterns in backend development encourage storing sensitive data in places where access control is only enforced at the infrastructure boundary — DynamoDB encryption at rest protects disks, but not application data, and once PII enters any of the database solutions inside the system, every principal with read access becomes a major threat.

This article demonstrates a production-tested pattern — encrypting sensitive data categorized as PII (user’s home address, email, name, surname, date of birth etc.) in AWS Lambda before they are stored in DynamoDB, using customer-managed KMS keys and infrastructure entirely defined in Python AWS CDK. The goal of this pattern is to ensure that if ever a data breach happens to your application, that the data is encrypted security inside DynamoDB, and that only chosen parts of the system can decrypt the data if necessary. This pattern can be adjusted to be used with any other database solution available on AWS.

This post will have most important snippets of the whole pattern, but the full code, where the Lambda handles all CRUD operations with DynamoDB, is available here.

Why are we going over this pattern?

There are many examples of existing serverless systems built on AWS Lambda and DynamoDB where some of the data, which gets stored inside the database, qualifies as PII and must not be stored in a human-readable format. As previously mentioned, the infrastructure in this pattern is defined using AWS CDK in Python, with a Lambda runtime in Python as well. Encryption keys are stored in AWS Key Management Service (KMS) and are customer-managed with explicit key policies, together with a strictly defined Lambda execution role.

This design does not attempt to solve encrypted search or complex querying over sensitive fields, as those problems require different tradeoffs. The scope here is to deliberately limit access to certain fields which are categorized as PII and could give you a legal headache in case of a data breach.

Moreover, DynamoDB server-side encryption is a must in any environment, but insufficient for PII or any field which needs protection by encryption. It protects against physical and low-level threats, but it does nothing to prevent plaintext access through IAM — any resource which has IAM access to the table and can scan or query the data, which can be a weakness in your infrastructure.

The only way to tip the scales in your favor here is to make sure that the sensitive fields are never stored in plaintext anywhere in your infrastructure. Once encryption happens in the application layer, any database solution becomes a persistence layer for ciphertext, and access to the data is no longer enough to expose PII. Data decryption process then becomes an explicit operation, which only authorized compute resources with correct IAM permissions can decrypt the data.

From this point forward, you are responsible for key rotation, correct IAM scoping and making sure that the application doesn’t leak any plain text information.

Architecture Overview

At a high level, the flow is simple:

User’s payload enter a Lambda function from API Gateway
Sensitive fields are encrypted immediately using AWS KMS
Encrypted data is written to DynamoDB

When data must be read and interpreted, the Lambda function explicitly decrypts the fields using the same KMS key from the encryption context.

Basically, there are 3 crucial operations:

Lambda calling KMS to encrypt and decrypt
Lambda reading and writing encrypted attributes in DynamoDB
IAM enforcing which principals are allowed to perform those operations

KMS Key Design in AWS CDK

In this pattern, the symmetric KMS key is the foundation. If its policy is overly permissive, field-level encryption becomes meaningless. The following CDK code is an example of creating a single customer-managed key dedicated for encryption and decryption processes for one service, with key rotation enabled and environment separation handled outside the snippet.

# Creation of a symmetric-key in KMS
pii_key = kms.Key(
    self,
    "PiiEncryptionKey",
    alias="pii-encryption-key",
    description="KMS key for encrypting PII data (latitude/longitude)",
    enable_key_rotation=True,
    removal_policy=RemovalPolicy.DESTROY,  # RETAIN for prod
)

At this layer, the key policy should be strict. Humans should not have decrypt permissions by default, and the Lambda execution role should be the primary and only consumer. CDK makes it easy to wire this correctly, but it also makes it easy to accidentally grant too much access if you are not careful enough.

Enabling Lambda IAM Access via Least Privilege

With the key in place, the next constraint is IAM. Encryption only works if decryption is tightly controlled. The Lambda execution role should have permission to encrypt and decrypt using exactly one KMS key and to read and write only the DynamoDB table it needs.

The following CDK code shows the Lambda function definition and the explicit grants for DynamoDB and KMS. There are no wildcards and no implicit permissions.

# CDK code for the Lambda + IAM Least Privilege Access
pii_handler = _lambda.Function(
    self,
    "PiiHandler",
    runtime=_lambda.Runtime.PYTHON_3_12,
    handler="handler.handler",
    code=_lambda.Code.from_asset(
        os.path.join(os.path.dirname(__file__), "PIIHandler"),
    ),
    layers=[powertools_layer],
    timeout=Duration.seconds(15),
    memory_size=256,
    environment={
        "POWERTOOLS_SERVICE_NAME": "pii-service",
        "POWERTOOLS_LOG_LEVEL": "INFO",
        "PII_TABLE_NAME": pii_table.table_name,
        "KMS_KEY_ARN": pii_key.key_arn,
        "ALLOWED_ORIGIN": "*",
    }
)

# Enable the PIIHandler Lambda access to DynamoDB table
# and the KMS key to encrypt/decrypt the data
pii_table.grant_read_write_data(pii_handler)
pii_key.grant_encrypt_decrypt(pii_handler)

How to encrypt and decrypt data inside the AWS Lambda?

All encryption and decryption logic lives in the Lambda function, close to where data enters and leaves the system. The key rule is that plaintext PII should exist only in memory and only for as long as necessary.

The following Python Lambda code uses the KMS client to directly to encrypt and decrypt fields which are sensitive.

import json
import os
import base64

from boto3 import client, resource

# Environment variables, AWS Client and AWS resources
TABLE_NAME = os.environ.get("PII_TABLE_NAME")
KMS_KEY_ARN = os.environ.get("KMS_KEY_ARN")

KMS_CLIENT = client('kms')
DDB_RESOURCE = resource('dynamodb')
TABLE = DDB_RESOURCE.Table(TABLE_NAME)

def encrypt_data(plaintext: str) -> str:
        """Encrypt data using KMS and return base64-encoded ciphertext."""
    response = KMS_CLIENT.encrypt(
        KeyId=KMS_KEY_ARN,
        Plaintext=plaintext.encode("utf-8"),
        EncryptionContext={
            "purpose": "pii-location-encryption",
            "service": "pii-service",
        },
    )
    return base64.b64encode(response["CiphertextBlob"]).decode("utf-8")

def decrypt_data(encrypted_data: str) -> str:
        """Decrypt base64-encoded ciphertext using KMS."""
    ciphertext = base64.b64decode(encrypted_data)
    response = KMS_CLIENT.decrypt(
        CiphertextBlob=ciphertext,
        KeyId=KMS_KEY_ARN,
        EncryptionContext={
            "purpose": "pii-location-encryption",
            "service": "pii-service",
        },
    )
    return response["Plaintext"].decode("utf-8")

Writing Encrypted Data to DynamoDB

Encryption happens before saving data anywhere in the system and only the PIIHandler Lambda has access to encrypt and decrypt the data, meaning that DynamoDB never sees plaintext latitude or longitude values, only base64-encoded cipher text.

The following code snippet shows you the usage of the encrypt_data method I’ve shown above and the process of saving the sensitive information inside the DynamoDB table:

def handle_create_location(event) -> dict:
        """Handle POST /locations - Store encrypted PII location data."""
    event_body = json.loads(event.get('body'))

    user_id = event_body["user_id"]
    latitude = event_body["latitude"]
    longitude = event_body["longitude"]

    # ..

    # Encrypt PII fields (latitude and longitude)
    coordinates = json.dumps({"latitude": latitude, "longitude": longitude})
    encrypted_coordinates = encrypt_data(coordinates)

    item = {
        "pk": f"USER#{user_id}",
        "sk": f"LOCATION#{timestamp}",
        "encrypted_coordinates": encrypted_coordinates
    }

    TABLE.put_item(Item=item)

    # ...

AWS KMS Operational Considerations and Pricing

Now, to talk about some of the considerations. This approach does introduce some latency and cost because of the KMS calls, and fetching encrypted data inside the DynamoDB will increase item size overall. However, these are normal and predictable tradeoffs, which are rarely a limiting factor unless you have a system which is already operating at a very high scale.

More importantly, this pattern makes PII access, or any access to sensitive fields for that matter, explicit. Every decrypt operation is visible in CloudTrail, and access can be revoked centrally by modifying the KMS key policy. That property is far more valuable in production than marginal performance gains.

Regarding pricing, when writing this in February 2026, the price for one KMS key is $1USD per month (prorated hourly) and operations cost only $0.03USD per 10 000 requests, more information about pricing can be found by clicking on the link here. It’s worth mentioning that KMS operations are free for 20 000 operations in every AWS region which has KMS.

This pattern has key rotation enabled, meaning that every 365 days a new key will be generated and used for encrypting new data, which increases your cost per month by $1 USD. You don’t have to worry about the older data encrypted with the old key — KMS will save your older key which gets replaced and it will still be used to decrypt your data.

Conclusion

Field-level encryption in Lambda isn't advanced security — it should be a baseline for any system handling PII information.

The pattern above is made simple on purpose, but the constraints it enforces are fundamental: DynamoDB read access no longer implies an insight into sensitive data and any compliance audits shift straight to concrete log analysis, making your job easier to do.

As we’ve already discussed, the tradeoffs are real but manageable — KMS client calls add tens of milliseconds per operation and are relatively cheap (with a very generous Free tier as well!), meaning that this overhead is totally worth it because of the risk mitigation it provides.

The real question in a production environment isn’t whether encryption exists, but whether which computing resources have access to your KMS key and which resources are authorized to decrypt the sensitive data. By using this pattern, your future self (and your legal team if you have it in your company!) will thank you when the inevitable security questionnaire arrives.

Deploying LibreChat on Amazon ECS using Terraform

Anthony Wat — Mon, 06 Apr 2026 01:56:39 +0000

Introduction

Generative AI has fundamentally shifted how we approach work, from writing and coding to research and problem-solving. For the past year, I used ChatGPT Business almost daily at work to improve my writing and research. However, I noticed limitations like fabrication and confirmation bias, so I wanted to explore how other non-OpenAI models perform. Additionally, my organization is consolidating on Microsoft 365 Copilot, which doesn't match ChatGPT's capabilities for my needs. This led me to search for a self-hosted, ChatGPT-like platform with flexibility in model choices.

I also needed it to be web-based for team members to access. As an AWS advocate, I wanted to leverage a diverse set of foundational models that Amazon Bedrock has to offer, and to host the platform using primarily AWS services. Based on my research, the three main options are LibreChat, Open WebUI, and AnythingLLM. Given that LibreChat is more feature-rich, customizable, and seemingly easier to deploy, I decided to give it a try and share my experience.

Without further ado, let's walk through the solution architecture and how it addresses my requirements.

Architecture Overview

The main design principle of the solution is to be cost-effective initially while allowing flexibility to scale in the future. Minimizing cost also means reducing operational overhead, not just service charges.

While cramming all components into an Amazon Lightsail instance is the cheapest option, it would need to be re-architected to scale horizontally. Deploying to an Amazon EC2 instance provides more flexibility, but it requires manual LibreChat installation and VM management. Ultimately, I decided to take a more modern approach and adopt a componentized architecture depicted in the following diagram:

This architecture uses the following technologies:

MongoDB Atlas - The LibreChat database runs on MongoDB Atlas using the Free (M0) cluster tier. It's technically free, runs in AWS, and is sufficient as a starter database engine as recommended by LibreChat.
Amazon ECS with AWS Fargate - The LibreChat application runs as a container with 512 (0.5) vCPU and 1 GB memory using 64-bit ARM architecture, which is sufficient when not enabling too many LibreChat features. Secrets are stored in AWS Systems Manager (SSM) Parameter Store (free), and configurations are stored in an Amazon S3 bucket to avoid additional shared storage services.
Application Load Balancer (ALB) - The public-facing endpoint. Although there is a running cost, its simpler TLS setup and support for scaling and AWS WAF integration makes it worthwhile.
fck-nat - Provides NAT gateway functionality using a pair of EC2 t4g.nano instances instead of AWS-managed NAT Gateways. This significantly reduces NAT gateway data transfer charges, making it a cost-effective option for modest traffic volumes.

The monthly cost of this architecture should be about $50 USD in us-east-1 including moderate Bedrock model use. To prevent surprise bills, set up a budget in AWS Budgets and create a cost monitor in AWS Cost Anomaly Detection.

Now that we have a good understanding of the architecture, let's go through the LibreChat concepts and prerequisites before we look at the Terraform configuration.

Deploying a MongoDB Atlas Database

Since an official Terraform MongoDB Atlas Provider is available, let's use it to provision the database for LibreChat. If you have not already done so, sign up for a new account using the Get Started button on the MongoDB website.

Once you've completed sign-up, log in to the MongoDB Atlas Console. MongoDB Atlas automatically creates an organization and a sample project named Project 0. Since we'll use Terraform to create a new project, feel free to delete this sample project. You may also edit the organization name in Organizational Settings as needed.

Next, create an API key at the organization level for the Terraform provider. In the MongoDB Atlas Console, go to the organization level view and select Identity & Access > Applications in the left menu. On the Application page, select the Service Accounts tab and click Create service account:

On the Create Service Account page, enter a name (for example, "terraform") and a description (for example, "Service account for Terraform"), keep the client secret expiration as recommended, select the Organization Project Creator permission, and click Create. Copy both the client ID and secret from the next page for use with Terraform:

We're now ready to write the Terraform configuration to:

Create a new project
Create a new Free (M0) cluster with AWS as the backing provider
Create a database user for LibreChat
Add the NAT Gateway's public IP to the project IP Access List for security

To avoid hardcoding credentials, set the service account credentials as environment variables before running Terraform using MONGODB_ATLAS_CLIENT_ID and MONGODB_ATLAS_CLIENT_SECRET as per the provider configuration.

The following Terraform configuration provisions these MongoDB Atlas resources. Note that some attributes are provided as variables for flexibility, and the IP access list CIDR block refers to the NAT service elastic IPs (since egress traffic from LibreChat container goes through the fck-nat instances or NAT gateways):

resource "mongodbatlas_project" "librechat" {
  org_id = var.atlas_org_id
  name   = var.atlas_project_name
}

resource "mongodbatlas_advanced_cluster" "librechat" {
  project_id   = mongodbatlas_project.librechat.id
  name         = var.atlas_cluster_name
  cluster_type = "REPLICASET"

  replication_specs = [
    {
      region_configs = [
        {
          electable_specs = {
            instance_size = "M0"
          }
          provider_name         = "TENANT"
          backing_provider_name = "AWS"
          region_name           = var.atlas_region_name
          priority              = 7
        }
      ]
    }
  ]
}

resource "mongodbatlas_project_ip_access_list" "librechat" {
  for_each   = var.use_fck_nat ? aws_eip.fck_nat : aws_nat_gateway.zonal
  project_id = mongodbatlas_project.librechat.id
  cidr_block = "${each.value.public_ip}/32"
  comment    = "ECS egress CIDR for LibreChat"
}

resource "random_password" "atlas_db" {
  length = 16
}

resource "mongodbatlas_database_user" "librechat" {
  project_id         = mongodbatlas_project.librechat.id
  username           = var.atlas_db_username
  password           = random_password.atlas_db.result
  auth_database_name = "admin"

  roles {
    role_name     = "readWriteAnyDatabase"
    database_name = "admin"
  }
}

Although using an IAM role to authenticate the Atlas database user would be more secure, it unfortunately doesn't work with the off-the-shelf LibreChat Docker image because it lacks the aws4 module required by the Mongoose library for AWS authentication. To avoid rebuilding a new container image, we'll stick with password authentication for now.

Strategies for Managing LibreChat Configuration

LibreChat uses two main sources of configuration: environment variables and LibreChat YAML. Environment variables are typically provided via a .env file created from the example in LibreChat's GitHub repository. Most LibreChat configuration is done using environment variables, and the LibreChat YAML file references these variables using the ${} notation for values such as API keys.

Additionally, the LibreChat YAML file (librechat.yaml) is typically placed in the application folder or mounted as an override in a containerized environment. However, it can also be provided in other locations by specifying the configuration path using the CONFIG_PATH environment variable.

Running LibreChat as a container introduces some challenges:

Building custom images is inefficient - While it's possible to build a LibreChat container image with .env and librechat.yaml baked in, rebuilding the image for every configuration change is inefficient. It's ideal to use the official LibreChat image from Docker Hub.
Managing many environment variables is difficult - Setting environment variables directly without using .env is hard to manage due to the sheer number of variables to configure, even when omitting those irrelevant to your use case.
Security concerns - Providing security-sensitive information as plain-text environment variables is not a security best practice.

ECS provides features to address these concerns:

Environment variable files - ECS supports passing environment variables via an environment variable file stored in S3. Since LibreChat already uses this format, it's a perfect fit.
Secrets management - AWS Secrets Manager or AWS Systems Manager (SSM) Parameter Store can securely pass sensitive data to containers as environment variables, avoiding hardcoded credentials.

To keep costs low, we will define all sensitive data as SSM Parameter Store parameters of type SecureString, while keeping all other configuration in a .env file provided to the container via ECS.

Providing the LibreChat YAML file to the container is more complicated. While the standard approach uses persistent storage options like Amazon EFS, that's cost-prohibitive for managing a single file. A better approach is using a sidecar container to write the file to a task-level shared volume before the main container starts.

For this, I implemented a sidecar container using busybox to decode a base64-encoded librechat.yaml (provided as an environment variable) and write it to /config/librechat.yaml in the task-level shared volume. The LibreChat container references this path using the CONFIG_PATH environment variable. The resulting container definitions (defined in the ECS task definition) are shown below:

  container_definitions = jsonencode([
    {
      name      = "init-librechat-config"
      image     = "public.ecr.aws/docker/library/busybox:1.36"
      essential = false

      command = [
        "sh",
        "-lc",
        "mkdir -p /config && printf '%s' \"$LIBRECHAT_YAML_B64\" | base64 -d > /config/librechat.yaml"
      ]

      environment = [
        {
          name  = "LIBRECHAT_YAML_B64"
          value = filebase64("${path.module}/librechat/${var.librechat_version}/librechat.yaml")
        }
      ]

      mountPoints = [
        {
          sourceVolume  = "librechat-config"
          containerPath = "/config"
          readOnly      = false
        }
      ]

      logConfiguration = {
        logDriver = "awslogs"
        options = {
          "awslogs-group"         = aws_cloudwatch_log_group.librechat.name
          "awslogs-region"        = var.region
          "awslogs-stream-prefix" = "librechat"
        }
      }
    },
    {
      name      = "librechat"
      image     = "librechat/librechat:v0.8.4"
      essential = true

      dependsOn = [
        {
          containerName = "init-librechat-config"
          condition     = "SUCCESS"
        }
      ]

      portMappings = [
        {
          containerPort = 3080
          protocol      = "tcp"
        }
      ]

      secrets = [
        {
          name      = "CREDS_KEY"
          valueFrom = aws_ssm_parameter.librechat_creds_key.arn
        },
        {
          name      = "CREDS_IV"
          valueFrom = aws_ssm_parameter.librechat_creds_iv.arn
        },
        {
          name      = "JWT_SECRET"
          valueFrom = aws_ssm_parameter.librechat_jwt_secret.arn
        },
        {
          name      = "JWT_REFRESH_SECRET"
          valueFrom = aws_ssm_parameter.librechat_jwt_refresh_secret.arn
        },
        {
          name      = "MEILI_MASTER_KEY"
          valueFrom = aws_ssm_parameter.librechat_meili_master_key.arn
        },
        {
          name      = "MONGO_URI"
          valueFrom = aws_ssm_parameter.librechat_mongo_uri.arn
        }
      ]

      environment = [
        {
          name  = "CONFIG_PATH"
          value = "/config/librechat.yaml"
        }
      ]

      environmentFiles = [
        {
          value = aws_s3_object.librechat_dot_env.arn
          type  = "s3"
        }
      ]

      mountPoints = [
        {
          sourceVolume  = "librechat-config"
          containerPath = "/config"
          readOnly      = true
        }
      ]

      logConfiguration = {
        logDriver = "awslogs"
        options = {
          "awslogs-group"         = aws_cloudwatch_log_group.librechat.name
          "awslogs-region"        = var.region
          "awslogs-stream-prefix" = "librechat"
        }
      }
    }
  ])

Now that we have a strategy for managing LibreChat configuration, let's define the minimal set of environment variables and LibreChat YAML as a starting point.

Preparing the Starter Environment File

LibreChat provides starter files .env.example and librechat.example.yaml that we'll use as the basis for our configuration. Let's start with the environment file.

Environment File (.env)

Copy the .env.example file to a local folder and comment out any sensitive values that will be provided as SSM Parameter Store environment variables specified below. Although individually defined environment variables take precedence over variables in environment files per the Amazon ECS Developer Guide, commenting them out avoids confusion.

In addition to MONGO_URI (the MongoDB connection string), LibreChat recommends adjusting any "secret" values from their default value for added security:

CREDS_IV - 16-byte Initialization Vector (IV) (32 characters in hex) for securely storing credentials
CREDS_KEY - 32-byte key (64 characters in hex) for securely storing credentials
JWT_SECRET - 32-byte key (64 characters in hex) as the JWT secret key
JWT_REFRESH_SECRET - 32-byte key (64 characters in hex) as the JWT refresh secret key
MEILI_MASTER_KEY - 16-byte key (32 characters in hex) as the MeiliSearch master key (required only if message and conversation search is enabled)

LibreChat provides a Credentials Generator to generate cryptographically secure random values for these secrets. Store them as SSM Parameter Store SecureString parameters or generate and store them directly using Terraform.

Next, adjust these environment variables for proper and secure operation:

HOST - Set to 0.0.0.0 to listen on all network interfaces, allowing ALB access
CONSOLE_JSON - Set to true to write logs to CloudWatch in JSON format for easier querying
ALLOW_REGISTRATION - Set to false to disable self-registration (we will create users with the create user script instead)
SEARCH - Set to false to disable message and conversation search (we're only demonstrating a minimal setup)

For the AI provider, we'll enable AWS Bedrock. Set ENDPOINTS to bedrock (the .env.example enables all pre-configured endpoints except Bedrock). Then configure the Bedrock endpoint by setting the following environment variables:

BEDROCK_AWS_DEFAULT_REGION - Set to your Bedrock region (e.g., us-east-1)
BEDROCK_AWS_MODELS - Set to us.amazon.nova-2-lite-v1:0 to use Amazon Nova Lite as a starting point (referring to the US Nova Lite system-defined inference profile)
OPENAI_API_KEY - Comment out to ensure that LibreChat does not try to make any calls to OpenAI APIs even if the endpoint is disabled via ENDPOINTS.
ANTHROPIC_API_KEY - Comment out to ensure that LibreChat does not try to make any calls to Anthropic APIs even if the endpoint is disabled via ENDPOINTS.
GOOGLE_KEY - Comment out to ensure that LibreChat does not try to make any calls to Google APIs even if the endpoint is disabled via ENDPOINTS.
ASSISTANTS_API_KEY - Comment out to ensure that LibreChat does not try to make any calls to OpenAI Assistants APIs even if the endpoint is disabled via ENDPOINTS.

Note that we'll be using the ECS task IAM role to allow LibreChat to seamlessly call Bedrock APIs, so we don't need to set BEDROCK_AWS_ACCESS_KEY_ID nor BEDROCK_AWS_SECRET_ACCESS_KEY.

LibreChat YAML File (librechat.yaml)

Copy the librechat.example.yaml file to a local folder. For this minimal setup, we won't need to define custom endpoints or configure advanced settings. However, to prevent custom endpoints like Groq and Mistral AI from appearing in the UI, comment out the custom key in the endpoints block and set it to an empty array [].

With all environment variables set in the environment file or SSM Parameter Store, and the librechat.yaml file prepared, we're ready to tie everything together with Terraform.

Building the Terraform Configuration

Since this blog post is getting quite long, let's focus on the key design elements of the Terraform configuration. You can find the complete Terraform configuration and source code in the 1_ecs_basic directory in this GitHub repository. Here are the descriptions for each Terraform configuration file:

atlas.tf - Defines the MongoDB Atlas resources, as explained in the earlier section of this blog post.
vpc.tf - Defines the VPC infrastructure for the solution. Key design elements include:
- The VPC design follows a three-tier architecture across two AZs. Although the database subnets are currently not in use, they can be utilized for AWS cache and database services in the future.
- There is built-in support for either fck-nat (default) or NAT Gateways, depending on your preference.
s3.tf - Defines the S3 bucket that hosts the LibreChat files and the S3 object for the .env file. Key design elements include:
- A ready-to-use .env file is included in the librechat/v0.8.4 folder. You can edit this file if using the same version, or upload a new .env file when you upgrade LibreChat (be sure to change the librechat_version variable).
- This S3 bucket may be used in the future as the LibreChat file storage backend, hence the .env file is placed in the config subfolder for better separation.
ecs.tf - Defines all ECS and related resources to run LibreChat as an ECS service. Key design elements include:
- A ready-to-use librechat.yaml file is included in the librechat/v0.8.4 folder. You can edit this file if using the same version, or upload a new librechat.yaml file when you upgrade LibreChat (be sure to change the librechat_version variable).
- The LibreChat credentials are generated by first creating a random password using the random_password ephemeral resource, then defining an aws_ssm_parameter resource with the write-only value storing the hash of the password (SHA1 or SHA256). You can replace this logic if you prefer to manually store the generated credentials in SSM Parameter Store first.
- The IAM policy for the ECS task role, aws_iam_role_policy.ecs_task_librechat, contains permissions to invoke Bedrock models and manage marketplace subscription of third-party models.
- Service auto-scaling is defined for future use, but for now, it is scaled to 1 for cost control.
- The LibreChat container runs on TCP port 3080 by default.
- CloudWatch logs are streamed to the /ecs/librechat log group.
alb.tf - Defines the ALB resources as the public-facing endpoint for LibreChat. Key design elements include:
- A HTTPS listener is defined with HTTP redirect to ensure security. Consequently, you must import or create a TLS certificate in AWS Certificate Manager (ACM), then pass the certificate's ARN using the alb_certificate_arn variable.
- HTTPS also requires a custom host name, so you must pass the DNS name using the librechat_dns_name variable.
- The target group checks the ECS task health using LibreChat's /health endpoint.

Deploying the Solution

After cloning the GitHub repository, deploy the solution as follows:

From the root of the cloned GitHub repository, navigate to 1_ecs_basic.
Set the MONGODB_ATLAS_CLIENT_ID and MONGODB_ATLAS_CLIENT_SECRET to the Atlas service account credentials created in the earlier section.
Configure your AWS credentials using aws configure for IAM, or aws configure sso for IAM Identity Center. The profile name will be provided as a Terraform variable.
Copy terraform.tfvars.example as terraform.tfvars and update the variables to match your configuration.
Run terraform init and terraform apply.

It is advised to check the ECS service status in the AWS Management Console to ensure that the tasks run successfully. Failed tasks will cause a perpetual restart due to the required capacity of 1, and overlooking this may lead to unexpected cost consequences. Check the task logs in the CloudWatch log group /ecs/librechat for errors if needed.

Once the configuration is applied, create the CNAME record for the ALB DNS name. If you manage your domain/subdomain using a public hosted zone in Amazon Route 53, you can also create an alias record pointing to the ALB. Lastly, go to the custom host name for LibreChat and ensure that the login page loads successfully.

Creating a User and Validating LibreChat

Since self-registration is disabled, we need to use the create user script to create the first user. The easiest way is to use ECS Exec in the ECS console to run the script in the librechat container in the task configuration. Here's a screenshot showing where to find the Connect button to open an interactive session:

Once CloudShell opens and connects to the container's shell, run the following command to start the user creation wizard:

npm run create-user

Enter the user's information as prompted, and a user will be created in LibreChat's database. Here's an example of creating a user for John Doe:

Now you're ready to log in. Open your LibreChat application URL and log in using the credentials you just created. Upon successful login, accept LibreChat's terms of service. You should see the Nova 2 Lite model already selected at the top, since it's the only configured model.

Let's test the setup with a simple prompt:

Tell me a joke about AWS.

If LibreChat responds with a joke, you've successfully completed the setup! Here's the joke I received, which honestly suggests that the Nova model could use some additional training with a better comedy dataset...

Summary

Congratulations, you now have your own LibreChat instance in AWS! You're now ready to start exploring and expanding its capabilities. While this setup gives you a functional chat interface, there's much more you can do to enhance its features. Here are some examples:

Configure more Bedrock models to unlock diverse capabilities and customization, balancing functionality, performance, and cost
Enable web search to allow LibreChat to search the internet and retrieve relevant information, enhancing conversations
Build custom AI assistants using AI Agents and integrate with various built-in and MCP tools to elevate capabilities and user experience

As your LibreChat usage grows, it's imperative to align your architecture with the AWS Well-Architected Framework. Here are some examples to improve security and operational robustness:

Streamline authentication by integrating with an Identity Provider (IDP)
Enable caching, session storage, and horizontal scaling in LibreChat using Redis and compute auto-scaling
Scale file storage with an Amazon S3 storage backend

Given the vast possibilities, I will start a new blog series about LibreChat and how to best use AWS services and best practices to enable these capabilities. If you enjoyed this post, stay tuned for new content at the Avangards Blog. Thanks so much for reading, and I hope you have fun chatting with LibreChat!

AWS Architecture Center: guía práctica para diseñar arquitecturas bien fundamentadas en AWS

Jose Luis Ariza — Mon, 06 Apr 2026 01:26:20 +0000

Introducción

¿Alguna vez te has preguntado cómo estructurar correctamente una arquitectura en AWS recurriendo solo a la experiencia previa sin fallar en el intento?

¿Dónde encontrar ejemplos reales, patrones validados y decisiones de diseño bien fundamentadas?

Cuando un arquitecto de soluciones comienza a trabajar en AWS, uno de los mayores retos no es aprender los servicios, sino entender cómo combinarlos correctamente para construir sistemas escalables, resilientes y eficientes.

Aquí es donde AWS proporciona una ventaja significativa:

👉 Se trata del AWS Architecture Center.

Y no, no se trata únicamente de una librería de iconos (que también nos brinda un paquete de íconos para descargar).

AWS Architecture Center: el punto de partida

El AWS Architecture Center es un portal oficial donde AWS centraliza:

arquitecturas de referencia

patrones de diseño

mejores prácticas

guías para tomar decisiones

No es solo inspiración visual.
Es una forma de pensar arquitectura con criterio.

🔗 Recurso

👉 https://aws.amazon.com/architecture/

🔁 Puntos importantes del portal oficial

Después de explorar el portal oficial, armé una estructura que me permitió entender mejor cómo está organizada toda la información y cómo aprovecharla en el diseño de soluciones.

AWS Architecture Center
│
├── AWS Reference Architecture Diagrams
├── AWS Prescriptive Guidance
│  → AWS Patrones de orientación prescriptiva
│  → Patrones, arquitecturas e implementaciones de diseño en la nube
└── AWS Well-Architected Framework

🏗️ AWS Reference Architecture Diagrams

Son un conjunto de arquitecturas de referencia diseñadas y validadas por AWS que representan soluciones reales a problemas comunes, utilizando servicios cloud de forma coherente y siguiendo buenas prácticas.

Cada arquitectura de referencia incorpora:

diagramas detallados (end-to-end)
descripción del caso de uso
servicios involucrados
flujos de interacción entre componentes
consideraciones de diseño

💡 Insight clave

Las Reference Architectures no son plantillas rígidas, sino puntos de partida validados que deben adaptarse al contexto de cada solución.

🔗 Recurso

👉 https://aws.amazon.com/architecture/reference-architecture-diagrams/

Les dejo un ejemplo de un diagrama que me gusto mucho como se integran los servicios y la explicación paso a paso: Diagrama de referencia AWS

🚀 AWS Prescriptive Guidance

La AWS Prescriptive Guidance es un conjunto de recursos desarrollados por expertos de AWS que proporcionan recomendaciones prácticas, patrones y guías detalladas para ayudar a diseñar, migrar y modernizar soluciones en la nube.

Se organiza en dos grandes bloques:

AWS Prescriptive Guidance
│
├── Guides (Guías)
└── Patterns (Patrones)

📘 Guides (Guías)

Las guías están orientadas a proporcionar una visión más amplia y estratégica.

Ofrecen orientación de planificación e implementación, centrada en mejores prácticas y herramientas, dirigida a arquitectos, gerentes, líderes técnicos y ejecutivos.

🔧 Patterns (Patrones)

Los patrones están enfocados en la implementación práctica.

Incluyen pasos, arquitecturas, herramientas y código para implementar escenarios comunes de migración, optimización y modernización, dirigidos a perfiles técnicos y constructores.

🔗 Recurso

👉 https://aws.amazon.com/es/prescriptive-guidance/

💡 Como aporte adicional, si quieres profundizar en cómo aplicar patrones en AWS, te recomiendo estos recursos oficiales:

Cloud Design Patterns (conceptos de arquitectura)

Prescriptive Guidance Patterns (implementación práctica)

⚖️ AWS Well-Architected Framework

El AWS Well-Architected Framework es el conjunto de principios, buenas prácticas y criterios definidos por AWS para evaluar y mejorar la calidad de una arquitectura en la nube.

A diferencia de otros recursos como patrones o arquitecturas de referencia, su objetivo no es decirte qué construir, sino ayudarte a responder una pregunta clave:

👉 ¿Está bien diseñada mi arquitectura?

El framework se basa en seis pilares que cubren los aspectos clave de cualquier sistema:

Excelencia Operativa: Capacidad para ejecutar y monitorear sistemas, y mejorar continuamente los procesos para entregar valor.

Seguridad: Protección de datos, sistemas y activos mediante la gestión de identidades y controles de detección de amenazas.

Fiabilidad: Garantía de que el sistema se recupere de fallos y funcione correctamente ante cambios o demandas de carga.

Eficiencia del Rendimiento: Uso eficiente de los recursos informáticos para cumplir con los requisitos y adaptarse a la evolución tecnológica.

Optimización de Costos: Capacidad para evitar gastos innecesarios y optimizar el uso de los recursos para maximizar el retorno de inversión.

Sostenibilidad: Enfoque en minimizar el impacto ambiental y la huella de carbono mediante la eficiencia energética en la nube.

🔗 Recurso

👉 https://aws.amazon.com/architecture/well-architected/

🎓 Aporte: formación en AWS Well-Architected (AWS Skill Builder)

💡 Como complemento a este contenido, AWS también ofrece formación oficial a través de AWS Skill Builder, donde puedes profundizar en el uso del Well-Architected Framework y aplicarlo en escenarios reales.

👉 AWS Well-Architected Foundations

Building a Document Processing Pipeline with S3, Textract, Step Functions and EventBridge

Renaldi — Sun, 05 Apr 2026 23:00:00 +0000

This is one of my favorite AWS patterns to demo because it is both visually compelling and production-relevant. It shows event-driven architecture, orchestration, asynchronous AI/ML service integration, scale-out processing, human-in-the-loop review, and operational discipline in one workflow.

In this post, I will walk through an end-to-end implementation of a document processing pipeline built with:

Amazon S3 for document ingress and result storage
Amazon Textract for OCR and structured extraction
AWS Step Functions for orchestration (including Distributed Map for batch scale)
Amazon EventBridge for event routing and downstream integration

I will also cover:

Async Textract orchestration
Batch scaling with Distributed Map
Result storage and audit trail
Human review step
Cost and throughput tuning
Architecture and code walkthrough

Why this pattern is so effective

In real teams, document processing is rarely just “OCR a file and store JSON.” We usually need to handle:

Multi-page PDFs and asynchronous processing
Bursty uploads (for example, end-of-day batch drops)
Traceability for compliance and audits
Human review for low-confidence or ambiguous fields
Clean integration points for downstream systems

This architecture solves those concerns in a way that is easy to demonstrate and scale.

What we are building

I am designing the pipeline around two operating modes:

Single-document event-driven processing
- A file lands in S3
- EventBridge triggers the workflow
- The workflow runs end-to-end for that document
Batch processing mode
- A manifest (JSON/CSV/list of document keys) is provided
- Step Functions uses Distributed Map to fan out child workflows
- Each child workflow processes one document independently

This keeps the core processing logic consistent while letting me scale from demos to production.

Architecture Overview

At a high level, the flow is:

A document is uploaded to an S3 input bucket
S3 emits an event to EventBridge
EventBridge starts a Step Functions parent workflow
The parent workflow prepares a manifest (single item or batch list)
A Distributed Map launches child workflows for each document
Each child workflow:
- validates input
- starts async Textract
- waits/polls for completion
- retrieves and normalizes results
- stores raw + normalized outputs
- writes audit records
- routes low-confidence cases to human review
- emits a processed event to EventBridge
Downstream systems consume the processed event

End-to-End Walkthrough

1) Document ingress with S3 + EventBridge

I use S3 as the system of record for inbound documents. The upload path usually looks like:

s3://my-docs-incoming/raw/YYYY/MM/DD/<tenant>/<document>.pdf

I prefer a structured key naming convention because it helps with:

lifecycle policies
tenant scoping
troubleshooting
cost attribution
replay operations

S3 can publish object events to EventBridge, and I use an EventBridge rule to filter only the prefixes and file types that should trigger processing (for example, PDFs in raw/).

Example EventBridge rule pattern (S3 object created)

{
  "source": ["aws.s3"],
  "detail-type": ["Object Created"],
  "detail": {
    "bucket": {
      "name": ["my-docs-incoming"]
    },
    "object": {
      "key": [{ "prefix": "raw/" }]
    }
  }
}

From there, EventBridge starts the Step Functions parent state machine.

2) Parent orchestration with Step Functions

I use a Standard workflow for the parent because:

document jobs can run for a while
I may have retries and waits
I may pause for human review (callback pattern)
I want richer execution history

The parent workflow does a few things:

Parses the incoming event (or batch request)
Creates a manifest for processing (even if only one document)
Sets concurrency controls (important for Textract/Lambda quotas)
Launches a Distributed Map

Why Distributed Map matters

For real batches, a normal Map state can become limiting because of concurrency and history size. Distributed Map gives me child workflow executions, better scaling, and cleaner observability for each document.

3) Child workflow per document (core pipeline)

Each child workflow processes a single document from start to finish.

Core steps in the child workflow

Validate input
- file type
- bucket/key exists
- metadata (tenant, doc type, correlation ID)
- idempotency key check
Start Textract asynchronously
- call StartDocumentAnalysis
- capture JobId
- store correlation in audit trail
Wait for completion
- polling loop (simple and explicit for Step Functions)
- or callback/SNS-based completion (production optimization)
Retrieve paginated results
- call GetDocumentAnalysis until NextToken is exhausted
Normalize + enrich
- convert Textract blocks into domain JSON
- score confidence thresholds
- apply business rules
Store outputs
- raw output
- normalized output
- processing summary
- audit entries
Human review (if needed)
- pause with task token callback
- reviewer approves/edits/rejects
Emit completion event
- publish a custom EventBridge event
- downstream systems subscribe independently

4) Async Textract orchestration (practical implementation)

Textract async processing is the correct choice for multi-page documents and larger workloads. The key point is that I do not try to synchronously block a Lambda while waiting for Textract.

Instead, I let Step Functions own the waiting and retry logic.

Two patterns I use in practice

Pattern A (implemented in this post): Step Functions polling loop

This is the simplest pattern to explain and demo.

StartDocumentAnalysis
Wait
GetDocumentAnalysis (status check)
loop until SUCCEEDED / FAILED

Pros:

Easy to understand
No extra callback plumbing
Great for demos and many production cases

Cons:

More state transitions
Not the most efficient if jobs are very long

Pattern B (production optimization): SNS/SQS + callback

Textract async operations notify completion via SNS. I can correlate JobId, then call SendTaskSuccess to resume the Step Functions execution.

Pros:

Fewer polling transitions
More event-driven

Cons:

More moving parts (SNS/SQS/Lambda/token correlation)

For this article, I am implementing Pattern A for clarity, and I will still preserve auditability and scalability.

5) Result storage and audit trail (what I store and why)

This is where many demos stop too early. I do not want a pipeline that only prints extraction results to logs.

I store four classes of artifacts:

A. Raw input (immutable)

Original document in S3 (raw/)

B. Raw Textract result snapshot

Either Textract output JSON (if using OutputConfig)
Or consolidated raw blocks written by my retrieval Lambda

C. Normalized business result

A clean JSON model that downstream systems can actually use, for example:

{
  "documentId": "INV-2026-000123",
  "documentType": "invoice",
  "vendorName": "Acme Pty Ltd",
  "invoiceDate": "2026-02-25",
  "totalAmount": 1285.40,
  "currency": "AUD",
  "fields": [
    {
      "name": "invoice_number",
      "value": "INV-2026-000123",
      "confidence": 98.7,
      "source": "textract"
    }
  ],
  "review": {
    "required": false,
    "reason": null
  }
}

D. Audit trail (DynamoDB + events)

I keep an audit table with entries like:

correlation ID
S3 bucket/key/version
execution ARN
Textract JobId
timestamps (started/completed)
status transitions
retry count
reviewer ID + decision (if reviewed)

This makes troubleshooting and compliance conversations much easier.

6) Human review step (callback pattern)

I like to add a human review step because it turns the demo into a real workflow.

When I trigger human review

I send documents for human review if:

confidence is below threshold for required fields
required fields are missing
document type classification is ambiguous
business validations fail (for example, total does not match line items)

How the callback works

Step Functions reaches HumanReviewRequired
It invokes a Lambda using waitForTaskToken
That Lambda stores a review task (with the task token) in DynamoDB and optionally emits a notification event
A reviewer UI or ops tool loads the pending task
Reviewer approves/edits/rejects
API Gateway + Lambda calls:
- SendTaskSuccess (approve / approved edits)
- SendTaskFailure (reject / unrecoverable issue)

This lets me keep the workflow paused cleanly without polling the UI.

7) EventBridge for decoupled downstream integration

I use EventBridge in two places:

Ingress routing

S3 object-created events trigger the parent workflow

Egress publishing

The child workflow publishes custom events like:
- document.processed
- document.review.required
- document.failed

This is a huge win because downstream systems can subscribe independently:

search indexing
analytics pipelines
case management
notifications
data quality dashboards

No tight coupling to the core document processing workflow.

Implementation Discussion

Below is a concrete implementation outline with code.

State machine design (parent + child)

I prefer splitting the logic into:

Parent workflow: batching and fan-out
Child workflow: per-document processing

This keeps each workflow easier to reason about and test.

Parent workflow (ASL fragment with Distributed Map)

{
  "Comment": "Parent workflow for document batch orchestration",
  "StartAt": "BuildManifest",
  "States": {
    "BuildManifest": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${BuildManifestFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "Next": "ProcessDocumentsDistributed"
    },
    "ProcessDocumentsDistributed": {
      "Type": "Map",
      "Label": "ProcessDocuments",
      "ItemsPath": "$.documents",
      "MaxConcurrencyPath": "$.maxConcurrency",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "STANDARD"
        },
        "StartAt": "StartChildExecution",
        "States": {
          "StartChildExecution": {
            "Type": "Task",
            "Resource": "arn:aws:states:::states:startExecution.sync:2",
            "Arguments": {
              "StateMachineArn": "${ChildStateMachineArn}",
              "Input": {
                "document.$": "$$.Map.Item.Value",
                "batchContext.$": "$.batchContext"
              }
            },
            "End": true
          }
        }
      },
      "Next": "BatchSummary"
    },
    "BatchSummary": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${BatchSummaryFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "End": true
    }
  }
}

Notes on this design

MaxConcurrencyPath lets me tune concurrency dynamically per batch.
I can set lower concurrency for production if Textract quotas are tighter than my desired fan-out.
Child executions give me better isolation and clearer failure analysis.

Child workflow (ASL fragment for async Textract + human review + EventBridge)

{
  "Comment": "Child workflow for processing one document",
  "StartAt": "ValidateDocument",
  "States": {
    "ValidateDocument": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${ValidateDocumentFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "Next": "StartTextract"
    },
    "StartTextract": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:textract:startDocumentAnalysis",
      "Arguments": {
        "DocumentLocation": {
          "S3Object": {
            "Bucket.$": "$.document.bucket",
            "Name.$": "$.document.key"
          }
        },
        "FeatureTypes": ["FORMS", "TABLES"],
        "ClientRequestToken.$": "$.idempotencyKey",
        "JobTag.$": "$.jobTag"
      },
      "ResultPath": "$.textractStart",
      "Next": "WaitBeforeStatusCheck",
      "Retry": [
        {
          "ErrorEquals": [
            "Textract.ThrottlingException",
            "Textract.ProvisionedThroughputExceededException",
            "States.TaskFailed"
          ],
          "IntervalSeconds": 2,
          "BackoffRate": 2.0,
          "MaxAttempts": 5
        }
      ]
    },
    "WaitBeforeStatusCheck": {
      "Type": "Wait",
      "Seconds": 10,
      "Next": "CheckTextractStatus"
    },
    "CheckTextractStatus": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:textract:getDocumentAnalysis",
      "Arguments": {
        "JobId.$": "$.textractStart.JobId",
        "MaxResults": 1
      },
      "ResultPath": "$.textractStatus",
      "Next": "TextractComplete?"
    },
    "TextractComplete?": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.textractStatus.JobStatus",
          "StringEquals": "SUCCEEDED",
          "Next": "CollectTextractPages"
        },
        {
          "Variable": "$.textractStatus.JobStatus",
          "StringEquals": "FAILED",
          "Next": "MarkFailed"
        },
        {
          "Variable": "$.textractStatus.JobStatus",
          "StringEquals": "PARTIAL_SUCCESS",
          "Next": "CollectTextractPages"
        }
      ],
      "Default": "WaitBeforeStatusCheck"
    },
    "CollectTextractPages": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${CollectTextractPagesFnArn}",
        "Payload": {
          "jobId.$": "$.textractStart.JobId",
          "document.$": "$.document",
          "executionArn.$": "$$.Execution.Id"
        }
      },
      "OutputPath": "$.Payload",
      "Next": "NormalizeAndScore"
    },
    "NormalizeAndScore": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${NormalizeAndScoreFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "Next": "NeedsHumanReview?"
    },
    "NeedsHumanReview?": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.review.required",
          "BooleanEquals": true,
          "Next": "CreateHumanReviewTask"
        }
      ],
      "Default": "PublishProcessedEvent"
    },
    "CreateHumanReviewTask": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Arguments": {
        "FunctionName": "${CreateHumanReviewTaskFnArn}",
        "Payload": {
          "taskToken.$": "$$.Task.Token",
          "document.$": "$.document",
          "result.$": "$",
          "executionArn.$": "$$.Execution.Id"
        }
      },
      "TimeoutSeconds": 604800,
      "ResultPath": "$.humanReviewDecision",
      "Next": "ApplyHumanReviewDecision"
    },
    "ApplyHumanReviewDecision": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${ApplyHumanReviewDecisionFnArn}",
        "Payload": {
          "pipelineResult.$": "$",
          "reviewDecision.$": "$.humanReviewDecision.Payload"
        }
      },
      "OutputPath": "$.Payload",
      "Next": "PublishProcessedEvent"
    },
    "PublishProcessedEvent": {
      "Type": "Task",
      "Resource": "arn:aws:states:::events:putEvents",
      "Arguments": {
        "Entries": [
          {
            "Source": "com.example.documents",
            "DetailType": "document.processed",
            "EventBusName": "${EventBusName}",
            "Detail": {
              "documentId.$": "$.document.id",
              "bucket.$": "$.document.bucket",
              "key.$": "$.document.key",
              "status": "PROCESSED",
              "review.$": "$.review",
              "output.$": "$.output"
            }
          }
        ]
      },
      "Next": "FinalizeAudit"
    },
    "FinalizeAudit": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${FinalizeAuditFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "End": true
    },
    "MarkFailed": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Arguments": {
        "FunctionName": "${MarkFailedFnArn}",
        "Payload.$": "$"
      },
      "OutputPath": "$.Payload",
      "End": true
    }
  }
}

Lambda: Collect paginated Textract results and store raw output (Python)

This Lambda consolidates paginated GetDocumentAnalysis responses, writes a raw artifact to S3, and returns a pointer for downstream normalization.

import json
import os
from datetime import datetime, timezone

import boto3

textract = boto3.client("textract")
s3 = boto3.client("s3")

RAW_RESULTS_BUCKET = os.environ["RAW_RESULTS_BUCKET"]
RAW_RESULTS_PREFIX = os.environ.get("RAW_RESULTS_PREFIX", "textract-raw")


def lambda_handler(event, context):
    job_id = event["jobId"]
    document = event["document"]
    execution_arn = event["executionArn"]

    blocks = []
    next_token = None
    first_page_metadata = None
    final_status = None
    warnings = []

    while True:
        kwargs = {"JobId": job_id, "MaxResults": 1000}
        if next_token:
            kwargs["NextToken"] = next_token

        resp = textract.get_document_analysis(**kwargs)
        final_status = resp.get("JobStatus", final_status)

        if first_page_metadata is None:
            first_page_metadata = {
                "DocumentMetadata": resp.get("DocumentMetadata", {}),
                "AnalyzeDocumentModelVersion": resp.get("AnalyzeDocumentModelVersion"),
            }

        blocks.extend(resp.get("Blocks", []))
        warnings.extend(resp.get("Warnings", []))

        next_token = resp.get("NextToken")
        if not next_token:
            break

    now = datetime.now(timezone.utc).isoformat()
    raw_key = (
        f"{RAW_RESULTS_PREFIX}/"
        f"{document.get('tenant', 'unknown')}/"
        f"{document['id']}/"
        f"{job_id}.json"
    )

    payload = {
        "jobId": job_id,
        "document": document,
        "executionArn": execution_arn,
        "collectedAt": now,
        "jobStatus": final_status,
        "metadata": first_page_metadata,
        "warnings": warnings,
        "blocks": blocks,
    }

    s3.put_object(
        Bucket=RAW_RESULTS_BUCKET,
        Key=raw_key,
        Body=json.dumps(payload).encode("utf-8"),
        ContentType="application/json",
    )

    return {
        "document": document,
        "textract": {
            "jobId": job_id,
            "jobStatus": final_status,
            "rawResultsS3": {
                "bucket": RAW_RESULTS_BUCKET,
                "key": raw_key,
            },
            "blockCount": len(blocks),
            "warningsCount": len(warnings),
        },
    }

Lambda: Normalize and score extracted data (Python)

This is where I transform Textract’s generic block model into a domain result that downstream systems can consume.

import json
import os
from decimal import Decimal

import boto3

s3 = boto3.client("s3")

NORMALIZED_BUCKET = os.environ["NORMALIZED_BUCKET"]
NORMALIZED_PREFIX = os.environ.get("NORMALIZED_PREFIX", "normalized")


def _load_json_from_s3(bucket: str, key: str) -> dict:
    obj = s3.get_object(Bucket=bucket, Key=key)
    return json.loads(obj["Body"].read())


def _extract_lines_and_words(blocks):
    # Minimal example; in production I usually build maps by block Id and relationships.
    lines = []
    for b in blocks:
        if b.get("BlockType") == "LINE":
            lines.append({
                "text": b.get("Text", ""),
                "confidence": b.get("Confidence", 0.0)
            })
    return lines


def _simple_invoice_heuristics(lines):
    # Demo heuristics only. Replace with rule engine / ML classifier as needed.
    text_blob = "\n".join([l["text"] for l in lines])
    avg_conf = sum(l["confidence"] for l in lines) / max(len(lines), 1)

    result = {
        "documentType": "unknown",
        "fields": [],
        "review": {
            "required": False,
            "reason": None
        }
    }

    if "invoice" in text_blob.lower():
        result["documentType"] = "invoice"

    # Example field extraction heuristic placeholder
    # In production, I'd parse key-value pairs and tables from FORM/TABLE blocks
    result["fields"].append({
        "name": "avg_line_confidence",
        "value": round(avg_conf, 2),
        "confidence": round(avg_conf, 2),
        "source": "textract"
    })

    # Review policy example
    if avg_conf < 90:
        result["review"]["required"] = True
        result["review"]["reason"] = "Average OCR confidence below threshold"

    return result


def lambda_handler(event, context):
    document = event["document"]
    textract_ptr = event["textract"]["rawResultsS3"]

    raw = _load_json_from_s3(textract_ptr["bucket"], textract_ptr["key"])
    blocks = raw["blocks"]

    lines = _extract_lines_and_words(blocks)
    derived = _simple_invoice_heuristics(lines)

    normalized = {
        "document": document,
        "textract": {
            "jobId": raw["jobId"],
            "jobStatus": raw["jobStatus"],
            "warnings": raw.get("warnings", []),
            "blockCount": len(blocks)
        },
        "classification": {
            "documentType": derived["documentType"]
        },
        "fields": derived["fields"],
        "review": derived["review"],
        "output": {
            "rawResultsS3": textract_ptr
        }
    }

    out_key = f"{NORMALIZED_PREFIX}/{document.get('tenant', 'unknown')}/{document['id']}.json"
    s3.put_object(
        Bucket=NORMALIZED_BUCKET,
        Key=out_key,
        Body=json.dumps(normalized).encode("utf-8"),
        ContentType="application/json"
    )

    normalized["output"]["normalizedS3"] = {
        "bucket": NORMALIZED_BUCKET,
        "key": out_key
    }

    return normalized

Lambda: Create human review task (callback token storage) (Python)

This Lambda stores the task token so a reviewer can resume the workflow later.

import json
import os
import time
from datetime import datetime, timezone

import boto3

ddb = boto3.resource("dynamodb")
events = boto3.client("events")

REVIEW_TABLE = ddb.Table(os.environ["REVIEW_TABLE"])
EVENT_BUS_NAME = os.environ["EVENT_BUS_NAME"]


def lambda_handler(event, context):
    task_token = event["taskToken"]
    document = event["document"]
    result = event["result"]
    execution_arn = event["executionArn"]

    review_id = f"{document['id']}#{int(time.time())}"
    now = datetime.now(timezone.utc).isoformat()

    item = {
        "reviewId": review_id,
        "status": "PENDING",
        "createdAt": now,
        "documentId": document["id"],
        "executionArn": execution_arn,
        "taskToken": task_token,
        "document": document,
        "resultSummary": {
            "reason": result["review"]["reason"],
            "documentType": result.get("classification", {}).get("documentType", "unknown"),
            "normalizedOutput": result["output"].get("normalizedS3")
        }
    }

    REVIEW_TABLE.put_item(Item=item)

    events.put_events(
        Entries=[
            {
                "Source": "com.example.documents",
                "DetailType": "document.review.required",
                "EventBusName": EVENT_BUS_NAME,
                "Detail": json.dumps({
                    "reviewId": review_id,
                    "documentId": document["id"],
                    "reason": result["review"]["reason"]
                })
            }
        ]
    )

    # For waitForTaskToken, Lambda can return immediately after storing task metadata.
    # The workflow stays paused until SendTaskSuccess/SendTaskFailure is called.
    return {
        "reviewId": review_id,
        "status": "PENDING"
    }

Lambda: Reviewer callback endpoint (API Gateway -> Lambda -> Step Functions) (Python)

This Lambda receives the human decision and resumes the waiting execution.

import json
import os
from datetime import datetime, timezone

import boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
sfn = boto3.client("stepfunctions")

REVIEW_TABLE = ddb.Table(os.environ["REVIEW_TABLE"])


def _response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body)
    }


def lambda_handler(event, context):
    try:
        body = json.loads(event.get("body") or "{}")
        review_id = body["reviewId"]
        action = body["action"]  # approve | reject
        reviewer = body.get("reviewer", "unknown")
        corrected_fields = body.get("correctedFields", {})

        item_resp = REVIEW_TABLE.get_item(Key={"reviewId": review_id})
        item = item_resp.get("Item")
        if not item:
            return _response(404, {"message": "Review task not found"})

        if item["status"] != "PENDING":
            return _response(409, {"message": f"Review task already {item['status']}"})

        task_token = item["taskToken"]
        now = datetime.now(timezone.utc).isoformat()

        if action == "approve":
            sfn.send_task_success(
                taskToken=task_token,
                output=json.dumps({
                    "reviewId": review_id,
                    "decision": "APPROVED",
                    "reviewer": reviewer,
                    "reviewedAt": now,
                    "correctedFields": corrected_fields
                })
            )
            new_status = "APPROVED"
        elif action == "reject":
            sfn.send_task_failure(
                taskToken=task_token,
                error="HumanReviewRejected",
                cause=json.dumps({
                    "reviewId": review_id,
                    "reviewer": reviewer,
                    "reviewedAt": now
                })
            )
            new_status = "REJECTED"
        else:
            return _response(400, {"message": "Invalid action"})

        REVIEW_TABLE.update_item(
            Key={"reviewId": review_id},
            UpdateExpression="SET #s = :s, reviewer = :r, reviewedAt = :t",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={
                ":s": new_status,
                ":r": reviewer,
                ":t": now
            }
        )

        return _response(200, {"message": "Review decision recorded", "status": new_status})

    except KeyError as e:
        return _response(400, {"message": f"Missing field: {e.args[0]}"})
    except ClientError as e:
        return _response(500, {"message": "AWS error", "detail": str(e)})
    except Exception as e:
        return _response(500, {"message": "Unexpected error", "detail": str(e)})

CDK example (Python) for S3 -> EventBridge -> Step Functions start

This is a simplified snippet to show the core wiring. In a production stack, I would break this into constructs (ingress, orchestration, review, storage, observability).

from aws_cdk import (
    Stack,
    aws_s3 as s3,
    aws_events as events,
    aws_events_targets as targets,
    aws_stepfunctions as sfn,
    aws_iam as iam,
)
from constructs import Construct


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

        input_bucket = s3.Bucket(
            self,
            "InputBucket",
            event_bridge_enabled=True
        )

        parent_state_machine = sfn.StateMachine.from_state_machine_arn(
            self,
            "ParentStateMachine",
            "arn:aws:states:ap-southeast-2:123456789012:stateMachine:doc-parent"
        )

        rule = events.Rule(
            self,
            "S3ObjectCreatedRule",
            event_pattern=events.EventPattern(
                source=["aws.s3"],
                detail_type=["Object Created"],
                detail={
                    "bucket": {"name": [input_bucket.bucket_name]},
                    "object": {"key": [{"prefix": "raw/"}]}
                }
            )
        )

        # Start execution on matching events
        rule.add_target(targets.SfnStateMachine(parent_state_machine))

In a real deployment, I also add:

explicit execution input transformation

IAM least privilege policies

DLQs / retry policies on targets where appropriate

environment-based prefix filtering (dev/test/prod)

Data model and audit design (recommended)

I strongly recommend treating auditability as a first-class feature.

DynamoDB audit table (example keys)

PK: DOC#<documentId>

SK: EVENT#<timestamp>#<eventType>

Example audit events

INGESTED
WORKFLOW_STARTED
TEXTRACT_STARTED
TEXTRACT_SUCCEEDED
NORMALIZED
HUMAN_REVIEW_REQUIRED
HUMAN_REVIEW_APPROVED
PUBLISHED
FAILED

This pattern gives me an append-only lineage timeline per document.

Error handling and retries (what I do on purpose)

This pipeline is asynchronous and distributed, so failures are normal. I design for them.

Where I retry

Textract throttling / throughput exceptions
transient Lambda errors
EventBridge PutEvents partial failures (Step Functions can fail the task and retry)
downstream S3 write failures

Where I do not blindly retry

invalid file format
missing required metadata
corrupted documents
deterministic business-rule violations

Failure routing

I emit a document.failed event and write a terminal audit entry with:

failure type
step name
error code
correlation IDs

That makes operations and replay much easier.

Cost and Throughput Tuning

This is where the architecture becomes production-grade.

1) Choose Step Functions workflow type intentionally

I use Standard for the main orchestration because:

async waits
human callback pauses
richer history and control

If I want to optimize cost for very high-volume short preprocessing tasks, I may split out a lightweight Express workflow in front, but I keep the long-running document path in Standard.

2) Tune Distributed Map concurrency to downstream capacity

Distributed Map can scale very high, but the bottleneck is usually downstream service quotas (Textract, Lambda concurrency, DynamoDB write capacity, etc.).

What I do:

make concurrency configurable (MaxConcurrencyPath)
start conservatively in production
load test with realistic document sizes and page counts
request quota increases before large launches

3) Reduce unnecessary state transitions

Polling is simple, but excessive polling increases state transitions and cost.

Practical tuning:

increase wait interval after the first few polls (progressive backoff)
use doc-size-aware polling intervals (longer waits for large PDFs)
consider SNS/SQS callback for long-running jobs

4) Store outputs once, reuse many times

I avoid re-running Textract when:

a replay is only needed for normalization logic changes
a downstream consumer fails
a human review needs to re-open a prior result

By storing raw output + normalized output + audit events, I can replay downstream steps without paying for re-extraction.

5) Separate hot path and review path

I do not force all documents through human review.
Most documents should:

extract
validate
publish
finish

Only low-confidence or exception cases enter the review path. This keeps cost and latency predictable.

6) Add lifecycle policies

For S3 cost management, I usually apply different retention windows:

raw uploads: longer retention (compliance dependent)
raw Textract artifacts: medium retention
normalized results: longer (business value)
debug-only artifacts: shorter

Security, Compliance, and Operational Notes

IAM least privilege

Give each component only what it needs:

Step Functions can call specific Lambdas / Textract / EventBridge
Lambdas can read/write only the intended buckets/prefixes
Reviewer API can only access review table + Step Functions callback APIs

Encryption

I enable encryption at rest for:

S3 buckets
DynamoDB tables
logs (when applicable)

If needed, I also use KMS for tighter key control and auditability.

PII considerations

Document pipelines often contain sensitive data. I plan for:

access controls by tenant
minimal logging (no full document contents in logs)
review UI redaction where needed
retention policies aligned to policy/legal requirements

Observability

I instrument:

Step Functions execution failure alarms
Lambda error/throttle alarms
custom metrics (documents processed, review rate, average pages/doc, latency)
audit events for business-level tracing

Demo Tips (if you are presenting this live)

This pattern demos extremely well if you show all three views:

Upload a document
- drag-and-drop into S3 (or app UI)
Workflow progress
- Step Functions execution graph
- Distributed Map child runs for batches
Outputs
- raw Textract JSON pointer
- normalized JSON in S3
- audit trail item(s)
- human review queue (triggered by a low-confidence sample)

If I want a smooth live demo, I prepare:

one “clean” invoice (no review needed)
one noisy/scanned doc (human review path)
one small batch manifest (3 to 10 docs) to show Distributed Map fan-out

Extensions I would add next

Once this base pipeline is running, it becomes a great platform for extensions:

Document classification before extraction (route to different extraction templates)
Schema validation per document type
OpenSearch indexing for searchable archives
Comprehend / Bedrock post-processing for summaries and entity normalization
Tenant-specific extraction configs (thresholds, required fields, routing)
Replay tooling for partial reprocessing from audit trail checkpoints

Closing Thoughts

I like this architecture because it is not just a toy OCR demo. It demonstrates a proper event-driven, auditable, scalable workflow that teams can actually build on.

The combination of:

S3 for durable ingress
Textract for extraction
Step Functions for orchestration
Distributed Map for batch scale
EventBridge for decoupling
callback-based human review for exceptions

gives a strong foundation for production document intelligence systems.

If I were publishing this as a developer advocacy post, I would also include a sample repository with:

CDK deployment
test documents
reviewer UI mock
load test script
dashboards/alarms

That turns the article into something readers can run and evolve.

References

Amazon Textract async APIs and async processing overview (start/get patterns, async job model, supported file types, SNS notification model for async operations)
GetDocumentAnalysis pagination (MaxResults, NextToken) and paginated retrieval behavior
Textract output storage and OutputConfig (custom S3 output and optional KMS for output)
Textract quotas and Service Quotas guidance
S3 -> EventBridge integration and S3 EventBridge event structure
Step Functions Distributed Map (when to use it, high concurrency, child workflows, map runs, input/history limits)
Step Functions callback/task token pattern and long waits for external/human steps
Step Functions optimized EventBridge integration (events:putEvents) and failure handling behavior
Step Functions pricing model (state transitions for Standard workflows)

Rebuilding TLS, Part 2 — Adding Integrity to the Channel

Dmytro Huz — Sun, 05 Apr 2026 21:42:10 +0000

In the first part of this series, we built our first fake secure channel.

We took a simple socket-based client and server, wrapped their communication in AES-CTR with a shared secret key, and got something that already looked much more serious than plain TCP. The traffic stopped being transparent. A passive observer could no longer read the request and response directly.

That was real progress.

But it still had a fatal flaw.

The receiver had no way to know whether the encrypted message had been changed on the way.

Encryption hid the bytes.

It did not protect their meaning.

So in this part, we will fix that.

We will first add a MAC so the receiver can detect tampering. Then we will make the record layer a little less naive by adding a sequence number. And after that, we will take one more step toward the real world and move to AEAD, because that is how modern secure protocols usually protect records.

We still will not have real TLS when we are done.

But we will have a much more serious record layer than the one from Part 1.

What we will build in this part

The plan for this article is simple:

briefly introduce MACs
add HMAC to our encrypted record format
make tampering detectable
add a sequence number to each record
explain why sequence numbers matter
then move from our hand-built “encrypt + MAC” construction to AEAD, because that is the approach real systems usually use

Just like in Part 1, I want to keep the pattern simple:

explain the idea
show the code
explain what changed
explain what is still broken

Why encryption still was not enough

At the end of Part 1, our protocol already had one real property:

confidentiality against passive observers

That mattered.

But it still failed against active attackers.

Because AES-CTR by itself does not provide integrity, an attacker could modify ciphertext and the receiver would still decrypt it and trust the result. That was the main lesson of the first article:

confidentiality is not integrity

So the next missing property is obvious.

The receiver needs a way to verify that the message arrived unchanged.

That is what a MAC gives us.

A very short note on MACs

MAC stands for Message Authentication Code.

Very roughly, it is a cryptographic tag computed over a message using a secret key.

The sender computes the tag and sends it together with the message.

The receiver recomputes the tag and compares it with the one that was received.

If the tags match, the receiver can trust that:

the message was not modified
and it was created by someone who knows the MAC key

If the tags do not match, the message must be rejected.

In this article, we will use HMAC-SHA256.

I do not want to go too deep into HMAC itself here, because the goal of this series is to understand TLS as a protocol. But if you want a deeper explanation of MACs and HMAC, I already wrote about them in my cryptography series, and I’ll link that here: https://www.dmytrohuz.com/p/building-own-mac-part-3-reinventing

So for our purposes, the important idea is simple:

encryption hides the message

MAC protects the message from silent modification

That is the missing half we need.

Adding HMAC to the channel

Let’s start by upgrading the record format from Part 1.

In Part 1, our protected payload was basically:

nonce || ciphertext

Now we will add a MAC tag:

nonce || ciphertext || tag

And the sender will compute the HMAC over:

nonce || ciphertext

So the full logic becomes:

Sender

encrypt plaintext with AES-CTR
compute HMAC over nonce || ciphertext
send nonce || ciphertext || tag

Receiver

read nonce || ciphertext || tag
recompute HMAC over nonce || ciphertext
compare tags
only if they match, decrypt the ciphertext
otherwise reject the message

There is one more small improvement I want to make here.

Instead of using one key for everything, we will already separate them:

one key for encryption
one key for HMAC

This is still a toy setup, but it is better design than reusing the same bytes for every cryptographic job.

HMAC helpers

import os
import hmac
import hashlib

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

# ---------------------------------------------------------------------------
# Keys — hardcoded for educational purposes.
# In a real protocol, these would be derived from a key exchange (e.g.,
# Diffie-Hellman), not embedded in source code.
# ---------------------------------------------------------------------------

# 32-byte (256-bit) key for AES-256-CTR encryption.
ENC_KEY = b"0123456789ABCDEF0123456789ABCDEF"

# 32-byte key for HMAC-SHA256.  Separate from the encryption key.
MAC_KEY = b"HMAC_KEY_FOR_PART2_DEMO_1234567"

# HMAC-SHA256 produces a 32-byte (256-bit) tag.
TAG_LEN = 32

# AES-CTR nonce is 16 bytes (128 bits).
NONCE_LEN = 16

def encrypt_then_mac(plaintext: bytes) -> bytes:
    """Encrypt a plaintext and append an HMAC tag.

    Returns: nonce (16 B) || ciphertext (N B) || tag (32 B)
    """

    # Step 1: Generate a fresh random nonce for AES-CTR.
    # A new nonce MUST be used for every record — reusing a nonce with
    # the same key completely breaks CTR-mode security.
    nonce = os.urandom(NONCE_LEN)

    # Step 2: Encrypt the plaintext with AES-256-CTR.
    cipher = Cipher(algorithms.AES(ENC_KEY), modes.CTR(nonce))
    encryptor = cipher.encryptor()
    ciphertext = encryptor.update(plaintext) + encryptor.finalize()

    # Step 3: Compute HMAC-SHA256 over (nonce || ciphertext).
    # New in Part 2: we authenticate the encrypted record before sending it.
    # The HMAC input includes the nonce so an attacker cannot swap nonces
    # between records without detection.
    mac_input = nonce + ciphertext
    tag = hmac.new(MAC_KEY, mac_input, hashlib.sha256).digest()

    print(f"  [crypto_hmac] encrypt_then_mac:")
    print(f"    nonce    = {nonce.hex()[:32]}...")
    print(f"    ct_len   = {len(ciphertext)} bytes")
    print(f"    tag      = {tag.hex()[:32]}...")

    # Step 4: Assemble the wire format.
    return nonce + ciphertext + tag

def verify_then_decrypt(payload: bytes) -> bytes:
    """Verify the HMAC tag, then decrypt if valid.

    Expects: nonce (16 B) || ciphertext (N B) || tag (32 B)
    Raises ValueError if the tag does not match.
    """

    # Step 1: Parse the record into its components.
    # The tag is always the last 32 bytes.  The nonce is the first 16.
    # Everything in between is ciphertext.
    if len(payload) < NONCE_LEN + TAG_LEN:
        raise ValueError("Record too short to contain nonce + tag")

    nonce = payload[:NONCE_LEN]
    ciphertext = payload[NONCE_LEN:-TAG_LEN]
    received_tag = payload[-TAG_LEN:]

    # Step 2: Recompute the HMAC over (nonce || ciphertext).
    mac_input = nonce + ciphertext
    expected_tag = hmac.new(MAC_KEY, mac_input, hashlib.sha256).digest()

    # Step 3: Compare tags using constant-time comparison.
    # hmac.compare_digest() prevents timing side-channel attacks.
    # A naive `==` comparison can leak information about which byte
    # position differs first, allowing an attacker to forge a valid
    # tag byte by byte.
    if not hmac.compare_digest(received_tag, expected_tag):
        print("  [crypto_hmac] *** MAC VERIFICATION FAILED — record rejected ***")
        raise ValueError("HMAC verification failed — record has been tampered with")

    print("  [crypto_hmac] MAC verification: OK")

    # Step 4: Decrypt only after verification succeeds.
    # This is the key benefit of encrypt-then-MAC: we never process
    # unauthenticated ciphertext.
    cipher = Cipher(algorithms.AES(ENC_KEY), modes.CTR(nonce))
    decryptor = cipher.decryptor()
    plaintext = decryptor.update(ciphertext) + decryptor.finalize()

    return plaintext

This is the first big improvement over Part 1.

The important change is not just that we added a tag.

It is that the receiver no longer blindly trusts ciphertext and only then discovers what it means. Now the receiver first checks whether the record is authentic and unchanged.

That is a very different protocol posture.

Updating the client and server

Now let’s plug this into the channel.

HMAC-based client

import socket

from framing import send_record, recv_record
from crypto_hmac import encrypt_then_mac, verify_then_decrypt

HOST = "127.0.0.1"
PORT = 9001

# A toy HTTP-like request — same spirit as Part 1.
request = (
    "GET /transfer?to=bob&amount=100 HTTP/1.1\r\nHost: localhost\r\n\r\n"
).encode("utf-8")

print("=" * 60)
print("Part 2 — HMAC Client (encrypt-then-MAC)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((HOST, PORT))
    print(f"Connected to {HOST}:{PORT}")

    # ----- SEND REQUEST -----
    print("\n--- Sending request ---")
    protected = encrypt_then_mac(request)
    send_record(client, protected)
    print(f"  Record sent ({len(protected)} bytes on wire)")

    # ----- RECEIVE RESPONSE -----
    print("\n--- Receiving response ---")
    raw_response = recv_record(client)
    response = verify_then_decrypt(raw_response)
    print(f"\n  Decrypted response:\n  {response.decode('utf-8')}")

print("\nDone.")

HMAC-based server

import socket

from framing import send_record, recv_record
from crypto_hmac import encrypt_then_mac, verify_then_decrypt

HOST = "127.0.0.1"
PORT = 9001

print("=" * 60)
print("Part 2 — HMAC Server (encrypt-then-MAC)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    # SO_REUSEADDR lets us restart the server immediately without waiting
    # for the OS to release the port from TIME_WAIT state.
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(1)
    print(f"Listening on {HOST}:{PORT}")

    conn, addr = server.accept()
    with conn:
        print(f"Connected by {addr}")

        # ----- RECEIVE REQUEST -----
        print("\n--- Receiving request ---")
        raw_request = recv_record(conn)

        try:
            request = verify_then_decrypt(raw_request)
        except ValueError as e:
            # New in Part 2: if the MAC fails, we reject the record loudly.
            # In Part 1 we had no way to detect tampering at all.
            print(f"\n  *** REJECTED: {e} ***")
            print("  Connection closed — refusing to process tampered data.")
        else:
            print(f"\n  Decrypted request:\n  {request.decode('utf-8')}")

            # ----- SEND RESPONSE -----
            print("--- Sending response ---")
            response = (
                "HTTP/1.1 200 OK\r\n"
                "Content-Type: text/plain\r\n"
                "Content-Length: 13\r\n\r\n"
                "hello, client"
            ).encode("utf-8")

            protected = encrypt_then_mac(response)
            send_record(conn, protected)
            print(f"  Record sent ({len(protected)} bytes on wire)")

print("\nDone.")

The shape of the channel is still familiar.

That matters.

We did not replace the whole design.

We strengthened one missing property.

That is how protocol evolution should feel.

Let’s check it on the wire.

We try to start the server and client, which we just created.

Here is our client request’s data.

-- Sending request ---
[crypto_hmac] encrypt_then_mac:
nonce = 387f0065f8915133473597d8cef15f34...
ct_len = 61 bytes
tag = b7f0db369e5d2a221a18f2d167b5b3a8...
Record sent (109 bytes on wire)

Detecting tampering

Now let’s revisit the failure from Part 1.

Previously, if someone modified the ciphertext, the receiver would still decrypt it and accept modified plaintext.

Now that should no longer work.

Here is a tiny tampering demo:

# tampering_demo_hmac.py
from crypto_hmac import encrypt_then_mac, verify_then_decrypt

original = b"amount=100"
protected = encrypt_then_mac(original)

tampered = bytearray(protected)
tampered[20] ^= 0x08  # flip one bit somewhere in the encrypted body

try:
    result = verify_then_decrypt(bytes(tampered))
    print("Unexpected success:", result)
except ValueError as e:
    print("Tampering detected:", e)

Now the result should be rejection, not silent acceptance.

That is exactly what we wanted.

This is the moment where our channel stops being merely “encrypted” and starts being “protected.”

Because now the receiver does not just recover bytes. It verifies them first.

That is a serious step.

Why we also need a sequence number

At this point, we fixed the big flaw from Part 1: silent tampering.

But the record layer is still naive.

Why?

Because even with a valid HMAC, the receiver still has no sense of record position or freshness.

Imagine an attacker records one valid protected message and sends it again later.

The HMAC is still valid.

The ciphertext is still valid.

And unless the receiver keeps some state, it may accept the same record again.

That means integrity alone is not the whole story.

We also need some sense of:

order
position
repetition
replay

This is where sequence numbers come in.

A sequence number is just a counter that increases with every record:

first record = 0
next = 1
next = 2
and so on

We then include that sequence number in the authenticated data, so the receiver does not just verify “these bytes were protected,” but also “these bytes belong in this position in the stream.”

That makes the record layer much less naive.

It still does not solve every replay problem in every possible system. But for our toy protocol, it is a very good next step.

Updating the record format

Now our record becomes:

seq || nonce || ciphertext || tag

And our HMAC input becomes:

seq || nonce || ciphertext

So the sender and receiver now both need a little bit of state:

the sender tracks the next sequence number to send
the receiver tracks the next sequence number it expects

This is one of those moments where secure transport starts looking more like a real protocol and less like “some crypto around a socket.”

Sequence-aware HMAC helper

# crypto_hmac_seq.py
import os
import hmac
import hashlib
import struct

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

# ---------------------------------------------------------------------------
# Keys — same as crypto_hmac.py, hardcoded for education.
# ---------------------------------------------------------------------------
ENC_KEY = b"0123456789ABCDEF0123456789ABCDEF"
MAC_KEY = b"HMAC_KEY_FOR_PART2_DEMO_1234567"

TAG_LEN = 32  # HMAC-SHA256 output: 32 bytes (256 bits)
NONCE_LEN = 16  # AES-CTR nonce: 16 bytes (128 bits)
SEQ_LEN = 8  # Sequence number: 8 bytes (64-bit unsigned integer)

def protect_record(seq: int, plaintext: bytes) -> bytes:
    """Encrypt a plaintext record and attach a sequence-aware HMAC tag.

    Args:
        seq:       The current send-side sequence number (0, 1, 2, …).
        plaintext: The message to protect.

    Returns:
        seq (8 B) || nonce (16 B) || ciphertext (N B) || tag (32 B)
    """

    # Pack the sequence number as an 8-byte big-endian unsigned integer.
    # "!Q" = network byte order, unsigned 64-bit.
    seq_bytes = struct.pack("!Q", seq)

    # Generate a fresh AES-CTR nonce.
    nonce = os.urandom(NONCE_LEN)

    # Encrypt the plaintext.
    cipher = Cipher(algorithms.AES(ENC_KEY), modes.CTR(nonce))
    encryptor = cipher.encryptor()
    ciphertext = encryptor.update(plaintext) + encryptor.finalize()

    # Compute HMAC over (seq || nonce || ciphertext).
    # The sequence number is included in the MAC input so the integrity
    # check also covers record order/position in the stream.
    mac_input = seq_bytes + nonce + ciphertext
    tag = hmac.new(MAC_KEY, mac_input, hashlib.sha256).digest()

    return seq_bytes + nonce + ciphertext + tag

def verify_and_unprotect(expected_seq: int, payload: bytes) -> bytes:
    """Verify the HMAC and sequence number, then decrypt.

    Args:
        expected_seq: The sequence number the receiver expects next.
        payload:      The raw bytes received: seq || nonce || ct || tag.

    Returns:
        The decrypted plaintext.

    Raises:
        ValueError if the MAC is invalid or the sequence number is wrong.
    """

    min_len = SEQ_LEN + NONCE_LEN + TAG_LEN
    if len(payload) < min_len:
        raise ValueError("Record too short")

    # Step 1: Parse the record.
    seq_bytes = payload[:SEQ_LEN]
    nonce = payload[SEQ_LEN : SEQ_LEN + NONCE_LEN]
    ciphertext = payload[SEQ_LEN + NONCE_LEN : -TAG_LEN]
    received_tag = payload[-TAG_LEN:]

    # Step 2: Recompute HMAC over (seq || nonce || ciphertext).
    mac_input = seq_bytes + nonce + ciphertext
    expected_tag = hmac.new(MAC_KEY, mac_input, hashlib.sha256).digest()

    # Step 3: Constant-time tag comparison.
    if not hmac.compare_digest(received_tag, expected_tag):
        print("  [crypto_hmac_seq] *** MAC VERIFICATION FAILED ***")
        raise ValueError("HMAC verification failed — record tampered or replayed")

    print("  [crypto_hmac_seq] MAC verification: OK")

    # Step 4: Check the sequence number matches what we expect.
    # Even though the MAC already covers the sequence number (so an
    # attacker cannot change it without invalidating the MAC), we still
    # explicitly verify that it matches our counter.  This catches
    # replayed or reordered records that carry a valid MAC but belong
    # to a different position in the stream.
    (received_seq,) = struct.unpack("!Q", seq_bytes)
    if received_seq != expected_seq:
        print(
            f"  [crypto_hmac_seq] *** SEQUENCE MISMATCH: "
            f"got {received_seq}, expected {expected_seq} ***"
        )
        raise ValueError(
            f"Sequence number mismatch: got {received_seq}, expected {expected_seq}"
        )

    print(
        f"  [crypto_hmac_seq] Sequence number: {received_seq} (expected {expected_seq}) — OK"
    )

    # Step 5: Decrypt.
    cipher = Cipher(algorithms.AES(ENC_KEY), modes.CTR(nonce))
    decryptor = cipher.decryptor()
    plaintext = decryptor.update(ciphertext) + decryptor.finalize()

    return plaintext

And now the channel has a bit more memory.

Not just “is this record authentic?”

But also “is this the record I expected next?”

That is a real protocol improvement.

Updating the client and server

Now let’s plug this into the channel.

Sequence-aware HMAC-based client

# client_v2_hmac_seq.py
import socket

from framing import send_record, recv_record
from crypto_hmac_seq import protect_record, verify_and_unprotect

HOST = "127.0.0.1"
PORT = 9003

# Sequence counters — sender and receiver each maintain their own.
# The sender increments after each record sent.
# The receiver expects consecutive values starting from 0.
send_seq = 0
recv_seq = 0

# A toy HTTP-like request — same spirit as Part 1.
request = (
    "GET /transfer?to=bob&amount=100 HTTP/1.1\r\nHost: localhost\r\n\r\n"
).encode("utf-8")

print("=" * 60)
print("Part 2 — HMAC + Sequence Numbers Client (Stage 2)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((HOST, PORT))
    print(f"Connected to {HOST}:{PORT}")

    # ----- SEND REQUEST -----
    print(f"\n--- Sending request (send_seq={send_seq}) ---")
    protected = protect_record(send_seq, request)
    send_record(client, protected)
    send_seq += 1
    print(f"  Record sent ({len(protected)} bytes on wire)")

    # ----- RECEIVE RESPONSE -----
    print(f"\n--- Receiving response (expecting recv_seq={recv_seq}) ---")
    raw_response = recv_record(client)

    try:
        response = verify_and_unprotect(recv_seq, raw_response)
        recv_seq += 1
        print(f"\n  Decrypted response:\n  {response.decode('utf-8')}")
    except ValueError as e:
        print(f"\n  *** REJECTED: {e} ***")

print("\nDone.")

Sequence-aware HMAC-based server

# server_v2_hmac_seq.py
import socket

from framing import send_record, recv_record
from crypto_hmac_seq import protect_record, verify_and_unprotect

HOST = "127.0.0.1"
PORT = 9003

# Sequence counters.
# The server's recv_seq tracks the client's send_seq, and vice versa.
send_seq = 0
recv_seq = 0

print("=" * 60)
print("Part 2 — HMAC + Sequence Numbers Server (Stage 2)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(1)
    print(f"Listening on {HOST}:{PORT}")

    conn, addr = server.accept()
    with conn:
        print(f"Connected by {addr}")

        # ----- RECEIVE REQUEST -----
        print(f"\n--- Receiving request (expecting recv_seq={recv_seq}) ---")
        raw_request = recv_record(conn)

        try:
            request = verify_and_unprotect(recv_seq, raw_request)
            recv_seq += 1
        except ValueError as e:
            # Rejection: either the MAC is invalid, the sequence number
            # is wrong, or the data was tampered with / replayed.
            print(f"\n  *** REJECTED: {e} ***")
            print("  Connection closed — refusing to process invalid data.")
        else:
            print(f"\n  Decrypted request:\n  {request.decode('utf-8')}")

            # ----- SEND RESPONSE -----
            print(f"--- Sending response (send_seq={send_seq}) ---")
            response = (
                "HTTP/1.1 200 OK\r\n"
                "Content-Type: text/plain\r\n"
                "Content-Length: 13\r\n\r\n"
                "hello, client"
            ).encode("utf-8")

            protected = protect_record(send_seq, response)
            send_record(conn, protected)
            send_seq += 1
            print(f"  Record sent ({len(protected)} bytes on wire)")

print("\nDone.")

Why real-world systems usually do not stop here

At this point, we have something much stronger than Part 1.

We have:

encryption
integrity protection
message authentication
sequence-aware records

That is already a meaningful protocol.

But if you look at how real systems are usually built, they do not normally stop at manually composing:

AES-CTR
HMAC-SHA256
explicit sequence-aware record protection

Why?

Because modern systems usually prefer a single primitive that gives confidentiality and integrity together.

That is where AEAD comes in.

We separated these properties on purpose because it makes the protocol easier to understand.

But the real world usually packages them together.

A very short note on AEAD

AEAD stands for Authenticated Encryption with Associated Data.

That sounds heavier than it really is.

The practical idea is simple:

An AEAD construction gives us:

encryption
integrity/authentication of the encrypted message
and the ability to authenticate extra metadata that should not be encrypted

Common examples are:

AES-GCM
ChaCha20-Poly1305

This is much closer to how modern secure protocols protect records.

It is also why I wanted to include AEAD in this part. If we stopped only at “encrypt + HMAC,” we would understand the missing property better, but we would still be one step away from how modern systems actually package it.

So now we take that final step.

Moving our channel to AEAD

For the AEAD version, I will use AES-GCM.

The high-level idea is:

the plaintext gets encrypted
integrity/authentication is built in
and we can include extra metadata as associated data

In our case, the sequence number is a good example of associated data.

That means:

it does not need to be encrypted
but it should still be authenticated

AEAD-based helper

# crypto_aead.py
import os
import struct

from cryptography.hazmat.primitives.ciphers.aead import AESGCM

# ---------------------------------------------------------------------------
# Key — a single 256-bit key for AES-GCM.
# With AEAD, we do NOT need separate encryption and MAC keys — the
# algorithm handles both internally.
# ---------------------------------------------------------------------------
AEAD_KEY = b"AEAD_KEY_PART2_DEMO_FOR_AES_GCM!"  # 32 bytes → AES-256-GCM

# AES-GCM nonce length: 12 bytes is the recommended (and most efficient) size.
NONCE_LEN = 12

# Sequence number: 8 bytes (64-bit unsigned integer), same as Stage 2.
SEQ_LEN = 8

def protect_record_aead(seq: int, plaintext: bytes) -> bytes:
    """Seal a plaintext record with AES-GCM.

    Args:
        seq:       The current send-side sequence number.
        plaintext: The message to protect.

    Returns:
        seq (8 B) || nonce (12 B) || ciphertext_and_tag (N+16 B)
    """

    # Pack the sequence number as associated data.
    # The sequence number is authenticated but sent in the clear — the
    # receiver needs it to know which counter value to expect.
    seq_bytes = struct.pack("!Q", seq)

    # Generate a random 12-byte nonce for AES-GCM.
    nonce = os.urandom(NONCE_LEN)

    # Create an AESGCM instance with our key.
    aesgcm = AESGCM(AEAD_KEY)

    # Encrypt and authenticate in one call.
    # AESGCM.encrypt(nonce, data, associated_data) returns
    # ciphertext || 16-byte authentication tag as a single bytes object.
    # The associated_data (seq_bytes) is authenticated but NOT encrypted.
    ciphertext_and_tag = aesgcm.encrypt(nonce, plaintext, seq_bytes)

    print(f"  [crypto_aead] protect_record_aead:")
    print(f"    seq      = {seq}")
    print(f"    nonce    = {nonce.hex()}")
    print(
        f"    sealed   = {len(ciphertext_and_tag)} bytes "
        f"(plaintext {len(plaintext)} + tag 16)"
    )

    return seq_bytes + nonce + ciphertext_and_tag

def unprotect_record_aead(expected_seq: int, payload: bytes) -> bytes:
    """Verify and decrypt an AES-GCM sealed record.

    Args:
        expected_seq: The sequence number the receiver expects next.
        payload:      seq (8 B) || nonce (12 B) || ciphertext_and_tag.

    Returns:
        The decrypted plaintext.

    Raises:
        ValueError if the sequence number is wrong.
        cryptography.exceptions.InvalidTag if decryption/auth fails.
    """

    min_len = SEQ_LEN + NONCE_LEN + 16  # at least seq + nonce + tag
    if len(payload) < min_len:
        raise ValueError("Record too short")

    # Step 1: Parse the record.
    seq_bytes = payload[:SEQ_LEN]
    nonce = payload[SEQ_LEN : SEQ_LEN + NONCE_LEN]
    ciphertext_and_tag = payload[SEQ_LEN + NONCE_LEN :]

    # Step 2: Check the sequence number.
    (received_seq,) = struct.unpack("!Q", seq_bytes)
    if received_seq != expected_seq:
        print(
            f"  [crypto_aead] *** SEQUENCE MISMATCH: "
            f"got {received_seq}, expected {expected_seq} ***"
        )
        raise ValueError(
            f"Sequence number mismatch: got {received_seq}, expected {expected_seq}"
        )

    print(
        f"  [crypto_aead] Sequence number: {received_seq} "
        f"(expected {expected_seq}) — OK"
    )

    # Step 3: Decrypt and verify in one call.
    # AESGCM.decrypt(nonce, data, associated_data) verifies the auth tag
    # and decrypts.  If anything was tampered with — the ciphertext, the
    # tag, or the associated data — it raises InvalidTag.
    aesgcm = AESGCM(AEAD_KEY)
    plaintext = aesgcm.decrypt(nonce, ciphertext_and_tag, seq_bytes)

    print(f"  [crypto_aead] AEAD decryption: OK ({len(plaintext)} bytes)")

    return plaintext

This code is noticeably simpler.

That is one of the big practical advantages of AEAD.

Instead of manually:

encrypting
computing HMAC
verifying HMAC
then decrypting

we use one primitive that already combines confidentiality and integrity.

And the sequence number fits naturally as associated data.

AEAD-based client

# client_v2_aead.py
import socket

from framing import send_record, recv_record
from crypto_aead import protect_record_aead, unprotect_record_aead

HOST = "127.0.0.1"
PORT = 9002

# Sequence counters — sender and receiver each maintain their own.
# The sender increments after each record sent.
# The receiver expects consecutive values starting from 0.
send_seq = 0
recv_seq = 0

# A toy HTTP-like request.
request = (
    "GET /transfer?to=bob&amount=100 HTTP/1.1\r\nHost: localhost\r\n\r\n"
).encode("utf-8")

print("=" * 60)
print("Part 2 — AEAD Client (AES-GCM)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((HOST, PORT))
    print(f"Connected to {HOST}:{PORT}")

    # ----- SEND REQUEST -----
    print(f"\n--- Sending request (send_seq={send_seq}) ---")
    protected = protect_record_aead(send_seq, request)
    send_record(client, protected)
    send_seq += 1
    print(f"  Record sent ({len(protected)} bytes on wire)")

    # ----- RECEIVE RESPONSE -----
    print(f"\n--- Receiving response (expecting recv_seq={recv_seq}) ---")
    raw_response = recv_record(client)

    try:
        response = unprotect_record_aead(recv_seq, raw_response)
        recv_seq += 1
        print(f"\n  Decrypted response:\n  {response.decode('utf-8')}")
    except Exception as e:
        print(f"\n  *** REJECTED: {e} ***")

print("\nDone.")

AEAD-based server

# server_v2_aead.py
import socket

from framing import send_record, recv_record
from crypto_aead import protect_record_aead, unprotect_record_aead

HOST = "127.0.0.1"
PORT = 9002

# Sequence counters.
# The server's recv_seq tracks the client's send_seq, and vice versa.
send_seq = 0
recv_seq = 0

print("=" * 60)
print("Part 2 — AEAD Server (AES-GCM)")
print("=" * 60)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((HOST, PORT))
    server.listen(1)
    print(f"Listening on {HOST}:{PORT}")

    conn, addr = server.accept()
    with conn:
        print(f"Connected by {addr}")

        # ----- RECEIVE REQUEST -----
        print(f"\n--- Receiving request (expecting recv_seq={recv_seq}) ---")
        raw_request = recv_record(conn)

        try:
            request = unprotect_record_aead(recv_seq, raw_request)
            recv_seq += 1
        except Exception as e:
            # AEAD rejection: either the auth tag is invalid, the sequence
            # number is wrong, or the data was tampered with.
            print(f"\n  *** REJECTED: {e} ***")
            print("  Connection closed — refusing to process invalid data.")
        else:
            print(f"\n  Decrypted request:\n  {request.decode('utf-8')}")

            # ----- SEND RESPONSE -----
            print(f"--- Sending response (send_seq={send_seq}) ---")
            response = (
                "HTTP/1.1 200 OK\r\n"
                "Content-Type: text/plain\r\n"
                "Content-Length: 13\r\n\r\n"
                "hello, client"
            ).encode("utf-8")

            protected = protect_record_aead(send_seq, response)
            send_record(conn, protected)
            send_seq += 1
            print(f"  Record sent ({len(protected)} bytes on wire)")

print("\nDone.")

This version is already much closer to how modern secure transport actually protects records.

Not identical to TLS, of course. But structurally much closer.

What we gained

At this point, our channel is much stronger than the one from Part 1.

We now have:

confidentiality
integrity protection
authenticated records
sequence-aware message handling
a much more realistic record protection design through AEAD

That is a big improvement.

The receiver is no longer just decrypting whatever arrives and trusting the result. Now the receiver can reject modified or structurally unexpected records.

That is a real protocol boundary.

What is still broken

And yet, even now, we are still very far from real TLS.

Because the biggest assumption in our design is still untouched:

both sides already share the necessary secret keys

That means we still do not know how to solve the next real problem:

how do two strangers establish fresh secrets?
how does the client know it is talking to the right server?
how do we scale beyond hardcoded shared secrets?
how do we build trust instead of assuming it?

We improved record protection a lot.

But we still do not have a real way to establish trust.

That is the next wall.

Summary

In this part, we took the encrypted but still incomplete channel from Part 1 and made it much more serious.

First, we added HMAC, which gave the receiver a way to detect tampering.

Then, we added a sequence number, which made the record layer less naive and bound records to their place in the stream.

Finally, we moved to AEAD, because in real-world systems confidentiality and integrity are usually protected together, not assembled manually from separate pieces.

So this article had two goals:

understand the missing property explicitly
then move toward the real-world shape of the solution

That is why we did not stop at HMAC.

But even after all of this, we still depend on one assumption that makes the whole thing unrealistic:

we are still starting with pre-shared secret keys.

And that is exactly what the next part will attack.

Next part — getting rid of the pre-shared key

So we stop here.

We now have a much better record layer than in Part 1. But we are still relying on a hardcoded shared secret, and that is not how real secure communication between strangers on the internet works.

In the next part, we will stop assuming both sides already share a secret.

We will start building a real handshake and establish fresh session keys instead.

That still will not give us full TLS.

But it will take us much closer to the real shape of the protocol.

Final code

I’ll put the full code for this part on GitHub here:

[GitHub link to final code]

Forem: AWS Community Builders

Add Chat AI Summary Using Amazon Bedrock and HTTP Response Streaming

Architecture

Amazon Bedrock

HTTP Response Streaming with API Gateway

Demo

Conclusions

Handling File Uploads in AWS Lambda with Powertools OpenAPI (From Limitation to Production Feature)

Introduction

The Problem: File Uploads Break the Abstraction

The Goal: Make File Uploads First-Class

The Solution: File() Parameter Support

Two Ways to Work with Files

1. Raw bytes

2. Rich file object

Combining Files with Form Data

What Changed Under the Hood

API Gateway Gotcha (Important)

Before vs After

Before

After

Why This Matters

Open Source Insight

Final Thoughts

References

I Ditched Prisma for Raw SQL (And My Queries Got 10x Faster)

What Prisma Actually Does to Your Queries

The N+1 Problem Prisma Doesn't Fully Solve

The Performance Numbers

Migrating Without Rewriting Everything

What You Actually Lose

When to Keep Prisma

Production Observability for Kubernetes on AWS using OpenTelemetry Operator

Why Observability is Critical in Kubernetes

Observability Architecture Overview

Architecture Flow

Key Components of the Stack

OpenTelemetry Operator

OpenTelemetry Collector

Prometheus (Metrics)

Grafana Tempo (Traces)

Loki (Logs)

Grafana

Deploying on AWS (EKS-Based Architecture)

Auto-Instrumentation Example

OpenTelemetry Collector Config

Correlation Workflow

Production Best Practices

Common Pitfalls

Real-World Example

Production Debugging Scenario

Scenario

Step 1: Detect the Issue (Metrics)

Step 2: Trace the Request (Traces)

Step 3: Drill Down into Spans

Step 4: Inspect Logs

Step 5: Root Cause Identified

Step 6: Resolution

Outcome

Key Insight

Final Thoughts

MCP Development with Python, Gemini CLI, and Amazon AWS ECS Express

Aren’t There a Billion Python MCP Demos?

What Is Python?

Python Version Management

Amazon ECS Express Configuration

Gemini CLI

Testing the Gemini CLI Environment

Node Version Management

Python MCP Documentation

Where do I start?

Setup the Basic Environment

Hello World with HTTP Transport

Running the Python Code

Deploy to ECS Express

Gemini CLI settings.json

Validation with Gemini CLI

Validate Project Setup with Gemini CLI

Summary

25 Internal Knowledge and Productivity Agent Patterns on AWS You Can Steal Right Now

The Solution: `File()` Parameter Support