🛡️ AWS CDK Guardrails Workflow - Complete Exercise Summary

What we built: A production-grade serverless guardrails pipeline using AWS CDK with Python

Cost: $0.00 (Free Tier)

---

📋 Table of Contents

1. Architecture Overview
2. CDK Fundamentals
3. Project Setup
4. Lambda Functions
5. The Stack Code
6. Deployment
7. Testing
8. AWS Console Visualization
9. Cost Management
10. Cleanup
11. Quick Reference

💡 Sticky Analogy: Message Screening Facility

Think of it as a message screening facility:

• 📬 Message arrives (user query)
• 🚨 Security Guard #1 checks: "Is this message safe to process?"
• 🤖 Robot Writer generates a response
• 🚨 Security Guard #2 checks: "Is this response appropriate?"
• 📁 Filing Cabinet stores the approved result (DynamoDB)

AWS Services Used

Service	Purpose	Free Tier
Lambda	Compute for guards and mock LLM	1M requests/month
Step Functions	Orchestrates the workflow	4,000 transitions/month
DynamoDB	Stores approved results	25 GB storage
S3	CDK deployment artifacts	5 GB

---

🧱 CDK Fundamentals

What is AWS CDK?

Infrastructure as Code - Define your cloud resources using Python (or other languages) instead of clicking through the AWS Console.

Analogy: CDK is like an architect's blueprint. You describe WHAT to build, and CDK figures out HOW to build it in AWS.

Key CDK Modules

from aws_cdk import (
    aws_stepfunctions as sfn,        # Workflow STRUCTURE
    aws_stepfunctions_tasks as tasks, # Actual WORK
    aws_lambda as lambda_,            # Compute functions
    aws_dynamodb as dynamodb,         # Database
)

💡 Sticky Analogy: Factory vs Machines

aws_stepfunctions (sfn) = The conveyor belt system

• Decides where things go
• "If quality check passes, go left. If fails, go right."

aws_stepfunctions_tasks (tasks) = The machines on the belt

• Actually do the work
• "This machine welds", "This machine paints"

One makes decisions. One does work.

CDK Module Comparison Table

Module	What It Does	Example
sfn.Choice	Decision point	"If safe, continue. Else, reject."
sfn.Fail	Stop and report error	"Input Rejected"
tasks.LambdaInvoke	Call a Lambda function	"Run InputGuard"
dynamodb.Table	Create database table	Store results

---

🚀 Project Setup

Prerequisites Installed

# Node.js (required for CDK CLI)
brew install node

# CDK CLI
npm install -g aws-cdk

# AWS CLI
brew install awscli

# Configure credentials
aws configure set aws_access_key_id YOUR_KEY
aws configure set aws_secret_access_key YOUR_SECRET
aws configure set region us-east-1

Project Initialization

mkdir -p ~/guardrails-demo
cd ~/guardrails-demo
cdk init app --language python

Project Structure Created

~/guardrails-demo/
├── app.py                    # Entry point
├── cdk.json                  # CDK configuration
├── requirements.txt          # Python dependencies
├── .venv/                    # Virtual environment
├── lambda/                   # Our Lambda code (we create this)
│   ├── input_guard/
│   ├── gpt4_mock/
│   ├── output_guard/
│   └── store_result/
└── guardrails_demo/
    └── guardrails_demo_stack.py  # THE MAIN STACK CODE

---

⚡ Lambda Functions

1. Input Guard (lambda/input_guard/index.py)

Purpose: Block malicious inputs BEFORE they reach the LLM (saves money!)

# lambda/input_guard/index.py
import json

def handler(event, context):
    """Mock input guardrail - checks for bad words"""
    user_input = event.get("query", "")
    
    # Simple mock: reject if contains "bad" or "hack"
    blocked_words = ["bad", "hack", "inject"]
    is_safe = not any(word in user_input.lower() for word in blocked_words)
    
    return {
        "query": user_input,
        "safe": is_safe,
        "stage": "input_guard"
    }

2. GPT-4 Mock (lambda/gpt4_mock/index.py)

Purpose: Simulate LLM response (in production, this calls GPT-4)

# lambda/gpt4_mock/index.py
import json

def handler(event, context):
    """Mock LLM - returns a fake response"""
    user_query = event.get("query", "")
    
    # Fake LLM response (in production, you would call GPT-4 here)
    fake_response = "This is a helpful answer to: " + user_query
    
    return {
        "query": user_query,
        "response": fake_response,
        "stage": "gpt4_generation"
    }

3. Output Guard (lambda/output_guard/index.py)

Purpose: Check LLM output for toxic/harmful content

# lambda/output_guard/index.py
import json

def handler(event, context):
    """Mock output guardrail - checks LLM response for bad content"""
    response = event.get("response", "")
    query = event.get("query", "")
    
    # Simple mock: reject if contains toxic words
    blocked_words = ["toxic", "harmful", "dangerous"]
    is_safe = not any(word in response.lower() for word in blocked_words)
    
    return {
        "query": query,
        "response": response,
        "safe": is_safe,
        "stage": "output_guard"
    }

4. Store Result (lambda/store_result/index.py)

Purpose: Save approved responses to DynamoDB

# lambda/store_result/index.py
import json
import boto3
import os
import uuid

def handler(event, context):
    """Store the final result in DynamoDB"""
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table(os.environ["TABLE_NAME"])
    
    request_id = str(uuid.uuid4())
    
    table.put_item(Item={
        "request_id": request_id,
        "query": event.get("query", ""),
        "response": event.get("response", ""),
        "status": "completed"
    })
    
    return {
        "request_id": request_id,
        "status": "stored"
    }

---

📦 The Stack Code

This is the main infrastructure definition file.

guardrails_demo/guardrails_demo_stack.py

from aws_cdk import (
    Duration,
    Stack,
    aws_stepfunctions as sfn,
    aws_stepfunctions_tasks as tasks,
    aws_lambda as lambda_,
    aws_dynamodb as dynamodb,
)
from constructs import Construct

class GuardrailsDemoStack(Stack):

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

        # ========== DynamoDB Table ==========
        results_table = dynamodb.Table(self, "ResultsTable",
            partition_key=dynamodb.Attribute(
                name="request_id",
                type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
        )

        # ========== Lambda Functions ==========
        input_guard = lambda_.Function(self, "InputGuard",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=lambda_.Code.from_asset("lambda/input_guard"),
            timeout=Duration.seconds(30)
        )

        gpt4_mock = lambda_.Function(self, "GPT4Mock",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=lambda_.Code.from_asset("lambda/gpt4_mock"),
            timeout=Duration.seconds(60)
        )

        output_guard = lambda_.Function(self, "OutputGuard",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=lambda_.Code.from_asset("lambda/output_guard"),
            timeout=Duration.seconds(30)
        )

        store_result = lambda_.Function(self, "StoreResult",
            runtime=lambda_.Runtime.PYTHON_3_11,
            handler="index.handler",
            code=lambda_.Code.from_asset("lambda/store_result"),
            timeout=Duration.seconds(10),
            environment={"TABLE_NAME": results_table.table_name}
        )
        results_table.grant_write_data(store_result)

        # ========== Step Functions Tasks ==========
        input_task = tasks.LambdaInvoke(self, "Check Input",
            lambda_function=input_guard,
            output_path="$.Payload"
        )

        gpt4_task = tasks.LambdaInvoke(self, "Generate Response",
            lambda_function=gpt4_mock,
            output_path="$.Payload"
        )

        output_task = tasks.LambdaInvoke(self, "Check Output",
            lambda_function=output_guard,
            output_path="$.Payload"
        )

        store_task = tasks.LambdaInvoke(self, "Store Result",
            lambda_function=store_result,
            output_path="$.Payload"
        )

        # ========== Failure States ==========
        reject_input = sfn.Fail(self, "Input Rejected",
            cause="Input failed safety check"
        )

        reject_output = sfn.Fail(self, "Output Rejected",
            cause="Output failed safety check"
        )

        # ========== Choice States ==========
        check_input_safe = sfn.Choice(self, "Input Safe?")
        check_output_safe = sfn.Choice(self, "Output Safe?")

        # ========== Wire Together ==========
        definition = (
            input_task
            .next(check_input_safe
                .when(sfn.Condition.boolean_equals("$.safe", False), reject_input)
                .otherwise(gpt4_task
                    .next(output_task)
                    .next(check_output_safe
                        .when(sfn.Condition.boolean_equals("$.safe", False), reject_output)
                        .otherwise(store_task)
                    )
                )
            )
        )

        sfn.StateMachine(self, "GuardrailsWorkflow",
            definition_body=sfn.DefinitionBody.from_chainable(definition),
            timeout=Duration.minutes(5)
        )

---

🚢 Deployment

Step 1: Bootstrap (One-Time Setup)

cd ~/guardrails-demo
source .venv/bin/activate
cdk bootstrap

Analogy: Like setting up the construction site before building - creates storage shed (S3 bucket) and worker badges (IAM roles).

Step 2: Deploy

cdk deploy --require-approval never

What happens:

1. CDK reads your stack code
2. Generates CloudFormation template
3. Creates all resources in AWS
4. Shows progress: CREATE_IN_PROGRESS → CREATE_COMPLETE
5. Final: ✅ GuardrailsDemoStack

Key CDK Commands Reference

Command	Purpose
cdk init app --language python	Create new project
cdk bootstrap	One-time AWS setup
cdk synth	Preview CloudFormation template
cdk deploy	Deploy to AWS
cdk diff	Compare deployed vs current
cdk destroy	Delete all resources

---

🧪 Testing

Test 1: Happy Path ("hello" → Should Succeed)

# Create test input file (avoids shell escaping issues)
cat > ~/test-input.json << 'JSONEOF'
{"query":"hello"}
JSONEOF

# Start execution
aws stepfunctions start-execution \
  --state-machine-arn "arn:aws:states:us-east-1:ACCOUNT:stateMachine:NAME" \
  --input file://~/test-input.json

# Check result
aws stepfunctions describe-execution --execution-arn "arn:..."

Result: "status": "SUCCEEDED"

Test 2: Rejection Path ("hack" → Should Fail)

cat > ~/test-bad.json << 'JSONEOF'
{"query":"hack"}
JSONEOF

aws stepfunctions start-execution \
  --state-machine-arn "arn:..." \
  --input file://~/test-bad.json

Result:

{
  "status": "FAILED",
  "cause": "Input failed safety check"
}

✅ Guardrails worked! The "hack" query was blocked before reaching GPT-4.

Verify Data in DynamoDB

# List tables
aws dynamodb list-tables

# Scan table
aws dynamodb scan --table-name "GuardrailsDemoStack-ResultsTable-XXXXX"

Result:

{
  "Items": [
    {
      "query": {"S": "hello"},
      "response": {"S": "This is a helpful answer to: hello"},
      "status": {"S": "completed"}
    }
  ]
}

---

🖥️ AWS Console Visualization

How to Access

1. AWS Console → Search "Step Functions"
2. Click your state machine
3. Click any execution to see visual flow

Visual Elements

Element	Meaning
🟢 Green	Succeeded step
🔴 Red	Failed/rejected
🔶 Diamond	Choice state (decision point)
🟠 Orange	Task execution

---

💰 Cost Management

Check Costs Programmatically

# Check budget status
aws budgets describe-budgets --account-id YOUR_ACCOUNT_ID

# Check free tier usage
aws freetier get-free-tier-usage

Our Usage Summary

Service	Usage	Free Tier Limit	Cost
Lambda	~10 invocations	1M/month	$0.00
Step Functions	3 executions	4,000/month	$0.00
DynamoDB	2 writes, 1 scan	25 GB	$0.00
S3	~5 MB	5 GB	$0.00
Total			$0.00

---

🗑️ Cleanup

Delete All Resources

cd ~/guardrails-demo
source .venv/bin/activate
cdk destroy --force

Why Some Resources Are Skipped

CDK defaults to RETAIN for stateful resources:

Resource	Why Retained
DynamoDB	Contains data (prevents accidental loss)
CloudWatch Logs	Contains audit trails

Analogy: Like a shredder that refuses to shred family photos, even if they're in the "to shred" pile. Better safe than sorry!

To Force Delete (Dev/Test Only!)

from aws_cdk import RemovalPolicy

table = dynamodb.Table(self, "Table",
    removal_policy=RemovalPolicy.DESTROY  # ⚠️ Deletes on cdk destroy
)

---

📚 Quick Reference

Complete Command Sequence

# 1. Setup
mkdir ~/guardrails-demo && cd ~/guardrails-demo
cdk init app --language python
source .venv/bin/activate
pip install -r requirements.txt

# 2. Create Lambda folders
mkdir -p lambda/{input_guard,gpt4_mock,output_guard,store_result}

# 3. Deploy
cdk bootstrap          # One-time
cdk deploy --require-approval never

# 4. Test
aws stepfunctions start-execution --state-machine-arn "..." --input file://test.json
aws stepfunctions describe-execution --execution-arn "..."

# 5. Cleanup
cdk destroy --force

💡 All Sticky Analogies

Concept	Analogy
CDK	Architect's blueprint - describes WHAT, CDK builds HOW
sfn vs tasks	Factory: conveyor belt (decisions) vs machines (work)
Workflow	Message screening facility with security guards
cdk bootstrap	Setting up construction site before building
SageMaker Serverless	Magic kitchen that appears when needed
RemovalPolicy.RETAIN	Shredder that refuses to shred family photos
Idempotency	Elevator button - press 1x or 100x, same result

---

✅ Exercise Complete!

What you accomplished:

• Built production-grade guardrails workflow
• Deployed with Infrastructure as Code
• Tested both success and failure paths
• Verified visually in AWS Console
• Managed costs ($0.00)
• Cleaned up properly