DEV Community: Dandi Pangestu

Discover how Python, Large Language Models (LLM), and Model Context Protocol (MCP) come together to create an intelligent reporting app. This project simplifies data analysis and delivers automated reports with insightful visualizations.

Dandi Pangestu — Sat, 17 May 2025 13:28:24 +0000

Dandi Pangestu

May 17 '25

How to Build an AI Reporting App with Python, LLM, and Model Context Protocol (MCP)

#python #ai #llm #mcp

Comments

7 min read

How to Build an AI Reporting App with Python, LLM, and Model Context Protocol (MCP)

Dandi Pangestu — Sat, 17 May 2025 12:35:04 +0000

Introduction

In recent years, AI-powered applications have revolutionized the tech industry by automating complex workflows and introducing intelligence at scale. Among these advancements, AI-based reporting systems have emerged as a transformative tool, enabling organizations to extract meaningful insights from data with unprecedented ease and flexibility. These systems are reshaping how businesses approach data analysis, offering intuitive solutions for generating reports tailored to specific requirements.

Two pivotal technologies drive the success of these systems: Large Language Models (LLM) and Model Context Protocol (MCP). LLM, such as Anthropic's cutting-edge model, excel at processing and generating human-like text, making them ideal for interpreting user queries and producing actionable analytical insights. On the other hand, MCP (Model Context Protocol) provides a framework for seamless interaction between AI models and external systems, ensuring that data context is preserved and effectively leveraged throughout the analysis process. Together, LLM and MCP form the backbone of modern AI-driven data solutions.

In this article, we will explore a hands-on mini-project that demonstrates how to build an AI-powered reporting application. This project empowers users to generate flexible, criteria-driven reports effortlessly. By simply typing their reporting criteria in natural language, users can tap into the power of an LLM (Anthropic) and MCP servers to produce analytical results and visualizations automatically. The system integrates two MCP servers:

A Postgres MCP server for querying data sources.
A custom MCP server for generating visualizations.

This architecture enables dynamic, user-driven reporting while minimizing manual setup, showcasing how AI can simplify complex data analysis workflows. Whether you're a data enthusiast or a developer, this project offers insights into leveraging Python, LLM, and MCP to create smarter, more efficient reporting solutions.

You can find the complete code for this project in the Github repository.

Architecture

The architecture of the AI-based reporting app is designed to seamlessly integrate user queries, AI-powered analysis, and dynamic data visualization. At the core of the system is the App, which acts as the central orchestrator. The workflow begins with users submitting a query, which is processed by the app through the following key steps:

Connection to MCP Servers:

The app establishes connections with two specialized MCP servers:
- Postgres MCP Server for querying structured data sources.
- Visualization MCP Server for creating and managing visual representations of the data.
Tool Integration:

Tools from the MCP servers are dynamically added to the app, enabling flexibility in data manipulation and visualization.
Query to LLM:

The app sends the user query to an LLM (Anthropic) for advanced natural language processing and analysis.
LLM Tool Calls:

If the LLM identifies the need for external tools, it makes a call to the appropriate MCP server (e.g., for data queries or visualizations).
Response Generation:

The app consolidates the LLM’s output and any results from the MCP servers, returning a comprehensive response to the user.

Here is the flow diagram:

Prerequisites

Before starting this project, ensure that you have the following tools and resources set up:

Python v3.12 or later: The application is built using Python, so you'll need the latest version installed to take full advantage of its features and compatibility.
Postgres Database: The reporting app relies on a Postgres database as the data source. Make sure to have Postgres installed and running on your system.
Postgres MCP Server: This project uses a Postgres MCP Server to query data from the database. MCP (Model Context Protocol) enhances the context management capabilities of the app by allowing dynamic interactions between the app and external data sources.
Anthropic API Key: The app integrates with Anthropic's LLM (Large Language Model) for natural language understanding and data analysis. You’ll need an API key from Anthropic to access their services.
Node v23.10 or later: Node.js is required to set up and run the Postgres MCP Server. Ensure you have the latest version installed to avoid compatibility issues.

Additionally, an example table schema is provided in the repository to serve as the data source for reporting. You can find the schema here. To get started, execute the SQL script included in the repository to set up the table in your Postgres database. This table will act as the example dataset for the reporting app.

How to Run

Follow these simple steps to set up and run the application:

1. Clone the Repository

First, clone the project repository to your local machine and navigate into the project directory:

$ git clone https://github.com/Dandi-Pangestu/ai-based-reporting-app.git
$ cd ai-based-reporting-app

2. Install Dependencies

Ensure you have Python installed, then use pip to install the required dependencies for the project:

$ pip install .

This will install all the necessary libraries and dependencies listed in the project.

3. Set Up Environment Variables

The application relies on specific environment variables to connect to external services like the database, Anthropic API, and visualization server. To configure these:

Navigate to the api folder and locate the .env.example file.
Copy the .env.example file to a new .env file:

   $ cp api/.env.example api/.env

Open the .env file and fill out the following variables based on your environment:

   ANTHROPIC_API_KEY=your_anthropic_api_key
   DATABASE_HOST=your_database_host
   DATABASE_USERNAME=your_database_username
   DATABASE_PASSWORD=your_database_password
   DATABASE_NAME=your_database_name
   DATABASE_PORT=your_database_port
   VISUALIZATION_SERVER_PATH=your_visualization_server_path

Replace placeholders such as your_anthropic_api_key with the actual API key specific to your setup, and update your_visualization_server_path with the absolute path to the visualization_server.py file, which is located in the mcp_server folder of your project directory.

4. Run the Application

Start the application using the following command:

$ uvicorn main:app --reload

This command runs the application using Uvicorn, a fast ASGI server, with auto-reload enabled for development.

5. Access the Application

Once the server is running, you can access the API on your local machine by navigating to:

http://localhost:8000

The application will now be live and ready to process user queries for generating reports.

Testing Application

Below are the key API endpoints available in the application:

1. Get Available Servers

Retrieve the list of MCP servers connected to the application.

curl --location 'http://localhost:8000/servers'

Response:

2. Get Available Tools

Retrieve the list of tools dynamically added to the application from MCP servers.

curl --location 'http://localhost:8000/tools'

Response:

3. Query the AI Reporting App

Send a query to the application to generate results and visualizations based on the specified criteria.

curl --location 'http://localhost:8000/query' \
--header 'Content-Type: application/json' \
--data '{
    "query": "List the top 3 organization id with the highest average messages count from metrics_rooms. Please generate chart visualization with type bar and also please give me analysis for that result."
}'

Response:

In the /query endpoint, one of the outputs is a base64-encoded chart visualization. This visualization is embedded as a base64 string in the response, representing a chart (e.g., a bar chart or line chart) generated based on your query. You can please copy-paste that base64 to the online base64 image decoder, for example:

The /query endpoint also provides an AI-generated analysis that summarizes trends, highlights key findings, and offers observations based on the query results, making the report easier to understand.

Analysis of the Results:

1. Organization with ID "51160f9a-253c-4db2-a564-80af1321874f" has the highest average message count of 25 messages, showing the most active communication within their rooms.
2. The second highest is organization "b223aab1-68cc-49e6-9b52-b459a0c917e4" with an average of 20 messages, showing moderate but still significant communication activity.
3. The third organization "b462ea35-c804-4cc6-b7a5-535023924ee1" averages 18 messages, which is still considerable but shows less communication intensity compared to the top two.

Key Observations:
- There is a clear hierarchy in communication intensity among these organizations
- The difference between the highest and lowest is 7 messages on average
- All three organizations maintain relatively high message counts, suggesting active engagement in their communication rooms
- The distribution shows a gradual decrease in average message counts rather than sharp drops, indicating consistent usage patterns across these top organizations

This data suggests that these organizations have established effective communication practices, with the top organization particularly standing out in terms of message exchange frequency.

Conclusion

AI-powered reporting applications are transforming the way organizations analyze and interpret their data. By combining the capabilities of Large Language Models (LLM) and Model Context Protocol (MCP), we can create flexible, user-driven systems that simplify complex workflows and deliver meaningful insights effortlessly.

In this article, we explored the architecture and functionality of an AI-based reporting app built with Python, Anthropic’s LLM, and MCP servers. We discussed how the application allows users to generate custom, criteria-driven reports by integrating a Postgres MCP server for querying data and a custom MCP server for creating visualizations. By walking through the process of running and testing the app, we demonstrated how this system leverages cutting-edge technologies to streamline reporting tasks.

Whether you are a developer looking to build similar applications or a data enthusiast exploring new tools for data analysis, this project serves as a practical example of how AI and modern protocols can be applied to solve real-world problems. The combination of Python, LLM, and MCP offers a powerful framework for building smarter, more intuitive solutions.

We hope this guide has provided valuable insights into the potential of AI in reporting systems. Feel free to explore the repository and experiment with the application to see its capabilities firsthand. Happy coding!

Discover how PixelCascade leverages AWS fanout architecture for scalable, serverless image processing. Automate tasks like resizing, formatting & metadata extraction! 🚀 Check out the full guide and architecture insights now!

Dandi Pangestu — Thu, 01 May 2025 07:53:50 +0000

Dandi Pangestu

May 1 '25

PixelCascade: Building a Scalable Image Processing Pipeline with AWS Fanout Architecture

#aws #architecture #serverless #lambda

Comments

15 min read

PixelCascade: Building a Scalable Image Processing Pipeline with AWS Fanout Architecture

Dandi Pangestu — Thu, 01 May 2025 04:53:50 +0000

Introduction

PixelCascade is a mini-project that demonstrates how to build a scalable image processing pipeline using AWS services like S3, Lambda, SNS, SQS, and DynamoDB. By leveraging a fanout architecture, this project allows users to upload images in formats such as JPEG, PNG, and GIF, and process them into formatted outputs, resized versions, and extracted metadata.

This article provides a step-by-step guide to implementing PixelCascade, from architecture design to testing the pipeline. You can find the complete source code for this project on: GitHub Repository.

Background of the Project and Architecture

The primary objective of PixelCascade is to create a scalable and efficient workflow for processing images. Below are the project goals, workflow, and architecture details.

Goals

User Uploads: Accept images in JPEG, PNG, and GIF formats.
Outputs:
- Formatted images
- Converted images
- Extracted metadata
Scalability: Handle high volumes of image uploads without bottlenecks.

Workflow

Image Upload: Users upload a master image to the master folder in an S3 bucket.
Lambda Trigger: A Lambda function processes the S3 upload event and publishes a message to an SNS topic.
SNS Fanout: The SNS topic fans out the message to three SQS queues:
- Format Conversion Queue: Converts the master image to JPEG format and stores it in S3.
- Metadata Extraction Queue: Extracts metadata from the master image and stores it in DynamoDB.
- Resizing Queue: Resizes the master image into thumbnail, medium, and large sizes and stores them in S3.

Architecture Diagram

Requirements

Before setting up PixelCascade, ensure you have the following:

AWS Account: Access to AWS services like S3, Lambda, SNS, SQS, and DynamoDB.
Golang: Required for writing and deploying Lambda functions.

Step-by-Step Implementation

1. Create an SNS Topic

Create an SNS topic named new-image-uploaded to handle messages for every master image uploaded.

Steps:

Navigate to SNS Console in AWS.
Click Create Topic.
Select Standard as the topic type.
Enter new-image-uploaded as the topic name.
Click Create Topic.

2. Create an S3 Bucket and Folders

Create an S3 bucket named pixel-cascade with the following folders:

master: For storing user-uploaded images.
converted: For storing converted images.
resized: For storing resized images.

Steps:

Navigate to S3 Console in AWS.
Click Create Bucket and name it pixel-cascade.
Open the bucket and create three folders: master, converted, and resized.

3. Create and Deploy the Job Publisher Lambda

The Job Publisher Lambda is triggered by S3 events and publishes messages to the SNS topic.

Steps:

Create an IAM Policy:

Name: LambdaJobPublisherPolicy.
Grant permissions to read from S3 and publish to SNS.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents",
        "logs:CreateLogGroup",
        "logs:CreateLogStream"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sns:Publish"
      ],
      "Resource": "arn:aws:sns:*:*:new-image-uploaded"
    }
  ]
}

Create an IAM Role:

Name: LambdaJobPublisherExecutionRole.
Attach the LambdaJobPublisherPolicy.

Write the Lambda Code:

Use Golang to write a Lambda function that:
- Processes S3 events.
- Publishes messages to the new-image-uploaded SNS topic.

package main

import (
    "context"
    "encoding/json"
    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/sns"
    "log"
)

var newImageUploadedTopicArn = "arn:aws:sns:us-east-1:116981782358:new-image-uploaded"

func handler(ctx context.Context, s3Event events.S3Event) error {
    sdkConfig, err := config.LoadDefaultConfig(ctx)
    if err != nil {
        log.Printf("failed to load default config: %s\n", err)
        return err
    }
    snsClient := sns.NewFromConfig(sdkConfig)

    for _, record := range s3Event.Records {
        bucket := record.S3.Bucket.Name
        key := record.S3.Object.URLDecodedKey
        log.Printf("success receive s3 event trigger for bucket: %s and key: %s\n", bucket, key)

        payload := struct {
            Bucket string `json:"bucket"`
            Key    string `json:"key"`
        }{
            Bucket: bucket,
            Key:    key,
        }

        m, err := json.Marshal(payload)
        if err != nil {
            log.Printf("failed to marshal payload: %s\n", err)
            return err
        }
        message := string(m)

        resp, err := snsClient.Publish(ctx, &sns.PublishInput{
            Message:  &message,
            TopicArn: &newImageUploadedTopicArn,
        })
        if err != nil {
            log.Printf("failed to publish message to SNS: %s\n", err)
            return err
        }

        log.Printf("message sent to SNS with message ID: %s\n", *resp.MessageId)
    }

    return nil
}

func main() {
    lambda.Start(handler)
}

Add Trigger:

Set the trigger to the pixel-cascade bucket.
Prefix: master/.

Deploy the Lambda:

Create lambda function.
Package the code into a .zip file and upload it to the AWS Lambda Console.

$ cd job-publisher-function
$ GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o bootstrap -tags lambda.norpc main.go
$ zip job-publisher-function.zip bootstrap

4. Create SQS Queues

Create three SQS queues: format-conversion-queue, metadata-extraction-queue, and resizing-queue.

Steps:

Navigate to SQS Console in AWS.
Click Create Queue and choose Standard Queue.
Enter the queue name and repeat for the other two queues.

5. Edit Access Policies for SQS Queues

Allow the new-image-uploaded SNS topic to send messages to the SQS queues.

6. Create SNS Subscriptions

Subscribe each SQS queue to the new-image-uploaded SNS topic.

Steps:

Navigate to the SNS Console.
Open the new-image-uploaded topic.
Click Create Subscription and set the protocol to SQS.
Provide the ARN of each SQS queue.

7. Deploy Format Converter Lambda

The Format Converter Lambda converts the master image to JPEG and stores it in the converted folder.

Steps:

Create an IAM Policy:

Name: LambdaFormatConverterPolicy.
Grant permissions to read from SQS and write to S3.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents",
        "logs:CreateLogGroup",
        "logs:CreateLogStream"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Resource": "arn:aws:s3:::pixel-cascade/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:ReceiveMessage"
      ],
      "Resource": "arn:aws:sqs:*:*:format-conversion-queue"
    }
  ]
}

Create an IAM Role:

Name: LambdaFormatConverterExecutionRole.
Attach the LambdaFormatConverterPolicy.

Write the Lambda Code:

Use Golang to write a Lambda function that:
- Reads messages from the format-conversion-queue.
- Converts the image to JPEG format.
- Stores the result in the converted folder.

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "image"
    _ "image/gif"
    "image/jpeg"
    _ "image/png"
    "log"
    "path/filepath"
    "strings"
)

var s3Client *s3.Client

const (
    ConvertedFolder = "converted/"
)

type messageBody struct {
    Bucket string `json:"bucket"`
    Key    string `json:"key"`
}

func handler(ctx context.Context, event events.SQSEvent) error {
    sdkConfig, err := config.LoadDefaultConfig(ctx)
    if err != nil {
        log.Printf("failed to load default config: %s\n", err)
        return err
    }
    s3Client = s3.NewFromConfig(sdkConfig)

    for _, record := range event.Records {
        log.Printf("received message from SQS with messageID: %s, body: %s\n", record.MessageId, record.Body)

        var body messageBody
        err := json.Unmarshal([]byte(record.Body), &body)
        if err != nil {
            log.Printf("failed to unmarshal this messageID %s: %s\n", record.MessageId, err)
            return err
        }

        objOutput, err := getObject(ctx, body.Bucket, body.Key)
        if err != nil {
            log.Printf("failed to get object from S3: %s\n", err)
            return err
        }
        defer objOutput.Body.Close()
        log.Printf("content type: %s\n", *objOutput.ContentType)
        log.Printf("content length: %d\n", *objOutput.ContentLength)

        // Decode the image
        img, format, err := image.Decode(objOutput.Body)
        if err != nil {
            log.Printf("failed to decode image: %s\n", err)
            return err
        }
        log.Printf("decoded image format: %s\n", format)

        jpegImg, err := convertToJPEG(img)
        if err != nil {
            log.Printf("failed to convert image: %s\n", err)
            return err
        }

        err = putObject(ctx, body.Bucket, ConvertedFolder+extractFilename(body.Key)+".jpg", jpegImg)
        if err != nil {
            log.Printf("failed to put object to S3: %s\n", err)
            return err
        }
    }

    return nil
}

// extractFilename extracts the filename from the input string
func extractFilename(input string) string {
    // Extract the base name (e.g., aaa.jpg)
    base := filepath.Base(input)
    // Remove the file extension (e.g., aaa)
    name := strings.TrimSuffix(base, filepath.Ext(base))
    return name
}

// getObject retrieves an object from S3
func getObject(ctx context.Context, bucket string, key string) (*s3.GetObjectOutput, error) {
    output, err := s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: &bucket,
        Key:    &key,
    })
    if err != nil {
        log.Printf("failed to get object from S3: %s\n", err)
        return nil, err
    }

    log.Printf("successfully retrieved object from S3 bucket: %s and key: %s\n", bucket, key)

    return output, nil
}

// putObject uploads an object to S3
func putObject(ctx context.Context, bucket string, key string, body []byte) error {
    _, err := s3Client.PutObject(ctx, &s3.PutObjectInput{
        Bucket: &bucket,
        Key:    &key,
        Body:   bytes.NewReader(body),
    })
    if err != nil {
        log.Printf("failed to put object to S3: %s\n", err)
        return err
    }

    log.Printf("successfully put object to S3 bucket: %s and key: %s\n", bucket, key)

    return nil
}

// convertToJPEG converts an image to JPEG format
func convertToJPEG(img image.Image) ([]byte, error) {
    var buf bytes.Buffer

    // Set JPEG options (quality 100)
    options := &jpeg.Options{Quality: 100}

    err := jpeg.Encode(&buf, img, options)
    if err != nil {
        log.Printf("failed to convert image to JPEG: %s\n", err)
        return nil, err
    }
    return buf.Bytes(), nil
}

func main() {
    lambda.Start(handler)
}

Add Trigger:

Set the trigger to the format-conversion-queue.

Set Timeout:

Edit the Lambda configuration to set the execution timeout to 30 seconds.

Deploy the Lambda:

Create lambda function.
Package the code into a .zip file and upload it to the AWS Lambda Console.

$ cd format-converter-function
$ GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o bootstrap -tags lambda.norpc main.go
$ zip format-converter-function.zip bootstrap

8. Deploy Pixel Resizer Lambda

The Pixel Resizer Lambda resizes the master image into three sizes (thumbnail, medium, and large) and stores them in the resized folder.

Steps:

Create an IAM Policy:

Name: LambdaPixelResizerPolicy.
Grant permissions to read from SQS and write to S3.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents",
        "logs:CreateLogGroup",
        "logs:CreateLogStream"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Resource": "arn:aws:s3:::pixel-cascade/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:ReceiveMessage"
      ],
      "Resource": "arn:aws:sqs:*:*:resizing-queue"
    }
  ]
}

Create an IAM Role:

Name: LambdaPixelResizerExecutionRole.
Attach the LambdaPixelResizerPolicy.

Write the Lambda Code:

Use Golang to write a Lambda function that:
- Reads messages from the resizing-queue.
- Resizes the image into three sizes (thumbnail, medium, large).
- Stores the results in the resized folder.

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/nfnt/resize"
    "image"
    _ "image/gif"
    "image/jpeg"
    _ "image/png"
    "log"
    "path/filepath"
    "strings"
    "sync"
)

var s3Client *s3.Client

const (
    ResizedFolder = "resized/"
)

type messageBody struct {
    Bucket string `json:"bucket"`
    Key    string `json:"key"`
}

type resizeTask struct {
    SizeName string
    Width    uint
}

func handler(ctx context.Context, event events.SQSEvent) error {
    sdkConfig, err := config.LoadDefaultConfig(ctx)
    if err != nil {
        log.Printf("failed to load default config: %s\n", err)
        return err
    }
    s3Client = s3.NewFromConfig(sdkConfig)

    for _, record := range event.Records {
        log.Printf("received message from SQS with messageID: %s, body: %s\n", record.MessageId, record.Body)

        var body messageBody
        err := json.Unmarshal([]byte(record.Body), &body)
        if err != nil {
            log.Printf("failed to unmarshal this messageID %s: %s\n", record.MessageId, err)
            return err
        }

        objOutput, err := getObject(ctx, body.Bucket, body.Key)
        if err != nil {
            log.Printf("failed to get object from S3: %s\n", err)
            return err
        }
        defer objOutput.Body.Close()
        log.Printf("content type: %s\n", *objOutput.ContentType)
        log.Printf("content length: %d\n", *objOutput.ContentLength)

        // Decode the image
        img, format, err := image.Decode(objOutput.Body)
        if err != nil {
            log.Printf("failed to decode image: %s\n", err)
            return err
        }
        log.Printf("decoded image format: %s\n", format)

        // Resize the image
        resizeTasks := []resizeTask{
            {"thumbnail", 100},
            {"medium", 500},
            {"large", 1000},
        }

        var wg sync.WaitGroup

        for _, task := range resizeTasks {
            wg.Add(1)
            go func(task resizeTask) {
                defer wg.Done()

                resizedImg := resize.Resize(task.Width, 0, img, resize.Lanczos3)
                if err != nil {
                    log.Printf("failed to resize image: %s\n", err)
                    return
                }

                jpegImg, err := convertToJPEG(resizedImg)
                if err != nil {
                    log.Printf("failed to convert image: %s\n", err)
                    return
                }

                err = putObject(ctx, body.Bucket, ResizedFolder+extractFilename(body.Key)+"/"+task.SizeName+".jpg", jpegImg)
                if err != nil {
                    log.Printf("failed to put object to S3: %s\n", err)
                    return
                }
            }(task)
        }

        wg.Wait()

        log.Printf("all resize tasks completed for image: %s\n", body.Key)
    }

    return nil
}

// extractFilename extracts the filename from the input string
func extractFilename(input string) string {
    // Extract the base name (e.g., aaa.jpg)
    base := filepath.Base(input)
    // Remove the file extension (e.g., aaa)
    name := strings.TrimSuffix(base, filepath.Ext(base))
    return name
}

// getObject retrieves an object from S3
func getObject(ctx context.Context, bucket string, key string) (*s3.GetObjectOutput, error) {
    output, err := s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: &bucket,
        Key:    &key,
    })
    if err != nil {
        log.Printf("failed to get object from S3: %s\n", err)
        return nil, err
    }

    log.Printf("successfully retrieved object from S3 bucket: %s and key: %s\n", bucket, key)

    return output, nil
}

// putObject uploads an object to S3
func putObject(ctx context.Context, bucket string, key string, body []byte) error {
    _, err := s3Client.PutObject(ctx, &s3.PutObjectInput{
        Bucket: &bucket,
        Key:    &key,
        Body:   bytes.NewReader(body),
    })
    if err != nil {
        log.Printf("failed to put object to S3: %s\n", err)
        return err
    }

    log.Printf("successfully put object to S3 bucket: %s and key: %s\n", bucket, key)

    return nil
}

// convertToJPEG converts an image to JPEG format
func convertToJPEG(img image.Image) ([]byte, error) {
    var buf bytes.Buffer

    // Set JPEG options (quality 100)
    options := &jpeg.Options{Quality: 100}

    err := jpeg.Encode(&buf, img, options)
    if err != nil {
        log.Printf("failed to convert image to JPEG: %s\n", err)
        return nil, err
    }
    return buf.Bytes(), nil
}

func main() {
    lambda.Start(handler)
}

Add Trigger:

Set the trigger to the resizing-queue.

Set Timeout:

Edit the Lambda configuration to set the execution timeout to 30 seconds.

Deploy the Lambda:

Create lambda function.
Package the code into a .zip file and upload it to the AWS Lambda Console.

$ cd pixel-resizer-function
$ GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o bootstrap -tags lambda.norpc main.go
$ zip pixel-resizer-function.zip bootstrap

9. Create a DynamoDB Table

Create a DynamoDB table named pixel-metadata with name as the partition key.

10. Deploy Metadata Extractor Lambda

The Metadata Extractor Lambda generates metadata from the master image and stores it in DynamoDB.

Steps:

Create an IAM Policy:

Name: LambdaMetadataExtractorPolicy.
Grant permissions to read from SQS and write to DynamoDB.

Create an IAM Role:

Name: LambdaMetadataExtractorExecutionRole.
Attach the LambdaMetadataExtractorPolicy.

Write the Lambda Code:

Use Golang to write a Lambda function that:
- Reads messages from the metadata-extraction-queue.
- Extracts metadata from the image.
- Stores the metadata in the DynamoDB table.

package main

import (
    "context"
    "encoding/json"
    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/dynamodb"
    "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "image"
    _ "image/gif"
    _ "image/jpeg"
    _ "image/png"
    "io"
    "log"
    "os"
    "path/filepath"
    "strconv"
    "strings"
    "time"
)

var s3Client *s3.Client
var dynamoClient *dynamodb.Client
var tableName = "pixel-metadata"

type messageBody struct {
    Bucket string `json:"bucket"`
    Key    string `json:"key"`
}

func handler(ctx context.Context, event events.SQSEvent) error {
    sdkConfig, err := config.LoadDefaultConfig(ctx)
    if err != nil {
        log.Printf("failed to load default config: %s\n", err)
        return err
    }
    s3Client = s3.NewFromConfig(sdkConfig)
    dynamoClient = dynamodb.NewFromConfig(sdkConfig)

    for _, record := range event.Records {
        log.Printf("received message from SQS with messageID: %s, body: %s\n", record.MessageId, record.Body)

        var body messageBody
        err := json.Unmarshal([]byte(record.Body), &body)
        if err != nil {
            log.Printf("failed to unmarshal this messageID %s: %s\n", record.MessageId, err)
            return err
        }

        metadata, err := generateMetadata(ctx, body.Bucket, body.Key)
        if err != nil {
            log.Printf("failed to extract metadata: %s\n", err)
            return err
        }

        // Creating DynamoDB item
        err = createDynamoDBItem(ctx, map[string]types.AttributeValue{
            "name":     &types.AttributeValueMemberS{Value: extractFilename(body.Key)},
            "bucket":   &types.AttributeValueMemberS{Value: body.Bucket},
            "metadata": &types.AttributeValueMemberM{Value: convertToAttributeValue(metadata)},
        })
        if err != nil {
            log.Printf("failed to create DynamoDB item: %s\n", err)
            return err
        }
    }

    return nil
}

// generateMetadata generates metadata for the image file
func generateMetadata(ctx context.Context, bucket, key string) (map[string]string, error) {
    objOutput, err := getObject(ctx, bucket, key)
    if err != nil {
        log.Printf("failed to get object from S3: %s\n", err)
        return nil, err
    }
    defer objOutput.Body.Close()
    log.Printf("content type: %s\n", *objOutput.ContentType)
    log.Printf("content length: %d\n", *objOutput.ContentLength)

    // Read all bytes from S3 object
    bodyBytes, err := io.ReadAll(objOutput.Body)
    if err != nil {
        log.Printf("failed to read object body: %s\n", err)
        return nil, err
    }

    // Write bytes to temp file
    localFilePath := filepath.Join("/tmp", filepath.Base(key))
    err = os.WriteFile(localFilePath, bodyBytes, 0644)
    if err != nil {
        log.Printf("failed to write to local file: %s\n", err)
        return nil, err
    }
    defer os.Remove(localFilePath)

    // Open file for reading
    file, err := os.Open(localFilePath)
    if err != nil {
        log.Printf("failed to open local file: %s\n", err)
        return nil, err
    }
    defer file.Close()

    // Decode image to get dimensions
    cfg, format, err := image.DecodeConfig(file)
    if err != nil {
        log.Printf("failed to decode image: %s\n", err)
        return nil, err
    }

    // Open the file again to extract EXIF data
    fileInfo, err := os.Stat(localFilePath)
    if err != nil {
        log.Printf("failed to get file info: %s\n", err)
        return nil, err
    }

    // Construct metadata map
    metadata := map[string]string{
        "width":         strconv.Itoa(cfg.Width),
        "height":        strconv.Itoa(cfg.Height),
        "format":        format,
        "file_size":     strconv.FormatInt(fileInfo.Size(), 10), // File size in bytes, // File size in bytes
        "file_name":     fileInfo.Name(),
        "last_modified": fileInfo.ModTime().Format(time.RFC3339),
    }

    return metadata, nil
}

// convertToAttributeValue converts a map of string to AttributeValue
func convertToAttributeValue(data map[string]string) map[string]types.AttributeValue {
    attributeValue := make(map[string]types.AttributeValue)
    for key, value := range data {
        attributeValue[key] = &types.AttributeValueMemberS{Value: value}
    }
    return attributeValue
}

// extractFilename extracts the filename from the input string
func extractFilename(input string) string {
    // Extract the base name (e.g., aaa.jpg)
    base := filepath.Base(input)
    // Remove the file extension (e.g., aaa)
    name := strings.TrimSuffix(base, filepath.Ext(base))
    return name
}

// getObject retrieves an object from S3
func getObject(ctx context.Context, bucket string, key string) (*s3.GetObjectOutput, error) {
    output, err := s3Client.GetObject(ctx, &s3.GetObjectInput{
        Bucket: &bucket,
        Key:    &key,
    })
    if err != nil {
        log.Printf("failed to get object from S3: %s\n", err)
        return nil, err
    }

    log.Printf("successfully retrieved object from S3 bucket: %s and key: %s\n", bucket, key)

    return output, nil
}

// createDynamoDBItem creates a new item in DynamoDB
func createDynamoDBItem(ctx context.Context, item map[string]types.AttributeValue) error {
    _, err := dynamoClient.PutItem(ctx, &dynamodb.PutItemInput{
        TableName: &tableName,
        Item:      item,
    })
    if err != nil {
        log.Printf("failed to put item in DynamoDB: %s\n", err)
        return err
    }

    log.Printf("successfully put item in DynamoDB table: %s\n", tableName)

    return nil
}

func main() {
    lambda.Start(handler)
}

Add Trigger:

Set the trigger to the metadata-extraction-queue.

Set Timeout:

Edit the Lambda configuration to set the execution timeout to 30 seconds.

Deploy the Lambda:

Create lambda function.
Package the code into a .zip file and upload it to the AWS Lambda Console.

$ cd metadata-extractor-function
$ GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o bootstrap -tags lambda.norpc main.go
$ zip metadata-extractor-function.zip bootstrap

11. Test End-to-End

Upload an image to the master folder.

Monitor Lambda execution logs in CloudWatch.

Verify the converted and resized folders in S3.

Check the metadata in DynamoDB.

Conclusion

PixelCascade demonstrates how to build a scalable, serverless image processing pipeline using AWS services. The fanout architecture ensures scalability and decoupling, making the system robust and adaptable to high workloads. By following this guide, you can build and customize your own version of PixelCascade. Happy building!

How to Build a Scalable and Cost-Effective Mass Email Service with AWS SES, SQS, Lambda, S3, and CloudWatch

Dandi Pangestu — Sat, 12 Apr 2025 16:51:31 +0000

Sending mass emails efficiently and reliably is a common challenge for developers and businesses. Through my recent project, I built a scalable mass email service leveraging AWS services like SES, SQS, Lambda, S3, and CloudWatch. This article will walk you through the project, from the architecture to the step-by-step implementation. The source code and resources can be found in the this GitHub repository https://github.com/Dandi-Pangestu/aws-mass-email-service.

Why This Project?

Email campaigns can be tricky to scale, especially when handling high volumes, ensuring deliverability, and maintaining reliability. This project aims to address those challenges by leveraging Amazon Web Services (AWS), enabling a robust and scalable solution for mass email campaigns.

Architecture Overview

The architecture for this project is designed to ensure scalability, fault tolerance, and monitoring. Below is the architecture diagram:

Components in the Architecture

Amazon SES (Simple Email Service): Responsible for sending emails.
Amazon SQS (Simple Queue Service): Adds a queue to handle email requests asynchronously.
AWS Lambda: There are two lambda functions in this project. The first one is for receiving event notification from s3 and then extract the recipient email addresses which exists in the csv file. The second one is for sending each email address to Amazon SES.
Amazon S3: Stores recipient email addresses in CSV form.
Amazon CloudWatch: Amazon CloudWatch is utilized in this project for application logging, enabling the monitoring of both success and error events during the email processing pipeline. It ensures transparency and provides actionable insights by capturing logs for successfully executed processes as well as errors that might occur, making debugging and system optimization more efficient.

Real-World Use Cases

This scalable AWS mass email service can be applied in various real-world scenarios, such as:

Marketing Campaigns: Businesses can use this service to send promotional emails to thousands, tens of thousands, or even millions of users. The architecture ensures that the emails are sent reliably without overwhelming the system.
Transactional Emails: E-commerce platforms can use this setup to send order confirmations, shipping updates, and other transactional emails. The asynchronous processing ensures that alerts are sent promptly.
Notifications System: Organizations can send alerts to their users, such as password reset emails, security notifications, or event reminders. The monitoring capabilities of CloudWatch make it easy to track the success and failure rates of such notifications (https://docs.aws.amazon.com/ses/latest/dg/monitor-using-event-publishing.html).
Newsletter Distribution: Content creators and publishers can distribute newsletters to their subscribers efficiently, with the ability to customize email templates stored in S3.
Community Announcements: Non-profits and community organizations can send announcements to their members, ensuring that the communication reaches everyone in the network.

Pricing Prediction for AWS Services

Before implementing any project, understanding the potential costs is crucial. Here's a breakdown of the pricing for the main AWS services used in this project.

Amazon SES
- Free Tier: 62,000 emails per month when sent from an Amazon EC2 instance.
- Pricing Beyond Free Tier: $0.10 per 1,000 emails sent.
- Additional Costs: $0.12 per GB of attachments.
Amazon SQS
- Free Tier: 1 million requests per month.
- Pricing Beyond Free Tier: $0.40 per 1 million requests (for standard queues).
- Data Transfer Costs: Additional charges apply for data transfer out beyond the free tier.
AWS Lambda
- Free Tier: 1 million free requests and 400,000 GB-seconds of compute time per month.
- Pricing Beyond Free Tier:
  - $0.20 per 1 million requests.
  - $0.00001667 per GB-second of compute time.
Amazon S3
- Free Tier: 5 GB of standard storage.
- Pricing Beyond Free Tier:
  - $0.023 per GB for the first 50 TB of standard storage.
  - $0.005 per 1,000 PUT, COPY, POST, or LIST requests.
  - $0.0004 per 1,000 GET and SELECT requests.

Cost Example
For a campaign sending 100,000 emails monthly with attachments:

SES: $10 for emails + $1.20 for 10 GB of attachments.
SQS: $0.40 for 1 million requests (assuming 1 email = 1 request).
Lambda: $0.60 for 3 million requests (batch processing) and 2.5 GB-seconds of compute.
S3: Approximately $0.50 for 20 GB of templates and logs.

Estimated Total Monthly Cost: ~$12.50 (excluding free tier).
Costs can vary depending on usage, so always use the AWS Pricing Calculator (https://calculator.aws/#/) for accurate predictions.

Step-by-Step Implementation

Instead of re-explaining the implementation process here, I've detailed each step in the tutorials section (https://github.com/Dandi-Pangestu/aws-mass-email-service/tree/main/tutorials) of the GitHub repository. The tutorials provide:

Detailed step-by-step instructions on setting up the AWS services.
Code snippets for configuration and deployment.
Examples of how to test the service.

Challenges and How They Were Solved

Handling Large Email Volumes: By using SQS and Lambda, the system can process emails in batches, avoiding bottlenecks.
Improving Email Deliverability: Optimizing SES settings, such as ensuring SPF, DKIM, and DMARC records, helped improve email deliverability rates.
Monitoring and Debugging: CloudWatch provided actionable insights into system health and email delivery metrics, making it easier to identify and resolve issues.

What Makes This Solution Scalable?

The decoupled architecture allows each component to scale independently. For example:

SQS ensures reliable queuing of email requests.
Lambda scales out automatically to process messages in parallel.
SES handles high email volumes efficiently, adhering to AWS limits.

Conclusion

This project demonstrates how AWS services can be combined to build a powerful, scalable mass email service. Whether you're a developer or a business owner, this solution can help streamline your email campaigns while ensuring reliability and scalability. You can explore the project further in the GitHub repository https://github.com/Dandi-Pangestu/aws-mass-email-service. Feel free to adapt it to your needs and share your feedback!

Next Steps

Future improvements could include:

Adding a user-friendly interface for managing email campaigns.
Implementing analytics for campaign performance.
Add functionality to handle transactional emails, allowing the system to send emails triggered from any external source or application.

If this article inspired or helped you in any way, please leave a ❤️ and share it with your network. Happy coding!