The 5 AWS Services Every Developer Must Know to Build AI Products in 2026 (With Real Projects)

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: Amazon Bedrock : the AI brain AWS Lambda : the serverless execution engine Amazon API Gateway : the front door Amazon DynamoDB : the fast, flexible database 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: Model Input (per 1K tokens) Output (per 1K tokens) Claude 3 Haiku $0.00025 $0.00125 Claude 3.5 Haiku $0.0008 $0.004 Claude 3.5 Sonnet $0.003 $0.015 Amazon Titan Text Lite $0.00015 $0.0002 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: Feature HTTP API REST API Cost $1/million requests $3.50/million requests Latency Lower Higher API key auth ❌ ✅ JWT auth ✅ ✅ Request/response transformation Limited Full Best for Most AI APIs Enterprise API management 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.