- [Architecture Overview](#architecture)
- [Prerequisites & Tools](#prereqs)
- [SAM Template – Defining the Functions](#sam-template)
- [Lambda Code – Image Ingestion, Processing & Summarization](#code)
- [IAM Role – Permissions (including Bedrock invoke)](#iam)
- [Deploy, Test & Debug](#deploy)
- [Observability & Cost Tips](#monitor)
- [📚 References & Further Reading](#references)
- [Your Turn](#your-turn)

ComponentPurpose
**S3 – raw‑images‑bucket‑<yourname>**Stores original uploads from the front‑end. Triggers the `ImageProcessor` Lambda on `s3:ObjectCreated:*`.
**S3 – optimized‑images‑bucket‑<yourname>**Destination for resized, compressed JPEG/WEBP files and for AI‑generated captions.
**Lambda – ImageProcessor**Downloads the source image, runs a lightweight PyTorch model (e.g., MobileNetV3), resizes & compresses, stores results, and writes a metadata record to DynamoDB.
**Lambda – SummaryAPI**Exposes a `GET /summary` endpoint (see Shattered.io step 9) that calls Amazon Bedrock `InvokeModel` to generate a natural‑language summary of the day's processed assets.
**DynamoDB – image‑metadata**Tracks processing status, dimensions, model confidence, and Bedrock summary references.
**API Gateway**Routes CRUD and `/summary` requests to the appropriate Lambda functions.

- AWS account with **AdministratorAccess** (or equivalent scoped policy).
- AWS CLI v2, SAM CLI v2.12+, Docker (for local Lambda container builds).
- Python 3.12 installed locally (matching the Lambda runtime).
- Git (optional, for version control).

mkdir ai-image-pipeline && cd ai-image-pipeline
sam init --runtime python3.12 --name image-pipeline --app-template hello-world

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >-
  AI‑Powered Serverless Image Processing Pipeline – Part 2.
  Deploys two Python Lambdas: ImageProcessor and SummaryAPI.

Globals:
  Function:
    Timeout: 30
    Runtime: python3.12
    MemorySize: 1024
    Environment:
      Variables:
        RAW_BUCKET: !Ref RawImagesBucket
        OPT_BUCKET: !Ref OptimizedImagesBucket
        METADATA_TABLE: !Ref ImageMetadataTable

Resources:
  # -------------------------------------------------
  # 1️⃣ S3 Buckets (created in Part 1, imported here)
  # -------------------------------------------------
  RawImagesBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Retain

  OptimizedImagesBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Retain

  # -------------------------------------------------
  # 2️⃣ DynamoDB Table for metadata
  # -------------------------------------------------
  ImageMetadataTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: image-metadata
      AttributeDefinitions:
        - AttributeName: image_id
          AttributeType: S
      KeySchema:
        - AttributeName: image_id
          KeyType: HASH
      BillingMode: PAY_PER_REQUEST

  # -------------------------------------------------
  # 3️⃣ Execution Role – grants S3, DynamoDB, Bedrock
  # -------------------------------------------------
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: ImagePipelinePermissions
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - s3:DeleteObject
                Resource:
                  - !Sub arn:aws:s3:::${RawImagesBucket}/*
                  - !Sub arn:aws:s3:::${OptimizedImagesBucket}/*
              - Effect: Allow
                Action:
                  - dynamodb:PutItem
                  - dynamodb:GetItem
                  - dynamodb:Query
                Resource: !GetAtt ImageMetadataTable.Arn
              - Effect: Allow
                Action:
                  - bedrock:InvokeModel   # 
    - The `LambdaExecutionRole` includes `bedrock:InvokeModel` (as required by the [Shattered.io tutorial](https://shattered.io/aws-lambda-serverless-tutorial-2026)).
    - Both functions share the same role to keep the example concise; in production you would split them.
    - The `/summary` route is wired exactly like the CRUD routes, satisfying the “wire this into a new GET /summary route” instruction.

### Lambda Code – Image Ingestion, Processing & Summarization

We’ll keep the code in two separate folders for clarity: `src/image_processor/` and `src/summary_api/`. Each folder contains a `requirements.txt` and an `app.py` (or `summary.py`) file. The SAM build step packages them into separate Lambda layers automatically.

#### 1️⃣ Image Processor (src/image_processor/requirements.txt)

#### 2️⃣ Image Processor Handler (src/image_processor/app.py)

# 2️⃣ Download original image
raw_obj = s3.get_object(Bucket=bucket, Key=key)
raw_bytes = raw_obj['Body'].read()
img = Image.open(BytesIO(raw_bytes)).convert('RGB')

# 3️⃣ Run inference – we only need top‑1 class for demo
input_tensor = preprocess(img).unsqueeze(0)  # Shape: [1, 3, 224, 224]
with torch.inference_mode():
    logits = model(input_tensor)
    prob, idx = torch.softmax(logits, dim=1).max(1)
class_id = idx.item()
confidence = prob.item()

# 4️⃣ Resize & compress (WebP 80% quality)
resized = img.resize((800, 800), Image.LANCZOS)
out_buffer = BytesIO()
resized.save(out_buffer, format='WEBP', quality=80)
out_buffer.seek(0)

# 5️⃣ Store optimized version
opt_key = f'optimized/{uuid.uuid4()}.webp'
s3.put_object(
    Bucket=os.getenv('OPT_BUCKET'),
    Key=opt_key,
    Body=out_buffer,
    ContentType='image/webp'
)

# 6️⃣ Persist metadata
image_id = str(uuid.uuid4())
metadata = {
    'image_id': image_id,
    'original_key': key,
    'optimized_key': opt_key,
    'class_id': class_id,
    'confidence': round(confidence, 4),
    'processed_at': context.aws_request_id,
}
table.put_item(Item=metadata)

# 7️⃣ Return a concise response
return {
    'statusCode': 200,
    'body': json.dumps({
        'message': 'Image processed',
        'image_id': image_id,
        'class_id': class_id,
        'confidence': confidence,
        'optimized_key': opt_key
    })
}

**Why this model?** MobileNetV3‑small weighs  70 % top‑1 accuracy on ImageNet – ideal for a “quick‑look” classification before we hand‑off to a more heavyweight Bedrock model for captioning (handled later).

#### 3️⃣ Summary API (src/summary_api/requirements.txt)

#### 4️⃣ Summary API Handler (src/summary_api/summary.py)

# 2️⃣ Pull metadata for today
response = table.scan(
    FilterExpression='processed_at BETWEEN :start AND :end',
    ExpressionAttributeValues={
        ':start': start_of_day.isoformat(),
        ':end': now.isoformat()
    }
)
items = response.get('Items', [])
if not items:
    return {
        'statusCode': 200,
        'body': json.dumps({'summary': 'No images processed today.'})
    }

# 3️⃣ Build prompt for Bedrock
prompt = f"""
You are an assistant summarizing image‑processing activity for a serverless pipeline.
Today ({now.date()}) we processed {len(items)} images. Provide a concise bullet list
with:
• Total images
• Most common predicted class (by class_id)
• Average confidence
• Any failures (none in this demo)
"""
# 4️⃣ Call Bedrock (Claude‑3.0 or fallback)
model_id = os.getenv('BEDROCK_MODEL', 'anthropic.claude-3-0-sonnet')
try:
    resp = bedrock.invoke_model(
        body=json.dumps({
            'prompt': prompt,
            'max_tokens': 256,
            'temperature': 0.3,
            'top_p': 0.9
        }),
        modelId=model_id,
        contentType='application/json',
        accept='application/json'
    )
    summary_text = json.loads(resp['body'].read())['completion']
except Exception as e:
    # Graceful fallback – simple Python aggregation
    summary_text = f"Processed {len(items)} images. Unable to call Bedrock: {str(e)}"

return {
    'statusCode': 200,
    'body': json.dumps({'summary': summary_text})
}

Notice how we reuse the same DynamoDB table for a lightweight query. In a real‑world scenario you would add a `GSI` on `processed_at` for efficient range scans.

### IAM Role – Permissions (including Bedrock invoke)

The `LambdaExecutionRole` defined earlier already grants the three core permissions. For completeness, here is a trimmed‑down policy you can attach to any custom role if you prefer a more granular approach:

Make sure the role is attached to both Lambda functions in the SAM template (`Role: !GetAtt LambdaExecutionRole.Arn`).

### Deploy, Test & Debug

#### 1️⃣ Build the application

During the build SAM reads each `requirements.txt`, creates a Lambda layer, and resolves the `torch` binary wheels compatible with the Amazon Linux 2023 runtime (the base image for Python 3.12 Lambdas in 2026).

2️⃣ Deploy to a test stack</h

---
*Originally published at [https://artificial-inteligence.phptutorial.co.in](https://artificial-inteligence.phptutorial.co.in/ai-powered-serverless-image-processing-pipeline-part-2-setting-up-serverless-functions-with-aws-lambda-python/)*