Module 4: LocalStack — AWS Local Development

8. Project: LocalStack AI Pipeline

Next →

Project description

This is the integrator project for Module 4. You'll build a complete AI pipeline running entirely on LocalStack inside Docker Compose: an S3 bucket receives documents, a Lambda processes them with an LLM, and the result is written to another S3 bucket. Everything orchestrated with a Python script that demonstrates the end-to-end flow. It's the culmination of everything you learned in this module: LocalStack setup, local S3, local Lambda, pipeline, environment switching, and debugging.

Why it matters: This pipeline is the central artifact of Phase 2. In Module 5, you'll work with the same services (S3, Lambda) but in a real AWS context. In Module 6, this pipeline's code is abstracted to run against LocalStack OR AWS with the same codebase — and the transition will be trivial because environment switching is already implemented. In Module 8 (Integrator Project), LocalStack remains your local development environment.

Project objective

Produce a functional LocalStack AI Pipeline that:

  1. Runs entirely on LocalStack inside Docker Compose
  2. Has two S3 buckets: ai-input (documents) and ai-output (results)
  3. Has an ai-document-processor Lambda that reads from S3, processes with GPT-4o-mini, and writes to S3
  4. Supports environment switching: the same code works against LocalStack or AWS
  5. Has setup, run, and verification scripts
  6. Includes a diagnostic script that checks the health of the whole pipeline
  7. Initializes automatically with LocalStack init scripts

Module Recap

CapsuleConceptWhere you use it in the project
02LocalStack in Docker ComposeCompose file with LocalStack as a service
03Local S3 with boto3Input/output buckets, upload/download
04Lambda on LocalStackDeploy and run the processor function
05S3 + Lambda pipelineComplete flow input → process → output
06Environment switchingAWS_ENDPOINT_URL controls the target
07Debugging LocalStackDiagnostic and recovery scripts

Technical Specifications

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Docker Compose                           │
│                                                               │
│  ┌──────────┐   ┌──────────┐   ┌──────────────────────────┐ │
│  │  FastAPI  │   │  Redis   │   │      LocalStack          │ │
│  │  (api)    │   │ (cache)  │   │                          │ │
│  │  :8000    │   │  :6379   │   │  S3: ai-input            │ │
│  └──────────┘   └──────────┘   │  S3: ai-output           │ │
│                                 │  Lambda: ai-doc-processor │ │
│                                 │  :4566                    │ │
│                                 └──────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

Pipeline Flow:
  Upload doc → S3 (ai-input)
       ↓
  Invoke Lambda (ai-document-processor)
       ↓
  Lambda reads from S3 → GPT-4o-mini → writes to S3
       ↓
  Download result ← S3 (ai-output)

Required files

localstack-ai-pipeline/
├── api/
│   ├── main.py                  # FastAPI with the M2 endpoints
│   ├── config.py                # Settings with Pydantic
│   ├── aws_clients.py           # AWS client factory
│   ├── requirements.txt         # Dependencies
│   └── Dockerfile               # API container
├── lambda/
│   ├── processor.py             # Lambda handler
│   ├── requirements.txt         # Lambda dependencies (openai, boto3)
│   └── package/                 # Packaging directory
├── scripts/
│   ├── setup.sh                 # Complete pipeline setup
│   ├── deploy-lambda.sh         # Deploy/update the Lambda
│   ├── run-pipeline.py          # Runs the pipeline end-to-end
│   ├── diagnose.sh              # System diagnosis
│   └── demo.sh                  # Complete demo in one command
├── init-scripts/
│   └── setup.sh                 # LocalStack init script (creates buckets)
├── data/
│   └── sample-documents/        # Test documents
├── docker-compose.yml           # Complete Compose
├── .env                         # Variables (OPENAI_API_KEY)
├── .env.example                 # Template
└── .gitignore                   # Ignores .env, package/, *.zip

Complete Code

docker-compose.yml

services:
  api:
    build:
      context: ./api
    ports:
      - "${API_PORT:-8000}:8000"
    env_file:
      - .env
    environment:
      - REDIS_URL=redis://cache:6379
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
      - AWS_DEFAULT_REGION=us-east-1
    depends_on:
      cache:
        condition: service_healthy
      localstack:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s
    restart: unless-stopped

  cache:
    image: redis:7-alpine
    command: redis-server --appendonly yes --maxmemory 128mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    restart: unless-stopped

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
      - DEBUG=0
      - LAMBDA_EXECUTOR=docker
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - localstack_data:/var/lib/localstack
      - /var/run/docker.sock:/var/run/docker.sock
      - ./init-scripts:/etc/localstack/init/ready.d
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 20s
    restart: unless-stopped

volumes:
  redis_data:
  localstack_data:

init-scripts/setup.sh

#!/bin/bash
echo "=== LocalStack Init: Creating resources ==="

awslocal s3 mb s3://ai-input 2>/dev/null || true
awslocal s3 mb s3://ai-output 2>/dev/null || true

echo "Buckets created:"
awslocal s3 ls

echo "=== LocalStack Init: Complete ==="

api/aws_clients.py

import boto3
import os
import logging

logger = logging.getLogger(__name__)

AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL")


def _base_kwargs():
    kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
    return kwargs


def get_s3_client():
    client = boto3.client("s3", **_base_kwargs())
    target = f"LocalStack ({AWS_ENDPOINT_URL})" if AWS_ENDPOINT_URL else "AWS"
    logger.debug(f"S3 client → {target}")
    return client


def get_lambda_client():
    client = boto3.client("lambda", **_base_kwargs())
    target = f"LocalStack ({AWS_ENDPOINT_URL})" if AWS_ENDPOINT_URL else "AWS"
    logger.debug(f"Lambda client → {target}")
    return client


def is_local():
    return AWS_ENDPOINT_URL is not None

api/config.py

from pydantic_settings import BaseSettings
from pydantic import field_validator
from typing import Optional


class Settings(BaseSettings):
    openai_api_key: str
    redis_url: str = "redis://cache:6379"
    environment: str = "development"
    log_level: str = "debug"
    cache_ttl: int = 3600
    model_name: str = "gpt-4o-mini"
    max_tokens: int = 500
    aws_endpoint_url: Optional[str] = None
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"

    @field_validator("openai_api_key")
    @classmethod
    def validate_key(cls, v):
        if not v or "replace" in v.lower():
            raise ValueError("Set a real OPENAI_API_KEY in .env")
        return v

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    class Config:
        env_file = ".env"


settings = Settings()

api/main.py

import hashlib
import json
import logging
import time
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from openai import OpenAI
import redis

from config import settings
from aws_clients import get_s3_client, get_lambda_client, is_local

logging.basicConfig(level=getattr(logging, settings.log_level.upper()))
logger = logging.getLogger(__name__)

app = FastAPI(title="AI Pipeline API", version="1.0.0")
openai_client = OpenAI(api_key=settings.openai_api_key, timeout=30.0)
cache = redis.Redis.from_url(settings.redis_url, decode_responses=True)


class AskRequest(BaseModel):
    prompt: str
    max_tokens: int = 500
    use_cache: bool = True


class ProcessRequest(BaseModel):
    document_key: str
    force: bool = False


@app.get("/health")
def health():
    checks = {"api": "up"}

    try:
        cache.ping()
        checks["redis"] = "up"
    except Exception:
        checks["redis"] = "down"

    try:
        s3 = get_s3_client()
        s3.list_buckets()
        checks["localstack_s3"] = "up"
    except Exception:
        checks["localstack_s3"] = "down"

    overall = "healthy" if all(v == "up" for v in checks.values()) else "degraded"
    status_code = 200 if overall == "healthy" else 503

    return JSONResponse(
        status_code=status_code,
        content={
            "status": overall,
            "environment": "localstack" if is_local() else "aws",
            "services": checks,
        },
    )


@app.post("/ask")
def ask(request: AskRequest):
    cache_key = f"ask:{hashlib.md5(f'{request.prompt}:{request.max_tokens}'.encode()).hexdigest()}"

    if request.use_cache:
        try:
            cached = cache.get(cache_key)
            if cached:
                return json.loads(cached) | {"cached": True}
        except Exception:
            pass

    try:
        response = openai_client.chat.completions.create(
            model=settings.model_name,
            messages=[{"role": "user", "content": request.prompt}],
            max_tokens=request.max_tokens,
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"LLM error: {str(e)}")

    result = {
        "answer": response.choices[0].message.content,
        "model": settings.model_name,
        "tokens_used": response.usage.total_tokens,
        "cached": False,
    }

    try:
        cache.setex(cache_key, settings.cache_ttl, json.dumps(result))
    except Exception:
        pass

    return result


@app.post("/process")
def process_document(request: ProcessRequest):
    """Invokes Lambda to process a document from S3."""
    lambda_client = get_lambda_client()

    try:
        response = lambda_client.invoke(
            FunctionName="ai-document-processor",
            InvocationType="RequestResponse",
            Payload=json.dumps({
                "document_key": request.document_key,
                "force": request.force,
            }),
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Lambda invocation failed: {str(e)}")

    result = json.loads(response["Payload"].read())
    body = json.loads(result.get("body", "{}"))

    if result.get("statusCode", 500) != 200:
        raise HTTPException(status_code=result["statusCode"], detail=body)

    return body


@app.get("/documents")
def list_documents():
    """Lists pending documents in S3 input."""
    s3 = get_s3_client()
    response = s3.list_objects_v2(Bucket=settings.s3_input_bucket, Prefix="documents/")
    return {
        "bucket": settings.s3_input_bucket,
        "documents": [
            {"key": obj["Key"], "size": obj["Size"]}
            for obj in response.get("Contents", [])
        ],
    }


@app.get("/results")
def list_results():
    """Lists processed results in S3 output."""
    s3 = get_s3_client()
    response = s3.list_objects_v2(Bucket=settings.s3_output_bucket, Prefix="results/")
    return {
        "bucket": settings.s3_output_bucket,
        "results": [
            {"key": obj["Key"], "size": obj["Size"]}
            for obj in response.get("Contents", [])
        ],
    }

api/requirements.txt

fastapi==0.115.0
uvicorn==0.30.0
openai>=1.0.0
redis==5.0.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
boto3>=1.34.0

api/Dockerfile

FROM python:3.11-slim

RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

lambda/processor.py

import json
import os
import time
import boto3
from openai import OpenAI

AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL")

openai_client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", ""),
    timeout=50,
    max_retries=1,
)

MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini")
INPUT_BUCKET = os.environ.get("INPUT_BUCKET", "ai-input")
OUTPUT_BUCKET = os.environ.get("OUTPUT_BUCKET", "ai-output")


def get_s3_client():
    kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = "test"
        kwargs["aws_secret_access_key"] = "test"
    return boto3.client("s3", **kwargs)


def handler(event, context):
    start_time = time.time()
    s3 = get_s3_client()

    document_key = event.get("document_key", "")
    if not document_key:
        if "body" in event:
            try:
                body = json.loads(event["body"])
                document_key = body.get("document_key", "")
            except (json.JSONDecodeError, TypeError):
                pass

    if not document_key:
        return _response(400, {"error": "document_key is required"})

    force = event.get("force", False)

    # Check if already processed
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    if not force:
        try:
            existing = s3.get_object(Bucket=OUTPUT_BUCKET, Key=output_key)
            result = json.loads(existing["Body"].read().decode("utf-8"))
            return _response(200, {
                "status": "already_processed",
                "output": f"s3://{OUTPUT_BUCKET}/{output_key}",
                "tokens_used": result.get("tokens_used", 0),
            })
        except Exception:
            pass

    # Read document
    try:
        doc = s3.get_object(Bucket=INPUT_BUCKET, Key=document_key)
        document_text = doc["Body"].read().decode("utf-8")
    except Exception as e:
        return _response(404, {
            "error": f"Document not found: {str(e)}",
            "bucket": INPUT_BUCKET,
            "key": document_key,
        })

    # Process with LLM
    system_prompt = (
        "Analyze the following business document. Respond in JSON with these fields:\n"
        "- title: title or subject of the document\n"
        "- category: document type (report, email, proposal, ticket, other)\n"
        "- summary: summary in 2-3 sentences\n"
        "- key_points: list of 3-5 important points\n"
        "- urgency: high, medium, or low\n"
        "- suggested_actions: list of 1-3 recommended actions"
    )

    try:
        llm_response = openai_client.chat.completions.create(
            model=MODEL_NAME,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": document_text},
            ],
            max_tokens=800,
            response_format={"type": "json_object"},
        )
        analysis = llm_response.choices[0].message.content
        tokens_used = llm_response.usage.total_tokens
    except Exception as e:
        return _response(502, {"error": f"LLM failed: {str(e)}"})

    duration_ms = round((time.time() - start_time) * 1000)

    # Build result
    result = {
        "source_document": document_key,
        "analysis": json.loads(analysis),
        "metadata": {
            "model": MODEL_NAME,
            "tokens_used": tokens_used,
            "duration_ms": duration_ms,
            "processed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
            "environment": "localstack" if AWS_ENDPOINT_URL else "aws",
        },
    }

    # Write to S3
    try:
        s3.put_object(
            Bucket=OUTPUT_BUCKET,
            Key=output_key,
            Body=json.dumps(result, indent=2, ensure_ascii=False),
            ContentType="application/json",
        )
    except Exception as e:
        return _response(500, {"error": f"Cannot write result: {str(e)}"})

    return _response(200, {
        "status": "processed",
        "input": f"s3://{INPUT_BUCKET}/{document_key}",
        "output": f"s3://{OUTPUT_BUCKET}/{output_key}",
        "analysis": result["analysis"],
        "tokens_used": tokens_used,
        "duration_ms": duration_ms,
    })


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

lambda/requirements.txt

openai>=1.0.0
boto3>=1.34.0

.env.example

OPENAI_API_KEY=sk-proj-replace-with-your-key
ENVIRONMENT=development
LOG_LEVEL=debug
CACHE_TTL=3600
MODEL_NAME=gpt-4o-mini
MAX_TOKENS=500
API_PORT=8000

.gitignore

.env
lambda/package/
lambda/*.zip
data/output/
__pycache__/
*.pyc

Operation Scripts

scripts/setup.sh

#!/bin/bash
set -e

echo "=== Setup: LocalStack AI Pipeline ==="

echo "1. Copying .env..."
if [ ! -f .env ]; then
  cp .env.example .env
  echo "   Created .env from .env.example"
  echo "   IMPORTANT: edit .env with your real OPENAI_API_KEY"
else
  echo "   .env already exists"
fi

echo "2. Creating the init scripts directory..."
mkdir -p init-scripts
chmod +x init-scripts/setup.sh 2>/dev/null || true

echo "3. Packaging Lambda..."
mkdir -p lambda/package
pip install -r lambda/requirements.txt -t lambda/package/ --quiet
cp lambda/processor.py lambda/package/
cd lambda/package
zip -r ../processor.zip . -q
cd ../..
echo "   Package: $(ls -lh lambda/processor.zip | awk '{print $5}')"

echo "4. Starting Docker Compose..."
docker compose up -d --build

echo "5. Waiting for health checks..."
for i in {1..30}; do
  ALL_HEALTHY=true
  for service in api cache localstack; do
    STATUS=$(docker compose ps $service --format json 2>/dev/null | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('Health',''))" 2>/dev/null || echo "")
    if [ "$STATUS" != "healthy" ]; then
      ALL_HEALTHY=false
    fi
  done
  if [ "$ALL_HEALTHY" = "true" ]; then
    echo "   All services healthy"
    break
  fi
  echo "   Waiting... ($i/30)"
  sleep 3
done

echo "6. Deploying Lambda to LocalStack..."
source .env 2>/dev/null || true

awslocal lambda create-function \
  --function-name ai-document-processor \
  --runtime python3.11 \
  --handler processor.handler \
  --zip-file fileb://lambda/processor.zip \
  --role arn:aws:iam::000000000000:role/lambda-role \
  --timeout 90 \
  --memory-size 768 \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    INPUT_BUCKET=ai-input,
    OUTPUT_BUCKET=ai-output,
    AWS_ENDPOINT_URL=http://host.docker.internal:4566,
    AWS_DEFAULT_REGION=us-east-1
  }" 2>/dev/null || \
awslocal lambda update-function-code \
  --function-name ai-document-processor \
  --zip-file fileb://lambda/processor.zip > /dev/null

echo "7. Uploading test documents..."
mkdir -p data/sample-documents

cat > data/sample-documents/report-q1.txt << 'EOF'
Q1 2026 Quarterly Report — Technology Department

Summary: We completed the migration to microservices, reducing deploy times from 4h to 15min.
The AI ticket-classification system reduced resolution time by 40%.

Achievements: Migration to Kubernetes, automated CI/CD, AI classification system.
Next steps: Monitoring with Prometheus, expand AI to customer support.
EOF

cat > data/sample-documents/support-email.txt << 'EOF'
From: customer@company.com
Subject: Urgent billing problem

I've had an incorrect charge for 3 months. The Premium service costs $49/month but
I'm charged $79. I've contacted support twice with no resolution. I need a correction
and a refund of the difference. If this isn't resolved this week, I'll cancel the service.
EOF

cat > data/sample-documents/technical-proposal.txt << 'EOF'
Proposal: Recommendation System for E-commerce

Objective: Improve CTR by 20% with recommendations based on embeddings.
Stack: FastAPI + Redis + OpenAI Embeddings.
Timeline: 6 weeks. Budget: $15,000.
Team: 2 ML Engineers + 1 Backend Developer.
Estimated ROI: $50,000/year in increased sales.
EOF

for f in data/sample-documents/*.txt; do
  FILENAME=$(basename "$f")
  awslocal s3 cp "$f" "s3://ai-input/documents/$FILENAME"
done

echo "   Documents uploaded to S3"

echo ""
echo "=== Setup complete ==="
echo ""
echo "Services:"
docker compose ps --format "table {{.Name}}\t{{.Status}}\t{{.Ports}}"
echo ""
echo "S3 Buckets:"
awslocal s3 ls
echo ""
echo "Lambda Functions:"
awslocal lambda list-functions --query 'Functions[].FunctionName' --output text
echo ""
echo "Documents in S3:"
awslocal s3 ls s3://ai-input/documents/
echo ""
echo "Next step: python scripts/run-pipeline.py"

scripts/run-pipeline.py

#!/usr/bin/env python3
"""Runs the complete AI pipeline against LocalStack."""
import boto3
import json
import time
import sys

ENDPOINT = "http://localhost:4566"
KWARGS = {
    "endpoint_url": ENDPOINT,
    "aws_access_key_id": "test",
    "aws_secret_access_key": "test",
    "region_name": "us-east-1",
}

s3 = boto3.client("s3", **KWARGS)
lambda_client = boto3.client("lambda", **KWARGS)


def list_pending_documents():
    response = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
    return [obj["Key"] for obj in response.get("Contents", [])]


def process_document(document_key, force=False):
    payload = {"document_key": document_key, "force": force}

    start = time.time()
    response = lambda_client.invoke(
        FunctionName="ai-document-processor",
        InvocationType="RequestResponse",
        Payload=json.dumps(payload),
    )
    elapsed = round((time.time() - start) * 1000)

    result = json.loads(response["Payload"].read())
    body = json.loads(result.get("body", "{}"))

    return {
        "status_code": result.get("statusCode", 500),
        "body": body,
        "invoke_ms": elapsed,
    }


def download_result(output_path):
    parts = output_path.replace("s3://ai-output/", "")
    response = s3.get_object(Bucket="ai-output", Key=parts)
    return json.loads(response["Body"].read().decode("utf-8"))


def main():
    force = "--force" in sys.argv

    print("=" * 60)
    print("  LocalStack AI Pipeline — Document Processor")
    print("=" * 60)
    print(f"\nEnvironment: LocalStack ({ENDPOINT})")
    print(f"Force re-process: {force}\n")

    documents = list_pending_documents()
    print(f"Documents found: {len(documents)}\n")

    if not documents:
        print("No documents in ai-input/documents/")
        print("Upload documents with: awslocal s3 cp file.txt s3://ai-input/documents/")
        return

    results = []
    for i, doc_key in enumerate(documents, 1):
        doc_name = doc_key.split("/")[-1]
        print(f"[{i}/{len(documents)}] Processing: {doc_name}")

        result = process_document(doc_key, force=force)
        status = result["body"].get("status", "unknown")
        tokens = result["body"].get("tokens_used", 0)

        if result["status_code"] == 200:
            print(f"    Status: {status}")
            print(f"    Tokens: {tokens}")
            print(f"    Time: {result['invoke_ms']}ms")

            if "analysis" in result["body"]:
                analysis = result["body"]["analysis"]
                if isinstance(analysis, dict):
                    print(f"    Category: {analysis.get('category', 'N/A')}")
                    print(f"    Urgency: {analysis.get('urgency', 'N/A')}")
        else:
            print(f"    ERROR: {result['body']}")

        results.append(result)
        print()

    # Summary
    print("=" * 60)
    print("  SUMMARY")
    print("=" * 60)

    success = sum(1 for r in results if r["status_code"] == 200)
    total_tokens = sum(r["body"].get("tokens_used", 0) for r in results)
    total_time = sum(r["invoke_ms"] for r in results)

    print(f"\n  Documents processed: {success}/{len(results)}")
    print(f"  Total tokens:        {total_tokens}")
    print(f"  Total time:          {total_time}ms")
    print(f"  Estimated cost:      ~${total_tokens * 0.0000015:.4f} (gpt-4o-mini)")
    print(f"  LocalStack cost:     $0.00")
    print()

    # List results in S3
    print("Results in S3:")
    response = s3.list_objects_v2(Bucket="ai-output", Prefix="results/")
    for obj in response.get("Contents", []):
        print(f"  s3://ai-output/{obj['Key']} ({obj['Size']} bytes)")


if __name__ == "__main__":
    main()

scripts/diagnose.sh

#!/bin/bash
echo "=== DIAGNOSIS: LocalStack AI Pipeline ==="
echo ""

echo "1. Docker Compose"
docker compose ps --format "table {{.Name}}\t{{.Status}}"
echo ""

echo "2. LocalStack Health"
curl -s http://localhost:4566/_localstack/health | python3 -m json.tool 2>/dev/null || echo "   Not available"
echo ""

echo "3. API Health"
curl -s http://localhost:8000/health | python3 -m json.tool 2>/dev/null || echo "   Not available"
echo ""

echo "4. S3 Buckets"
awslocal s3 ls 2>/dev/null || echo "   S3 not available"
echo ""

echo "5. S3 Input Documents"
awslocal s3 ls s3://ai-input/documents/ 2>/dev/null || echo "   Bucket empty or doesn't exist"
echo ""

echo "6. S3 Output Results"
awslocal s3 ls s3://ai-output/results/ 2>/dev/null || echo "   No results"
echo ""

echo "7. Lambda Functions"
awslocal lambda list-functions --query 'Functions[].{Name:FunctionName,Runtime:Runtime,Memory:MemorySize,Timeout:Timeout}' --output table 2>/dev/null || echo "   Lambda not available"
echo ""

echo "8. Recent Errors"
ERRORS=$(docker compose logs localstack --tail 50 2>&1 | grep -i "error\|exception" | tail -3)
if [ -n "$ERRORS" ]; then
  echo "$ERRORS"
else
  echo "   No recent errors"
fi
echo ""

echo "9. Resources"
docker stats --no-stream --format "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}" 2>/dev/null | grep -E "NAME|api|cache|localstack"
echo ""

echo "=== END OF DIAGNOSIS ==="

scripts/demo.sh

#!/bin/bash
set -e

echo "╔════════════════════════════════════════════╗"
echo "║  LocalStack AI Pipeline — Complete Demo    ║"
echo "╚════════════════════════════════════════════╝"
echo ""

# Verify everything is running
echo "Checking services..."
curl -s http://localhost:8000/health > /dev/null 2>&1 || { echo "API not available. Run: bash scripts/setup.sh"; exit 1; }
curl -s http://localhost:4566/_localstack/health > /dev/null 2>&1 || { echo "LocalStack not available"; exit 1; }
echo "Services OK"
echo ""

# List documents
echo "Available documents:"
curl -s http://localhost:8000/documents | python3 -c "
import json, sys
data = json.load(sys.stdin)
for doc in data['documents']:
    print(f\"  - {doc['key']} ({doc['size']} bytes)\")
"
echo ""

# Process a document via the API
echo "Processing a document via the API..."
RESULT=$(curl -s -X POST http://localhost:8000/process \
  -H "Content-Type: application/json" \
  -d '{"document_key": "documents/support-email.txt", "force": true}')

echo "$RESULT" | python3 -c "
import json, sys
data = json.load(sys.stdin)
print(f\"  Status: {data.get('status', 'unknown')}\")
print(f\"  Tokens: {data.get('tokens_used', 0)}\")
print(f\"  Duration: {data.get('duration_ms', 0)}ms\")
if 'analysis' in data:
    a = data['analysis']
    print(f\"  Category: {a.get('category', 'N/A')}\")
    print(f\"  Urgency: {a.get('urgency', 'N/A')}\")
    print(f\"  Summary: {a.get('summary', 'N/A')[:100]}...\")
"
echo ""

# List results
echo "Results in S3:"
curl -s http://localhost:8000/results | python3 -c "
import json, sys
data = json.load(sys.stdin)
for r in data['results']:
    print(f\"  - {r['key']} ({r['size']} bytes)\")
"
echo ""

echo "Pipeline ran successfully"
echo "AWS cost: \$0.00 (LocalStack)"

Step by Step to Build

1. Create the structure (2 min)

mkdir -p localstack-ai-pipeline/{api,lambda/package,scripts,init-scripts,data/sample-documents}
cd localstack-ai-pipeline

2. Create the files (15 min)

Copy the files from the previous sections into their corresponding locations.

3. Configure the environment (2 min)

cp .env.example .env
# Edit .env with your real OPENAI_API_KEY

4. Make the scripts executable (1 min)

chmod +x scripts/*.sh
chmod +x init-scripts/setup.sh

5. Complete setup (5 min)

bash scripts/setup.sh

6. Run the pipeline (3 min)

python scripts/run-pipeline.py

7. Interactive demo (2 min)

bash scripts/demo.sh

8. Diagnosis (1 min)

bash scripts/diagnose.sh

Completeness Checklist

Infrastructure

  • docker compose up starts api, cache, and localstack without errors
  • All three services report healthy in docker compose ps
  • Init scripts create the buckets automatically on startup
  • The ai-document-processor Lambda is deployed on LocalStack

Functional pipeline

  • Test documents are in s3://ai-input/documents/
  • python scripts/run-pipeline.py processes all the documents
  • The results appear in s3://ai-output/results/
  • The analyses include: title, category, summary, key points, urgency
  • Re-running without --force returns "already_processed"
  • Re-running with --force reprocesses correctly

API

  • GET /health returns the status of api, redis, and localstack
  • POST /process invokes Lambda and returns the result
  • GET /documents lists documents in S3 input
  • GET /results lists results in S3 output
  • POST /ask works with the cache (Redis)

Environment switching

  • The code uses AWS_ENDPOINT_URL to determine the target
  • Without AWS_ENDPOINT_URL, boto3 would point to real AWS
  • There's no environment if/else in the business code
  • The bucket names are configurable via environment variables

Scripts and operation

  • scripts/setup.sh configures the whole system from scratch
  • scripts/run-pipeline.py runs the pipeline end-to-end
  • scripts/diagnose.sh reports the status of all the components
  • scripts/demo.sh demonstrates the pipeline in one command
  • .gitignore excludes .env, package/, and *.zip

Quality

  • Lambda returns clear errors (400 for invalid input, 502 for LLM failure)
  • The results include metadata (model, tokens, duration, environment)
  • Deduplication works (doesn't reprocess without force=true)
  • The system recovers from restarts (init scripts + deploy scripts)

Project Troubleshooting

"setup.sh fails at 'Deploying Lambda'"

# Verify that LocalStack is healthy
docker compose ps localstack

# Verify the zip exists and has content
ls -lh lambda/processor.zip
unzip -l lambda/processor.zip | head -5

# Verify that OPENAI_API_KEY is in .env
grep OPENAI_API_KEY .env

"run-pipeline.py returns 502 for all the documents"

# The LLM can't connect — check the API key
awslocal lambda get-function-configuration \
  --function-name ai-document-processor \
  --query 'Environment.Variables.OPENAI_API_KEY'
# If it's empty or a placeholder, update it:
source .env
awslocal lambda update-function-configuration \
  --function-name ai-document-processor \
  --environment "Variables={OPENAI_API_KEY=${OPENAI_API_KEY},...}"

"Lambda can't read from S3"

# Check Lambda's endpoint URL
awslocal lambda get-function-configuration \
  --function-name ai-document-processor \
  --query 'Environment.Variables.AWS_ENDPOINT_URL'

# If LAMBDA_EXECUTOR=docker: it should be http://host.docker.internal:4566
# If LAMBDA_EXECUTOR=local: it should be http://localhost:4566

"The results have empty analyses"

# Verify that response_format=json_object works
# Some models don't support json_object mode
# Try without it or use a model that supports it

Connection with the Guide

What you built

LocalStack AI Pipeline
├── Docker Compose (3 services: api, cache, localstack)
├── S3 Storage (ai-input → documents, ai-output → results)
├── Lambda Processor (reads S3 → GPT-4o-mini → writes S3)
├── FastAPI Gateway (endpoints to operate the pipeline)
├── Environment Switching (AWS_ENDPOINT_URL)
├── Init Scripts (automatic setup)
└── Operation Scripts (setup, run, diagnose, demo)

What comes next

Module 5 (AWS Services for AI):
├── Same pipeline, now with depth on real S3 and Lambda
├── Integration with SageMaker basics
└── Your LocalStack scripts remain your dev environment

Module 6 (Cloud Migration Patterns):
├── The environment switching from this module is the basis
├── Your code already works on LocalStack — now you take it to AWS
├── Migration = change AWS_ENDPOINT_URL + configure IAM
└── Confidence: "I already tested everything locally"

Module 8 (Integrator Project):
├── LocalStack remains your development environment
├── This module's pipeline integrates with the final system
└── You develop on LocalStack, you deploy to production

Summary

  • You built a complete AI pipeline on LocalStack inside Docker Compose (api, cache, localstack).
  • You integrated S3 (ai-input and ai-output buckets) with Lambda for document processing with GPT-4o-mini.
  • The Lambda reads documents from S3, analyzes them with the LLM, and writes structured results to S3.
  • You implemented environment switching with AWS_ENDPOINT_URL: the same code works against LocalStack or AWS.
  • You configured LocalStack init scripts to create buckets automatically on startup.
  • You integrated the pipeline with FastAPI, Redis, and operation scripts (setup, run-pipeline, diagnose, demo).

Project Resources

  1. LocalStack Documentation — Complete official documentation
  2. LocalStack Docker Compose — Setup with Compose
  3. boto3 S3 Reference — Complete S3 API
  4. boto3 Lambda Reference — Complete Lambda API
  5. LocalStack Init Hooks — Initialization scripts
  6. OpenAI JSON Mode — response_format=json_object
  7. Docker Compose Override — Compose per environment
  8. FastAPI + boto3 Best Practices — Settings and configuration
← Previous7. Debugging LocalStackNext →1. Introduction: AWS Services for AI (S3, Lambda, SageMaker Basics)