DEV Community: Amr Abed

Self-Host an AI Agent Using OpenClaw and WhatsApp Cloud API (2026)

Amr Abed — Sun, 04 Oct 2026 13:40:19 +0000

Set up a WhatsApp Business number for your AI agent in under 30 minutes — no third-party providers, no ban risk, using official WhatsApp Cloud API

OpenClaw + WhatsApp

You shouldn’t need to pay a middleman just to put your AI agent on WhatsApp — yet that’s the default everyone ends up recommending. After spending days wading through scattered resources, I finally got OpenClaw talking directly to the WhatsApp Cloud API. Here’s the exact path I took.

In this guide, I am sharing:

Prerequisites
How it works
A Step-by-step Guide
The resources I used

Prerequisites

You obviously need OpenClaw for this. To install it on your Linux/Mac server, use the following command:

curl -fsSL https://openclaw.ai/install.sh | bash

For other installation options, follow the instructions here.

Meta Developer Account

You will need to register as a Meta developer if you are not already.

Meta Business Portfolio

You will also need a Meta Business Portfolio. If you don’t have one, you can create it on the Meta Business Suite⁠.

Public Domain Name

You will need a public domain for WhatsApp to send callback requests to OpenClaw.

How It Works

The diagram below shows how the different services communicate with each other while keeping your local server secure.

Request/Response Flow between OpenClaw and WhatsApp

Here is how it works:

You send a message on WhatsApp
WhatsApp sends a notification to a webhook running on your server through the tunnel
The webhook passes the message to your plugin to process and pass to the OpenClaw agent
The agent processes replies to the message and passes the answer back to the plugin
The plugin then communicates with the WhatsApp Cloud API to send you back your response

Step-by-step Guide

Step 0 (One-Time Setup): Create a WhatsApp Business Account

From the Business Settings⁠, create a new WhatsApp account if you don’t have one:

Select WhatsApp accounts from the sidebar
Click the + Add button in the top-right corner and select Create a WhatsApp Business account
Enter a display name and select a category, e.g., Other. Make sure the display name aligns with the guidelines
Select Use a display name only and click Continue —WhatsApp will create a virtual number for you
Select the new WhatsApp account
Select the Phone Numbers tab
Click the created phone number to open the Phone Profile
Copy the ID and store it in a secure place to use later

Step 1: Create a Meta app & Connect WhatsApp Business account

From the App Dashboard, create a new Meta app if you don’t have one:

Click on Create App
Enter the app name and contact email
Select Connect with customers through WhatsApp use case
Select an existing Meta Business Portfolio or create one
Review details and click Create app
Select the app from the App Dashboard
Expand App settings near the bottom of the sidebar and select Basic
Show and copy the App secret and store it in a secure place to use later
Select Use cases from the sidebar
Select Connect with customers through WhatsApp and click Customize
In the API Setup section, select the WhatsApp Business Account

3. Select **Connect with customers through WhatsApp* use case*

Step 2: Generate an API Access Token

From the Business Settings⁠:

Select System users from the sidebar
Click the + Add button in the top-right corner to create a system user
Enter the name for the system user and select a role
Select the new system user and click Assign Assets
From Apps, select the app and toggle Test app
From WhatsApp accounts, select the account and toggle Everything
Click the Assign assets button
Click the Generate token button to generate the API token
Add the following permissions to the token: whatsapp_business_messaging and whatsapp_business_management
Copy the token and store it in a secure place to use later

Step 3: Set up a tunnel to your server

I am using CloudFlare for that, but feel free to use whichever setup you prefer. On your server, run:

cloudflared tunnel --hostname <your-subdomain> run --url tcp://localhost:3100

You can also do the same from the CloudFlare dashboard under Zero Trust → Networks → Connectors

Make sure you have the CloudFlareIP list added to the trustedProxies of the OpenClaw gateway. The gateway object of the openclaw.json should look something like this:

"gateway": {    "port": 18789,    "mode": "local",    "bind": "loopback",    "controlUi": {      "allowedOrigins": [        "${ALLOWED_ORIGIN}"      ]    },    "auth": {      "mode": "token",      "token": "${OPENCLAW_GATEWAY_TOKEN}"    },    "trustedProxies": [      "173.245.48.0/20",      "103.21.244.0/22",      "103.22.200.0/22",      "103.31.4.0/22",      "141.101.64.0/18",      "108.162.192.0/18",      "190.93.240.0/20",      "188.114.96.0/20",      "197.234.240.0/22",      "198.41.128.0/17",      "162.158.0.0/15",      "104.16.0.0/13",      "104.24.0.0/14",      "172.64.0.0/13",      "131.0.72.0/22"    ],    "tailscale": {      "mode": "off",      "resetOnExit": false    },    "nodes": {      "denyCommands": [        "camera.snap",        "camera.clip",        "screen.record",        "contacts.add",        "calendar.add",        "reminders.add",        "sms.send"      ]    }  }

Step 4: Install and configure the OpenClaw WhatsApp-Cloud plugin

I am using this unofficial OpenClaw WhatsApp plugin: openclaw-whatsapp-cloud-api. It is a TypeScript app that communicates with the WhatsApp Cloud API. If you prefer, you can actually build your own by following the instructions here.

To install the plugin, use these commands:

cd ~/.openclawmkdir plugins && cd pluginsgit clone git@github.com:mcostantino-dev/openclaw-whatsapp-cloud-api.gitmv openclaw-channel-whatsapp-cloud openclaw-whatsappcd openclaw-whatsappnpm install && npm run buildopenclaw plugins install -l .

Note: the above commands are slightly different from the ones on the repo

Now, let’s configure the plugin using environment variables:

cd ~/.openclawnano .env

Copy and paste these values into your .env file:

# WhatsApp Cloud APIWHATSAPP_PHONE_NUMBER_ID=WHATSAPP_ACCESS_TOKEN=WHATSAPP_APP_SECRET=WHATSAPP_VERIFY_TOKEN=WHATSAPP_WEBHOOK_PATH=/webhook/whatsapp-cloudALLOWED_PHONE=

Fill the values in the file with the following:

WhatsApp Phone number ID: the phone number ID from step 0
WhatsApp App Secret: the app secret from step 1
WhatsApp Access Token: the token from step 2
WhatsApp Verify Token: any string you prefer to use as a verify token
Allowed Phone: your phone number

Next, open the openclaw.json file and update the whatsapp-cloudfields so the file looks like that:

"channels": {  ... // other channels here, if any  "whatsapp-cloud": {      "enabled": true,      "phoneNumberId": "${WHATSAPP_PHONE_NUMBER_ID}",      "accessToken": "${WHATSAPP_ACCESS_TOKEN}",      "appSecret": "${WHATSAPP_APP_SECRET}",      "verifyToken": "${WHATSAPP_VERIFY_TOKEN}",      "webhookPath": "${WHATSAPP_WEBHOOK_PATH}",      "webhookPort": 3100,      "dmPolicy": "allowlist",      "allowFrom": [        "${ALLOWED_PHONE}"      ]   }},  ... // other objects here, e.g., gateway, skills, etc."plugins": {   "load": {      "paths": [        "~/.openclaw/plugins/whatsapp-cloud"      ]    },    "entries": {      ... // other plugins here      "openclaw-whatsapp-cloud-api": {        "enabled": true      }    }  }

Finally, restart the OpenClaw gateway

openclaw gateway restart

You should now have a webhook listening on port 3100 for WhatsApp Cloud API callback requests

Step 5: Configure the webhook

From the App Dashboard:

Select the app
Select Use cases from the sidebar
Select Connect with customers through WhatsApp and click Customize
Select Configuration
For the Callback URL, enter https:///webhook/whatsapp-cloud
For the Verify token, enter the verify token you chose in step 4
The API verifies your webhook, then shows a list of webhook fields
Under Webhook fields, enable messages

Step 6: Test the connection

Open WhatsApp on your phone, search for the WhatsApp business phone number, and start chatting. Your OpenClaw agent should respond to you. Enjoy!

Troubleshooting

OpenClaw gateway failing to start

Make sure openclaw.json has no issues. Run the following command to check:

journalctl --user -u openclaw-gateway.service -f

It should give you a list of issues in the file.

Tunnel not connecting

Make sure the webhook is listening on your server on port 3100 using this command:

netstat -tap tcp | grep 3100

You should get something like:

tcp6  0  0 [::]:3100  [::]:*   LISTEN   123456/openclaw-gat

Webhook verification failure

Make sure your callback URL is reachable from the internet. Try running this command on your local:

curl "https://<your-subdomain>/webhook/whatsapp-cloud?hub.mode=subscribe&hub.challenge=123&hub.verify_token=<verify-token>"

Remember to replace and with your values. You should get 123 back

References

OpenClaw documentation
WhatsApp Cloud API Get Started
WhatsApp Business Accounts documentation
CloudFlare Tunnel Documentation
OpenClaw WhatsApp Plugin (unofficial): mcostantino-dev/openclaw-whatsapp-cloud-api

You should now have a fully functioning WhatsApp number for your OpenClaw agent, running securely through your own tunnel without third-party fees. What kind of AI agent are you building with OpenClaw? Let me know in the comments below! If you run into any issues during setup, drop a question, and I’ll help you troubleshoot.

Self-Host an AI Agent Using OpenClaw and WhatsApp Cloud API (2026) was originally published in Capsulat on Medium, where people are continuing the conversation by highlighting and responding to this story.

Your AI Code Can Be Elegant Too

Amr Abed — Fri, 02 Oct 2026 13:20:44 +0000

As the Machine Learning Engineering Manager at an AI-powered SaaS company, I get a front-row seat to the machine learning (ML) code written by data science teams. When I'm not reviewing models and production pipelines, I dabble in the occasional Kaggle competition, though I'm more of an enthusiastic competitor than a podium regular.

A recurring pattern quickly emerges: many brilliant data scientists come from mathematics, statistics, or academic research where code is treated merely as a vehicle to run an experiment. Python was adopted as a friendlier upgrade from R or MATLAB. The resulting code may hit top leaderboard accuracy, but in terms of software craftsmanship, it's often about as elegant as a spork.

"Working" code is no longer enough when models move from quick experiments to production pipelines.

Photo by Chris Ried on Unsplash

Why Should You Care About Clean ML Code?

"Clean code always looks like it was written by someone who cares."

— Michael Feathers, quoted in Robert C. Martin's Clean Code

When you write machine learning code, you're communicating with an audience: your future self six months from now, your teammates, and the engineers responsible for running your model in production.

My software journey began in 2003 as a Computer Engineering student at one of Egypt's top universities. Starting with C++ (rather than C) naturally trained me in object-oriented programming (OOP), separation of concerns, and clean architectural design.

Fast-forward to graduate school at Virginia Tech (2014–2017): I dove into machine learning and deep learning, tinkering with Pandas and early versions of TensorFlow (yes, versions 0.12 and 1.0 — Google, I’ll take that apology now).

Coming from that engineering background, I’ve always had a soft spot for clean code. For me, writing clean code is like signing a piece of art. That includes machine learning code. At one point, I even took a stab at rewriting DL4J (Deep Learning for Java) with a modern object-oriented architecture.

In modern MLOps environments, orchestrating workloads on platforms like MLflow, TFX, Kubeflow, and AWS SageMaker, sloppy code isn't just an eyesore; it's a real operational risk. Silent bugs, unreproducible splits, untracked dependencies, and unreadable transformations slow down teams and break downstream services.

The 3-Tier ML Code Maturity Model

To turn this philosophy into an actionable engineering workflow, I think of ML code quality as progressing through three distinct maturity tiers:

graph TD
    T1["Tier 1: Exploratory Script (Notebook Prototype)<br/>• Monolithic scripts, smashed imports, global variables<br/>• Unencapsulated state, unrepeatable cell execution"]
    T2["Tier 2: Clean Idiomatic Python (The 'Refactor')<br/>• Surgical imports, PEP 8 compliance, descriptive naming<br/>• Hyperparameter constants, clean logical stages"]
    T3["Tier 3: Modular Object-Oriented Architecture (Production)<br/>• Encapsulated classes (Pipeline, Classifier)<br/>• Strongly-typed Pydantic configuration models<br/>• Unit-testable, importable into APIs & MLOps pipelines"]

    T1 -->|Refactor & Clean| T2
    T2 -->|Encapsulate & Modularize| T3

Let's walk through the foundational rules that take you from Tier 1 to Tier 2, and then explore how object-oriented design elevates your code to Tier 3.

5 Foundational Rules for Clean ML Code

1. Learn Your Tools (Stop Reinventing Vectorized Operations)

Libraries like NumPy, Pandas, PyTorch, and TensorFlow provide highly optimized C and CUDA backends. Take the time to study the API references before resorting to Python for loops across lists.

I once reviewed a pipeline where a data scientist had written a 40-line nested loop to normalize features across a DataFrame. The whole thing could be replaced with a single sklearn.preprocessing.StandardScaler call — and it ran about 50x faster. If you find yourself writing custom nested loops to compute metrics or transform arrays, there's almost certainly an optimized, idiomatic operation already built for the job.

2. Adhere to Python Naming Conventions (PEP 8)

In linear algebra, $X$ is a feature matrix and $y$ is a target vector. But Python doesn't care about mathematical conventions — from the interpreter's perspective, both are just variables.

Variables and functions in Python use snake_case (lowercase with underscores) as defined by PEP 8:

“Variable names follow the same convention as function names.”

“Function names should be lowercase, with words separated by underscores as necessary to improve readability.”

3. Give Your Variables Descriptive Names

I can't count how many times I've opened a notebook and found df, df2, df_final, and df_final_v2 all living in the same file. Single-letter abbreviations and generic shorthand make notebooks needlessly hard to follow. Replace them with names that communicate intent:

Cryptic / Notebook Style	Descriptive & Explicit	Why It Matters
`df`	`raw_data` / `customer_churn_data`	Identifies the actual domain entity
`x`, `y`	`features`, `labels` (or `targets`)	Disambiguates inputs and prediction goals
`train_ds`, `val_ds`, `test_ds`	`training_data`, `validation_data`, `test_data`	Clear without requiring mental decoding
`m` or `clf`	`model` / `classifier`	Communicates role and purpose clearly

4. Be Precise and Surgical with Imports

Why import the entire numpy namespace when all you need is array and expand_dims? Why pull in all of pandas when you just need read_csv?

Instead of broad, monolithic imports:

import pandas as pd

df = pd.read_csv(file)

Be surgical:

from pandas import DataFrame, read_csv

data: DataFrame = read_csv(file)

Surgical imports get rid of redundant module prefixes and make your dependencies immediately transparent.

When you need functions with identical names from different packages (e.g., load from json and load from pickle), use explicit aliases:

from json import load as load_json
from pickle import load as load_pickle

The Pragmatic Rule: Clarity Over Dogmatism

I should be honest here — in data science, aliases like import numpy as np and import pandas as pd are virtually universal conventions. If you're manipulating dozens of array operations or DataFrame joins across a file, typing np.mean or pd.concat is perfectly fine and preserves helpful namespace context.

The problem arises when you import massive framework submodules wholesale:

# Cluttered: Forces repetitive layers.* prefixes everywhere
from keras import layers

layer = layers.Dense(64)

Versus:

# Clean: Direct, explicit, and self-documenting
from keras.layers import Dense

layer = Dense(64)

The goal isn't blind dogmatism—it's intentionality. Know when a namespace prefix adds genuine clarity versus when it simply introduces noise.

5. Graduate from Loose Notebooks to a Modern IDE

Databricks and Google Colab are great for initial experimentation, but they're rarely enough for building robust, production-grade systems.

Modern IDEs like Visual Studio Code and PyCharm support Jupyter notebooks natively while giving you first-class software engineering tools:

Version control with GitHub pull requests, branch protection, and diff reviews
Automated formatting & linting with Ruff: Written in Rust, Ruff has rapidly become the modern standard in Python engineering, replacing Black, Flake8, and isort simultaneously while running 10–100x faster.
Data & Configuration validation with Pydantic: Catch schema mismatches, type errors, and invalid hyperparameter values before running expensive multi-hour training runs.
Static type checking with Mypy or Pyright: Detect tensor dimension mistakes and invalid argument types at development time.
AI code assistance with GitHub Copilot and Gemini
Cloud compute integration for remote debugging on GPUs and TPUs

Pro Tip for Notebook Repositories: If your team must commit .ipynb files to Git, install nbstripout as a pre-commit hook. It automatically strips cell outputs, execution counts, and bloated base64 image strings before commits, turning unreadable multi-thousand-line JSON diffs into clean, reviewable code changes.

💡 Quick Win: Run pip install ruff && ruff check . on your project folder today. You'll instantly catch unused imports, undefined variables, and formatting inconsistencies in milliseconds.

Case Study: Refactoring a Transfer Learning Pipeline

Enough theory — let's look at a concrete example. Here's an image classification transfer learning workflow based on the TensorFlow/Keras documentation.

The "Before" Script (Tier 1: Exploratory Script)

This is what typical data science code looks like before any cleanup:

import keras
from keras import layers
import matplotlib.pyplot as plt
import numpy as np
from tensorflow import data as tf_data
import tensorflow_datasets as tfds

tfds.disable_progress_bar()
train_ds, validation_ds, test_ds = tfds.load(
    "cats_vs_dogs",
    # Reserve 10% for validation and 10% for test
    split=["train[:40%]", "train[40%:50%]", "train[50%:60%]"],
    as_supervised=True,  # Include labels
)
print(f"Number of training samples: {train_ds.cardinality()}")
print(f"Number of validation samples: {validation_ds.cardinality()}")
print(f"Number of test samples: {test_ds.cardinality()}")

plt.figure(figsize=(10, 10))
for i, (image, label) in enumerate(train_ds.take(9)):
    ax = plt.subplot(3, 3, i + 1)
    plt.imshow(image)
    plt.title(int(label))
    plt.axis("off")

resize_fn = keras.layers.Resizing(150, 150)
train_ds = train_ds.map(lambda x, y: (resize_fn(x), y))
validation_ds = validation_ds.map(lambda x, y: (resize_fn(x), y))
test_ds = test_ds.map(lambda x, y: (resize_fn(x), y))

augmentation_layers = [
    layers.RandomFlip("horizontal"),
    layers.RandomRotation(0.1),
]


def data_augmentation(x):
    for layer in augmentation_layers:
        x = layer(x)
    return x


train_ds = train_ds.map(lambda x, y: (data_augmentation(x), y))

batch_size = 64
train_ds = train_ds.batch(batch_size).prefetch(tf_data.AUTOTUNE).cache()
validation_ds = validation_ds.batch(batch_size).prefetch(tf_data.AUTOTUNE).cache()
test_ds = test_ds.batch(batch_size).prefetch(tf_data.AUTOTUNE).cache()

for images, labels in train_ds.take(1):
    plt.figure(figsize=(10, 10))
    first_image = images[0]
    for i in range(9):
        ax = plt.subplot(3, 3, i + 1)
        augmented_image = data_augmentation(np.expand_dims(first_image, 0))
        plt.imshow(np.array(augmented_image[0]).astype("int32"))
        plt.title(int(labels[0]))
        plt.axis("off")

base_model = keras.applications.Xception(
    weights="imagenet",  # Load weights pre-trained on ImageNet.
    input_shape=(150, 150, 3),
    include_top=False,
)  # Do not include the ImageNet classifier at the top.

# Freeze the base_model
base_model.trainable = False

inputs = keras.Input(shape=(150, 150, 3))
scale_layer = keras.layers.Rescaling(scale=1 / 127.5, offset=-1)
x = scale_layer(inputs)
x = base_model(x, training=False)
x = keras.layers.GlobalAveragePooling2D()(x)
x = keras.layers.Dropout(0.2)(x)  # Regularize with dropout
outputs = keras.layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

model.summary(show_trainable=True)

model.compile(
    optimizer=keras.optimizers.Adam(),
    loss=keras.losses.BinaryCrossentropy(from_logits=True),
    metrics=[keras.metrics.BinaryAccuracy()],
)

epochs = 2
print("Fitting the top layer of the model")
model.fit(train_ds, epochs=epochs, validation_data=validation_ds)

base_model.trainable = True
model.summary(show_trainable=True)

model.compile(
    optimizer=keras.optimizers.Adam(1e-5),  # Low learning rate
    loss=keras.losses.BinaryCrossentropy(from_logits=True),
    metrics=[keras.metrics.BinaryAccuracy()],
)

epochs = 1
print("Fitting the end-to-end model")
model.fit(train_ds, epochs=epochs, validation_data=validation_ds)

print("Test dataset evaluation")
model.evaluate(test_ds)

The Code Review Critique

I see several opportunities for cleanup here:

Unnecessary module imports: numpy is imported solely for expand_dims and array.
Heavy plotting imports: matplotlib.pyplot is imported in its entirety for just four functions (axis, figure, imshow, subplot).
Repeated module prefixes: Importing layers causes repetitive layers.* prefixes throughout the model definition.
Redundant package namespaces: The same issue affects keras.optimizers, keras.losses, and keras.metrics.
Vague variable names: train_ds, validation_ds, and test_ds can be renamed to training_data, validation_data, and test_data.
Hidden constants: batch_size is a constant hyperparameter, but defined as a mutable variable.
Reused and redundant variables: epochs is defined and immediately consumed, obscuring the parameter at the call site.
Unformatted structure: Lacks consistent code formatting (e.g., Ruff/Black) and organized import blocks.

Image by Glen Lipka on Commadot (inspired by Thom Holwerda’s post on OSNews)

The Refactored Script (Tier 2: Clean Idiomatic Python)

Here is the same code after applying the foundational rules from above:

from keras import Model
from keras.applications import Xception
from keras.layers import (
    Dense,
    Dropout,
    GlobalAveragePooling2D,
    Input,
    RandomFlip,
    RandomRotation,
    Rescaling,
    Resizing,
)
from keras.losses import BinaryCrossentropy
from keras.metrics import BinaryAccuracy
from keras.optimizers import Adam
from matplotlib.pyplot import axis, figure, imshow, subplot, title
from numpy import array, expand_dims
from tensorflow.data import AUTOTUNE
from tensorflow_datasets import disable_progress_bar, load

# 1. Load and split the dataset
disable_progress_bar()
training_data, validation_data, test_data = load(
    "cats_vs_dogs",
    split=["train[:40%]", "train[40%:50%]", "train[50%:60%]"],
    as_supervised=True,  # Include labels
)
print(f"Number of training samples: {training_data.cardinality()}")
print(f"Number of validation samples: {validation_data.cardinality()}")
print(f"Number of test samples: {test_data.cardinality()}")

# 2. Inspect training samples
figure(figsize=(10, 10))
for i, (image, label) in enumerate(training_data.take(9)):
    ax = subplot(3, 3, i + 1)
    imshow(image)
    title(int(label))
    axis("off")

# 3. Standardize image size
resize = Resizing(150, 150)
training_data = training_data.map(lambda x, y: (resize(x), y))
validation_data = validation_data.map(lambda x, y: (resize(x), y))
test_data = test_data.map(lambda x, y: (resize(x), y))

# 4. Data augmentation
augmentation_layers = [RandomFlip("horizontal"), RandomRotation(0.1)]


def data_augmentation(images):
    for layer in augmentation_layers:
        images = layer(images)
    return images


training_data = training_data.map(lambda x, y: (data_augmentation(x), y))

# 5. Batch and optimize pipeline
BATCH_SIZE = 64
training_data = training_data.batch(BATCH_SIZE).prefetch(AUTOTUNE).cache()
validation_data = validation_data.batch(BATCH_SIZE).prefetch(AUTOTUNE).cache()
test_data = test_data.batch(BATCH_SIZE).prefetch(AUTOTUNE).cache()

# 6. Inspect augmented batch
for images, labels in training_data.take(1):
    figure(figsize=(10, 10))
    first_image = images[0]
    for i in range(9):
        ax = subplot(3, 3, i + 1)
        augmented_image = data_augmentation(expand_dims(first_image, 0))
        imshow(array(augmented_image[0]).astype("int32"))
        title(int(labels[0]))
        axis("off")

# 7. Configure pre-trained base model
base_model = Xception(
    weights="imagenet",  # Load weights pre-trained on ImageNet
    input_shape=(150, 150, 3),
    include_top=False,  # Exclude ImageNet top classifier
)
base_model.trainable = False  # Freeze base weights for initial training

# 8. Build custom classification head
inputs = Input(shape=(150, 150, 3), name="input")
x = Rescaling(scale=1 / 127.5, offset=-1)(inputs)  # Scale [0, 255] to [-1, 1]
x = base_model(x, training=False)
x = GlobalAveragePooling2D()(x)
x = Dropout(0.2)(x)  # Regularize with dropout
outputs = Dense(1)(x)
model = Model(inputs, outputs)
model.summary(show_trainable=True)

# 9. Train custom head
model.compile(
    optimizer=Adam(),
    loss=BinaryCrossentropy(from_logits=True),
    metrics=[BinaryAccuracy()],
)
model.fit(training_data, epochs=2, validation_data=validation_data)

# 10. Fine-tune end-to-end
base_model.trainable = True  # Unfreeze base model
model.summary(show_trainable=True)
model.compile(
    optimizer=Adam(1e-5),  # Lower learning rate to preserve learned representations
    loss=BinaryCrossentropy(from_logits=True),
    metrics=[BinaryAccuracy()],
)
model.fit(training_data, epochs=1, validation_data=validation_data)

# 11. Evaluate on test set
model.evaluate(test_data)

The Power of Explicit Imports

Here's what I really love about surgical imports — they give you a high-level summary of your entire model architecture right at the top of the file:

from keras import Model
from keras.applications import Xception
from keras.layers import (
    Dense,
    Dropout,
    GlobalAveragePooling2D,
    Input,
    RandomFlip,
    RandomRotation,
    Rescaling,
)
from keras.losses import BinaryCrossentropy
from keras.metrics import BinaryAccuracy
from keras.optimizers import Adam

Within seconds of opening the file, any engineer or reviewer can tell:

Base Model: Transfer learning with pre-trained Xception
Layers: Rescaling, GlobalAveragePooling2D, Dropout, and Dense
Data Augmentation: RandomFlip and RandomRotation
Optimization Strategy: Adam optimizer, BinaryCrossentropy loss, and BinaryAccuracy metric

No need to dig through hundreds of lines of notebook code just to figure out what model is being trained.

Taking It to the Next Level: Object-Oriented ML Architecture (Tier 3)

Tier 2 is a dramatic improvement, but flat procedural scripts still have real limitations when they need to live inside production software. I learned this the hard way when one of my teams tried to deploy a Tier 2 script behind a FastAPI endpoint — importing the module triggered a 2 GB dataset download on every cold start.

Here are the three problems that keep showing up:

Global State Pollution: Variables like base_model, model, and training_data float in global module scope. In notebooks or long-running worker processes, this leads to memory leaks and accidental state bleeding.
Untestable Code: You can't write isolated unit tests for your data augmentation or model construction without running the entire training pipeline end-to-end.
No Reusability: If an API engineer needs to serve inference from your trained model in FastAPI or AWS Lambda microservices, they can't cleanly import your model logic without triggering dataset downloads and training routines.

This is where Object-Oriented Design (OOP), Separation of Concerns, and strongly-typed Pydantic models come in.

Architectural Component Flow

graph TD
    Config["TrainingConfig<br/>(batch_size, image_size, epochs, lr)"]
    Config -->|config| Pipeline["ImageDatasetPipeline<br/>• Download & Split<br/>• Resize & Augment<br/>• Batch & Cache (AUTOTUNE)"]
    Config -->|config| Classifier["TransferLearningClassifier<br/>• Base Xception (Frozen)<br/>• Custom Dense Head<br/>• Head Train / Fine-tune"]
    Pipeline -->|train, val, test| Classifier
    Classifier --> Model["Evaluated Model<br/>(Inference Ready)"]

Designing the Components

I like to decompose this into three focused responsibilities:

TrainingConfig: An immutable, validated Pydantic model holding all hyperparameters and configurations.
ImageDatasetPipeline: Handles downloading, splitting, caching, and augmenting dataset batches.
TransferLearningClassifier: Handles building the neural network, compiling, training, fine-tuning, and evaluating.

Here's the Tier 3 implementation:

from keras import Model
from keras.applications import Xception
from keras.layers import (
    Dense,
    Dropout,
    GlobalAveragePooling2D,
    Input,
    RandomFlip,
    RandomRotation,
    Rescaling,
    Resizing,
)
from keras.losses import BinaryCrossentropy
from keras.metrics import BinaryAccuracy
from keras.optimizers import Adam
from numpy import ndarray
from pydantic import BaseModel, ConfigDict
from tensorflow import Tensor
from tensorflow.data import AUTOTUNE, Dataset
from tensorflow_datasets import disable_progress_bar, load


class TrainingConfig(BaseModel):
    """Hyperparameters and runtime settings for the model pipeline."""

    model_config = ConfigDict(frozen=True)

    image_size: tuple[int, int] = (150, 150)
    batch_size: int = 64
    initial_epochs: int = 2
    fine_tune_epochs: int = 1
    fine_tune_learning_rate: float = 1e-5
    seed: int = 42


class ImageDatasetPipeline:
    """Encapsulates data ingestion, preprocessing, and augmentation."""

    def __init__(self, config: TrainingConfig) -> None:
        self.config = config
        self.resize = Resizing(*config.image_size)
        self.augmentation = [RandomFlip("horizontal"), RandomRotation(0.1)]

    def _augment(self, image: Tensor) -> Tensor:
        for layer in self.augmentation:
            image = layer(image)
        return image

    def prepare(
        self, dataset_name: str = "cats_vs_dogs"
    ) -> tuple[Dataset, Dataset, Dataset]:
        """Loads and prepares train, validation, and test dataset splits."""
        disable_progress_bar()
        train, val, test = load(
            dataset_name,
            split=["train[:40%]", "train[40%:50%]", "train[50%:60%]"],
            as_supervised=True,
        )

        def process(ds: Dataset, augment: bool = False) -> Dataset:
            ds = ds.map(lambda x, y: (self.resize(x), y))
            if augment:
                ds = ds.map(lambda x, y: (self._augment(x), y))
                ds = ds.shuffle(buffer_size=1000, seed=self.config.seed)
            return ds.batch(self.config.batch_size).prefetch(AUTOTUNE).cache()

        return process(train, augment=True), process(val), process(test)


class TransferLearningClassifier:
    """Encapsulates model architecture, training, and evaluation lifecycle."""

    def __init__(self, config: TrainingConfig) -> None:
        self.config = config
        self.base_model = Xception(
            weights="imagenet",
            input_shape=(*config.image_size, 3),
            include_top=False,
        )
        self.model = self._build_model()

    def _build_model(self) -> Model:
        """Constructs the transfer learning model with a custom classification head."""
        self.base_model.trainable = False  # Freeze base weights initially
        inputs = Input(shape=(*self.config.image_size, 3), name="input")
        x = Rescaling(scale=1 / 127.5, offset=-1)(inputs)
        x = self.base_model(x, training=False)
        x = GlobalAveragePooling2D()(x)
        x = Dropout(0.2)(x)
        outputs = Dense(1)(x)
        return Model(inputs, outputs)

    def train_head(self, train_data: Dataset, val_data: Dataset) -> None:
        """Trains only the newly added top classification layers."""
        self.model.compile(
            optimizer=Adam(),
            loss=BinaryCrossentropy(from_logits=True),
            metrics=[BinaryAccuracy()],
        )
        self.model.fit(
            train_data,
            epochs=self.config.initial_epochs,
            validation_data=val_data,
        )

    def fine_tune(self, train_data: Dataset, val_data: Dataset) -> None:
        """Unfreezes the base model and fine-tunes with a low learning rate."""
        self.base_model.trainable = True
        self.model.compile(
            optimizer=Adam(self.config.fine_tune_learning_rate),
            loss=BinaryCrossentropy(from_logits=True),
            metrics=[BinaryAccuracy()],
        )
        self.model.fit(
            train_data,
            epochs=self.config.fine_tune_epochs,
            validation_data=val_data,
        )

    def evaluate(self, test_data: Dataset) -> dict[str, float]:
        """Evaluates model performance on unseen test data."""
        return self.model.evaluate(test_data, return_dict=True)

    def predict(self, data: Dataset) -> ndarray:
        """Generates prediction probabilities for input samples."""
        return self.model.predict(data)

Running the Modular Pipeline

Now look at how clean the execution becomes:

if __name__ == "__main__":
    config = TrainingConfig()

    pipeline = ImageDatasetPipeline(config)
    training_data, validation_data, test_data = pipeline.prepare()

    classifier = TransferLearningClassifier(config)
    classifier.train_head(training_data, validation_data)
    classifier.fine_tune(training_data, validation_data)
    classifier.evaluate(test_data)

    # For downstream inference or API serving:
    predictions = classifier.predict(test_data)

Every component now has a single, well-defined role:

Want to swap data augmentation strategies? Touch only ImageDatasetPipeline.
Want to try a different learning rate or image resolution? Change one value in TrainingConfig.
Want to write a unit test for image resizing? Test pipeline.resize — no GPU, no massive neural network initialization.
Want to deploy inference to a FastAPI microservice or production AWS Lambda template? Import TransferLearningClassifier directly and call predict().

💡 What Happened to the Data Visualizations?

You might notice that Tier 3 drops the matplotlib code from earlier. That's intentional. In production architecture, visualization is a downstream consumer, not a pipeline dependency. Your core model training and inference pipelines should be headless, lightweight, and free of plotting libraries. When you want to inspect data or generate confusion matrices, write a dedicated evaluation script or spin up a lightweight notebook that imports TransferLearningClassifier.

The 3-Tier ML Architecture Cheat Sheet

Here's how the three tiers compare side by side as a quick reference for technical reviews and design specs:

Dimension	Notebook Script (Tier 1)	Idiomatic Script (Tier 2)	Modular OOP Architecture (Tier 3)
State Scope	Leaks into global namespace	Top-level module scope	Strictly encapsulated in class instances
Hyperparameters	Hardcoded magic numbers	Module constants (`BATCH_SIZE`)	Strongly-typed immutable Pydantic model
Unit Testing	Impossible without executing all cells	Difficult; relies on global state	Trivially testable; components mockable
Reusability	Copy-pasting cells	Copying script file	Importable module into FastAPI, Celery, or Kubeflow
Type Safety	None	Partial hints	End-to-end type annotations (`Dataset`, `tuple`)

The Clean ML Checklist

Before submitting an ML pull request or moving notebook code into production, I run through this checklist. It doubles as a quick health scorecard — if you can't confidently check most of these, the code isn't production-ready yet.

[ ] Reproducibility: Can a new engineer clone the repo and run the entire pipeline with a single command? Are random seeds set explicitly?
[ ] Descriptive Naming: Are variables named for their domain roles (features, target, customer_data) rather than single letters (x, y, df)?
[ ] Surgical Imports: Are you importing only the functions, classes, and layers you actually use?
[ ] Separation of Concerns: Is data loading separated from model definition and training logic?
[ ] Encapsulated State: Are models and pipelines encapsulated in classes rather than loose global variables?
[ ] Configurability: Are hyperparameters externalized into a typed, validated Pydantic model rather than hardcoded?
[ ] Testability: Can you test your data transformations without spinning up a GPU?
[ ] Modularity: Can your trained model be imported into an API without triggering a training run?
[ ] Linter & Formatter: Did you run ruff check and ruff format before committing?

Conclusion

Writing clean code in machine learning isn't pedantic nitpicking — it directly impacts your team's velocity, prevents subtle data leaks, and bridges the gap between quick prototypes and reliable production systems.

Moving from loose notebook cells to clean Python scripts — and ultimately to modular, object-oriented pipelines — is how data science code becomes resilient software. Whether it's a weekend Kaggle submission or an enterprise ML pipeline, your teammates (and your future self) will thank you.

What's the biggest friction point your team faces when taking ML code from notebook experiments to production? Do you enforce OOP pipelines, or do you prefer functional scripts? I'd love to hear your thoughts in the comments below!

Stop Writing Lambda Boilerplate

Amr Abed — Thu, 01 Oct 2026 02:30:28 +0000

Introducing the cur8d/lambda open-source repository — a collection of production-ready Python Lambda templates for Bedrock Agent, REST API, GraphQL, DynamoDB Stream, EventBridge, S3, and SQS scenarios. These templates come pre-integrated with AWS Lambda Powertools, AWS CDK, Pydantic, and a robust testing infrastructure.

Every serverless developer knows the feeling. You spin up a new Lambda project, and before writing a single line of business logic, you're already copying boilerplate from a previous project: setting up AWS Lambda Powertools, wiring up structured logging, configuring X-Ray tracing, and building yet another Repository class for DynamoDB access.

This repetitive setup isn't just tedious — it's a source of inconsistency across projects and a hidden risk. Teams often skip best practices under time pressure, and the "quick Lambda" becomes the undocumented, untested function that nobody dares touch six months later.

That's why I built cur8d/lambda — a collection of production-ready, plug-and-play Python Lambda templates for some of the common real-world scenarios on AWS.

The Problem with Lambda Bootstrapping

AWS Lambda's promise is simplicity: deploy a function, it scales. But production Lambda functions require significantly more than a handler file. You need:

Structured logging that integrates with CloudWatch Insights
Distributed tracing with AWS X-Ray
Metrics for operational visibility
Batch processing for event sources like SQS and DynamoDB Streams
Strong typing to catch bugs at development time, not runtime
Infrastructure as Code, so your function is reproducible and auditable
Comprehensive tests with AWS service mocking

Most developers piece this together manually each time — or worse, they copy a "good enough" version from a previous project and gradually accumulate inconsistencies.

Introducing cur8d/lambda

cur8d/lambda is a GitHub template repository that gives you a complete, opinionated, production-grade starting point for Python Lambda development. It currently covers seven real-world scenarios out of the box:

Bedrock Agent — Handle function-based actions from Amazon Bedrock Agents
GraphQL API — Resolve AppSync GraphQL requests
REST API — Handle API Gateway REST requests
DynamoDB Stream — Batch-process stream events with failure handling
EventBridge — React to events and call external APIs
S3 to SQS — Queue messages on S3 object changes
SQS to DynamoDB — Batch-process SQS messages into DynamoDB

Each template is a fully self-contained starting point. Pick your scenario, rename the project, and start writing business logic immediately.

What’s Pre-Wired for You

AWS Lambda Powertools

Every template integrates AWS Lambda Powertools for Python — AWS's first-party library for Lambda best practices. This gives you:

Structured JSON logging with correlation IDs automatically injected
CloudWatch EMF Metrics for operational dashboards without custom code
AWS X-Ray Tracing with automatic subsegment creation
Batch processing utilities for SQS and DynamoDB Streams, including partial failure handling
Parameter and Secrets loading with caching built in
Event source data classes for type-safe event parsing

You don't configure any of this — it's already there. More on that later.

Pydantic Data Modeling

All input/output models are defined using Pydantic, giving you automatic validation, serialization, and type safety.

from uuid import uuid4

from pydantic import BaseModel, Field
from pydantic.alias_generators import to_camel


class Item(BaseModel, alias_generator=to_camel, populate_by_name=True):
    id: str = Field(
        description="Unique item identifier", default_factory=lambda: str(uuid4())
    )
    name: str = Field(description="Human-readable item name")

The alias_generator=to_camel configuration means the JSON payload from API Gateway uses camelCase while the Python code uses snake_case — matching JSON conventions on the wire and Python conventions in code, with zero manual mapping.

Infrastructure as Code with AWS CDK

Every template ships with a matching AWS CDK stack under infra/stacks/. Deploy any stack with a single command:

make deploy STACK=api

No manual console clicks. No documentation to update. The infrastructure is the code.

Clean Architecture with the Repository Pattern

All database access is encapsulated in a Repository class, separating your business logic from AWS service calls. This makes testing trivial and keeps your handler functions focused on what matters: the business logic.

Testing That Actually Tests Things

The test suite uses:

pytest for test orchestration
moto for AWS service mocking (DynamoDB, SQS, S3 — all mocked locally)
Hypothesis for property-based testing, automatically generating edge-case inputs
coverage for coverage reporting

Run the entire suite with:

make test

Code Quality, Automated

ruff for linting and formatting (replacing Flake8, isort, Black)
pyright for static type checking
pre-commit hooks to enforce quality before every commit
Dependabot for automated dependency updates

Developer Experience

Dev Containers support means you can open the repo in VS Code and have the entire development environment running in Docker — Python, Poetry, CDK, and all tools pre-installed—no setup required.
MkDocs for auto-generated API documentation published to GitHub Pages
GitHub Actions workflows for CI/CD and documentation deployment
AI Agent guidelines (AGENTS.md) providing context for AI coding assistants working in the repo

AWS Lambda Powertools — In Practice

Every template integrates AWS Lambda Powertools for Python as a first-class citizen — not as an afterthought bolted on after the fact. Here's how it actually shapes each template.

Batch Processing with Partial Failure Handling

This is where Powertools saves the most code — and prevents the most production incidents.

Without Powertools, batch processing from SQS or streams requires you to manually track which records succeeded and which failed, then construct a batchItemFailures response telling Lambda which messages to retry. Get it wrong and you either re-process successfully handled messages (causing duplicates) or silently drop failed ones.

The SQS template uses BatchProcessor to handle this automatically:

from aws_lambda_powertools.utilities.batch import (
    BatchProcessor,
    EventType,
    process_partial_response,
)
from aws_lambda_powertools.utilities.batch.types import PartialItemFailureResponse

processor = BatchProcessor(event_type=EventType.SQS)


def handle_record(record) -> None:
    try:
        ...
    except Exception as exc:
        logger.error("Failed to process record", exc_info=exc)
        raise  # Re-raise so BatchProcessor marks it as a failed item


def main(event, context) -> PartialItemFailureResponse:
    return process_partial_response(
        event=event,
        record_handler=handle_record,
        processor=processor,
        context=context,
    )

If handle_record raises for one message, BatchProcessor catches it, logs the failure, continues processing the rest of the batch, and returns the correct batchItemFailures response. Your DLQ only receives the genuinely failed messages — not the whole batch.

The same pattern applies to DynamoDB Streams, where handling partial failures is equally critical to prevent stream stalls:

from aws_lambda_powertools.utilities.batch import (
    BatchProcessor,
    EventType,
    process_partial_response,
)

processor = BatchProcessor(event_type=EventType.DynamoDBStreams)


def handle_record(record):
    try:
        ...
    except Exception as exc:
        logger.error("Failed to process record", exc_info=exc)
        raise  # Re-raise so BatchProcessor marks it as a failed item


def main(event, context):
    return process_partial_response(
        event=event,
        record_handler=handle_record,
        processor=processor,
        context=context,
    )

Event Parsing and Handling

Lambda Powertools really shines here. Whether it’s handling a Bedrock Agent tool use, REST API request, or a GraphQL query, Powertools provides resolvers that eliminate custom event parsing.

The Bedrock Agent template uses the BedrockAgentFunctionResolver and the @app.tool decorator:

from aws_lambda_powertools.event_handler import BedrockAgentFunctionResolver
from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent

app = BedrockAgentFunctionResolver()


@app.tool(name="getItem", description="Gets item details by ID")
def get_item(item_id: str) -> dict: ...


@app.tool(
    name="createItem",
    description="Creates a new item with name and optional description",
)
def create_item(item_id: str, name: str, description: "str | None = None) -> dict: ..."


def main(event: BedrockAgentEvent, context: LambdaContext) -> dict:
    """Lambda entry point for the Bedrock Agent handler"""
    return app.resolve(event, context)

The REST API template uses the APIGatewayRestResolver and Flask-like decorators — meaning if you've used Flask or FastAPI, you'll feel right at home:

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.api_gateway import Response

app = APIGatewayRestResolver()


@app.get("/items/<id>")
def get_item(id: str) -> Response:
    """Retrieve an item by ID"""
    ...

    return Response(status_code=200, content_type="application/json", body=dumps(item))


@app.post("/items")
def create_item() -> Response:
    """Create a new item from the request body"""
    body = app.current_event.json_body
    ...
    return Response(status_code=201, content_type="application/json", body=dumps(item))


def main(event: dict, context: LambdaContext) -> dict:
    """Lambda entry point for the API Gateway handler"""
    return app.resolve(event, context)

The EventBridge template uses the event_parser to ensure your events match your Pydantic models before your logic even runs:

from aws_lambda_powertools.utilities.parser import event_parser
from aws_lambda_powertools.utilities.parser.models import EventBridgeModel


@event_parser(model=EventBridgeModel)
def main(event: EventBridgeModel, context) -> None:
    handle(event)

Combining Powertools with Pydantic

One of the most useful patterns across the templates is combining Powertools event data classes with Pydantic models for end-to-end type safety.

Powertools parses the raw Lambda event into a typed object (so you're not indexing into raw dicts). Pydantic then validates and deserializes the business payload within that event:

from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from pydantic import ValidationError

app = APIGatewayRestResolver()


@app.post("/items")
def create_item() -> Response:
    # Powertools resolves the typed API Gateway envelope
    body = app.current_event.json_body
    # Pydantic validates the business payload
    try:
        item = Item.model_validate(body)
    except ValidationError as exc:
        return Response(
            status_code=422,
            content_type="application/json",
            body=dumps({"error": "..."}),
        )
    ...
    return Response(
        status_code=201, content_type="application/json", body=dumps(item.dump())
    )

In the Bedrock Agent template, the same pattern applies via BedrockAgentFunctionResolver — Powertools resolves the agent event, and typed function signatures replace manual parameter extraction:

# templates/agent/handler.py
from aws_lambda_powertools.event_handler import BedrockAgentFunctionResolver
from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent
from aws_lambda_powertools.utilities.typing import LambdaContext

from templates.agent.models import Item

repository = Repository(settings.table_name)
app = BedrockAgentFunctionResolver()


@app.tool(name="getItem", description="Gets item details by ID")
def get_item(item_id: str) -> dict:
    item = repository.get_item(item_id)
    if not item:
        return {"error": f"Item {item_id} not found"}
    return item


@app.tool(
    name="createItem",
    description="Creates a new item with name and optional description",
)
def create_item(item_id: str, name: str, description: "str | None = None) -> dict:"
    item = Item(id=item_id, name=name, description=description)
    repository.put_item(item.model_dump())
    return item.model_dump(by_alias=True, exclude_none=True)


def main(event: BedrockAgentEvent, context: LambdaContext) -> dict:
    """Lambda entry point for the Bedrock Agent handler"""
    return app.resolve(event, context)

Fetching Parameters and Secrets

In the EventBridge template, the handler calls an external API using an API token stored in the Secrets Manager. With Lambda Powertools, that adds exactly three lines of code:

from aws_lambda_powertools.utilities.parameters import SecretsProvider

secret_provider = SecretsProvider()

token = secret_provider.get(secret_name)

The same goes for loading configuration from the Parameter Store (SSMProvider), AppConfig (AppConfigProvider), or a DynamoDB table (DynamoDBProvider):

from aws_lambda_powertools.utilities.parameters import SSMProvider

provider = SSMProvider()

configuration = provider.get(key)

Structured Logging That Travels With Your Request

The logger is initialized once at module level and injected into every handler via the @logger.inject_lambda_context decorator. This automatically appends the Lambda context (function name, version, cold start flag, and AWS request ID) to every log line — without you writing a single line of configuration code.

from aws_lambda_powertools import Logger

logger = Logger(service="api")


@logger.inject_lambda_context(log_event=True)
def handler(event: APIGatewayProxyEventV2, context: LambdaContext) -> dict:
    logger.info("Processing request", path=event.path, method=event.http_method)
    ...

The output in CloudWatch is structured JSON from the first line:

{
  "level": "INFO",
  "message": "Processing request",
  "path": "/users",
  "method": "GET",
  "service": "api",
  "function_name": "my-api-handler",
  "cold_start": true,
  "aws_request_id": "abc-123"
}

This matters at scale: CloudWatch Logs Insights can now query your logs by any of these fields. Finding all cold starts, or all errors on a specific path, becomes a one-line query instead of a regex hunt through raw strings.

Tracing Without Instrumentation Noise

X-Ray tracing is applied via two decorators that require no additional configuration:

from aws_lambda_powertools import Tracer

tracer = Tracer(service=settings.service_name)


# On methods — creates child subsegments in X-Ray
@tracer.capture_method
def handle_record(record: SQSRecord) -> None:
    repository.put_item(process(record.body))


# On the Lambda handler — creates the root X-Ray segment
@tracer.capture_lambda_handler
def main(event: dict, context: LambdaContext) -> dict: ...

The @tracer.capture_lambda_handler decorator creates the root X-Ray segment, and @tracer.capture_method on Repository methods creates child subsegments — so your X-Ray service map shows handler → repository → DynamoDB as distinct, timed segments. You can see exactly where latency lives without writing a single xray_recorder.begin_subsegment() call.

CloudWatch Metrics Without Custom Namespaces

Most Lambda functions either skip custom metrics entirely — relying on the default Lambda service metrics — or end up with a tangled mix of boto3 CloudWatch put_metric_data calls scattered across the codebase. Neither approach is great in production.Powertools solves this with the Metrics utility, which uses CloudWatch Embedded Metric Format (EMF) to emit custom metrics as structured log lines. No boto3 calls, no extra latency, no IAM permissions beyond what CloudWatch Logs already has.

Each template initializes a Metrics instance at module level with a shared namespace and service name:

from aws_lambda_powertools import Metrics
from aws_lambda_powertools.metrics import MetricUnit

metrics = Metrics(namespace=settings.metrics_namespace, service=settings.service_name)


def process(event, context):
    ...
    metrics.add_metric(name="Processed", unit=MetricUnit.Count, value=1)


@metrics.log_metrics
def main(event: dict, context: LambdaContext) -> dict:
    return process(event, context)

The @metrics.log_metrics decorator flushes all buffered metrics at the end of each invocation as a single EMF-formatted log line — CloudWatch automatically parses it and makes those metrics available in dashboards and alarms without any additional infrastructure.

capture_cold_start_metric=True is a one-liner that adds a ColdStart metric to every function automatically. In production, tracking cold starts per function gives you a clear signal for when to invest in provisioned concurrency — no custom instrumentation needed.

Getting Started in Under 5 Minutes

1. Create your repository

Click Use this template on GitHub to create a new repository from the template.

2. Rename the project

After cloning, run the setup command once:

make project NAME="new-name" DESCRIPTION="New description" AUTHOR="Your Name" EMAIL="you@example.com" GITHUB="yourusername"

This renames all references throughout the codebase in a single step.

3. Install and run

make install      # Install dependencies via Poetry
make precommit    # Install pre-commit hooks
make test         # Run tests with coverage

4. Deploy

That's it. You're shipping production-grade Lambda code, not maintaining boilerplate.

make deploy STACK=api    # Deploy the REST API scenario

Design Decisions Worth Noting

Python 3.14+: The templates target the latest Python runtime. Lambda supports Python 3.14, and you should be using it.
Poetry over pip/requirements.txt: Deterministic dependency resolution and lock files eliminate the "works on my machine" problem. pyproject.toml is the single source of truth for the entire project configuration.
Makefile as the interface: All common operations — install, test, lint, deploy, docs — are exposed through make commands. This creates a consistent interface regardless of the underlying tools, making CI/CD pipeline configuration trivial.
camelCase for JSON, snake_case for Python: AWS services speak camelCase; Python speaks snake_case. Pydantic's alias_generator=to_camel bridges the gap automatically, keeping both sides idiomatic.

The Bedrock Agent Template

One template deserves special mention: the Bedrock Agent template. As generative AI workloads move to production on AWS, Lambda functions increasingly serve as the action executors behind Amazon Bedrock Agents. This template gives you a proper starting point for building agent action groups — with typed request/response models, error handling, and the exact response schema Bedrock expects.

It's production-ready for the AI-native workloads being built today.

Contributing

The repository is open source under the MIT license. Contributions are welcome — whether that's a new template scenario, improved infrastructure code, additional test coverage, or documentation improvements. Check out the contributing guidelines to get started.

If you're building on AWS Lambda and you find yourself writing the same setup for the third time, I hope cur8d/lambda saves you that time and helps you ship better, more consistent serverless code.

Ready to start? Head over to the repository and click "Use this template":

👉 github.com/cur8d/lambda

The Only AIP-C01 Study Plan You Need

Amr Abed — Thu, 01 Oct 2026 02:30:26 +0000

How I Passed as One of the First 5,000 — And the Exact Plan I Used

I recently passed the new AWS Certified Generative AI Developer Professional (AIP-C01) exam as an early adopter (one of the first 5,000 worldwide). Here’s the exact 6-week plan I used, so you don’t have to figure it out from scratch.

AWS Certified AWS Generative AI Developer Professional Badge

What you’ll find here is a complete, curated, sequenced study plan — the one I wish someone had handed me before I started. No fluff, just the resources that matter, in the order they matter.

In this guide, you’ll find:

The Fast Track (if you’re short on time)
Detailed Study Plan
Notion Tracking Template
Time-Saving Lab Strategies

Do I really need to prepare?

“I have been using AWS Generative AI for the last year or so. Do I really need to prepare?”

Short answer: Yes!

No matter how long you have been working with AWS Generative AI tools, you may not be using all the tools covered by the exam in your day-to-day experience, or you may not have encountered all the scenarios seen on the exam. That’s proven true for any AWS certification exam, not only this one. If you have taken any AWS certification exam before, you already know that. In short, being *AWS-*ome at what you do daily and passing an AWS certification exam are two different skills, though highly interconnected.

That said, a minimum level of familiarity with the exam topic and at least some of the tools covered is required to pass. No amount of preparation can make up for that. Always make sure you are prepared before scheduling and taking the exam.

How long does it take?

Plan for 6–8 weeks at roughly 2–3 hours per day, depending on your existing Bedrock experience.

What to expect?

If you’re wondering how hard the AIP-C01 is, it felt harder than the AWS Machine Learning Specialty exam but more focused than the Solutions Architect exam. Expect Bedrock depth and several multi-service scenario questions. My practice scores plateaued at around 75% for the first couple of weeks, then eventually rose to around 95%. Don’t panic if progress feels slow mid-way. The domains click together in the final stretch.

The Fast Track

If you are pressed for time:

Focus on the Bedrock User Guide and the Exam Prep Plan
The Generative AI for Developers videos are highly recommended for some exam tips and tricks
You can just read only the topics that address any knowledge gaps you find in the Advanced Learning Plan

Notion Tracking Template

I have created a Notion database of study-plan items to track my preparation progress and group, sort, or filter by certification, plan, format, and domain. I am sharing it here so others can benefit as well. The database is available as a template at amrabed.notion.site/aws-exam-prep. Feel free to duplicate and use right away.

The database has content for other exams as well. Remember to filter by certification for AWS Certified Generative AI Developer — Professional

Notion database for AWS Exam Preparation

Preparation Material

The AWS Training & Certification team recommends these two learning plans for preparation:

I added a couple more learning plans for more hands-on practice:

Both the AWS SimuLearn and the AWS Cloud Quest plans share the same labs. The list below points to the AWS SimuLearn labs, but you may want to do the Cloud Quest plan instead to earn a training badge.

Amazon Web Services (AWS) also produced a dedicated LinkedIn Live 4-episode series for the exam, Generative AI for Developers. I highly recommend watching them for some exam tips and tricks.

Last but definitely not least, the Bedrock User Guide is a must-read for this exam. I have highlighted the most relevant parts for each domain below.

Create an AWS skills profile if you don’t already have one. You will earn points and training badges during your preparation, and you want to showcase them.

AWS Skills Profile

Before you start

Go over the Exam Guide 🆓 to understand the exam structure
Take the Official Practice Question Set 🆓 to identify your knowledge gaps
Go over the Exam Prep Overview 🆓

The 6-Week Plan

Day 0: Exam guide + Practice Questions
Day 1–13: Domain 1 (31%)
Day 14–24: Domain 2 (26%)
Day 25–31: Domain 3 (20%)
Day 32–40: Domains 4 & 5 (23%)
Day 41–42: Final Review & Pretest

My Comprehensive Study Plan

I have curated the list below covering the readings, labs, videos, and practice questions from the material above in a recommended best-effort sequence that aligns with the exam guide and provides the most comprehensive preparation.

Most of the items on the list require a subscription. The free items are marked 🆓. The subscription is currently $29 per month — totally worth it! And, you can always try to complete all the paid items on the list within one month 😉

Please feel free to copy the list and use it as a preparation checklist.

Domain 1: FM Integration, Data Management, and Compliance

Documentation 🆓

Hands-on Labs

Overview

Explore the Amazon Bedrock Playgrounds 🆓
Get Started with Generative AI 🆓
Use AI Services with Amazon SageMaker 🆓
Introduction to Generative AI
Generative AI for Personalized Marketing

Foundation Models

Prompt Engineering

Knowledge Bases

Fine Tuning

Review and Questions

Domain 1 Review 🆓
Domain 1 Questions

Video 🆓

Generative AI For Developers | S1 E1 | Exam Overview and FM Integrations

Optional Reading

Domain 2: Implementation and Integration

Documentation 🆓

Hands-on Labs

Agentic AI

Model Deployment & Integration

Development Tools

Review and Questions

Video 🆓

Generative AI For Developers | S1 E2 | Implementation and Integration

Optional Reading

Domain 3: AI Safety, Security, and Governance

Documentation 🆓

Hands-on Labs

Review and Questions

Domain 3 Review 🆓
Domain 3 Questions

Video 🆓

Generative AI For Developers | S1 E3 | AI Safety, Security, and Governance

Optional Reading

Domain 4: Operational Efficiency and Optimization for AI Applications

Documentation 🆓

Review and Questions

Video 🆓

Generative AI For Developers | S1 E4 | Operations and Troubleshooting

Optional Reading

Domain 5: Testing, Validation, and Troubleshooting

Documentation 🆓

Evaluate models

Review and Questions

Video 🆓

Episode 4 also covers domain 5 in the last 10 minutes

Optional Reading

Final Review and Pretest

Go over the Exam Prep Summary 🆓
Retake: Official Practice Question Set 🆓, Domain 1 Questions, Domain 2 Questions, Domain 3 Questions, Domain 4 Questions, Domain 5 Questions
Take the Official Pretest

Bonus

Take the AWS Agentic AI Demonstrated microcredential
Play the Generative AI track on AWS Card Clash just for fun!

Would you still like more?

If you need even more training, you can always search the AWS Skill Builder content with filters like:

Roles: Generative AI Development Engineer
Format: Learning plan

AWS Skill Builder Gen AI Learning Plans

Time-Saving Tips

As you see in the study plan above, there is a lot to cover, ideally in no more than a month or two. So, you’ll need to save as much time as you can during your preparation. Here are a few tips and tricks based on my experience, especially if you decide to take the hands-on labs. You will notice that most of the lab time is consumed by provisioning the environment, deploying model endpoints, and running Jupyter notebooks. Here is how you can finish the labs faster.

No idle time

While the lab environment is being provisioned, don’t wait for provisioning to complete before starting. Go through the instructions for all three stages (Learn, Practice, and DIY) to hit the ground running once the environment is ready.

Parallelize tasks

Many DIY sections do not depend on the practice sections, which means you can complete both practice and DIY together. For instance, the practice section may ask you to start a 2-instance cluster, which the DIY asks to scale up to 3 instances. You can just go ahead and start a 3-instance cluster right away. Moreover, some steps are independent of one another, so open multiple tabs and continue to the next step while the current one completes.

Run all cells

For the Jupyter notebooks, skim through the notebook to understand the concepts, add any missing configuration or code, then run all cells rather than cell by cell.

Split screen

This is not technically a preparation tip, but it helps. Once the AWS sandbox environment is loaded, split the screen between the lab window with instructions and the AWS environment window to follow along more quickly.

When to schedule the exam

I prefer to schedule the exam only when I can answer at least 90% of the questions correctly, with each question taking no more than 90 seconds on average. That gives me confidence that I can pass the exam smoothly, even under the added stress.

I only schedule the exam when I can hit 90%+ accuracy, spending no more than 90 seconds per question. That’s my personal go/no-go signal.