Why Most AWS Tutorials Miss the Point

If you search "learn AWS," you will find courses that cover 50+ services across 40+ hours of video. You finish the course knowing what Route 53 is, what Glacier does, and why you might use Lightsail, but you still cannot build an AI product because you have spent most of your time on services you will not need for months.

Here is the honest truth: to build real, production-ready AI products on AWS in 2026, you primarily need five services. These five services appear in almost every AI architecture you will encounter. They work together in a pattern that repeats across different use cases. Once you understand them, you can build most of what you see being built in enterprise AI teams.

The five services are:

1. Amazon Bedrock: the AI brain
2. AWS Lambda: the serverless execution engine
3. Amazon API Gateway: the front door
4. Amazon DynamoDB: the fast, flexible database
5. Amazon CloudWatch: the observability layer

This guide explains each one from scratch, shows you where it fits in AI architectures, and ends with a complete project that uses all five together. By the end, you will have built something real.

Service 1: Amazon Bedrock, The AI Brain

What It Is

Amazon Bedrock is the AWS service that gives you access to large AI models, Claude (from Anthropic), Llama (from Meta), Mistral, and Amazon's own Titan models, via a simple API.

Before Bedrock, using AI models in production meant either:

• Paying for OpenAI's API and managing a separate API key outside AWS
• Running open-source models on EC2 GPU instances (expensive and complex to maintain)
• Using SageMaker (powerful but complex, designed for data scientists training custom models)

Bedrock changes this. You call a model the same way you call any other AWS service, using your existing AWS credentials, the same IAM permissions you already understand, and billing that goes straight to your AWS account. No separate sign-up, no separate API key, no separate billing.

What It Costs

This matters for planning. Bedrock pricing is per-token:

A token is roughly 4 characters of text. A 1,000-word article is approximately 1,300 tokens. At Claude 3 Haiku pricing, processing that article costs $0.0004, less than a tenth of a cent.

For context: an enterprise application making 10,000 requests per day with average 500 tokens per request would spend roughly $0.625/day on Claude 3 Haiku. That is $228/year, not free, but very affordable.

Where It Fits in AI Architecture

User request
     ↓
Lambda (your business logic)
     ↓
Bedrock (AI model call)
     ↓
AI response back to Lambda
     ↓
Lambda formats and returns response to user

Bedrock is always in the middle. It never faces the internet directly. Lambda calls it on behalf of users.

Enable Bedrock and Make Your First Call

bash

# Step 1: Enable model access in the AWS Console
# Go to: AWS Console → Amazon Bedrock → Model access → Manage model access
# Enable: Anthropic Claude 3 Haiku
# Approval is usually instant

# Step 2: Test from the command line
aws bedrock-runtime invoke-model \
  --region us-east-1 \
  --model-id anthropic.claude-3-haiku-20240307-v1:0 \
  --content-type application/json \
  --accept application/json \
  --body '{
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 300,
    "messages": [
      {"role": "user", "content": "Explain what Amazon Bedrock is in 2 sentences for a beginner."}
    ]
  }' \
  output.json && cat output.json | python3 -m json.tool

You should see a response like:

json

{
  "content": [{"text": "Amazon Bedrock is an AWS service that gives developers easy access to powerful AI models from multiple companies through a simple API. It handles all the infrastructure complexity so you can focus on building applications rather than managing AI model deployments."}],
  "usage": {"input_tokens": 25, "output_tokens": 47}
}

Your First Bedrock Project: AI Text Analyser

python

# bedrock_demo.py
import boto3
import json

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

def analyse_text(text: str) -> dict:
    """
    Analyse any text and return sentiment, key topics, and a one-line summary.
    Great for analysing customer feedback, reviews, or documents.
    """
    prompt = f"""Analyse this text and return ONLY valid JSON:
{{
  "sentiment": "POSITIVE|NEUTRAL|NEGATIVE",
  "confidence": 0.0-1.0,
  "keyTopics": ["topic1", "topic2", "topic3"],
  "summary": "one sentence summary",
  "wordCount": number
}}

Text to analyse:
{text}"""

    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-haiku-20240307-v1:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 400,
            "temperature": 0.1,
            "messages": [{"role": "user", "content": prompt}]
        })
    )

    result = json.loads(response['body'].read())
    content = result['content'][0]['text'].replace('```json', '').replace('```', '').strip()

    return {
        'analysis': json.loads(content),
        'tokensUsed': result['usage']['input_tokens'] + result['usage']['output_tokens'],
        'estimatedCostUsd': round(
            (result['usage']['input_tokens'] / 1000 * 0.00025) +
            (result['usage']['output_tokens'] / 1000 * 0.00125), 8
        )
    }

# Test it
sample_text = """
The new product update is fantastic! The interface is so much cleaner
and the performance improvements are noticeable. The team did a great
job. Only small issue is the export feature is a bit slower than before,
but overall a massive improvement.
"""

result = analyse_text(sample_text)
print(json.dumps(result, indent=2))

Run it:

bash

pip install boto3
python3 bedrock_demo.py

What this teaches you: How Bedrock works, how to structure prompts for JSON output, and how to calculate costs. These patterns appear in every AI application you will build.

Service 2: AWS Lambda, The Serverless Execution Engine

What It Is

AWS Lambda runs your code without you having to manage any servers. You write a function. AWS runs it when triggered. You pay only for the milliseconds your code actually runs.

Before serverless, deploying backend code meant:

• Provisioning an EC2 instance
• Installing the runtime
• Configuring security groups
• Setting up auto-scaling
• Managing operating system patches
• Paying for the server 24/7 even when nobody is using it

Lambda eliminates all of this. You write the function, deploy it, and Lambda handles everything else. It scales automatically, from zero requests to a million requests without any configuration changes.

Key Lambda Concepts Every Developer Needs

The Handler Function

Every Lambda function has a handler, the entry point that receives the event and returns a response:

python

def lambda_handler(event, context):
    # event: the input (from API Gateway, S3 event, schedule, etc.)
    # context: metadata about the invocation (request ID, remaining time, etc.)

    # Your logic here

    return {
        'statusCode': 200,
        'body': 'Hello from Lambda'
    }

Cold Starts vs Warm Starts

When Lambda has not been invoked recently, the first invocation takes longer (100ms-1s extra) while AWS initialises the execution environment. This is called a cold start. Subsequent invocations on the same execution environment are warm and fast.

For AI applications: put your AWS client initialisations (boto3 clients, database connections) outside the handler function. They persist across warm invocations:

python

#  CORRECT — initialised once, reused across warm invocations
import boto3
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')

def lambda_handler(event, context):
    # bedrock and dynamodb are already initialised — fast!
    response = bedrock.invoke_model(...)
    return {'statusCode': 200}

# WRONG — re-initialised on every invocation
def lambda_handler(event, context):
    bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')  # Slow!
    response = bedrock.invoke_model(...)
    return {'statusCode': 200}

Memory and Timeout Settings

Lambda memory (128MB–10GB) also controls CPU allocation. More memory = more CPU = faster execution. For AI applications calling Bedrock:

• Memory: 512MB is a good starting point
• Timeout: 30 seconds (Bedrock can take 10-20 seconds for complex prompts)

What Lambda Costs

Lambda pricing has two components:

• Requests: $0.20 per million invocations (first 1 million free per month)
• Duration: $0.0000166667 per GB-second

A Lambda with 512MB running 1,000ms (1 second) costs $0.0000083 per invocation. At 100,000 invocations per month: $0.83. Essentially free for most applications.

Deploy Your First Lambda

bash

REGION="eu-west-1"
ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)

# Create execution role
cat > lambda-trust.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "lambda.amazonaws.com"},
    "Action": "sts:AssumeRole"
  }]
}
EOF

aws iam create-role \
  --role-name my-first-lambda-role \
  --assume-role-policy-document file://lambda-trust.json

# Attach basic execution policy (CloudWatch logging)
aws iam attach-role-policy \
  --role-name my-first-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

# Add Bedrock permission
cat > bedrock-policy.json << EOF
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": ["bedrock:InvokeModel"],
    "Resource": "arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-haiku-20240307-v1:0"
  }]
}
EOF

aws iam put-role-policy \
  --role-name my-first-lambda-role \
  --policy-name bedrock-access \
  --policy-document file://bedrock-policy.json

LAMBDA_ROLE_ARN=$(aws iam get-role \
  --role-name my-first-lambda-role \
  --query 'Role.Arn' --output text)

# Wait for IAM to propagate
sleep 10

# Write the Lambda function
cat > function.py << 'EOF'
import boto3
import json

# Initialised outside handler — reused across warm invocations
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

def lambda_handler(event, context):
    body = json.loads(event.get('body', '{}'))
    text = body.get('text', '')

    if not text:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'text is required'})
        }

    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-haiku-20240307-v1:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 200,
            "messages": [{"role": "user", "content": f"Summarise this in one sentence: {text}"}]
        })
    )

    result = json.loads(response['body'].read())
    summary = result['content'][0]['text']

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({'summary': summary})
    }
EOF

zip function.zip function.py

# Deploy
aws lambda create-function \
  --function-name my-first-ai-lambda \
  --runtime python3.12 \
  --role $LAMBDA_ROLE_ARN \
  --handler function.lambda_handler \
  --zip-file fileb://function.zip \
  --timeout 30 \
  --memory-size 512 \
  --region $REGION

echo "Lambda deployed"

# Test it directly
aws lambda invoke \
  --function-name my-first-ai-lambda \
  --payload '{"body": "{\"text\": \"Amazon Web Services is a subsidiary of Amazon that provides on-demand cloud computing platforms and APIs to individuals, companies, and governments.\"}"}' \
  --cli-binary-format raw-in-base64-out \
  response.json \
  --region $REGION

cat response.json

What this teaches you: The Lambda lifecycle, cold start optimisation, the handler pattern, and how to deploy and test a function. Every AI application you build will have Lambda functions following this pattern.

Service 3: Amazon API Gateway, The Front Door

What It Is

API Gateway creates HTTP endpoints that trigger your Lambda functions. Without API Gateway, your Lambda function has no URL, there is no way for a web browser, mobile app, or another service to call it over the internet.

API Gateway sits in front of your Lambda and:

• Gives it an HTTPS URL
• Handles authentication (API keys, JWT tokens)
• Enforces rate limits (how many requests per second)
• Logs all requests
• Handles CORS (Cross-Origin Resource Sharing, necessary for web browsers)

HTTP API vs REST API

AWS offers two types of API Gateway. For most AI applications, use HTTP API, it is simpler, faster, and cheaper:

Rule of thumb: Start with HTTP API. Switch to REST API only if you need built-in API key management.

Create an API Gateway for Your Lambda

bash

# Create HTTP API
API_ID=$(aws apigatewayv2 create-api \
  --name "my-ai-api" \
  --protocol-type HTTP \
  --cors-configuration \
    AllowOrigins='*' \
    AllowHeaders='Content-Type' \
    AllowMethods='POST,GET,OPTIONS' \
  --region $REGION \
  --query 'ApiId' \
  --output text)

echo "API ID: $API_ID"

# Connect API Gateway to your Lambda
INTEGRATION_ID=$(aws apigatewayv2 create-integration \
  --api-id $API_ID \
  --integration-type AWS_PROXY \
  --integration-uri arn:aws:lambda:$REGION:$ACCOUNT_ID:function:my-first-ai-lambda \
  --payload-format-version 2.0 \
  --region $REGION \
  --query 'IntegrationId' \
  --output text)

# Create a route — POST requests to /summarise call the Lambda
aws apigatewayv2 create-route \
  --api-id $API_ID \
  --route-key 'POST /summarise' \
  --target integrations/$INTEGRATION_ID \
  --region $REGION

# Deploy the API
aws apigatewayv2 create-stage \
  --api-id $API_ID \
  --stage-name production \
  --auto-deploy \
  --region $REGION

# Allow API Gateway to invoke the Lambda function
aws lambda add-permission \
  --function-name my-first-ai-lambda \
  --statement-id allow-api-gateway \
  --action lambda:InvokeFunction \
  --principal apigateway.amazonaws.com \
  --source-arn "arn:aws:execute-api:$REGION:$ACCOUNT_ID:$API_ID/*/*" \
  --region $REGION

# Get the URL
API_URL=$(aws apigatewayv2 get-api \
  --api-id $API_ID \
  --region $REGION \
  --query 'ApiEndpoint' \
  --output text)

echo ""
echo "Your API is live at: $API_URL/production/summarise"
echo ""

# Test it with curl
curl -X POST $API_URL/production/summarise \
  -H "Content-Type: application/json" \
  -d '{"text": "AWS Lambda is a serverless computing service that runs code without provisioning or managing servers, automatically scaling from zero to thousands of concurrent executions."}' \
  | python3 -m json.tool

Understanding the Request Flow

When a user calls your API, here is exactly what happens step by step:

1. User sends POST request to:
   https://abc123.execute-api.eu-west-1.amazonaws.com/production/summarise

2. API Gateway receives the request
   - Checks the route matches (POST /summarise → yes)
   - Applies any auth checks (none configured yet)
   - Applies rate limits (none configured yet)
   - Logs the request to CloudWatch

3. API Gateway calls your Lambda function with an event object:
   {
     "requestContext": {"http": {"method": "POST"}},
     "headers": {"content-type": "application/json"},
     "body": "{\"text\": \"AWS Lambda is...\"}",
     "rawPath": "/summarise"
   }

4. Lambda runs your handler function
   - Parses the body
   - Calls Bedrock
   - Returns a response dict

5. API Gateway receives the Lambda response
   - Returns it to the user as an HTTP response

Total time: typically 1-3 seconds for Bedrock calls

What this teaches you: The request/response flow that underpins every web-facing AI application. Understanding this flow is essential for debugging when things go wrong.

Service 4: Amazon DynamoDB, The Fast, Flexible Database

What It Is

DynamoDB is AWS's managed NoSQL database. It is designed for applications that need:

• Single-digit millisecond response times at any scale
• No servers to manage
• Pay-per-use pricing (you pay for reads and writes, not for capacity sitting idle)

For AI applications, DynamoDB is used for:

• Conversation history: storing multi-turn chat context
• Usage tracking: counting tokens per user for rate limiting
• Caching: storing previous AI responses to avoid re-calling Bedrock for identical inputs
• Results storage: saving AI analysis results for later retrieval

Key DynamoDB Concepts

Tables, Items, and Attributes

DynamoDB organises data differently from SQL databases:

SQL Database:
  TABLE: users
  COLUMNS: id, name, email, created_at
  ROW: 1, "Emma", "[email protected]", "2026-01-01"

DynamoDB:
  TABLE: users
  ITEM: {"userId": "u001", "name": "Emma", "email": "[email protected]",
         "createdAt": "2026-01-01", "preferences": {"theme": "dark"}}

Key difference: DynamoDB items can have different attributes.
Item 1 might have "preferences". Item 2 might not. That's fine.

Primary Key: The Most Important Concept

Every DynamoDB table has a primary key. Items are stored and retrieved by this key. Getting this right is the single most important DynamoDB design decision.

• Partition key only (simple): Good when you always look up items by one attribute. E.g., look up users by userId.
• Partition key + Sort key (composite): Good when you need to store multiple related items and query them together. E.g., look up all messages for a sessionId, sorted by timestamp.

For AI conversation history: sessionId (partition) + timestamp (sort) is the standard pattern.

On-Demand vs Provisioned Capacity

Use On-Demand (pay per read/write) unless you have predictable, high-volume traffic. On-demand is more expensive per operation but requires zero capacity planning and never throttles you unexpectedly.

Create a Conversation History Table

bash

# Table for storing AI conversation history
# Partition key: sessionId (groups all messages in a conversation)
# Sort key: timestamp (orders messages chronologically within a session)
aws dynamodb create-table \
  --table-name ai-conversations \
  --attribute-definitions \
    AttributeName=sessionId,AttributeType=S \
    AttributeName=timestamp,AttributeType=S \
  --key-schema \
    AttributeName=sessionId,KeyType=HASH \
    AttributeName=timestamp,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --region $REGION

# Wait for table to be active
aws dynamodb wait table-exists --table-name ai-conversations --region $REGION

echo "Table created"

# Add some test data
aws dynamodb put-item \
  --table-name ai-conversations \
  --item '{
    "sessionId": {"S": "session-001"},
    "timestamp": {"S": "2026-06-15T10:00:00Z"},
    "role": {"S": "user"},
    "content": {"S": "What is Amazon Bedrock?"},
    "tokenCount": {"N": "8"}
  }' \
  --region $REGION

aws dynamodb put-item \
  --table-name ai-conversations \
  --item '{
    "sessionId": {"S": "session-001"},
    "timestamp": {"S": "2026-06-15T10:00:05Z"},
    "role": {"S": "assistant"},
    "content": {"S": "Amazon Bedrock is an AWS service that provides access to foundation AI models via API."},
    "tokenCount": {"N": "23"}
  }' \
  --region $REGION

# Retrieve the conversation
aws dynamodb query \
  --table-name ai-conversations \
  --key-condition-expression 'sessionId = :sid' \
  --expression-attribute-values '{":sid": {"S": "session-001"}}' \
  --region $REGION | python3 -m json.tool

Using DynamoDB for Conversation Memory in Lambda

python

# conversation_lambda.py
# A conversational AI that remembers what was said earlier in the session

import boto3
import json
import time
from datetime import datetime, timezone

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')
table = dynamodb.Table('ai-conversations')

SYSTEM_PROMPT = "You are a helpful assistant. Be concise and friendly."
MAX_HISTORY_MESSAGES = 10  # Keep last 10 messages for context


def get_conversation_history(session_id: str) -> list:
    """Retrieve the last N messages for a session"""
    response = table.query(
        KeyConditionExpression=boto3.dynamodb.conditions.Key('sessionId').eq(session_id),
        ScanIndexForward=True,  # oldest first
        Limit=MAX_HISTORY_MESSAGES
    )
    items = response.get('Items', [])

    # Convert DynamoDB items to Bedrock message format
    return [
        {"role": item['role'], "content": item['content']}
        for item in items
    ]


def save_message(session_id: str, role: str, content: str, token_count: int = 0):
    """Save a message to DynamoDB"""
    table.put_item(Item={
        'sessionId': session_id,
        'timestamp': datetime.now(timezone.utc).isoformat(),
        'role': role,
        'content': content,
        'tokenCount': token_count
    })


def lambda_handler(event, context):
    body = json.loads(event.get('body', '{}'))
    session_id = body.get('sessionId', 'default-session')
    user_message = body.get('message', '').strip()

    if not user_message:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'message is required'})
        }

    # Get conversation history from DynamoDB
    history = get_conversation_history(session_id)

    # Add the new user message to history
    messages = history + [{"role": "user", "content": user_message}]

    # Call Bedrock with the full conversation context
    response = bedrock.invoke_model(
        modelId='anthropic.claude-3-haiku-20240307-v1:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 500,
            "system": SYSTEM_PROMPT,
            "messages": messages
        })
    )

    result = json.loads(response['body'].read())
    assistant_reply = result['content'][0]['text']
    input_tokens = result['usage']['input_tokens']
    output_tokens = result['usage']['output_tokens']

    # Save both messages to DynamoDB for future context
    save_message(session_id, 'user', user_message, input_tokens)
    save_message(session_id, 'assistant', assistant_reply, output_tokens)

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'reply': assistant_reply,
            'sessionId': session_id,
            'usage': {
                'inputTokens': input_tokens,
                'outputTokens': output_tokens
            }
        })
    }

Test a multi-turn conversation:

bash

# Turn 1
curl -X POST $API_URL/production/chat \
  -H "Content-Type: application/json" \
  -d '{"sessionId": "test-session-001", "message": "My name is Emma ,and I am a cloud architect."}'

# Turn 2 — the AI should remember your name
curl -X POST $API_URL/production/chat \
  -H "Content-Type: application/json" \
  -d '{"sessionId": "test-session-001", "message": "What is my name and job title?"}'

The AI remembers your name because DynamoDB stores the conversation history, which is included in every subsequent request. Without DynamoDB, the AI would start fresh every time and forget everything.

What this teaches you: How to give AI memory using DynamoDB, the conversation context pattern, and the query-by-partition-key pattern you will use constantly in AI applications.

Service 5: Amazon CloudWatch, The Observability Layer

What It Is

CloudWatch is AWS's monitoring, logging, and alerting service. When your AI application is running in production, CloudWatch tells you:

• What is happening right now (metrics)
• What happened in the past (logs)
• When something goes wrong (alarms)

Without CloudWatch, your AI application is a black box. You will not know it has a problem until a user complains. With CloudWatch, you will know before users notice.

The Three CloudWatch Concepts You Need

1. Logs: What Happened

Every Lambda function automatically sends its output to CloudWatch Logs. Every print() statement in your Lambda appears in CloudWatch. Every error. Every exception stack trace.

python

# In your Lambda function
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    logger.info(f"Processing request: {context.aws_request_id}")

    try:
        result = call_bedrock(...)
        logger.info(f"Bedrock call succeeded: {result['totalTokens']} tokens")
        return success_response(result)

    except Exception as e:
        # This appears in CloudWatch with full stack trace
        logger.error(f"Bedrock call failed: {str(e)}", exc_info=True)
        return error_response(500, "Internal error")

Pro tip: use structured JSON logging. It makes CloudWatch Logs Insights queries possible:

python

import json

def log(level: str, event: str, **kwargs):
    """Write structured JSON log entry"""
    print(json.dumps({
        "level": level,
        "event": event,
        **kwargs
    }))

# Usage in your Lambda
log("INFO", "bedrock_call_complete",
    tokens=287,
    latency_ms=1842,
    cost_usd=0.00000897)

Now you can query your logs like a database:

sql

-- Find all requests that cost more than $0.01
fields @timestamp, cost_usd, tokens
| filter event = "bedrock_call_complete" and cost_usd > 0.01
| sort cost_usd desc

2. Metrics: How Things Are Performing

CloudWatch Metrics are numerical data points over time. Lambda automatically creates metrics for you: invocation count, error count, duration, throttles.

You can also create custom metrics for your AI-specific measurements:

python

cloudwatch = boto3.client('cloudwatch', region_name='eu-west-1')

def publish_ai_metrics(tokens: int, cost: float, latency: int, success: bool):
    """Publish custom metrics to CloudWatch"""
    cloudwatch.put_metric_data(
        Namespace='MyAIApp/Metrics',
        MetricData=[
            {
                'MetricName': 'TokensConsumed',
                'Value': tokens,
                'Unit': 'Count'
            },
            {
                'MetricName': 'EstimatedCostUSD',
                'Value': cost,
                'Unit': 'None'
            },
            {
                'MetricName': 'BedrockLatencyMs',
                'Value': latency,
                'Unit': 'Milliseconds'
            },
            {
                'MetricName': 'SuccessfulRequests' if success else 'FailedRequests',
                'Value': 1,
                'Unit': 'Count'
            }
        ]
    )

3. Alarms: Know Before Your Users Do

Alarms watch a metric and notify you when it crosses a threshold:

bash

# Alarm: Lambda error rate > 5% over 5 minutes
aws cloudwatch put-metric-alarm \
  --alarm-name "AI-App-High-Error-Rate" \
  --alarm-description "More than 5% of requests are failing" \
  --namespace AWS/Lambda \
  --metric-name Errors \
  --dimensions Name=FunctionName,Value=my-first-ai-lambda \
  --statistic Sum \
  --period 300 \
  --evaluation-periods 1 \
  --threshold 5 \
  --comparison-operator GreaterThanThreshold \
  --alarm-actions arn:aws:sns:$REGION:$ACCOUNT_ID:my-alert-topic \
  --region $REGION

# Alarm: Lambda duration P95 > 20 seconds (approaching timeout)
aws cloudwatch put-metric-alarm \
  --alarm-name "AI-App-High-Latency" \
  --alarm-description "P95 Lambda duration is approaching timeout" \
  --namespace AWS/Lambda \
  --metric-name Duration \
  --dimensions Name=FunctionName,Value=my-first-ai-lambda \
  --extended-statistic p95 \
  --period 300 \
  --evaluation-periods 2 \
  --threshold 20000 \
  --comparison-operator GreaterThanThreshold \
  --alarm-actions arn:aws:sns:$REGION:$ACCOUNT_ID:my-alert-topic \
  --region $REGION

# Alarm: Daily AI cost > $10
aws cloudwatch put-metric-alarm \
  --alarm-name "AI-App-Daily-Cost-High" \
  --alarm-description "Daily AI spend exceeded $10" \
  --namespace MyAIApp/Metrics \
  --metric-name EstimatedCostUSD \
  --statistic Sum \
  --period 86400 \
  --evaluation-periods 1 \
  --threshold 10 \
  --comparison-operator GreaterThanThreshold \
  --alarm-actions arn:aws:sns:$REGION:$ACCOUNT_ID:my-alert-topic \
  --region $REGION

echo "Alarms created"

Powerful CloudWatch Logs Insights Queries

Once your Lambda is writing structured logs, these queries become your debugging superpower:

sql

-- Average Bedrock latency over time
fields @timestamp, latency_ms
| filter event = "bedrock_call_complete"
| stats avg(latency_ms) as avg_latency by bin(1h)
| sort @timestamp asc

-- Top 10 most expensive requests today
fields @timestamp, cost_usd, tokens, request_id
| filter event = "bedrock_call_complete"
| filter @timestamp > ago(1d)
| sort cost_usd desc
| limit 10

-- Error breakdown by error type
fields @timestamp, error_type, error_message
| filter level = "ERROR"
| stats count(*) as error_count by error_type
| sort error_count desc

-- Success rate by hour
fields @timestamp
| filter event = "bedrock_call_complete" or event = "bedrock_call_failed"
| stats count(*) as total,
        sum(event = "bedrock_call_complete") as successes
  by bin(1h)
| sort @timestamp asc

The Final Project: All 5 Services Together

Now we build something that uses all five services in a real, complete application, a multi-turn AI assistant with memory, rate limiting, cost tracking, and monitoring.

python

# complete_ai_assistant.py
# Uses all 5 services: Bedrock, Lambda, DynamoDB, CloudWatch
# (API Gateway is the trigger — configured separately)

import boto3
import json
import time
import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ── Service 1: Amazon Bedrock ──────────────────
bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

# ── Service 4: Amazon DynamoDB ─────────────────
dynamodb = boto3.resource('dynamodb', region_name='eu-west-1')
conversations = dynamodb.Table('ai-conversations')
usage_tracking = dynamodb.Table('ai-usage')

# ── Service 5: Amazon CloudWatch ───────────────
cloudwatch = boto3.client('cloudwatch', region_name='eu-west-1')

MODEL_ID = 'anthropic.claude-3-haiku-20240307-v1:0'
DAILY_TOKEN_LIMIT = 100_000
MAX_HISTORY = 10


def structured_log(level: str, event: str, **kwargs):
    """Structured JSON logging for CloudWatch Logs Insights"""
    print(json.dumps({
        "level": level,
        "event": event,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **kwargs
    }))


def get_history(session_id: str) -> list:
    """Fetch conversation history from DynamoDB"""
    response = conversations.query(
        KeyConditionExpression=boto3.dynamodb.conditions.Key('sessionId').eq(session_id),
        ScanIndexForward=True,
        Limit=MAX_HISTORY
    )
    return [
        {"role": item['role'], "content": item['content']}
        for item in response.get('Items', [])
    ]


def save_messages(session_id: str, user_msg: str, assistant_msg: str):
    """Save both messages to DynamoDB"""
    now = datetime.now(timezone.utc).isoformat()
    ts_user = now
    ts_assistant = now.replace('Z', '.001Z') if 'Z' in now else now + '.001'

    with conversations.batch_writer() as batch:
        batch.put_item(Item={
            'sessionId': session_id,
            'timestamp': ts_user,
            'role': 'user',
            'content': user_msg
        })
        batch.put_item(Item={
            'sessionId': session_id,
            'timestamp': ts_assistant,
            'role': 'assistant',
            'content': assistant_msg
        })


def check_rate_limit(session_id: str, tokens_needed: int) -> bool:
    """Check and update daily token usage. Returns True if allowed."""
    today = datetime.now(timezone.utc).strftime('%Y-%m-%d')
    try:
        usage_tracking.update_item(
            Key={'sessionId': session_id, 'date': today},
            UpdateExpression='ADD tokensUsed :t',
            ConditionExpression='attribute_not_exists(tokensUsed) OR tokensUsed < :limit',
            ExpressionAttributeValues={':t': tokens_needed, ':limit': DAILY_TOKEN_LIMIT}
        )
        return True
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        return False


def publish_metrics(tokens: int, cost: float, latency: int, success: bool):
    """Publish metrics to CloudWatch"""
    try:
        cloudwatch.put_metric_data(
            Namespace='AIAssistant/Metrics',
            MetricData=[
                {'MetricName': 'TokensConsumed', 'Value': tokens, 'Unit': 'Count'},
                {'MetricName': 'EstimatedCostUSD', 'Value': cost, 'Unit': 'None'},
                {'MetricName': 'BedrockLatencyMs', 'Value': latency, 'Unit': 'Milliseconds'},
                {
                    'MetricName': 'SuccessfulRequests' if success else 'FailedRequests',
                    'Value': 1,
                    'Unit': 'Count'
                }
            ]
        )
    except Exception:
        pass  # Never fail the main request for metrics


def lambda_handler(event, context):
    """
    ── Service 3: API Gateway triggers this ──
    Receives POST /chat with {sessionId, message}
    Returns AI response with usage info
    """
    request_id = context.aws_request_id

    # Handle CORS
    if event.get('requestContext', {}).get('http', {}).get('method') == 'OPTIONS':
        return {'statusCode': 200,
                'headers': {'Access-Control-Allow-Origin': '*',
                            'Access-Control-Allow-Headers': 'Content-Type',
                            'Access-Control-Allow-Methods': 'POST,OPTIONS'},
                'body': ''}

    # Parse input
    try:
        body = json.loads(event.get('body', '{}'))
    except json.JSONDecodeError:
        return {'statusCode': 400,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': 'Invalid JSON'})}

    session_id = body.get('sessionId', 'default')
    user_message = body.get('message', '').strip()

    if not user_message or len(user_message) < 2:
        return {'statusCode': 400,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': 'message must be at least 2 characters'})}

    if len(user_message) > 2000:
        return {'statusCode': 400,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': 'message must not exceed 2000 characters'})}

    structured_log("INFO", "request_received",
                   request_id=request_id,
                   session_id=session_id,
                   message_length=len(user_message))

    # ── Service 4: DynamoDB — Check rate limit ──
    estimated_tokens = len(user_message) // 4 + 500
    if not check_rate_limit(session_id, estimated_tokens):
        structured_log("WARN", "rate_limit_exceeded", session_id=session_id)
        return {'statusCode': 429,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': f'Daily limit of {DAILY_TOKEN_LIMIT:,} tokens exceeded. Resets at midnight UTC.'})}

    # ── Service 4: DynamoDB — Load conversation history ──
    history = get_history(session_id)
    messages = history + [{"role": "user", "content": user_message}]

    # ── Service 1: Amazon Bedrock — Call Claude ──
    try:
        bedrock_start = time.time()

        response = bedrock.invoke_model(
            modelId=MODEL_ID,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 800,
                "temperature": 0.7,
                "system": "You are a helpful, knowledgeable assistant. Give clear, concise answers.",
                "messages": messages
            })
        )

        latency_ms = int((time.time() - bedrock_start) * 1000)
        result = json.loads(response['body'].read())
        reply = result['content'][0]['text']
        input_tokens = result['usage']['input_tokens']
        output_tokens = result['usage']['output_tokens']
        total_tokens = input_tokens + output_tokens
        cost_usd = round(
            (input_tokens / 1000 * 0.00025) + (output_tokens / 1000 * 0.00125), 8
        )

        structured_log("INFO", "bedrock_call_complete",
                       request_id=request_id,
                       session_id=session_id,
                       input_tokens=input_tokens,
                       output_tokens=output_tokens,
                       latency_ms=latency_ms,
                       cost_usd=cost_usd)

        # ── Service 4: DynamoDB — Save conversation ──
        save_messages(session_id, user_message, reply)

        # ── Service 5: CloudWatch — Publish metrics ──
        publish_metrics(total_tokens, cost_usd, latency_ms, True)

        # ── Service 3: API Gateway — Return response ──
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*',
                'X-Request-ID': request_id
            },
            'body': json.dumps({
                'reply': reply,
                'sessionId': session_id,
                'usage': {
                    'inputTokens': input_tokens,
                    'outputTokens': output_tokens,
                    'estimatedCostUsd': cost_usd,
                    'latencyMs': latency_ms
                }
            })
        }

    except bedrock.exceptions.ThrottlingException:
        publish_metrics(0, 0, 0, False)
        return {'statusCode': 503,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': 'AI service busy. Please retry in 30 seconds.'})}

    except Exception as e:
        structured_log("ERROR", "request_failed",
                       request_id=request_id,
                       error_type=type(e).__name__,
                       error=str(e))
        publish_metrics(0, 0, 0, False)
        return {'statusCode': 500,
                'headers': {'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*'},
                'body': json.dumps({'error': 'Something went wrong. Please try again.'})}

Deploy and test:

bash

# Create usage tracking table
aws dynamodb create-table \
  --table-name ai-usage \
  --attribute-definitions \
    AttributeName=sessionId,AttributeType=S \
    AttributeName=date,AttributeType=S \
  --key-schema \
    AttributeName=sessionId,KeyType=HASH \
    AttributeName=date,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --region $REGION

# Deploy the complete assistant
zip complete-assistant.zip complete_ai_assistant.py

aws lambda update-function-code \
  --function-name my-first-ai-lambda \
  --zip-file fileb://complete-assistant.zip \
  --region $REGION

# Add DynamoDB permissions to Lambda role
cat > dynamodb-policy.json << EOF
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "dynamodb:GetItem",
      "dynamodb:PutItem",
      "dynamodb:UpdateItem",
      "dynamodb:Query",
      "dynamodb:BatchWriteItem"
    ],
    "Resource": [
      "arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/ai-conversations",
      "arn:aws:dynamodb:$REGION:$ACCOUNT_ID:table/ai-usage"
    ]
  }]
}
EOF

aws iam put-role-policy \
  --role-name my-first-lambda-role \
  --policy-name dynamodb-access \
  --policy-document file://dynamodb-policy.json

# Test a multi-turn conversation
SESSION="test-$(date +%s)"

echo "=== Turn 1 ==="
curl -s -X POST $API_URL/production/summarise \
  -H "Content-Type: application/json" \
  -d "{\"sessionId\": \"$SESSION\", \"message\": \"I am building a startup that helps restaurants manage inventory using AI. Our main challenge is predicting demand accurately.\"}" \
  | python3 -c "import json,sys; d=json.load(sys.stdin); print('Reply:', d['reply'][:200])"

echo ""
echo "=== Turn 2 — referencing Turn 1 context ==="
curl -s -X POST $API_URL/production/summarise \
  -H "Content-Type: application/json" \
  -d "{\"sessionId\": \"$SESSION\", \"message\": \"What AWS services would you recommend for the challenge I just described?\"}" \
  | python3 -c "import json,sys; d=json.load(sys.stdin); print('Reply:', d['reply'][:300])"

The second response will reference the restaurant inventory startup context from Turn 1, because DynamoDB stored it and it was included in the second Bedrock call.

How These 5 Services Work Together, The Full Picture

USER
  │
  │ POST /chat {"sessionId": "abc", "message": "Hello"}
  ▼
SERVICE 3: API GATEWAY
  │ Receives HTTPS request
  │ Routes to Lambda
  │ Logs request
  ▼
SERVICE 2: AWS LAMBDA
  │ Parses request
  │
  ├─── SERVICE 4: DYNAMODB
  │    │ Checks rate limit
  │    │ Loads conversation history
  │
  ├─── SERVICE 1: AMAZON BEDROCK
  │    │ Sends messages + history to Claude
  │    │ Receives AI response
  │
  ├─── SERVICE 4: DYNAMODB
  │    │ Saves user message + AI reply
  │
  ├─── SERVICE 5: CLOUDWATCH
  │    │ Publishes cost/latency metrics
  │    │ Writes structured logs
  │
  │ Returns JSON response to API Gateway
  ▼
SERVICE 3: API GATEWAY
  │ Returns response to user
  ▼
USER receives: {reply, usage, sessionId}

Every production AI application you build will follow this pattern. The services change, the business logic changes, the prompt changes, but the flow is the same.

What to Learn Next

You now understand the five core services. Here is the natural progression:

Deepen your Bedrock knowledge: Learn Bedrock Knowledge Bases (RAG) from Article 1 of this series to connect AI to your own documents.

Add security to your APIs: Learn AWS Secrets Manager (for storing API keys safely) and AWS WAF (for blocking malicious requests at the API Gateway level).

Scale your database: Learn DynamoDB Global Tables (replicate across regions) and DynamoDB Streams (trigger Lambda when data changes).

Improve observability: Learn CloudWatch Container Insights (for EKS/ECS workloads) and X-Ray (distributed tracing across Lambda + Bedrock calls).

Certify your knowledge: The AWS Solutions Architect Associate (SAA-C03) certification validates your understanding of all five services plus the broader AWS ecosystem. It is the most valuable AWS certification for developers building on the platform.