8. Project: S3+Lambda AI Service

Project description

This is the integrator project of Module 5. You'll build a complete AI service that uses S3 as the data layer and Lambda as the compute layer. The service stores AI assets in S3 (versioned prompt templates, RAG documents, configuration), uses Lambda to read those assets, build contextualized prompts, invoke an LLM, and persist the responses for auditing. It includes IAM with least privilege, S3 event triggers, and documented cost estimation.

Why it matters: This service integrates everything you learned in the module's capsules: S3 as storage for AI (C02), Lambda for inference (C03), event-driven integration (C04), the SageMaker vs Lambda decision framework (C05), IAM least privilege (C06), and cost estimation (C07). It's a functional artifact that proves you know how to build a serverless AI service on AWS. In Module 6, you'll abstract this code so it works against LocalStack AND AWS with the same codebase — the foundation of your migration path.

---

Project objective

Produce a functional S3+Lambda AI Service that:

1. Stores versioned prompt templates in S3 (prompts/{name}/{version}/system.txt)
2. Stores RAG documents in S3 (documents/{collection}/{doc_id}.json)
3. Receives requests via API Gateway (POST /process) with a document and a prompt to use
4. Lambda reads the prompt template from S3, builds the prompt with the document, invokes the LLM
5. Persists the responses in S3 (responses/{date}/{request_id}.json) for auditing
6. Processes documents automatically via an S3 trigger (upload to documents/inbox/)
7. Has a health endpoint (GET /health) that verifies S3 connectivity and configuration
8. IAM configured with least privilege (Lambda only accesses the necessary prefixes)
9. Cost estimation documented for three usage scenarios

---

Module Recap

Capsule	Concept	How you use it in the project
02	S3 for AI Assets	Store prompts, documents, responses in S3
03	Lambda for Inference	Handler with retry, structured logging, error handling
04	S3+Lambda Integration	S3 trigger → Lambda processes → writes to S3
05	SageMaker Basics	Justification of why you use Lambda (not SageMaker)
06	IAM Least Privilege	Least-privilege roles and policies
07	Cost Estimation	Cost estimation for the service

---

Technical Specifications

Architecture

                       ┌──────────────────────────────────┐
  POST /process ──────→│                                  │
  GET /health ────────→│  API Gateway (HTTP API v2)       │
                       │  CORS + Throttling               │
                       └──────────┬───────────────────────┘
                                  │
                       ┌──────────┴───────────────────────┐
                       │   Lambda: ai-service-process      │
                       │   768MB / 120s / arm64            │
                       │                                   │
                       │   1. Parse request                │
                       │   2. Read prompt template (S3)    │
                       │   3. Read/receive document        │
                       │   4. Build contextual prompt      │
                       │   5. Invoke LLM (with retry)     │
                       │   6. Save response to S3          │
                       │   7. Return result                │
                       └──────────┬───────────────────────┘
                                  │
                  ┌───────────────┼───────────────┐
                  │               │               │
           ┌──────┴──────┐  ┌────┴───────┐  ┌────┴──────┐
           │  S3 Bucket  │  │  OpenAI    │  │ CloudWatch│
           │  ai-service │  │  API       │  │ Logs      │
           │  -assets    │  │            │  │           │
           └─────────────┘  └────────────┘  └───────────┘

                    S3 Bucket Structure:
                    ai-service-assets-{account}/
                    ├── prompts/
                    │   ├── summarizer/v1/system.txt
                    │   ├── summarizer/v2/system.txt
                    │   └── classifier/v1/system.txt
                    ├── documents/
                    │   ├── inbox/          ← S3 trigger
                    │   └── knowledge-base/
                    ├── responses/
                    │   └── 2026/03/08/
                    └── config/
                        └── service-config.json

Required endpoints

POST /process  → Processes a document with a prompt template
GET  /health   → Service status, S3 connectivity, configuration

Request/Response format

// POST /process — Request
{
  "prompt_name": "summarizer",
  "prompt_version": "v1",
  "document": {
    "id": "doc-001",
    "title": "Introduction to Serverless",
    "content": "Serverless computing lets you run code..."
  },
  "max_tokens": 500
}

// POST /process — Response (200)
{
  "request_id": "req-a1b2c3d4",
  "answer": "The key points of the document are...",
  "model": "gpt-4o-mini",
  "tokens_used": 342,
  "duration_ms": 3245,
  "prompt_used": "summarizer/v1",
  "response_key": "responses/2026/03/08/req-a1b2c3d4.json",
  "cold_start": false
}

// POST /process — Response (400)
{
  "error": "prompt_name is required"
}

// POST /process — Response (404)
{
  "error": "Prompt template 'analyzer/v3' not found in S3"
}

// GET /health — Response (200)
{
  "status": "healthy",
  "checks": {
    "lambda": "up",
    "s3_bucket": "accessible",
    "openai_key": "configured"
  },
  "config": {
    "bucket": "ai-service-assets-dev",
    "model": "gpt-4o-mini",
    "memory_mb": "768",
    "region": "us-east-1"
  },
  "prompt_templates": ["summarizer/v1", "summarizer/v2", "classifier/v1"]
}

Lambda configuration

Parameter	Process Function	Health Function
Runtime	Python 3.11	Python 3.11
Architecture	arm64	arm64
Memory	768 MB	128 MB
Timeout	120s	10s
Handler	process_handler.handler	health_handler.handler

---

Complete Code

src/process_handler.py

import json
import logging
import os
import time
import uuid
from datetime import datetime

import boto3
from openai import OpenAI, APITimeoutError, RateLimitError, APIConnectionError

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")
MODEL = os.environ.get("MODEL_NAME", "gpt-4o-mini")
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3"))

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", ""),
    max_retries=0,
)

CORS_HEADERS = {
    "Content-Type": "application/json",
    "Access-Control-Allow-Origin": os.environ.get("ALLOWED_ORIGIN", "*"),
    "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
}

IS_COLD_START = True
RETRYABLE = (APITimeoutError, RateLimitError, APIConnectionError)


def _response(status_code: int, body) -> dict:
    return {
        "statusCode": status_code,
        "headers": CORS_HEADERS,
        "body": json.dumps(body) if isinstance(body, dict) else body,
    }


def get_prompt_template(prompt_name: str, prompt_version: str) -> str:
    """Reads a prompt template from S3."""
    key = f"prompts/{prompt_name}/{prompt_version}/system.txt"
    try:
        response = s3.get_object(Bucket=BUCKET, Key=key)
        return response["Body"].read().decode("utf-8")
    except s3.exceptions.NoSuchKey:
        raise FileNotFoundError(f"Prompt template '{prompt_name}/{prompt_version}' not found in S3")


def get_document_from_s3(collection: str, doc_id: str) -> dict:
    """Reads a document from S3."""
    key = f"documents/{collection}/{doc_id}.json"
    try:
        response = s3.get_object(Bucket=BUCKET, Key=key)
        return json.loads(response["Body"].read().decode("utf-8"))
    except s3.exceptions.NoSuchKey:
        raise FileNotFoundError(f"Document '{collection}/{doc_id}' not found in S3")


def save_response(request_id: str, data: dict) -> str:
    """Saves the inference response to S3."""
    now = datetime.utcnow()
    key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
    payload = {
        "request_id": request_id,
        "timestamp": now.isoformat(),
        **data,
    }
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
    )
    return key


def invoke_llm(messages: list, max_tokens: int, context) -> dict:
    """Invokes an LLM with retry and exponential backoff."""
    last_error = None

    for attempt in range(1, MAX_RETRIES + 1):
        remaining_ms = context.get_remaining_time_in_millis()
        if remaining_ms < 15000:
            raise TimeoutError(f"Insufficient time: {remaining_ms}ms")

        try:
            start = time.time()
            response = openai_client.chat.completions.create(
                model=MODEL,
                messages=messages,
                max_tokens=max_tokens,
                timeout=(remaining_ms / 1000) - 10,
            )
            duration_ms = int((time.time() - start) * 1000)

            return {
                "content": response.choices[0].message.content,
                "model": response.model,
                "tokens_used": response.usage.total_tokens,
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens,
                "llm_duration_ms": duration_ms,
                "attempt": attempt,
            }

        except RETRYABLE as e:
            last_error = e
            if attempt < MAX_RETRIES:
                delay = 1.0 * (2 ** (attempt - 1))
                logger.warning(json.dumps({
                    "event": "llm_retry",
                    "attempt": attempt,
                    "error": str(e),
                    "delay_s": delay,
                }))
                time.sleep(delay)
        except Exception as e:
            raise

    raise last_error


def process_request(body: dict, context) -> dict:
    """Processes a complete inference request."""
    prompt_name = body.get("prompt_name", "").strip()
    prompt_version = body.get("prompt_version", "v1").strip()
    document = body.get("document", {})
    max_tokens = min(body.get("max_tokens", 500), 2000)

    if not prompt_name:
        raise ValueError("prompt_name is required")
    if not document or not document.get("content"):
        raise ValueError("document with 'content' field is required")

    prompt_template = get_prompt_template(prompt_name, prompt_version)

    doc_title = document.get("title", "Untitled")
    doc_content = document.get("content", "")

    full_prompt = f"{prompt_template}\n\nTitle: {doc_title}\n\nContent:\n{doc_content}"

    messages = [{"role": "user", "content": full_prompt}]
    llm_result = invoke_llm(messages, max_tokens, context)

    return {
        "answer": llm_result["content"],
        "model": llm_result["model"],
        "tokens_used": llm_result["tokens_used"],
        "prompt_used": f"{prompt_name}/{prompt_version}",
        "document_title": doc_title,
        "llm_duration_ms": llm_result["llm_duration_ms"],
        "retries": llm_result["attempt"] - 1,
    }


def process_s3_event(record: dict, context) -> dict:
    """Processes a document triggered by an S3 event."""
    from urllib.parse import unquote_plus

    source_bucket = record["s3"]["bucket"]["name"]
    source_key = unquote_plus(record["s3"]["object"]["key"])

    doc_response = s3.get_object(Bucket=source_bucket, Key=source_key)
    document = json.loads(doc_response["Body"].read().decode("utf-8"))

    prompt_name = document.get("prompt_name", os.environ.get("DEFAULT_PROMPT", "summarizer"))
    prompt_version = document.get("prompt_version", "v1")

    body = {
        "prompt_name": prompt_name,
        "prompt_version": prompt_version,
        "document": document,
        "max_tokens": document.get("max_tokens", 500),
    }

    return process_request(body, context)


def handler(event, context):
    """Main handler — handles API Gateway and S3 triggers."""
    global IS_COLD_START
    was_cold = IS_COLD_START
    IS_COLD_START = False

    start_time = time.time()
    request_id = f"req-{uuid.uuid4().hex[:8]}"

    # S3 trigger
    if "Records" in event and event["Records"][0].get("eventSource") == "aws:s3":
        results = []
        for record in event["Records"]:
            try:
                result = process_s3_event(record, context)
                response_key = save_response(request_id, result)
                result["response_key"] = response_key
                result["request_id"] = request_id
                results.append({"status": "success", **result})
            except Exception as e:
                logger.error(json.dumps({"event": "s3_trigger_error", "error": str(e)}))
                results.append({"status": "error", "error": str(e)})
        return {"processed": len(results), "results": results}

    # API Gateway
    method = event.get("requestContext", {}).get("http", {}).get("method", "")
    if method == "OPTIONS":
        return _response(200, "")

    try:
        body = json.loads(event.get("body", "{}"))
    except (json.JSONDecodeError, TypeError):
        return _response(400, {"error": "Invalid JSON body"})

    try:
        result = process_request(body, context)
    except ValueError as e:
        return _response(400, {"error": str(e)})
    except FileNotFoundError as e:
        return _response(404, {"error": str(e)})
    except TimeoutError as e:
        return _response(408, {"error": str(e)})
    except RETRYABLE as e:
        return _response(502, {"error": f"LLM unavailable: {str(e)}"})
    except Exception as e:
        logger.error(json.dumps({"event": "process_error", "error": str(e)}))
        return _response(500, {"error": f"Internal error: {str(e)}"})

    total_ms = int((time.time() - start_time) * 1000)

    response_key = save_response(request_id, result)

    logger.info(json.dumps({
        "event": "process_success",
        "request_id": request_id,
        "prompt": result["prompt_used"],
        "tokens": result["tokens_used"],
        "duration_ms": total_ms,
        "cold_start": was_cold,
    }))

    return _response(200, {
        "request_id": request_id,
        "answer": result["answer"],
        "model": result["model"],
        "tokens_used": result["tokens_used"],
        "duration_ms": total_ms,
        "prompt_used": result["prompt_used"],
        "response_key": response_key,
        "cold_start": was_cold,
    })

src/health_handler.py

import json
import os
import time

import boto3

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")


def handler(event, context):
    start = time.time()
    checks = {"lambda": "up"}

    # Check S3 bucket
    try:
        s3.head_bucket(Bucket=BUCKET)
        checks["s3_bucket"] = "accessible"
    except Exception:
        checks["s3_bucket"] = "unreachable"

    # Check OpenAI key
    openai_key = os.environ.get("OPENAI_API_KEY", "")
    if not openai_key:
        checks["openai_key"] = "missing"
    elif not openai_key.startswith("sk-"):
        checks["openai_key"] = "invalid_format"
    else:
        checks["openai_key"] = "configured"

    # List available prompt templates
    templates = []
    try:
        response = s3.list_objects_v2(Bucket=BUCKET, Prefix="prompts/", Delimiter="/")
        for prefix in response.get("CommonPrefixes", []):
            name = prefix["Prefix"].replace("prompts/", "").rstrip("/")
            versions_resp = s3.list_objects_v2(
                Bucket=BUCKET, Prefix=f"prompts/{name}/", Delimiter="/"
            )
            for v in versions_resp.get("CommonPrefixes", []):
                version = v["Prefix"].split("/")[-2]
                templates.append(f"{name}/{version}")
    except Exception:
        pass

    all_healthy = all(
        v in ("up", "accessible", "configured") for v in checks.values()
    )
    status = "healthy" if all_healthy else "degraded"
    status_code = 200 if all_healthy else 503

    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": os.environ.get("ALLOWED_ORIGIN", "*"),
        },
        "body": json.dumps({
            "status": status,
            "checks": checks,
            "config": {
                "bucket": BUCKET,
                "model": os.environ.get("MODEL_NAME", "gpt-4o-mini"),
                "memory_mb": os.environ.get("AWS_LAMBDA_FUNCTION_MEMORY_SIZE", "unknown"),
                "region": os.environ.get("AWS_REGION", "unknown"),
            },
            "prompt_templates": sorted(templates),
            "latency_ms": round((time.time() - start) * 1000, 1),
        }),
    }

src/requirements.txt

openai>=1.0.0
boto3>=1.28.0

src/seed_data.py

"""Script to populate S3 with the service's initial data."""
import boto3
import json
import os

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-service-assets-dev")


def seed():
    """Creates the bucket and uploads the initial data."""
    try:
        s3.create_bucket(Bucket=BUCKET)
        print(f"Bucket created: {BUCKET}")
    except Exception:
        print(f"Bucket already exists: {BUCKET}")

    # Prompt templates
    prompts = {
        "summarizer/v1": (
            "Generate a concise summary of the following document.\n"
            "Maximum 3 paragraphs. Include the most important points.\n"
            "Use clear, direct language."
        ),
        "summarizer/v2": (
            "Generate a summary of the following document in bullet format.\n"
            "- Maximum 5 bullets\n"
            "- Each bullet must be a complete sentence\n"
            "- Include numeric data if present\n"
            "- Prioritize actionable information"
        ),
        "classifier/v1": (
            "Classify the following document into one of these categories:\n"
            "- technical: technical documentation, guides, tutorials\n"
            "- business: reports, proposals, business analysis\n"
            "- support: tickets, FAQs, support\n"
            "- general: others\n\n"
            "Respond ONLY with a JSON: {\"category\": \"...\", \"confidence\": 0.95, \"reason\": \"...\"}"
        ),
    }

    for path, content in prompts.items():
        key = f"prompts/{path}/system.txt"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=content.encode("utf-8"),
            ContentType="text/plain",
        )
        print(f"  Uploaded: {key}")

    # Sample documents
    documents = [
        {
            "id": "doc-001",
            "title": "SDK Installation Guide",
            "content": (
                "To install the platform SDK, follow these steps: "
                "1. Run pip install platform-sdk in your terminal. "
                "2. Configure the credentials with platform configure --key YOUR_API_KEY. "
                "3. Verify the installation with platform status. "
                "The SDK supports Python 3.9+ and requires at least 100MB of free space."
            ),
        },
        {
            "id": "doc-002",
            "title": "Q1 2026 Report",
            "content": (
                "In Q1 2026, the platform reached 50,000 monthly active users, "
                "a 35% growth over Q4 2025. Monthly recurring revenue "
                "reached $125,000. The churn rate dropped from 8% to 5.2%. "
                "The main growth drivers were: the launch of the Enterprise plan, "
                "the Slack integration, and the referral program."
            ),
        },
        {
            "id": "doc-003",
            "title": "FAQ: Connection Problems",
            "content": (
                "Q: I can't connect to the API. A: Check that your API key is valid and hasn't expired. "
                "Q: I get a 429 error. A: You're exceeding the rate limit. Reduce the request frequency. "
                "Q: The responses are very slow. A: Check your internet connection and the latency to the endpoint."
            ),
        },
    ]

    for doc in documents:
        key = f"documents/knowledge-base/{doc['id']}.json"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=json.dumps(doc, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        print(f"  Uploaded: {key}")

    # Service config
    config = {
        "default_prompt": "summarizer",
        "default_version": "v1",
        "max_tokens": 500,
        "available_prompts": list(prompts.keys()),
    }
    s3.put_object(
        Bucket=BUCKET,
        Key="config/service-config.json",
        Body=json.dumps(config).encode("utf-8"),
        ContentType="application/json",
    )
    print("  Uploaded: config/service-config.json")

    print(f"\nSeed complete. Bucket: {BUCKET}")


if __name__ == "__main__":
    seed()

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: S3+Lambda AI Service — Module 5 Project

Parameters:
  OpenAiApiKey:
    Type: String
    NoEcho: true
  BucketName:
    Type: String
    Default: ai-service-assets-dev
  ModelName:
    Type: String
    Default: gpt-4o-mini
  AllowedOrigin:
    Type: String
    Default: "*"

Globals:
  Function:
    Runtime: python3.11
    Architectures: [arm64]

Resources:
  AIBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref BucketName
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256

  ServiceApi:
    Type: AWS::Serverless::HttpApi
    Properties:
      StageName: prod
      CorsConfiguration:
        AllowOrigins: [!Ref AllowedOrigin]
        AllowMethods: [POST, GET, OPTIONS]
        AllowHeaders: [Content-Type]
        MaxAge: 3600

  ProcessFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ai-service-process
      Handler: process_handler.handler
      CodeUri: ./src/
      MemorySize: 768
      Timeout: 120
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAiApiKey
          AI_BUCKET: !Ref BucketName
          MODEL_NAME: !Ref ModelName
          ALLOWED_ORIGIN: !Ref AllowedOrigin
          MAX_RETRIES: "3"
          DEFAULT_PROMPT: summarizer
          LOG_LEVEL: INFO
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref BucketName
        - Statement:
            - Sid: WriteResponses
              Effect: Allow
              Action: [s3:PutObject]
              Resource: !Sub "arn:aws:s3:::${BucketName}/responses/*"
      Events:
        ProcessRoute:
          Type: HttpApi
          Properties:
            ApiId: !Ref ServiceApi
            Path: /process
            Method: POST
        S3Trigger:
          Type: S3
          Properties:
            Bucket: !Ref AIBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: documents/inbox/
                  - Name: suffix
                    Value: .json

  HealthFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ai-service-health
      Handler: health_handler.handler
      CodeUri: ./src/
      MemorySize: 128
      Timeout: 10
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAiApiKey
          AI_BUCKET: !Ref BucketName
          MODEL_NAME: !Ref ModelName
          ALLOWED_ORIGIN: !Ref AllowedOrigin
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref BucketName
      Events:
        HealthRoute:
          Type: HttpApi
          Properties:
            ApiId: !Ref ServiceApi
            Path: /health
            Method: GET

Outputs:
  ApiUrl:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod"
  ProcessEndpoint:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod/process"
  HealthEndpoint:
    Value: !Sub "https://${ServiceApi}.execute-api.${AWS::Region}.amazonaws.com/prod/health"
  BucketName:
    Value: !Ref AIBucket

env.json

{
  "ProcessFunction": {
    "OPENAI_API_KEY": "sk-proj-your-key-here",
    "AI_BUCKET": "ai-service-assets-dev",
    "MODEL_NAME": "gpt-4o-mini",
    "ALLOWED_ORIGIN": "*",
    "MAX_RETRIES": "3",
    "DEFAULT_PROMPT": "summarizer",
    "LOG_LEVEL": "DEBUG"
  },
  "HealthFunction": {
    "OPENAI_API_KEY": "sk-proj-your-key-here",
    "AI_BUCKET": "ai-service-assets-dev",
    "MODEL_NAME": "gpt-4o-mini",
    "ALLOWED_ORIGIN": "*"
  }
}

events/process.json

{
  "version": "2.0",
  "routeKey": "POST /process",
  "headers": {"content-type": "application/json"},
  "requestContext": {
    "http": {"method": "POST", "path": "/prod/process"}
  },
  "body": "{\"prompt_name\": \"summarizer\", \"prompt_version\": \"v1\", \"document\": {\"id\": \"doc-001\", \"title\": \"Installation Guide\", \"content\": \"To install the SDK, run pip install platform-sdk. Configure with platform configure. Verify with platform status.\"}, \"max_tokens\": 300}"
}

---

Step by Step to Build

1. Create the structure (2 min)

mkdir -p s3-lambda-ai-service/src
mkdir -p s3-lambda-ai-service/events
mkdir -p s3-lambda-ai-service/tests
cd s3-lambda-ai-service

2. Copy the files (5 min)

Copy all the files from the previous section into their corresponding locations.

3. Seed data into S3 (3 min)

# With LocalStack
export AWS_ENDPOINT_URL=http://localhost:4566
export AWS_ACCESS_KEY_ID=test
export AWS_SECRET_ACCESS_KEY=test
export AI_BUCKET=ai-service-assets-dev

python src/seed_data.py

4. Local test with SAM (5 min)

sam build

# Test health
sam local invoke HealthFunction \
  --event events/health.json \
  --env-vars env.json

# Test process
sam local invoke ProcessFunction \
  --event events/process.json \
  --env-vars env.json

# Full API
sam local start-api --env-vars env.json

# In another terminal:
curl http://127.0.0.1:3000/health

curl -X POST http://127.0.0.1:3000/process \
  -H "Content-Type: application/json" \
  -d '{
    "prompt_name": "summarizer",
    "prompt_version": "v1",
    "document": {
      "id": "test-001",
      "title": "Test Document",
      "content": "This is a test document about serverless computing."
    }
  }'

5. Test the S3 trigger (5 min)

# Upload a document to the inbox (triggers Lambda)
aws --endpoint-url=http://localhost:4566 s3 cp \
  <(echo '{"id": "trigger-test", "title": "Trigger Test", "content": "Testing S3 trigger pipeline."}') \
  s3://ai-service-assets-dev/documents/inbox/trigger-test.json

# Verify that Lambda processed the document
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/responses/ --recursive

6. Verify responses in S3 (3 min)

# List responses
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/responses/ --recursive

# Read one response
aws --endpoint-url=http://localhost:4566 s3 cp \
  s3://ai-service-assets-dev/responses/2026/03/08/req-abc12345.json \
  - | python -m json.tool

7. Deploy (optional, if you have an AWS account)

sam deploy --guided
# Stack name: ai-service
# Region: us-east-1
# OpenAiApiKey: sk-proj-...
# BucketName: ai-service-assets-{your-account-id}
# Confirm changes: y

---

Cost Estimation for This Service

Scenario: Development/Testing

Parameters: 100 invocations/day, 768MB, 6s avg

S3:
├── Storage: 1GB × $0.023 = $0.02
├── PUTs: 3,000/month × $0.005/1K = $0.015
├── GETs: 10,000/month × $0.0004/1K = $0.004
└── S3 subtotal: $0.04

Lambda:
├── 3,000 inv/month → Free tier: $0.00
├── 13,500 GB-s → Free tier: $0.00
└── Lambda subtotal: $0.00

OpenAI (gpt-4o-mini):
├── Input: 3K × 200 tokens × $0.15/1M = $0.09
├── Output: 3K × 400 tokens × $0.60/1M = $0.72
└── OpenAI subtotal: $0.81

══════════════════════
TOTAL: ~$0.85/month
══════════════════════

Scenario: Light Production

Parameters: 5,000 invocations/day, 768MB, 8s avg

S3:
├── Storage: 10GB × $0.023 = $0.23
├── PUTs: 150K/month × $0.005/1K = $0.75
├── GETs: 500K/month × $0.0004/1K = $0.20
└── S3 subtotal: $1.18

Lambda:
├── 150K inv/month → Free tier: $0.00
├── 900K GB-s → Billable: 500K × $0.0000133334 = $6.67
└── Lambda subtotal: $6.67

OpenAI (gpt-4o-mini):
├── Input: 150K × 300 tokens × $0.15/1M = $6.75
├── Output: 150K × 500 tokens × $0.60/1M = $45.00
└── OpenAI subtotal: $51.75

══════════════════════════
TOTAL: ~$59.60/month
├── AWS infra: $7.85 (13%)
├── OpenAI API: $51.75 (87%)
══════════════════════════

Scenario: High Production

Parameters: 50,000 invocations/day, 768MB, 8s avg

S3:
├── Storage: 50GB × $0.023 = $1.15
├── PUTs: 1.5M/month × $0.005/1K = $7.50
├── GETs: 5M/month × $0.0004/1K = $2.00
└── S3 subtotal: $10.65

Lambda:
├── 1.5M inv/month → Billable: 500K × $0.0000002 = $0.10
├── 9M GB-s → Billable: 8.6M × $0.0000133334 = $114.67
└── Lambda subtotal: $114.77

OpenAI (gpt-4o-mini):
├── Input: 1.5M × 300 tokens × $0.15/1M = $67.50
├── Output: 1.5M × 500 tokens × $0.60/1M = $450.00
└── OpenAI subtotal: $517.50

══════════════════════════
TOTAL: ~$642.92/month
├── AWS infra: $125.42 (20%)
├── OpenAI API: $517.50 (80%)
══════════════════════════

---

Completeness Checklist

Code and structure

• src/process_handler.py implements POST /process with S3 read/write
• src/health_handler.py implements GET /health with S3 and config checks
• src/seed_data.py populates S3 with prompt templates and sample documents
• src/requirements.txt contains openai>=1.0.0 and boto3>=1.28.0
• template.yaml defines both functions, the S3 bucket, API Gateway, and the S3 trigger
• env.json exists with local variables (NOT in Git)
• events/process.json exists for testing with SAM

Functionality

• POST /process receives prompt_name + document and returns the LLM response
• POST /process reads the prompt template from S3 dynamically
• POST /process saves the response to S3 (responses/{date}/{request_id}.json)
• POST /process returns metadata: model, tokens, duration, prompt_used, response_key
• GET /health verifies S3, OpenAI key, lists available prompt templates
• S3 trigger processes documents uploaded to documents/inbox/*.json
• Errors handled: prompt not found (404), invalid JSON (400), LLM error (502)
• Retry with exponential backoff for transient LLM errors
• CORS headers on all responses

IAM and Security

• Process Function: S3 read on prompts/*, documents/*, config/*
• Process Function: S3 write only on responses/*
• Health Function: S3 read only
• Block Public Access enabled on the bucket
• Encryption at rest (AES256) enabled
• API keys in environment variables (not hardcoded)
• env.json in .gitignore

Testing

• seed_data.py runs without errors (creates bucket and data)
• sam local invoke works for both functions
• sam local start-api starts and responds on /process and /health
• S3 trigger processes a document and saves the response
• (Optional) Deploy to AWS works without errors

Cost Estimation

• Estimation documented for the development scenario (~$1/month)
• Estimation documented for light production (~$60/month)
• Estimation documented for high production (~$643/month)
• Identified that the OpenAI API is the dominant cost component

---

Connection with the Guide

What you built

S3+Lambda AI Service
├── Data Layer (S3)
│   ├── Versioned prompt templates
│   ├── RAG documents
│   ├── Responses for auditing
│   └── Service configuration
├── Compute Layer (Lambda)
│   ├── Process handler (POST /process)
│   ├── Health handler (GET /health)
│   ├── S3 trigger handler
│   ├── Retry with backoff
│   └── Structured logging
├── API Layer (API Gateway)
│   ├── CORS configured
│   └── Throttling
├── Security (IAM)
│   ├── Least-privilege policies
│   └── S3 encryption
└── Operations
    ├── Cost estimation per scenario
    └── Seed data script

What comes next in the upcoming modules

Module 6 (Cloud Migration Patterns):
├── This service works against LocalStack and AWS
├── You'll abstract the endpoint_url with environment variables
├── Same code, different infrastructure
├── Testing the migration path: LocalStack → AWS
└── The M5 S3+Lambda Service is the "target" of the migration

Module 7 (Alternative Platforms):
├── You compare the cost of this service on AWS vs Render/Railway/Fly.io
├── Decision matrix: when is AWS the better option?
└── The M5 cost estimation gives you the baseline

Module 8 (Integrator Project):
├── This service can be part of your final system
├── Or you can choose another platform based on M7
├── The M1 decision matrix + the M5 costs inform the decision
└── Your service is deployed to production

---

Project Troubleshooting

"seed_data.py fails with NoSuchBucket"

# Verify that LocalStack is running
curl http://localhost:4566/_localstack/health

# Verify the endpoint URL
echo $AWS_ENDPOINT_URL
# It should be: http://localhost:4566

# The script creates the bucket, but if there's a connection error:
aws --endpoint-url=http://localhost:4566 s3 mb s3://ai-service-assets-dev

"Process handler returns 404 for the prompt template"

# Verify that seed_data.py ran correctly
aws --endpoint-url=http://localhost:4566 s3 ls \
  s3://ai-service-assets-dev/prompts/ --recursive

# If there are no prompts, run seed_data.py again
python src/seed_data.py

"S3 trigger doesn't fire"

# In LocalStack, check the notification configuration
aws --endpoint-url=http://localhost:4566 \
  s3api get-bucket-notification-configuration \
  --bucket ai-service-assets-dev

# Verify that Lambda has the permission
aws --endpoint-url=http://localhost:4566 \
  lambda get-policy --function-name ai-service-process

"Response isn't saved to S3 after POST /process"

# Verify that Lambda has write permission on responses/
# In template.yaml, confirm:
# - Statement:
#     - Sid: WriteResponses
#       Effect: Allow
#       Action: [s3:PutObject]
#       Resource: !Sub "arn:aws:s3:::${BucketName}/responses/*"

---

Summary

• You built a complete AI service on real AWS with S3 as the data layer and Lambda as the compute layer.
• You stored versioned prompt templates in S3 and RAG documents; Lambda reads them, builds prompts, and persists responses.
• You configured S3 event triggers to automatically process documents uploaded to documents/inbox/.
• You defined IAM with least privilege: Lambda only accesses the necessary S3 prefixes (prompts, documents, responses).
• You deployed with SAM: encrypted S3 bucket, API Gateway, Process and Health functions, and an S3 trigger.
• You documented the cost estimation for development ($0.85/month), light production ($60/month), and high production (~$643/month).

---

Project Resources

1. AWS SAM CLI Reference — SAM CLI reference
2. S3 Event Notifications — S3 triggers
3. Lambda Python Handler — Handler reference
4. boto3 S3 Client — S3 SDK reference
5. OpenAI Python SDK — Official SDK
6. IAM Policy Reference — Policies
7. AWS Pricing Calculator — Cost calculator
8. LocalStack SAM Integration — SAM with LocalStack

---