But data privacy regulations are tightening globally, and users are growing increasingly conscious of their digital footprints. Treating privacy as a "legal afterthought" or something to fix later in production is no longer sustainable.

This is where Privacy by Design comes in.

Coined as a framework to integrate privacy proactively into the engineering lifecycle, Privacy by Design means your system architecture should naturally protect user data by default.

In this comprehensive guide, we'll walk through how to structurally build data privacy into your backend APIs using modern engineering patterns, code concepts, and intentional database designs.

Table of Contents

• Prerequisites

• The Principles of Privacy by Design for Developers

• Project Directory Structure

• Implement Strict Data Minimization at the Endpoint Layer

• Decouple PII with the Pseudonymization Token Pattern

• Beyond RBAC: Implementing Policy-Based Access Control (PBAC)

• Automate Data Retention with Database TTLs and Hooks

• Trust is the Ultimate Developer Metric

Prerequisites

Before diving into this tutorial, you should have the following:

• A foundational understanding of Node.js and JavaScript (ES6+).

• Familiarity with building basic RESTful APIs using Express.

• Essential knowledge of database interactions (SQL or NoSQL concepts).

  

Figure 1: The four pillars of Privacy by Design for backend APIs.

Diagram Breakdown:

• Center: Core security shield representing privacy-first system design.

• Top-Left (Data Minimization): Payload filtering at the endpoint layer.

• Top-Right (Pseudonymization): Decoupling PII via tokenization.

• Bottom-Left (PBAC): Purpose-driven, context-aware authorization rules.

• Bottom-Right (Retention & TTL): Automated data expiration via database hooks.

The Principles of Privacy by Design for Developers

Before writing code, we need to shift our mindset. Privacy by Design isn't about writing a better "Privacy Policy" page on a website. It means translating structural abstractions into practical engineering boundaries.

For an API engineer, this boils down to four distinct execution pillars:

• Proactive, not reactive: Preventing privacy data leaks before they happen rather than managing breaches after the fact.

• Privacy as the default: The user doesn't have to opt-in to remain private. The ecosystem protects them out of the box.

• End-to-end security: Data remains secure, structured, and minimized from ingestion to permanent deletion.

• Visibility and transparency: Keeping clean logs of what data is handled, where it lives, and why it is being used.

Let's look at how we can implement these four foundations directly inside our codebases.

Project Directory Structure

To give you a technical perspective of how these privacy patterns fit together cleanly inside a modular production application, we'll be referencing code across the following project layout:

api-privacy-design/
├── config/
│   └── database.js
├── middleware/
│   ├── auth.js
│   └── privacyPolicy.js
├── models/
│   ├── auditLog.js
│   └── user.js
├── services/
│   └── piiVault.js
├── validators/
│   └── userValidator.js
├── app.js
├── package.json
└── README.md

Implement Strict Data Minimization at the Endpoint Layer

The fundamental rule of data privacy is simple: If you don’t have it, you can't lose it.

The Payload Over-Inclusion Anti-Pattern

Many APIs accept massive, generic JSON payloads from frontend clients and save everything straight to the database. Developers often do this for convenience, using spread operators like ...req.body to quickly insert records without explicitly mapping variables.

// A poorly designed registration endpoint that grabs everything indiscriminately
app.post('/api/register', async (req, res) => {
  const userData = req.body; // Accepts full profile, tracking tokens, internal metadata, etc.
  const user = await Database.save('users', userData);
  res.status(201).json(user);
});

If an attacker manipulates the client-side request to pass an administrative flag like {"isAdmin": true, "internalDeviceID": "123"} within the payload, a naïve endpoint will process it.

Architectural View: Schema Validation

Data minimization means configuring endpoints to accept and store only what is strictly necessary for immediate business operations. To enforce this, explicit schema validation must occur right at the API gateway or controller boundary. This process ensures that only predefined, safe fields enter your internal system, and everything else is structurally ignored.

The Code Solution: Schema Hardening

By implementing strict verification models (using validation libraries like Joi, Zod, or Yup), any unmapped field injected by a client is dropped or triggers a validation block.

Under our validators/userValidator.js configuration file, we can define our schema parameters.

If your validation provider is Joi-compatible, you can leverage native options like { stripUnknown: true } to automatically clean incoming objects. If you choose to use an alternate validation framework like Zod, you can achieve an identical outcome by chaining the .strict() modifier to block requests that contain unmapped fields entirely.

Here's how we use Joi to drop non-explicit fields instantly at our router boundary:

const Joi = require('joi');

// Explicitly define the bare minimum schema required for validation
const registrationSchema = Joi.object({
  email: Joi.string().email().required(),
  password: Joi.string().min(8).required()
  // Any extra tracking metadata or unauthorized attributes injected here are dropped automatically
});

app.post('/api/register', async (req, res) => {
  try {
    // stripUnknown: true drops any properties not explicitly defined in the schema
    const validatedData = await registrationSchema.validateAsync(req.body, { stripUnknown: true });
    
    const user = await Database.save('users', validatedData);
    
    // Privacy Safeguard: Never return raw internal properties back to the client response
    res.status(201).json({ id: user.id, email: user.email }); 
  } catch (err) {
    res.status(400).json({ error: err.details[0].message });
  }
});

Decouple PII with the Pseudonymization Token Pattern

When handling Personally Identifiable Information (PII) like real names, phone numbers, or physical home addresses, storing them in plain text inside your primary application tables is a massive security liability.

Why Simple Encryption Isn't Enough

While encrypting columns at rest (using AES-256) is standard practice, application systems still face risks. If developers run analytical reporting, dump operational data into debugging environments, or accidentally output database rows to application logs (for example, console.log(userObject)), raw sensitive data can quickly spill across unauthorized environments.

The Tokenization Architecture

Instead of keeping PII alongside business tables, use a Pseudonymization Token Pattern. This separates identity from transactional records. This isn't just encryption anymore; it's an architectural separation of duties.

Implementing a Pseudonymized Data Flow

Let's trace how the processing layer functions. In our isolated services/piiVault.js module, we manage the interaction between our primary app controller and our secure token storage vault. When a record initialisation occurs, we isolate the raw PII elements, exchange them for an opaque reference token, and use that reference token as our relational key.

// services/piiVault.js
// Conceptual abstraction of a decoupled user account initialization
async function createUserProfile(incomingPayload) {
  const { email, phoneNumber, ...transactionalData } = incomingPayload;

  // 1. Dispatch the raw PII elements directly to an isolated, encrypted Vault Service
  const vaultResponse = await PiiVaultService.tokenize({
    email: email,
    phone: phoneNumber
  });

  // 2. The Vault returns a uniquely structured reference UUID token (non-reversible string)
  const piiToken = vaultResponse.token; // e.g., "pii_token_83912x"

  // 3. Save the primary application record using ONLY the token link
  const operationalRecord = {
    ...transactionalData,
    piiRefToken: piiToken,
    accountStatus: 'active',
    createdAt: new Date()
  };

  await Database.save('primary_users_table', operationalRecord);
  return { success: true, reference: piiToken };
}

If internal developers run analytical operations or debug logs on the primary backend database, they'll only ever see non-identifiable reference strings rather than actual user credentials.

Beyond RBAC: Implementing Policy-Based Access Control (PBAC)

Traditional Role-Based Access Control (RBAC) (checking patterns like if (user.role === 'admin')) is no longer granular enough for complex software architectures. Privacy engineering demands that access control systems evaluate not just who is trying to view a resource, but the purpose of the access request.

The Core Vectors of PBAC

By transitioning to Policy-Based Access Control (PBAC), your authorization system dynamically maps rules along three parameters:

• The Actor/User: Who is making the API call and what are their active permissions?

• The Resource: What specific classification of data is being requested?

• The Purpose: Why does the business logic need to process this data, and did the data owner explicitly give consent for it?

Building Contextual Privacy Governance Middleware

Let's see how this works in practice. To set the context for our route execution, we'll build a dedicated authorization middleware file under middleware/privacyPolicy.js. This middleware intercepts the incoming request vector, checks the destination resource owner's privacy choices, and matches it against the application's processing target context.

// middleware/privacyPolicy.js
// Middleware pattern verifying purpose-bound consent parameters
const verifyDataAccessPolicy = (requiredPurpose) => {
  return (req, res, next) => {
    const actor = req.user; // Instantiated from a previous authentication layer
    const resourceOwner = req.resourceOwner; // The target data record context
    
    // Extract the explicit consent array configured by the user
    const userConsentPolicies = resourceOwner.consents || []; // e.g., ['essential_auth', 'marketing_emails']

    // Check if the code execution purpose matches what the user explicitly consented to
    const hasExplicitConsent = userConsentPolicies.includes(requiredPurpose);
    
    // Elevate override permissions ONLY to specialized audit actors if explicitly tracked
    if (!hasExplicitConsent && actor.role !== 'System_Auditor') {
      return res.status(403).json({ 
        error: "Access Denied: The operation requested exceeds your current purpose-bound user consent profiles." 
      });
    }
    
    // Access authorized; proceed down the middleware track
    next();
  };
};

// Application Execution Target Route within app.js
app.get('/api/analytics/user-behavior/:userId', 
  fetchResourceOwnerMiddleware, 
  verifyDataAccessPolicy('analytics_tracking'), // Will throw a 403 if user opted out of tracking
  (req, res) => {
    res.json({ status: "Success", data: "Contextual processing executed." });
  }
);

Automate Data Retention with Database TTLs and Hooks

Data privacy frameworks emphasize that user data shouldn't sit inside database storage indefinitely. If a temporary log has fulfilled its purpose, or an account lifecycle ends, that data needs to disappear entirely.

The Automated Retention Lifecycle

Relying on teams to run manual script updates or cron cleanup commands is prone to failure. Privacy-by-design systems build data expiration rules directly into their data models.

Document Stores: Native TTL Indexes

If you're managing unstructured payload layers or tracking sessions in NoSQL document collections (like MongoDB), you can leverage Time-To-Live (TTL) index configurations to wipe records automatically down to the second.

We write this logic directly into our schemas inside models/auditLog.js:

// models/auditLog.js
const mongoose = require('mongoose');

const auditLogSchema = new mongoose.Schema({
  userId: mongoose.Schema.Types.ObjectId,
  apiAction: String,
  ipAddress: String,
  createdAt: { 
    type: Date, 
    default: Date.now, 
    expires: '90d' // MongoDB background threads automatically drop this document after 90 days
  }
});

const AuditLog = mongoose.model('AuditLog', auditLogSchema);

Relational Databases: Safe Masking with ORM Hooks

In relational database systems (such as PostgreSQL or MySQL using Sequelize/Prisma), cascading relationships make immediate hard deletions tricky.

To avoid breaking database foreign keys while still respecting user privacy, we implement a two-step pattern: an instantaneous lifecycle hook handles data anonymisation and masking, while a native database scheduled script handles data cleansing asynchronously.

We can implement this hook inside our relational schema files under models/user.js:

// models/user.js
const { Sequelize, DataTypes } = require('sequelize');
const sequelize = new Sequelize('sqlite::memory:');

const OperationalUser = sequelize.define('OperationalUser', {
  username: DataTypes.STRING,
  email: DataTypes.STRING,
  fullName: DataTypes.STRING,
  isDeleted: { type: DataTypes.BOOLEAN, defaultValue: false }
});

// Anonymize records on the fly before a soft-deletion hook runs
OperationalUser.addHook('beforeUpdate', async (user, options) => {
  if (user.isDeleted && user.changed('isDeleted')) {
    // Mask and overwrite confidential fields to protect privacy while keeping non-PII metrics intact
    user.email = `anonymized_user_${Date.now()}@privacy-protected.org`;
    user.fullName = "Anonymized Profile Data";
    user.username = `archived_node_${Math.floor(Math.random() * 100000)}`;
  }
});

Trust is the Ultimate Developer Metric

Building high-throughput applications quickly is a fantastic skill, but engineering architectures that stand the test of time requires building for trust.

When you embed consent boundaries, strict payload validation schema controls, decoupled identifier vaults, and automated database retention routines into your backend services from day one, you aren't just checking boxes off a security compliance form. You're building reliable, loosely coupled code systems that mitigate data breach risks and treat user privacy as a first-class citizen in application design.

Instead of interacting with the agent through a terminal, we’ll expose it over HTTP so multiple users can access it through a chat-style frontend interface. Each session will maintain its own conversation history and streamed responses.

The local AI agent will be built with LangChain v1, Ollama, Qwen, and Python, running on your own machine and ready to plug into larger applications without any per-call model API charges.

Table of Contents

• Background

• What is FastAPI?

• What is Streamlit?

• What Is Multi-User Support?

• Motivation and Architecture

• Step 1: Install Ollama and Pull the Model

• Step 2: Install Python Dependencies

• Step 3: Build the agent and API layer with FastAPI

• Step 4: Build Streamlit UI

• Step 5: Run the backend app

• Step 6: Run the frontend app

• Sample Output

• What to Improve Before Production

• Conclusion

Background

Many AI agents start out as simple Python scripts that run in a command-line terminal. You type a message, the agent responds, and everything happens in a single local session.

That setup is great for development and testing, but it becomes limiting when you want other people or applications to interact with the agent.

To make an AI agent truly useful, we need to expose it through an interface that other users can access. A REST API is a practical way to do that.

To follow this tutorial, you'll need Ollama installed on your machine. The tutorial works on macOS, Windows, and Linux. I'm using a MacBook Pro with 32 GB of RAM, but you can run this on a lower-memory machine by choosing a smaller Qwen model from Ollama.

What is FastAPI?

FastAPI is a Python web framework for building APIs. In this tutorial, it gives us a simple way to expose the agent over HTTP so other apps, scripts, or services can call it.

FastAPI is a good fit for AI apps because it gives us a clean boundary around the system. We define the request and response models in Python, FastAPI validates them automatically, and it turns HTTP requests into Python objects and Python objects back into JSON. It also generates interactive API docs for free and supports async endpoints, which is useful for AI workloads that may take longer to respond.

What is Streamlit?

Streamlit is a Python framework for building lightweight web interfaces with minimal frontend work. It lets us create interactive browser-based apps using normal Python code instead of HTML, CSS, and JavaScript.

In this tutorial, Streamlit sits on top of the FastAPI backend as a thin client. FastAPI exposes the AI agent over HTTP, and Streamlit gives us a simple UI for calling that API and displaying the results. That separation keeps the backend reusable while still making the agent easy to use in the browser.

What Is Multi-User Support?

Multi-user support means the AI agent can handle requests from more than one user while keeping each user’s session separate.

For example, User 1 asks the agent one question and User 2 asks a different question. The agent should remember the correct context for each user independently. Without multi-user support, all users may end up sharing the same conversation state, which can lead to mixed responses, incorrect memory, or overwritten context.

Motivation and Architecture

Turning an AI agent into an API is the natural next step after building it locally. A Python script is great for experimenting, but an API makes the agent reusable. And adding multi-user support makes the agent extensible to be used by others.

To keep things simple, we’ll use a small local agent powered by Ollama and Qwen. The agent has two tools: one for checking the current time and another for counting words.

FastAPI provides the HTTP layer by exposing one endpoint called /chat/stream. When the request comes in with a user message, Pydantic validates the request, LangChain handles the agent loop and tool calling, and the final answer is returned as stream. Streamlit sits on top of that API and acts as a frontend that sends requests to the API and displays the results.

Example request:

{ 
    "message": "How many words are in: LangChain makes tool calling easier",
    "user_id":"123e4567-e89b-12d3-a456-426614174000"
 }

Example response:

{
  "answer": "There are **5** words in LangChain makes tool calling easier."
}

The model runs locally through Ollama, so there are no per-call model API charges.

Step 1: Install Ollama and Pull the Model

To get started, install the Ollama application for your platform.

We’ll use Qwen as the chat model. I’m using qwen3.5:4b. If your machine has less RAM, you can use qwen3.5:0.8b instead.

ollama pull qwen3.5:4b

Step 2: Install Python Dependencies

Create a virtual environment and install the required packages:

python3 -m venv venv
source venv/bin/activate

pip install fastapi uvicorn streamlit requests langchain langchain-core langchain-ollama langgraph

If tutorial requires LangChain >= 1.0.0.

Step 3: Build the Agent and API Layer with FastAPI

This application has three main responsibilities. FastAPI exposes the HTTP endpoint, Pydantic validates the incoming request data, and LangChain runs the agent, including tool calling and short-term memory.

The user_id sent with each request is used as the thread identifier, allowing the checkpointer to keep each user’s conversation history separate. This memory is per session. So every new session will have its own memory.

Another important detail is that the agent is created only once at startup with agent = build_agent(). Reusing the same agent instance avoids rebuilding the model and tool list for every request, which reduces overhead and improves response times while still supporting multiple users.

Inside the /chat/stream endpoint, the backend uses LangChain’s stream_events(..., version="v3") to generate the response as a stream instead of waiting for the full answer all at once. FastAPI then wraps that stream in a StreamingResponse, so the frontend can receive the output gradually as it's produced. This makes the app feel much more interactive, because users can start reading the answer immediately while the rest is still being generated.

Put together, this gives you a lightweight backend that validates input, preserves separate memory for each user, and streams responses to the UI in real time.

Save the following code as app.py:

from datetime import datetime
from uuid import UUID

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

from pydantic import BaseModel

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver

CHAT_MODEL = "qwen3.5:4b"

SYSTEM_PROMPT = (
    "You are a helpful assistant with access to tools for getting the current time "
    "and counting words in text. "
    "Use tools when needed. If the question does not need a tool, answer directly."
)

# -----------------------------
# Request model
# -----------------------------

class ChatRequest(BaseModel):
    user_id: UUID
    message: str

# -----------------------------
# Tools
# -----------------------------

@tool
def current_time() -> str:
    """Return the current local date and time."""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


@tool
def word_count(text: str) -> int:
    """Count the number of words in a piece of text."""
    return len(text.split())


# -----------------------------
# Agent + checkpoint memory
# -----------------------------

# Store conversation history in short term memory
checkpointer = InMemorySaver()

def build_agent():
    model = ChatOllama(model=CHAT_MODEL, temperature=0)
    return create_agent(
        model=model,
        tools=[current_time, word_count],
        system_prompt=SYSTEM_PROMPT,
        checkpointer=checkpointer,
    )


agent = build_agent()

# -----------------------------
# Streaming endpoint
# -----------------------------

app = FastAPI()

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    def generate():
        run = agent.stream_events(
            {
                "messages": [{"role": "user", "content": req.message}],
            },
            config={
                "configurable": {
                    # Keep each user's short-term memory isolated
                    # by using their user_id as the thread ID.
                    "thread_id": str(req.user_id),
                }
            },
            version="v3",
        )

        for message in run.messages:
            for token in message.text:
                yield token

    return StreamingResponse(generate(), media_type="text/plain")

Step 4: Build Streamlit UI

The Streamlit code creates a simple chat interface for the AI agent and keeps each browser session tied to a unique user_id.

When the app first loads, it generates and stores a UUID in st.session_state, which is later sent to the backend so the agent can keep that user’s conversation history separate from other users. It also creates a chat_history list in session state so previous messages remain visible every time Streamlit reruns the script. The app then loops through that saved history and displays each message in a chat-style format using st.chat_message().

When the user enters a new message through st.chat_input(), the app immediately saves and displays it, then sends it to the backend API with a POST request to http://127.0.0.1:8001/chat/stream along with the session’s user_id.

The request is made with stream=True, which allows the response to arrive gradually instead of all at once. As each chunk of text is received from the backend, the code appends it to full_answer and updates a placeholder on the page, creating a live streaming effect. Once the response is complete, the final assistant message is stored in chat_history so it remains part of the conversation on the page

Save the below as streamlit_app.py

import uuid
import requests
import streamlit as st

API_URL = "http://127.0.0.1:8001/chat/stream"

st.title("Local AI Agent")

if "user_id" not in st.session_state:
    st.session_state.user_id = str(uuid.uuid4())

if "chat_history" not in st.session_state:
    st.session_state.chat_history = []

# Show previous messages
for item in st.session_state.chat_history:
    with st.chat_message(item["role"]):
        st.markdown(item["content"])

message = st.chat_input("Enter a message")

if message:
    # Save and show user message
    st.session_state.chat_history.append({"role": "user", "content": message})
    with st.chat_message("user"):
        st.markdown(message)

    # Stream assistant response
    full_answer = ""
    with st.chat_message("assistant"):
        placeholder = st.empty()

        # Send the reqeust to backend API via POST request
        with requests.post(
            API_URL,
            json={
                "message": message,
                "user_id": st.session_state.user_id,
            },
            stream=True,
        ) as response:
            response.raise_for_status()

            for chunk in response.iter_content(chunk_size=None, decode_unicode=True):
                if chunk:
                    full_answer += chunk
                    placeholder.markdown(full_answer)

    # Save final assistant response
    st.session_state.chat_history.append(
        {"role": "assistant", "content": full_answer}
    )

Step 5: Run the Backend App

Start the server with Uvicorn:

uvicorn app:app --reload --port 8001

Once the application starts, open:

• http://127.0.0.1:8001/

• http://127.0.0.1:8001/docs

The /docs endpoint is automatically generated by FastAPI using your Pydantic models. It provides an interactive interface where you can test the API without writing any client code.

You can send requests directly from curl. In your terminal, run these commands to invoke the API for the AI agent and check the output:

$ curl -X POST http://127.0.0.1:8001/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message":"What time is it?","user_id":"123e4567-e89b-12d3-a456-426614174000"}'

$ curl -X POST http://127.0.0.1:8001/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message":"How many words are in: LangChain makes tool calling easier","user_id":"123e4567-e89b-12d3-a456-426614174000"}'

$ curl -X POST "http://127.0.0.1:8001/chat/stream" \
-H "Content-Type: application/json" \
-d '{"message":"What is the capital of France?","user_id":"123e4567-e89b-12d3-a456-426614174000"}'

To stop the server, press Ctrl+C in the terminal.

Step 6: Run the Frontend App

In another terminal, go to the project directory:

source venv/bin/activate
streamlit run streamlit_app.py

That opens the frontend in your browser at http://localhost:8501/. Try the example prompts like "What is the capital of France". You should see the answer in a chat style interface.

The UI is calling the FastAPI endpoint and invoking the AI agent. You now have a working end to end application for your local AI agent that you can play with.

To stop the server, press Ctrl+C in the terminal.

Sample Output

The image below show two browser sessions of the app running side by side on the same endpoint. Each session is assigned a unique id, which allows the backend to maintain a separate conversation history for each user.

Even though both users ask the same question, “Who am I?”, the responses are different because each session’s answer is based on its own prior messages.

What to Improve Before Production

Although this application is fully functional, it's still intentionally minimal. It already supports a reusable FastAPI backend, a Streamlit chat interface, per-user conversation history, and streaming responses.

If you wanted to take it further, the next steps would be adding authentication, persistent storage, structured logging, monitoring, and more robust deployment setup.

It's also worth noting that if your goal is simply to get a polished self-hosted chat UI up and running quickly, you may not need to build the frontend yourself. Projects like LibreChat and Open WebUI already provide richer interfaces and broader features out of the box.

This tutorial takes a different approach: instead of adopting a full platform, it shows how to build a lightweight custom stack yourself so you can better understand the architecture and have more control over how the agent is exposed.

Conclusion

In this tutorial, we took a local AI agent, wrapped it in a FastAPI app, and used Streamlit UI on top of it.

This transforms the AI agent from a standalone script into a reusable service. Instead of only working in a terminal, it can now be accessed through a simple HTTP endpoint by other apps, scripts, or internal tools.

By assigning each session a unique id, the service can also maintain separate conversation history for multiple users, making it possible to support a chat-style interface with isolated memory per session.

From here, you can continue extending the same service by adding authentication or production-ready features. Happy tinkering!

If you enjoyed this tutorial, you can find more of my writing on my blog (recent posts include system design paper series), my work on my personal website, and updates on LinkedIn.

I pulled up the query logs expecting something subtle. It was not. The list endpoint had no tenant_id filter at all. Every tenant in the database could read every other tenant's projects. The application never threw an error. It just returned whatever was there.

Missing tenant filters don't throw errors. They return the wrong data without any complaint, and nothing in your logs will flag it. I've seen this run in production for weeks before a support ticket pointed anyone at the query logs.

When it does surface, who finds it first matters a lot. A customer noticing it is bad. A compliance auditor noticing it during a SOC 2 review is a different kind of problem.

Isolation built in from the start is a day of work. The time I spent helping a team retrofit it after a compliance review was considerably longer than that, and involved more customer emails than anyone wanted to write.

The stack is Node.js with PostgreSQL. CRUD is the easy part. Tenant isolation, RBAC, and audit logging take more care, and where those checks run in the stack matters. I put all three in middleware, before any route handler fires. A handler that never calls the isolation logic directly can't accidentally skip it.

Prerequisites

• Node.js 18+

• PostgreSQL 14+

• Basic knowledge of Express.js and JWT

What We Will Build

A multi-tenant Express REST API that enforces:

1. Tenant isolation: every database query scopes to the tenant_id from the verified JWT. The client can't influence which tenant the query runs against.

2. RBAC: four roles, each with a numeric level (SuperAdmin is highest, Viewer lowest). Middleware checks the level before the handler runs.

3. Audit logging: any write or sensitive read appends a row to the audit table. The app can't modify those rows afterward. The database enforces this directly. If a bug in the app tries to UPDATE an audit row, the database refuses it. Application-level enforcement alone can't give you that guarantee.

4. Per-tenant rate limiting: request counts in Redis, keyed to the tenant. I've seen IP-based limiting break an enterprise rollout when fifty users came through a single corporate proxy.

5. Tenant isolation tests: a dedicated test file that proves cross-tenant data can't leak. Wire it into CI and it catches broken isolation before it ships.

Table of Contents

1. How Multi-Tenancy Works

2. Architecture Overview

3. Database Schema Design

4. Project Setup

5. JWT Design for Multi-Tenancy

6. Auth and RBAC Middleware

7. The Tenant-Safe Repository Layer

8. Audit Logging Service

9. Per-Tenant Rate Limiting

10. Building the Routes

11. Testing Tenant Isolation

12. Troubleshooting

13. Wrapping Up

How Multi-Tenancy Works

This tutorial uses a shared database with row-level isolation: a tenant_id column on every table, a filter on every query. The database holds everyone's data together. The application decides what each tenant can see.

Two other approaches exist: schema-per-tenant and database-per-tenant. I've talked to teams on schema-per-tenant who ended up spending more engineering time on migration tooling than on their actual product. Database-per-tenant gives stronger guarantees but a connection pool that balloons with every new customer signup.

Neither scales cheaply. Row-level isolation scales further than most teams expect. The ones I know who moved off it did so years in, usually under specific regulatory pressure, not because the approach stopped working.

The one thing in this design that can't be optional: tenant_id must always come from the verified JWT. Not from the request body, not from the URL. Users control what they put in both of those. They don't control what gets signed into a JWT on your server.

Architecture Overview

HTTP Request
     │
     ▼
┌─────────────────────────────────────────┐
│           Express Middleware Stack       │
│                                         │
│  1. Rate Limiter (per tenant_id)        │
│  2. Auth Middleware (verify JWT)        │
│     └─► Extracts: userId, tenantId,    │
│          role, permissions              │
│  3. RBAC Middleware (check role)        │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│           Route Handler                  │
│                                         │
│  1. Call Repository (tenant-safe query) │
│  2. Call Audit Service (fire & forget)  │
│  3. Return response                     │
└──────────────┬──────────────────────────┘
               │
     ┌─────────┴──────────┐
     ▼                    ▼
┌─────────┐        ┌────────────┐
│ Projects│        │ Audit Logs │
│  Table  │        │   Table    │
│(+tenant)│        │(append only│
└─────────┘        └────────────┘

Rate limiting, auth, and RBAC all run before any handler sees the request. Writes pass through the audit service. The repository takes tenantId from req.user and the handler never touches tenant scoping directly, so there's no path around it.

Database Schema Design

-- Tenants table
CREATE TABLE tenants (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name        VARCHAR(255) NOT NULL,
  plan        VARCHAR(50) NOT NULL DEFAULT 'free', -- 'free', 'pro', 'enterprise'
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Users table
CREATE TABLE users (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id   UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
  email       VARCHAR(255) NOT NULL,
  role        VARCHAR(50) NOT NULL DEFAULT 'Member', -- 'SuperAdmin','TenantAdmin','Member','Viewer'
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  UNIQUE(tenant_id, email)
);

CREATE INDEX idx_users_tenant ON users(tenant_id);

-- Projects table (example resource — replace with your domain entity)
CREATE TABLE projects (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id   UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
  name        VARCHAR(255) NOT NULL,
  description TEXT,
  created_by  UUID NOT NULL REFERENCES users(id),
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_projects_tenant ON projects(tenant_id);

-- Audit log table (append-only — never UPDATE or DELETE rows here)
CREATE TABLE audit_logs (
  id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id   UUID NOT NULL,
  user_id     UUID NOT NULL,
  user_email  TEXT NOT NULL,
  user_role   TEXT NOT NULL,        -- role at time of action
  action      TEXT NOT NULL,        -- 'CREATE', 'UPDATE', 'DELETE', 'VIEW'
  resource    TEXT NOT NULL,        -- table name
  resource_id TEXT,
  old_values  JSONB,
  new_values  JSONB,
  ip_address  INET,
  user_agent  TEXT,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_audit_tenant ON audit_logs(tenant_id);
CREATE INDEX idx_audit_created ON audit_logs(created_at DESC);

-- Protect audit log at database level
-- Use a DO block so this runs safely in Docker where app_user is the superuser
DO $$
BEGIN
  IF current_user <> 'app_user' THEN
    REVOKE DELETE, UPDATE ON audit_logs FROM app_user;
  END IF;
END $$;

The REVOKE matters. Application bugs happen. If something in your codebase accidentally tries to UPDATE an audit row, you want the database to refuse it outright, not silently comply.

Project Setup

mkdir nodejs-multitenant-saas-api
cd nodejs-multitenant-saas-api
npm init -y
npm install express pg jsonwebtoken bcryptjs express-rate-limit rate-limit-redis ioredis dotenv
npm install --save-dev jest supertest

Starting PostgreSQL and Redis with Docker

Skip the local installs. One docker-compose.yml in the project root brings up both PostgreSQL and Redis:

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: saas_api
      POSTGRES_USER: app_user
      POSTGRES_PASSWORD: app_password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./schema.sql:/docker-entrypoint-initdb.d/01_schema.sql

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

volumes:
  postgres_data:

That schema.sql mount runs your SQL automatically when the container first starts. No psql required.

docker compose up -d

.env in the project root:

DATABASE_URL=postgresql://app_user:app_password@localhost:5432/saas_api
REDIS_URL=redis://localhost:6379
JWT_SECRET=your_random_secret_here
PORT=3000
NODE_ENV=development

Don't type a JWT_SECRET by hand. Run this to generate one:

node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"

File structure:

nodejs-multitenant-saas-api/
├── src/
│   ├── middleware/
│   │   ├── auth.js          # JWT verification + tenant extraction
│   │   ├── rbac.js          # Role enforcement
│   │   └── rateLimiter.js   # Per-tenant rate limiting
│   ├── services/
│   │   └── auditService.js  # Append-only audit logger
│   ├── repositories/
│   │   └── projectRepo.js   # Tenant-safe DB queries
│   ├── routes/
│   │   └── projects.js      # Route handlers
│   └── utils/
│       └── token.js         # JWT token generation
├── db/
│   ├── index.js             # PostgreSQL pool
│   └── redis.js             # Redis client
├── docker-compose.yml
├── app.js
├── server.js
└── tests/
    └── tenantIsolation.test.js

Boilerplate Files

There are four files the tutorial doesn't cover in detail, but the test file needs all of them to run:

// db/index.js
const { Pool } = require('pg');

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

pool.on('error', (err) => console.error('PostgreSQL error:', err.message));

module.exports = { pool };

// db/redis.js
const Redis = require('ioredis');

const redisClient = new Redis(process.env.REDIS_URL);

redisClient.on('error', (err) => console.error('Redis error:', err.message));

module.exports = { redisClient };

// app.js
require('dotenv').config();
const express = require('express');
const projectsRouter = require('./src/routes/projects');

const app = express();
app.use(express.json());

app.use('/api/projects', projectsRouter);

// Global error handler — must have 4 parameters to be recognised by Express
app.use((err, req, res, next) => {
  console.error(err.stack);
  res.status(500).json({ error: 'Internal server error' });
});

module.exports = app;

// server.js
const app = require('./app');

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => console.log(`Server running on port ${PORT}`));

bcryptjs is included for a login endpoint with proper password hashing. That part isn't covered here, but the GitHub repo has a working /api/auth/login example.

JWT Design for Multi-Tenancy

Both tenantId and role go into the JWT payload. Everything downstream reads from these two fields. Get them wrong, and nothing behaves correctly.

// Example JWT payload
{
  "userId": "usr_abc123",
  "tenantId": "ten_xyz789",
  "email": "alice@acme.com",
  "role": "TenantAdmin",
  "iat": 1720000000,
  "exp": 1720086400
}

The roles in order of privilege:

• SuperAdmin: cross-tenant access for your internal team only

• TenantAdmin: full access within their tenant

• Member: read and write within their tenant

• Viewer: read-only within their tenant

Generate a token (used for testing and your auth endpoint):

// src/utils/token.js
const jwt = require('jsonwebtoken');

function generateToken({ userId, tenantId, email, role }) {
  return jwt.sign(
    { userId, tenantId, email, role },
    process.env.JWT_SECRET,
    { expiresIn: '24h' }
  );
}

module.exports = { generateToken };

Auth and RBAC Middleware

The auth middleware does two things: verifies the JWT signature and extracts the tenant context into req.user.

That second part is what the entire system depends on. Every query downstream reads req.user.tenantId. The client has no say in what that value is. They send a token the server signed, and the server reads back what it put in.

// src/middleware/auth.js
const jwt = require('jsonwebtoken');

function authMiddleware(req, res, next) {
  const authHeader = req.headers.authorization;
  if (!authHeader?.startsWith('Bearer ')) {
    return res.status(401).json({ error: 'Missing or malformed Authorization header' });
  }

  const token = authHeader.split(' ')[1];

  try {
    const decoded = jwt.verify(token, process.env.JWT_SECRET);

    // tenantId always comes from the verified token — never req.body or req.params
    req.user = {
      userId:   decoded.userId,
      tenantId: decoded.tenantId,
      email:    decoded.email,
      role:     decoded.role,
    };

    next();
  } catch (err) {
    return res.status(401).json({ error: 'Invalid or expired token' });
  }
}

module.exports = { authMiddleware };

The RBAC middleware is separate from auth by design. Auth runs on every route. Role enforcement only applies where a minimum role is required. You pass the allowed roles to requireRole() and it compares the user's level against the hierarchy. A Viewer trying to delete something hits the 403 before the handler ever runs.

// src/middleware/rbac.js
const ROLE_HIERARCHY = {
  SuperAdmin:   4,
  TenantAdmin:  3,
  Member:       2,
  Viewer:       1,
};

// requireRole('TenantAdmin') — user must be TenantAdmin or higher
function requireRole(...roles) {
  return (req, res, next) => {
    const userLevel = ROLE_HIERARCHY[req.user?.role] ?? 0;
    const requiredLevel = Math.min(...roles.map(r => ROLE_HIERARCHY[r] ?? 999));

    if (userLevel < requiredLevel) {
      return res.status(403).json({
        error: 'Insufficient permissions',
        required: roles,
        current: req.user?.role,
      });
    }

    next();
  };
}

module.exports = { requireRole };

The Tenant-Safe Repository Layer

Isolation lives here. Every function takes tenantId as a required argument, pulled from req.user by the handler. There's no way to call these without providing a tenant scope. I've watched teams try to handle this with a URL parameter instead (GET /api/projects?tenantId=xyz) and call it isolated. It is not. Any client sends whatever it wants in a query string.

// src/repositories/projectRepo.js
const { pool } = require('../../db');

// List all projects for a tenant — tenantId is ALWAYS from the JWT
async function listProjects(tenantId) {
  const result = await pool.query(
    `SELECT id, name, description, created_by, created_at
     FROM projects
     WHERE tenant_id = $1
     ORDER BY created_at DESC`,
    [tenantId]
  );
  return result.rows;
}

// Get a single project — returns null if it belongs to a different tenant
// NOTE: Returns 404 (not 403) intentionally — don't reveal the resource exists
async function getProject(id, tenantId) {
  const result = await pool.query(
    `SELECT id, name, description, created_by, created_at
     FROM projects
     WHERE id = $1 AND tenant_id = $2`,
    [id, tenantId]
  );
  return result.rows[0] || null;
}

async function createProject({ tenantId, name, description, createdBy }) {
  const result = await pool.query(
    `INSERT INTO projects (tenant_id, name, description, created_by)
     VALUES ($1, $2, $3, $4)
     RETURNING *`,
    [tenantId, name, description, createdBy]
  );
  return result.rows[0];
}

async function updateProject(id, tenantId, updates) {
  const result = await pool.query(
    `UPDATE projects
     SET name = COALESCE($3, name),
         description = COALESCE($4, description),
         updated_at = NOW()
     WHERE id = $1 AND tenant_id = $2
     RETURNING *`,
    [id, tenantId, updates.name, updates.description]
  );
  return result.rows[0] || null;
}

async function deleteProject(id, tenantId) {
  const result = await pool.query(
    `DELETE FROM projects WHERE id = $1 AND tenant_id = $2 RETURNING id`,
    [id, tenantId]
  );
  return result.rows[0] || null;
}

module.exports = { listProjects, getProject, createProject, updateProject, deleteProject };

Notice what getProject does when Tenant A tries to fetch a Tenant B resource. The query runs with Tenant A's tenantId. The condition id = $1 AND tenant_id = $2 matches nothing, null comes back, and the handler sends a 404. Not a 403. A 403 tells the caller the resource exists, but they can't access it, which is information they shouldn't have.

Audit Logging Service

// src/services/auditService.js
const { pool } = require('../../db');

async function log({
  tenantId,
  userId,
  userEmail,
  userRole,          // role at time of action — roles change, log should not
  action,            // 'CREATE' | 'UPDATE' | 'DELETE' | 'VIEW'
  resource,          // table name
  resourceId = null,
  oldValues = null,
  newValues = null,
  ipAddress = null,
  userAgent = null,
}) {
  const query = `
    INSERT INTO audit_logs
      (tenant_id, user_id, user_email, user_role, action, resource,
       resource_id, old_values, new_values, ip_address, user_agent)
    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
  `;

  const values = [
    tenantId, userId, userEmail, userRole, action, resource,
    resourceId,
    oldValues  ? JSON.stringify(oldValues)  : null,
    newValues  ? JSON.stringify(newValues)  : null,
    ipAddress,
    userAgent,
  ];

  // Fire-and-forget — audit logging must never block or fail a user request
  pool.query(query, values).catch((err) => {
    console.error('[AuditService] Failed to write log:', err.message);
  });
}

module.exports = { log };

Capturing userRole at write time matters more than it looks. User roles change after the fact: someone gets demoted, a permission is revoked. If the log only records the user ID, you lose the context of what privilege they held when the action happened. Store the role at the time of the action, and you always know.

Per-Tenant Rate Limiting

IP-based rate limiting breaks down in SaaS. A corporate customer might route hundreds of users through a single NAT gateway, sharing one IP address. One heavy tenant throttles everyone else on that address.

I've watched teams discover this the hard way when an enterprise customer suddenly floods the API, and their other tenants start getting 429s with no explanation. Scope limits to tenant_id instead.

// src/middleware/rateLimiter.js
const rateLimit = require('express-rate-limit');
const { RedisStore } = require('rate-limit-redis');
const { redisClient } = require('../../db/redis');

// Rate limits by plan — extend as needed
const PLAN_LIMITS = {
  free:       { max: 100,  windowMs: 15 * 60 * 1000 }, // 100 req / 15 min
  pro:        { max: 500,  windowMs: 15 * 60 * 1000 }, // 500 req / 15 min
  enterprise: { max: 2000, windowMs: 15 * 60 * 1000 }, // 2000 req / 15 min
};

function createTenantRateLimiter(plan = 'free') {
  const limits = PLAN_LIMITS[plan] || PLAN_LIMITS.free;

  return rateLimit({
    windowMs: limits.windowMs,
    max: limits.max,
    // Key = tenant_id from verified JWT — NOT the IP address
    keyGenerator: (req) => `tenant:${req.user?.tenantId || req.ip}`,
    store: new RedisStore({
      sendCommand: (...args) => redisClient.call(...args),
    }),
    handler: (req, res) => {
      res.status(429).json({
        error: 'Too many requests',
        retryAfter: Math.ceil(limits.windowMs / 1000),
      });
    },
  });
}

// Default limiter for all API routes
const defaultLimiter = createTenantRateLimiter('free');

module.exports = { defaultLimiter, createTenantRateLimiter };

Building the Routes

This is where everything connects. Auth and rate limiting apply to the whole router. Role checks go on individual routes. The audit log fires after every write. tenantId never comes from the request body or URL. req.user.tenantId is the only source, set by the auth middleware from the verified token, so there's no path around it.

One practical detail for Express 4: it doesn't catch async errors automatically. Every handler wraps its logic in try/catch and passes failures to next(err). Skip that and an unhandled promise rejection returns a blank 500 with no log entry and no audit trail. The comment at the top of the router is a reminder that the pattern is intentional.

// src/routes/projects.js
const express = require('express');
const { authMiddleware }  = require('../middleware/auth');
const { requireRole }     = require('../middleware/rbac');
const { defaultLimiter }  = require('../middleware/rateLimiter');
const audit               = require('../services/auditService');
const repo                = require('../repositories/projectRepo');

const router = express.Router();

// All routes require authentication
router.use(authMiddleware);
router.use(defaultLimiter);

// Express 4 does not catch async errors automatically.
// Every handler must wrap await calls in try/catch and pass errors to next().
// Without this, an unhandled promise rejection silently returns 500
// with no useful message and no audit log entry.

// GET /api/projects — list all (Viewer and above)
router.get('/', async (req, res, next) => {
  try {
    const projects = await repo.listProjects(req.user.tenantId);

    audit.log({
      tenantId:   req.user.tenantId,
      userId:     req.user.userId,
      userEmail:  req.user.email,
      userRole:   req.user.role,
      action:     'VIEW',
      resource:   'projects',
      ipAddress:  req.ip,
      userAgent:  req.headers['user-agent'],
    });

    res.json(projects);
  } catch (err) {
    next(err);
  }
});

// GET /api/projects/:id — single project (Viewer and above)
router.get('/:id', async (req, res, next) => {
  try {
    const project = await repo.getProject(req.params.id, req.user.tenantId);
    if (!project) return res.status(404).json({ error: 'Not found' });
    res.json(project);
  } catch (err) {
    next(err);
  }
});

// POST /api/projects — create (Member and above)
router.post('/', requireRole('Member', 'TenantAdmin', 'SuperAdmin'), async (req, res, next) => {
  try {
    const { name, description } = req.body;
    if (!name) return res.status(400).json({ error: 'name is required' });

    const project = await repo.createProject({
      tenantId:    req.user.tenantId,
      name,
      description,
      createdBy:   req.user.userId,
    });

    audit.log({
      tenantId:    req.user.tenantId,
      userId:      req.user.userId,
      userEmail:   req.user.email,
      userRole:    req.user.role,
      action:      'CREATE',
      resource:    'projects',
      resourceId:  project.id,
      newValues:   project,
      ipAddress:   req.ip,
      userAgent:   req.headers['user-agent'],
    });

    res.status(201).json(project);
  } catch (err) {
    next(err);
  }
});

// PUT /api/projects/:id — update (Member and above)
router.put('/:id', requireRole('Member', 'TenantAdmin', 'SuperAdmin'), async (req, res, next) => {
  try {
    const oldProject = await repo.getProject(req.params.id, req.user.tenantId);
    if (!oldProject) return res.status(404).json({ error: 'Not found' });

    const updated = await repo.updateProject(req.params.id, req.user.tenantId, req.body);

    audit.log({
      tenantId:    req.user.tenantId,
      userId:      req.user.userId,
      userEmail:   req.user.email,
      userRole:    req.user.role,
      action:      'UPDATE',
      resource:    'projects',
      resourceId:  req.params.id,
      oldValues:   oldProject,
      newValues:   updated,
      ipAddress:   req.ip,
      userAgent:   req.headers['user-agent'],
    });

    res.json(updated);
  } catch (err) {
    next(err);
  }
});

// DELETE /api/projects/:id — TenantAdmin and above only
router.delete('/:id', requireRole('TenantAdmin', 'SuperAdmin'), async (req, res, next) => {
  try {
    const project = await repo.getProject(req.params.id, req.user.tenantId);
    if (!project) return res.status(404).json({ error: 'Not found' });

    await repo.deleteProject(req.params.id, req.user.tenantId);

    audit.log({
      tenantId:    req.user.tenantId,
      userId:      req.user.userId,
      userEmail:   req.user.email,
      userRole:    req.user.role,
      action:      'DELETE',
      resource:    'projects',
      resourceId:  req.params.id,
      oldValues:   project,
      ipAddress:   req.ip,
      userAgent:   req.headers['user-agent'],
    });

    res.json({ deleted: true });
  } catch (err) {
    next(err);
  }
});

module.exports = router;

Testing Tenant Isolation

Skip the isolation tests and you're flying blind. The application keeps running, nothing throws an error, but two customers are reading each other's data.

I've watched this sit undetected in production for months because nothing actually broke. The wrong data just showed up quietly. Automated tests on every pull request are the only reliable way to catch it early.

// tests/tenantIsolation.test.js
require('dotenv').config();  // must be first — loads DATABASE_URL and REDIS_URL
const request = require('supertest');
const app     = require('../app');
const { generateToken } = require('../src/utils/token');
const { pool }        = require('../db');
const { redisClient } = require('../db/redis');

// Test fixture: two isolated tenants, one project in Tenant B
async function seedTestData() {
  // Clean up from any previous run to avoid unique-constraint failures
  await pool.query(`DELETE FROM projects WHERE name LIKE 'TEST-%'`);
  await pool.query(`DELETE FROM tenants WHERE name IN ('Tenant A', 'Tenant B')`);

  const tenantA = (await pool.query(
    `INSERT INTO tenants (name, plan) VALUES ('Tenant A', 'pro') RETURNING id`
  )).rows[0].id;

  const tenantB = (await pool.query(
    `INSERT INTO tenants (name, plan) VALUES ('Tenant B', 'pro') RETURNING id`
  )).rows[0].id;

  const userA = (await pool.query(
    `INSERT INTO users (tenant_id, email, role) VALUES ($1, 'usera@a.com', 'Member') RETURNING id`,
    [tenantA]
  )).rows[0].id;

  // userB owns the project in Tenant B — satisfies the created_by FK constraint
  const userB = (await pool.query(
    `INSERT INTO users (tenant_id, email, role) VALUES ($1, 'userb@b.com', 'Member') RETURNING id`,
    [tenantB]
  )).rows[0].id;

  const projectB = (await pool.query(
    `INSERT INTO projects (tenant_id, name, created_by)
     VALUES ($1, 'TEST-Secret Project', $2) RETURNING id`,
    [tenantB, userB]
  )).rows[0].id;

  return { tenantA, tenantB, userA, projectB };
}

describe('Tenant Isolation', () => {
  let data;

  beforeAll(async () => {
    data = await seedTestData();
  });

  afterAll(async () => {
    await pool.query(`DELETE FROM tenants WHERE name IN ('Tenant A', 'Tenant B')`);
    await pool.end();
    await redisClient.quit();  // close Redis connection so Jest exits cleanly
  });

  test('Tenant A user cannot read Tenant B project', async () => {
    const token = generateToken({
      userId:   data.userA,
      tenantId: data.tenantA,   // ← Tenant A token
      email:    'usera@a.com',
      role:     'Member',
    });

    const res = await request(app)
      .get(`/api/projects/${data.projectB}`)  // ← Tenant B's project ID
      .set('Authorization', `Bearer ${token}`);

    // Must be 404, not 200 or 403
    expect(res.status).toBe(404);
  });

  test('Tenant A user cannot list Tenant B projects', async () => {
    const token = generateToken({
      userId:   data.userA,
      tenantId: data.tenantA,
      email:    'usera@a.com',
      role:     'TenantAdmin',
    });

    const res = await request(app)
      .get('/api/projects')
      .set('Authorization', `Bearer ${token}`);

    expect(res.status).toBe(200);
    // Response must contain zero Tenant B projects
    const names = res.body.map(p => p.name);
    expect(names).not.toContain('TEST-Secret Project');
  });

  test('Viewer cannot delete a project', async () => {
    const token = generateToken({
      userId:   data.userA,
      tenantId: data.tenantA,
      email:    'usera@a.com',
      role:     'Viewer',         // ← Viewer role
    });

    const res = await request(app)
      .delete(`/api/projects/${data.projectB}`)
      .set('Authorization', `Bearer ${token}`);

    expect(res.status).toBe(403);
  });
});

Run the tests:

npm test

Three tests, three boundaries confirmed. Wire these into CI so they run on every pull request. A future refactor that quietly drops the tenant_id filter will get caught before it ships.

Troubleshooting

Tenant A can see Tenant B's data

One query is missing the AND tenant_id = $N clause. Search every repository file for SELECT statements and check each one. It's almost always this.

403 Forbidden on a route that should be accessible

The role string in the JWT doesn't match what requireRole() is checking. Check the exact string in the token payload. 'member' and 'Member' aren't the same thing. Paste your token into jwt.io and look at the role field directly.

Rate limiter isn't working

Redis is probably not connected. Log redisClient.status before the server starts. If it's not ready, the limiter has fallen back to in-memory, which means restarts reset all counters and tenant-scoped limiting stops working.

Audit log table growing very large

Expected behaviour. Audit tables grow, that's the point. Once it gets large, ship rows older than a year to S3 or Azure Blob and keep querying against a smaller hot table. Most compliance requirements want at least 12 months of accessible logs anyway. Just don't DELETE from the table itself.

jwt.verify throws JsonWebTokenError: invalid signature

The secret that signed the token doesn't match JWT_SECRET in the environment where you're verifying it. This comes up most when switching between environments or when a second service has a different value in its .env. Every service that calls jwt.verify needs the exact same secret. Copy it across, don't retype it.

Wrapping Up

The system you've built: row-level isolation in the repository, role checks before the handler runs, an audit table the app can't touch, and rate limits per tenant. That's the whole thing.

The tests are what I see dropped most often. Teams build the isolation, ship it, and never write something that actually proves cross-tenant data can't leak. Then a query gets refactored six months later and the tenant_id filter quietly disappears. CI catches it. Manual code review rarely does.

Schema-per-tenant comes up eventually if your product grows large enough. But not at the start. Row-level isolation handles more scale than most teams will ever hit, and it costs a fraction of the operational overhead.

The full working code is available on GitHub: nodejs-multitenant-saas-api

Role-Based Access Control sounds fancy, but the idea is simple: what you can do depends on who you are, not just that you're logged in. An admin deletes users. An editor creates posts. A regular user just reads. Same app, completely different experience depending on who's asking.

That's what we're building here. A REST API with three roles: JWT to carry those roles on every request, and a pair of middleware functions that check permissions before your route handlers even run. There's no database hit per request, and no if/else soup in your business logic.

By the end, you'll have three working roles (admin, editor, user) each locked to their own endpoints. More importantly, the pattern is transferable: once it clicks, you'll wire it into your next project without needing a tutorial.

Full source code on GitHub: github.com/ziaongit/nodejs-rbac-jwt-api

Table of Contents

• What You'll Learn

• Prerequisites

• What We'll Build

• Project Setup

• Setting Up the In-Memory Data Store

• Building the Auth Routes

• Building the RBAC Middleware

• Building the Protected Routes

• Putting It All Together

• Testing the API

• Key Takeaways

• Conclusion

What You'll Learn

• What RBAC is and how it differs from basic authentication

• How to embed roles in JWT payloads

• How to write reusable Express middleware for token verification and role checking

• How to protect API routes based on user roles

Prerequisites

• Node.js (v18+) installed

• Basic knowledge of Express.js

• Familiarity with how JWTs work (we'll cover the relevant parts)

• npm installed

What We'll Build

We'll build a REST API for a simple content management system with three user roles:

The API will expose these endpoints:

Project Setup

Create a new folder and initialize the project:

mkdir nodejs-rbac-jwt-api
cd nodejs-rbac-jwt-api
npm init -y

Install the dependencies:

npm install express jsonwebtoken bcryptjs dotenv
npm install --save-dev nodemon

Here's what each package does:

• express: web framework for building the API

• jsonwebtoken: creates and verifies JWTs

• bcryptjs: securely hashes passwords

• dotenv: reads your .env file so you're not hardcoding secrets in your source code

Update package.json to add start scripts:

"scripts": {
  "start": "node src/app.js",
  "dev": "nodemon src/app.js"
}

Create the project structure:

nodejs-rbac-jwt-api/
├── src/
│   ├── middleware/
│   │   └── auth.js
│   ├── routes/
│   │   ├── auth.js
│   │   ├── content.js
│   │   └── admin.js
│   ├── data/
│   │   └── users.js
│   └── app.js
├── .env
├── .env.example
└── package.json

Create your .env file:

JWT_SECRET=your_super_secret_key_change_this_in_production
PORT=3000

Important: Never commit your .env file to version control. Add it to .gitignore.

Setting Up the In-Memory Data Store

We don't have a database here, just an array in memory. The point was to keep the focus on RBAC, not spend half the tutorial on database config. In a real project, swap the array for whatever database you're already using.

Create src/data/users.js:

// In-memory users store
// In production, replace this with a real database (MongoDB, PostgreSQL, etc.)
const users = [];

const findUserByEmail = (email) => users.find((u) => u.email === email);
const findUserById = (id) => users.find((u) => u.id === id);
const createUser = (user) => {
  users.push(user);
  return user;
};
const getAllUsers = () => users.map(({ password, ...user }) => user);

module.exports = { findUserByEmail, findUserById, createUser, getAllUsers };

One thing worth noting: getAllUsers uses destructuring to drop the password before returning anything. Never send password fields in API responses, even hashed ones.

Building the Auth Routes

The auth routes handle registration and login. Login is where roles first enter the picture — we embed the user's role directly into the JWT payload.

Create src/routes/auth.js:

const express = require('express');
const bcrypt = require('bcryptjs');
const jwt = require('jsonwebtoken');
const { findUserByEmail, createUser } = require('../data/users');

const router = express.Router();

// POST /api/auth/register
router.post('/register', async (req, res) => {
  const { name, email, password, role } = req.body;

  if (!name || !email || !password) {
    return res.status(400).json({ message: 'Name, email, and password are required' });
  }

  if (findUserByEmail(email)) {
    return res.status(409).json({ message: 'Email already registered' });
  }

  // Only allow valid roles — default to 'user' if none provided
  const validRoles = ['user', 'editor', 'admin'];
  const assignedRole = validRoles.includes(role) ? role : 'user';

  const hashedPassword = await bcrypt.hash(password, 10);

  const newUser = {
    id: Date.now().toString(),
    name,
    email,
    password: hashedPassword,
    role: assignedRole,
  };

  createUser(newUser);

  res.status(201).json({
    message: 'User registered successfully',
    user: {
      id: newUser.id,
      name: newUser.name,
      email: newUser.email,
      role: newUser.role,
    },
  });
});

// POST /api/auth/login
router.post('/login', async (req, res) => {
  const { email, password } = req.body;

  if (!email || !password) {
    return res.status(400).json({ message: 'Email and password are required' });
  }

  const user = findUserByEmail(email);
  if (!user) {
    return res.status(401).json({ message: 'Invalid credentials' });
  }

  const isMatch = await bcrypt.compare(password, user.password);
  if (!isMatch) {
    return res.status(401).json({ message: 'Invalid credentials' });
  }

  // Issue JWT — embed role in the payload
  const token = jwt.sign(
    {
      id: user.id,
      email: user.email,
      role: user.role,   // ← This is the key part for RBAC
    },
    process.env.JWT_SECRET,
    { expiresIn: '24h' }
  );

  res.json({
    message: 'Login successful',
    token,
  });
});

module.exports = router;

The most important line is the JWT payload:

jwt.sign({ id, email, role }, process.env.JWT_SECRET, { expiresIn: '24h' })

By embedding role in the token, every subsequent request carries the user's permissions without requiring a database lookup. The server just verifies the token and reads the role from the payload.

Building the RBAC Middleware

This is the core of the system. We need two separate middleware functions:

1. verifyToken confirms the JWT is valid and attaches the decoded payload to req.user

2. checkRole confirms the user has the required role for a specific route

Keeping them separate gives you flexibility. Some routes only need authentication. Others need both authentication and a specific role.

Create src/middleware/auth.js:

const jwt = require('jsonwebtoken');

// Middleware 1: Verify the JWT token
const verifyToken = (req, res, next) => {
  const authHeader = req.headers['authorization'];
  const token = authHeader && authHeader.split(' ')[1]; // Expects: Bearer <token>

  if (!token) {
    return res.status(401).json({ message: 'Access denied. No token provided.' });
  }

  try {
    const decoded = jwt.verify(token, process.env.JWT_SECRET);
    req.user = decoded; // Attach decoded payload (including role) to request
    next();
  } catch (err) {
    return res.status(403).json({ message: 'Invalid or expired token.' });
  }
};

// Middleware 2: Check if user has one of the required roles
const checkRole = (...allowedRoles) => {
  return (req, res, next) => {
    if (!req.user) {
      return res.status(401).json({ message: 'Not authenticated.' });
    }

    if (!allowedRoles.includes(req.user.role)) {
      return res.status(403).json({
        message: `Access denied. Required role: ${allowedRoles.join(' or ')}. Your role: ${req.user.role}`,
      });
    }

    next();
  };
};

module.exports = { verifyToken, checkRole };

checkRole uses a rest parameter (...allowedRoles) so you can pass in one or multiple roles:

checkRole('admin')                  // only admin
checkRole('editor', 'admin')        // editor or admin
checkRole('user', 'editor', 'admin') // all roles

This makes route definitions clean and readable — the permissions are visible right at the route level.

Building the Protected Routes

Now let's wire up routes that use the middleware.

Create src/routes/content.js:

const express = require('express');
const { verifyToken, checkRole } = require('../middleware/auth');

const router = express.Router();

// In-memory content store
const content = [
  { id: '1', title: 'Getting Started with Node.js', author: 'admin' },
  { id: '2', title: 'Express Middleware Explained', author: 'editor' },
];

// GET /api/content — all authenticated users
router.get('/', verifyToken, checkRole('user', 'editor', 'admin'), (req, res) => {
  res.json({ content });
});

// POST /api/content — editors and admins only
router.post('/', verifyToken, checkRole('editor', 'admin'), (req, res) => {
  const { title } = req.body;

  if (!title) {
    return res.status(400).json({ message: 'Title is required' });
  }

  const newItem = {
    id: Date.now().toString(),
    title,
    author: req.user.email,
  };

  content.push(newItem);
  res.status(201).json({ message: 'Content created', item: newItem });
});

// DELETE /api/content/:id — admin only
router.delete('/:id', verifyToken, checkRole('admin'), (req, res) => {
  const index = content.findIndex((c) => c.id === req.params.id);

  if (index === -1) {
    return res.status(404).json({ message: 'Content not found' });
  }

  content.splice(index, 1);
  res.json({ message: 'Content deleted successfully' });
});

module.exports = router;

Notice how readable each route is:

router.delete('/:id', verifyToken, checkRole('admin'), handler)

You can understand the access control without reading the handler body. This is one of the key advantages of middleware-based RBAC: permissions live at the routing layer, not buried in business logic.

Create src/routes/admin.js:

const express = require('express');
const { verifyToken, checkRole } = require('../middleware/auth');
const { getAllUsers } = require('../data/users');

const router = express.Router();

// GET /api/admin/users — admin only
router.get('/users', verifyToken, checkRole('admin'), (req, res) => {
  res.json({ users: getAllUsers() });
});

module.exports = router;

Putting It All Together

Create src/app.js:

require('dotenv').config();
const express = require('express');

const authRoutes = require('./routes/auth');
const contentRoutes = require('./routes/content');
const adminRoutes = require('./routes/admin');

const app = express();

app.use(express.json());

// Routes
app.use('/api/auth', authRoutes);
app.use('/api/content', contentRoutes);
app.use('/api/admin', adminRoutes);

// Health check
app.get('/', (req, res) => {
  res.json({ message: 'RBAC API is running' });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
});

Testing the API

Start the server:

npm run dev

Step 1: Register Users with Different Roles

Register an admin:

curl -X POST http://localhost:3000/api/auth/register \
  -H "Content-Type: application/json" \
  -d '{"name": "Admin User", "email": "admin@example.com", "password": "password123", "role": "admin"}'

Register an editor:

curl -X POST http://localhost:3000/api/auth/register \
  -H "Content-Type: application/json" \
  -d '{"name": "Editor User", "email": "editor@example.com", "password": "password123", "role": "editor"}'

Register a regular user (no role specified — defaults to user):

curl -X POST http://localhost:3000/api/auth/register \
  -H "Content-Type: application/json" \
  -d '{"name": "Regular User", "email": "user@example.com", "password": "password123"}'

Step 2: Log in and Get a Token

curl -X POST http://localhost:3000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "password": "password123"}'

You'll get a response like:

{
  "message": "Login successful",
  "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

Copy the token.

Step 3: Test Role-based Access

Read content as a regular user (should succeed):

curl http://localhost:3000/api/content \
  -H "Authorization: Bearer YOUR_TOKEN_HERE"

Try creating content as a regular user (should fail — 403):

curl -X POST http://localhost:3000/api/content \
  -H "Authorization: Bearer YOUR_TOKEN_HERE" \
  -H "Content-Type: application/json" \
  -d '{"title": "New Article"}'

Response:

{
  "message": "Access denied. Required role: editor or admin. Your role: user"
}

Now log in as an editor and try the same POST request. It succeeds. Log in as admin and try the DELETE route. Only the admin token will work.

Step 4: Decode the JWT to See the Role

You can paste any token into jwt.io to inspect the payload. You'll see something like:

{
  "id": "1720300000000",
  "email": "admin@example.com",
  "role": "admin",
  "iat": 1720300000,
  "exp": 1720386400
}

The role field is exactly what checkRole reads on every protected request.

Key Takeaways

Roles live in the JWT payload. The role travels with the token — no extra DB call needed every time someone hits a protected route. It gets embedded at login and verified cryptographically on each request.

Middleware is composable. verifyToken and checkRole are separate, reusable functions. You can chain them on any route in any combination.

Permissions are visible at the route level. router.delete('/:id', verifyToken, checkRole('admin'), handler) tells you everything about access control before you even read the handler.

Before you ship this to production:

• The in-memory array was just to keep this tutorial focused — replace it with a real database before anything goes near production. A server restart wipes all your users right now.

• That 24h token expiry is too long. Cut it to 15 minutes and add refresh token rotation. A stolen token becomes useless fast.

• Re-validate roles from the DB on sensitive operations. A role change won't reflect in an existing token until it expires

• HTTPS, always

• If your permission logic grows beyond "check a role", look at casl. It handles attribute-level rules cleanly

Conclusion

The core of it fits in two middleware functions and a JWT payload. I've used this same pattern across several projects. And once you've built it yourself, you'll start spotting it everywhere, because almost every multi-user app needs some version of it.

Full source code on GitHub: github.com/ziaongit/nodejs-rbac-jwt-api

The boring stuff is what gets you, because the boring stuff is what everyone agrees to fix "later," and later has a way of never arriving.

My favorite personal example (favorite in the way a scar is your favorite) was an internal API that compared an access token with a plain equality check and had no limit on request size. It ran fine for a year. It ran fine right up until someone curious discovered they could both fingerprint the token comparison and post a body large enough to make the server sweat.

Neither bug was sophisticated. Both would have been a complete non-event if something had simply refused to let me do the wrong thing in the first place.

This tutorial shows you how to add practical guardrails around every HTTP API you build, regardless of framework. You'll write them by hand, in plain Node.js, with no dependencies. This will let you see exactly what each one does and why.

By the end, you'll have a small server that survives a lot more contact with the public internet than the version most of us shipped early in our careers.

This tutorial uses plain JavaScript so anyone can copy and run it. If you use TypeScript, you can add types afterward. You need Node 22 or newer, a basic understanding of HTTP requests and responses, and a terminal for testing the examples.

What We'll Cover:

1. Prerequisites

2. What You'll Build

3. How to Start with the Naïve Server

4. How to Limit the Request Body

5. How to Time Out Slow Requests

6. How to Parse JSON Safely and Block Prototype Pollution

7. How to Set Security Headers on Every Response

8. How to Compare Secrets in Constant Time

9. How to Validate Input as a Gate, Not a Suggestion

10. How to Fail Without Leaking and Log So You Can See It

11. How to Put It All Together

12. How to Handle CORS Correctly

13. What This Tutorial Doesn't Cover

14. Why Defaults Beat Checklists

15. An Honest Note on Frameworks

16. The Takeaway Checklist

Prerequisites

• Node.js 22 or newer

• Basic familiarity with HTTP requests and responses

• A terminal and curl, Postman, or a similar client

This isn't a guide to making your API unhackable. It's a practical guide to avoiding the easy attacks and building safer defaults.

What You'll Build

A plain Node.js API with request size limits, request timeouts, safe JSON parsing, security headers, timing-safe secret comparison, validation, and error handling.

How to Start with the Naïve Server

Here's the kind of server I wrote when I was younger and braver and wrong. It reads a JSON body and echoes it back. Pretend it's the start of a real API.

import http from "node:http";

const server = http.createServer((req, res) => {
  let body = "";
  req.on("data", (chunk) => (body += chunk));
  req.on("end", () => {
    const data = JSON.parse(body || "{}");
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(JSON.stringify({ youSent: data }));
  });
});

server.listen(3000, () => console.log("listening on http://localhost:3000"));

It works. You can curl it and it answers. It's also a small disaster, and here's the incomplete list of why:

• It reads an unbounded body into memory. Send it a few gigabytes and you have a denial-of-service with no effort.

• JSON.parse throws on bad input, which here means an uncaught exception that can take the process down.

• There's no timeout. A client that sends one byte per minute can hold a connection hostage.

• It sets no security headers and happily advertises that it's a Node server.

• It parses untrusted JSON straight into an object with no checks, which opens the door to prototype pollution downstream.

You're going to fix each of these issues. The fixes are small. The point is to make them habits, not heroics.

How to Limit the Request Body

The first rule of accepting input from strangers is to decide, in advance, how much input you're willing to accept. If you don't set a limit, the limit is "however much RAM the server has," and someone will find that out for you.

There are two layers here. The first is the Content-Length header, which the client sends to declare how big the body is. You can reject early based on it. But you must never trust it alone, because a client can lie or simply not send it.

The real defense is to count bytes as they stream in and stop the moment they cross your line.

const MAX_BODY_BYTES = 100 * 1024; // 100 KB is plenty for most JSON APIs

function readBody(req, limit = MAX_BODY_BYTES) {
  return new Promise((resolve, reject) => {
    // Cheap early rejection if the client is honest about being too big.
    const declared = Number(req.headers["content-length"]);
    if (Number.isFinite(declared) && declared > limit) {
      reject(httpError(413, "Payload too large"));
      return;
    }

    let size = 0;
    const chunks = [];

    req.on("data", (chunk) => {
      size += chunk.length;
      if (size > limit) {
        reject(httpError(413, "Payload too large"));
        req.destroy(); // stop reading; we are done with this client
        return;
      }
      chunks.push(chunk);
    });

    req.on("end", () => resolve(Buffer.concat(chunks)));
    req.on("error", reject);
  });
}

function httpError(statusCode, message) {
  return Object.assign(new Error(message), { statusCode });
}

A few things worth noticing here. You accumulate Buffer chunks and only join them at the end, rather than concatenating strings, because string concatenation forces an early decode and can mangle multibyte UTF-8 characters that happen to land on a chunk boundary.

You also call req.destroy() as soon as you cross the limit, so you don't keep pulling bytes you've already decided to refuse.

Pick a limit that matches the route. A JSON API that creates a user doesn't need a 50 MB body. A file upload endpoint is a different conversation, and there you would stream to disk or object storage instead of buffering in memory at all. The mistake is having no limit, not having the wrong one.

How to Time Out Slow Requests

Once you have a body limit, the next trick an attacker reaches for is to be slow instead of large. This is the family of attacks named after slowloris, a sad-looking primate that moves very slowly, which is rude to the animal but accurate about the attack.

The idea is to open many connections and feed them bytes at a glacial pace, never finishing, so the server keeps each one alive waiting politely. Do that enough times and you've exhausted the connection pool without sending anything that looks malicious.

Node has built-in defenses for this, and the defaults are generous and worth tightening for an API.

const server = http.createServer(handler);

// Total time allowed to receive the entire request (headers + body).
server.requestTimeout = 30_000; // 30 seconds

// Time allowed to receive just the headers. Slowloris lives here.
server.headersTimeout = 10_000; // 10 seconds

// Idle socket timeout: kill connections that go quiet.
server.setTimeout(60_000);

Those three lines handle the network layer. But there's a second kind of slow: your own handler. A database query that hangs, an outbound call to a third party that never answers, a regular expression that decided to think about its life choices. You want a ceiling on how long a single request is allowed to occupy a worker, and you want to be able to cancel the work when that ceiling is hit.

The modern tool for cancellation in Node is AbortController. Here's a small wrapper that gives every handler a deadline and a signal it can pass down to anything that supports cancellation, like fetch.

function withTimeout(handler, ms = 15_000) {
  return async (req, res) => {
    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), ms);
    try {
      await handler(req, res, controller.signal);
    } finally {
      clearTimeout(timer);
    }
  };
}

Now a handler can do await fetch(url, { signal }) and the request gets cut off if it blows the deadline, instead of camping on a worker forever.

The discipline to learn here is that any time you talk to something outside your process, you give it a deadline. Networks fail in the most boring way possible: by hanging, not by erroring. A timeout turns a hang into a clean error you can handle.

How to Parse JSON Safely and Block Prototype Pollution

This is the one people skip because it sounds theoretical, and then it shows up in a CVE with their stack in it.

First, the easy half. JSON.parse throws a SyntaxError on malformed input. In the naïve server, that throw was uncaught and could crash the process. So we wrap parsing and turn a parse failure into a clean 400.

function parseJson(buffer) {
  if (buffer.length === 0) return {};
  let text = buffer.toString("utf8");
  try {
    return JSON.parse(text, reviver);
  } catch {
    throw httpError(400, "Invalid JSON body");
  }
}

Now the interesting half: that reviver argument. Prototype pollution is an attack where a request payload reaches up and modifies Object.prototype, the object that almost every object in your program inherits from. If an attacker can set a property there, they can set it on effectively everything at once.

It's easier to believe once you see it. Here's a recursive merge function, the kind people write all the time to apply updates onto an existing record:

function merge(target, source) {
  for (const key in source) {
    if (source[key] && typeof source[key] === "object") {
      if (!target[key]) target[key] = {};
      merge(target[key], source[key]);
    } else {
      target[key] = source[key];
    }
  }
}

Looks harmless. Now feed it a payload an attacker controls:

const evil = JSON.parse('{"__proto__": {"isAdmin": true}}');
const account = {};
merge(account, evil);

console.log(account.isAdmin);   // undefined, account itself is fine
console.log(({}).isAdmin);      // true  <-- every object is now "admin"

That second line is the horror. You never touched ({}). You polluted the shared prototype, so a brand new empty object now reports isAdmin: true. If somewhere later your code does if (user.isAdmin) on an object that didn't explicitly set that field, congratulations! Everyone is an admin. The __proto__ key tricked the merge into walking up into the prototype that all objects share.

The defense is to refuse the dangerous keys before they ever get into your data. The cleanest way at parse time is the reviver, a function JSON.parse calls for every key as it builds the result. Return undefined for a key and it gets dropped.

const FORBIDDEN_KEYS = new Set(["__proto__", "constructor", "prototype"]);

function reviver(key, value) {
  if (FORBIDDEN_KEYS.has(key)) return undefined;
  return value;
}

That is it. Three key names, blocked at the door, and the merge above becomes harmless because the payload never carries __proto__ past the parser.

For extra defense in depth, you can also build internal lookup objects with Object.create(null), which creates an object with no prototype at all, or use a Map when the keys are user-controlled. And if you want a belt to go with the suspenders, Object.freeze(Object.prototype) early in your process start will make the whole class of attack fail loudly.

I wouldn't rely on freezing alone, because some libraries get unhappy about it, but blocking the keys costs you nothing and should be the default.

How to Set Security Headers on Every Response

Browsers will defend your users for you, but only if you tell them to. That instruction comes as a small set of response headers. For an API that returns JSON, the list is short and the defaults are strict, which is exactly how you want it.

function secureHeaders(res) {
  // Do not let the browser guess content types. Stops a JSON response
  // from being treated as HTML or a script.
  res.setHeader("X-Content-Type-Options", "nosniff");

  // Clickjacking defense: do not allow this response inside a frame.
  res.setHeader("X-Frame-Options", "DENY");
  res.setHeader("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'");

  // Do not leak the full URL (which may contain ids or tokens) on navigation.
  res.setHeader("Referrer-Policy", "no-referrer");

  // Only meaningful over HTTPS: force HTTPS for two years, including subdomains.
  res.setHeader("Strict-Transport-Security", "max-age=63072000; includeSubDomains");

  // Stop advertising what you are running. Free reconnaissance for nobody.
  res.removeHeader("X-Powered-By");
}

A quick tour, because cargo-culting headers is how you end up with a Content Security Policy that does nothing.

X-Content-Type-Options: nosniff stops the browser from second-guessing your Content-Type, which closes a sneaky path where a response gets reinterpreted as something executable. X-Frame-Options and the frame-ancestors directive both refuse to let your responses be embedded in a frame, which is the heart of clickjacking. For a pure JSON API, default-src 'none' is a strong and appropriate CSP, because an API has no business loading scripts, styles, or images.

Referrer-Policy: no-referrer keeps your URLs, which sometimes carry ids you would rather not gossip about, from being sent to other sites. Strict-Transport-Security only matters once you're on HTTPS, but once you are, it stops downgrade attacks by telling the browser to refuse plain HTTP.

And removing X-Powered-By is a tiny thing that just denies an attacker a free hint about what to throw at you.

The reason to wrap this in a function and call it on every response is that "every response" is the part humans forget. Make it one call you do at the top of the request, before you can get distracted.

How to Compare Secrets in Constant Time

Here's a bug that looks completely fine and is completely broken:

if (providedApiKey === expectedApiKey) {
  // grant access
}

The problem is that === on strings is allowed to be fast. It compares character by character and returns false the instant it finds a mismatch. That means a wrong guess that gets the first character right takes very slightly longer to reject than one that gets it wrong immediately.

That timing difference is tiny, but it's measurable over enough requests, and it lets an attacker recover a secret one character at a time. This is a real attack, it has a name (a timing attack), and the fix is built into Node.

You want a comparison whose running time doesn't depend on where the first difference is. Node gives you crypto.timingSafeEqual for exactly this. It has one sharp edge: it throws if the two buffers have different lengths, and length itself is a leak.

The clean way to handle both problems at once is to hash both inputs to a fixed size first, then compare the hashes.

import { timingSafeEqual, createHash } from "node:crypto";

function safeCompare(a, b) {
  // Hashing normalizes length (so timingSafeEqual is happy) and hides
  // the length of the real secret from a timing observer.
  const ha = createHash("sha256").update(String(a)).digest();
  const hb = createHash("sha256").update(String(b)).digest();
  return timingSafeEqual(ha, hb);
}

Use this anywhere you compare a value a stranger supplied against a secret you hold: API keys, webhook signatures, password reset tokens, session identifiers. The rule of thumb is simple: if being wrong about the comparison would let someone in, don't use ===.

One important caveat so you don't misuse this. For user passwords, don't store them and compare with this function. Passwords get hashed with a slow, purpose-built algorithm so that even if your database leaks, the hashes are expensive to crack.

Node ships scrypt for this in node:crypto, and bcrypt and argon2 are popular libraries. The safeCompare above is for comparing high-entropy secrets like tokens and keys, not for human-chosen passwords.

How to Validate Input as a Gate, Not a Suggestion

Everything so far has been about surviving hostile input at the transport level. Validation is about refusing input that doesn't match the shape your code expects, before that input reaches your business logic.

A surprising amount of "weird production behavior" is just a handler that assumed a field was a string and got an array, or assumed a number and got the string "NaN".

You can hand-roll validation for a small API, and it's healthy to see what that looks like before you reach for a library:

function expect(condition, message) {
  if (!condition) throw httpError(400, message);
}

function parseCreateUser(data) {
  expect(typeof data.email === "string" && data.email.includes("@"), "email is required");
  expect(typeof data.password === "string", "password is required");
  expect(data.password.length >= 12, "password must be at least 12 characters");
  // Return only the fields you actually want. Ignore everything else.
  return { email: data.email, password: data.password };
}

Notice the last line. You build a fresh object with only the fields you asked for, rather than passing data straight through. This quietly closes a mass-assignment hole, where a client sends {"email": "...", "password": "...", "role": "admin"} and a careless handler writes the whole object into the database, role included. If you only copy the fields you meant to accept, the extra ones never matter.

For anything beyond a few routes, a schema library pays for itself fast. Zod and Valibot are the popular choices, and both let you describe the shape once and get validation plus inferred types out of it.

import { z } from "zod";

const CreateUser = z
  .object({
    email: z.string().email(),
    password: z.string().min(12),
  })
  .strict(); // reject unknown keys instead of ignoring them

const result = CreateUser.safeParse(data);
if (!result.success) throw httpError(400, "Validation failed");
const user = result.data;

That .strict() call is doing the same mass-assignment defense as our hand-rolled version, but declaratively. Whether you hand-roll it or use a library, the principle is the same: input is guilty until proven to match a shape you defined on purpose.

How to Fail Without Leaking and Log So You Can See It

Errors are going to happen. The real question is what your server says when they do, and whether you can reconstruct what went wrong afterward.

There are two common ways to get this wrong, and they are opposites. Either you hand attackers a map of your internals, or you blind yourself during an incident.

The leaking version looks like this, and yes, I've shipped it:

catch (err) {
  res.writeHead(500);
  res.end(err.stack); // please do not
}

That stack trace can include file paths, library versions, query fragments, and sometimes secrets that got interpolated into an error message. It's a free briefing for whoever is poking at you.

The rule is blunt: a 500 should tell the client nothing useful, and tell you everything, through your logs.

The opposite failure is hiding the error so thoroughly that when it happens in production at 2am you have nothing to go on.

The fix for both problems is the same small idea: a request id. It's a short unique value you attach to each request, return to the client in a header, and include in every log line for that request. When a user reports "I got an error and it said request abc123," you can find exactly that request in your logs in seconds.

import { randomUUID } from "node:crypto";

function withRequestId(req, res) {
  const requestId = req.headers["x-request-id"] ?? randomUUID();
  res.setHeader("X-Request-Id", requestId);
  return requestId;
}

function log(level, requestId, message, extra = {}) {
  // Structured logs: one JSON object per line, easy to search and ship.
  console.log(
    JSON.stringify({ level, requestId, message, ...extra, at: new Date().toISOString() }),
  );
}

function sendError(res, err, requestId) {
  const status = err.statusCode ?? 500;
  const message = status === 500 ? "Internal Server Error" : err.message;
  if (status === 500) {
    log("error", requestId, "unhandled error", { stack: err.stack });
  }
  if (!res.headersSent) {
    res.writeHead(status, { "Content-Type": "application/json" });
  }
  res.end(JSON.stringify({ error: message, requestId }));
}

The client gets the requestId but never the details. They can quote it to support, but they can't read your stack trace. Accept an incoming X-Request-Id when a trusted upstream set one, so a single request keeps the same id as it moves across your services, but generate your own whenever it is missing. Structured logs, one JSON object per line, are worth the slight ugliness, because they're trivial to filter and feed into a log aggregator, which a pile of freeform console.log calls is not.

One note on environments. It's fine, even helpful, to return richer error detail when you run locally. Just gate it on an explicit environment check and make production the strict default, so the worst outcome of a misconfiguration is too little information leaked, never too much.

How to Put It All Together

None of these guardrails is impressive on its own. The power is in having all of them on, by default, on every route, so that the safe path is the path of least resistance.

Here is the naïve server from the start, rebuilt with everything we covered. It's still tiny. It's just no longer naïve.

import http from "node:http";
import { timingSafeEqual, createHash } from "node:crypto";

const MAX_BODY_BYTES = 100 * 1024;
const FORBIDDEN_KEYS = new Set(["__proto__", "constructor", "prototype"]);

function httpError(statusCode, message) {
  return Object.assign(new Error(message), { statusCode });
}

function reviver(key, value) {
  return FORBIDDEN_KEYS.has(key) ? undefined : value;
}

function readBody(req, limit = MAX_BODY_BYTES) {
  return new Promise((resolve, reject) => {
    const declared = Number(req.headers["content-length"]);
    if (Number.isFinite(declared) && declared > limit) {
      return reject(httpError(413, "Payload too large"));
    }
    let size = 0;
    const chunks = [];
    req.on("data", (chunk) => {
      size += chunk.length;
      if (size > limit) {
        reject(httpError(413, "Payload too large"));
        req.destroy();
        return;
      }
      chunks.push(chunk);
    });
    req.on("end", () => resolve(Buffer.concat(chunks)));
    req.on("error", reject);
  });
}

function parseJson(buffer) {
  if (buffer.length === 0) return {};
  try {
    return JSON.parse(buffer.toString("utf8"), reviver);
  } catch {
    throw httpError(400, "Invalid JSON body");
  }
}

function secureHeaders(res) {
  res.setHeader("X-Content-Type-Options", "nosniff");
  res.setHeader("X-Frame-Options", "DENY");
  res.setHeader("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'");
  res.setHeader("Referrer-Policy", "no-referrer");
  res.setHeader("Strict-Transport-Security", "max-age=63072000; includeSubDomains");
  res.removeHeader("X-Powered-By");
}

function safeCompare(a, b) {
  const ha = createHash("sha256").update(String(a)).digest();
  const hb = createHash("sha256").update(String(b)).digest();
  return timingSafeEqual(ha, hb);
}

function sendJson(res, status, payload) {
  if (!res.headersSent) {
    res.writeHead(status, { "Content-Type": "application/json" });
  }
  res.end(JSON.stringify(payload));
}

function sendError(res, err) {
  const status = err.statusCode ?? 500;
  // Never leak internal error details on a 500. Log them, do not ship them.
  const message = status === 500 ? "Internal Server Error" : err.message;
  if (status === 500) console.error(err);
  sendJson(res, status, { error: message });
}

const API_KEY = process.env.API_KEY ?? "dev-only-key";

async function handler(req, res) {
  secureHeaders(res);

  // A single protected route, as an example.
  if (req.method === "POST" && req.url === "/users") {
    const provided = req.headers["x-api-key"] ?? "";
    if (!safeCompare(provided, API_KEY)) {
      throw httpError(401, "Unauthorized");
    }

    const data = parseJson(await readBody(req));

    if (typeof data.email !== "string" || !data.email.includes("@")) {
      throw httpError(400, "email is required");
    }
    if (typeof data.password !== "string" || data.password.length < 12) {
      throw httpError(400, "password must be at least 12 characters");
    }

    // Only the fields we asked for. No mass assignment.
    const user = { email: data.email };
    return sendJson(res, 201, { created: user });
  }

  throw httpError(404, "Not found");
}

const server = http.createServer((req, res) => {
  handler(req, res).catch((err) => sendError(res, err));
});

server.requestTimeout = 30_000;
server.headersTimeout = 10_000;
server.setTimeout(60_000);

server.listen(3000, () => console.log("listening on http://localhost:3000"));

Read that top to bottom and notice how the security isn't a separate "security middleware" bolted on at the end. It's woven into the normal flow.

The body has a limit. The JSON is parsed safely. The headers go out every time. The API key check is timing-safe. The validation runs before any logic. The error handler refuses to leak internals. And the whole thing is still small enough to hold in your head, which matters, because security you can't understand is security you'll eventually disable by accident.

Try breaking it. Send a huge body and watch the 413. Send {"__proto__": {"isAdmin": true}} and confirm ({}).isAdmin is still undefined afterward. Send a wrong API key and notice you can't tell from the response time how close you were. That last one is invisible by design, which is the whole point.

How to Handle CORS Correctly

CORS, which stands for Cross-Origin Resource Sharing, is one of the most misunderstood security features in web development, and the misunderstanding is the dangerous kind.

Here's the part people get wrong: CORS doesn't protect your server. It's not a firewall. It's a browser feature that decides whether JavaScript running on one website is allowed to read the response from your API on another. Your server stays perfectly reachable from curl, from Postman, and from any other server, CORS headers or not.

What that means in practice is that the most common "fix" people apply is also the most common mistake:

res.setHeader("Access-Control-Allow-Origin", "*"); // understand this before you ship it

A wildcard says "any website's JavaScript may read my responses." For a genuinely public, read-only API with no credentials, that can be perfectly fine. For anything that uses cookies or returns data tied to a logged-in user, it's a mistake, and browsers will refuse to combine * with credentials anyway.

The correct approach is to allow only the origins you actually trust:

const ALLOWED_ORIGINS = new Set(["https://app.example.com"]);

function applyCors(req, res) {
  const origin = req.headers.origin;
  if (origin && ALLOWED_ORIGINS.has(origin)) {
    res.setHeader("Access-Control-Allow-Origin", origin);
    res.setHeader("Vary", "Origin"); // so a cache does not mix origins up
    res.setHeader("Access-Control-Allow-Credentials", "true");
  }
}

Keep this mental model: CORS loosens the browser's default protection in a controlled way. Setting it to * doesn't make your API more exposed to attacks from other servers, because servers were never restricted in the first place. It makes your data readable by any web page a victim happens to visit, which is a privacy and data-exposure decision, not a "make the console error go away" decision. Decide it on purpose, origin by origin.

What This Tutorial Doesn't Cover

Honesty time, because a tutorial that pretends to be totally complete is doing you a disservice. The guardrails above are the baseline, not the finish line.

Here's what's deliberately out of scope and where to look next:

• Authentication and authorization: You checked one API key. Real apps need sessions or tokens, and a real story for who is allowed to do what. That is a whole topic on its own.

• Rate limiting: A single client shouldn't be able to hammer your login route ten thousand times a minute. In-memory counters work for one instance. Behind a load balancer you need a shared store like Redis.

• Outbound request safety (SSRF): The moment your server makes requests to URLs a user supplied, you have a new attack surface: someone can point you at internal addresses or the cloud metadata endpoint. That deserves its own article.

• TLS: Everything HSTS-related assumes you actually terminate HTTPS somewhere, whether that's the runtime, a reverse proxy, or your platform.

• Logging and monitoring: You can't respond to what you can't see. Structured logs with request ids are the unglamorous foundation of every incident response that went well.

Each of these is a future tutorial, and each one follows the same philosophy as this one: make the safe choice the default, and make the unsafe choice something you have to go out of your way to do.

Why Defaults Beat Checklists

You might be wondering why I keep saying "by default" instead of just handing you a checklist and wishing you luck. The reason is that I've watched a lot of checklists lose to a deadline.

A checklist is a list of things a human has to remember to do, correctly, every single time, forever. That includes the junior dev who joined last week, and the senior dev who's exhausted and shipping a hotfix at midnight.

Security that depends on perfect human memory is security that quietly degrades the moment the team gets busy, which is precisely the moment an attacker is hoping for.

A default is a different kind of thing. A default is what happens when nobody does anything at all. If the safe behavior is the default, then forgetting produces a safe app. If the unsafe behavior is the default, then forgetting produces a vulnerability, and people forget constantly, because they're human and they have forty other things on their plate.

This is exactly why the helpers in this article are wrappers you call once at the top of a request, instead of steps you sprinkle through your handlers and hope you got them all. It's the same reason frameworks that take security seriously turn protections on and make you opt out, rather than leaving them off and making you opt in.

The wording sounds like a small difference. Measured across a real team over a real year, the difference in outcomes is enormous. Design it so the lazy path and the safe path are the same path, and you'll be surprised how secure "lazy" can be.

An Honest Note on Frameworks

If wiring all of this by hand every time sounds tedious, that's exactly the right instinct, and it's why some frameworks ship these protections on by default so you don't have to remember them.

Full disclosure: I maintain one of them, an open-source project called DaloyJS. I'm not here to sell it, and everything in this article is plain Node that works the same no matter what you build on.

I mention it only because the lesson that produced it is the lesson of this whole piece: the defaults are the product. A framework that makes you opt into safety will, statistically, be run by someone who forgot to.

Whether you use a framework or roll your own, copy the helpers above into your project today. They're dependency-free and they will quietly prevent a category of bad days.

The Takeaway Checklist

If you remember nothing else, remember this list and put it somewhere your team will see it:

• Limit the body: Count bytes as they stream, reject past your cap, never trust Content-Length alone.

• Time out everything: Tighten Node's request and header timeouts, and give every outbound call a deadline with AbortController.

• Parse JSON defensively: Catch parse errors into a clean 400, and strip __proto__, constructor, and prototype with a reviver.

• Set security headers on every response: Wrap them in one function so "every response" actually means every response.

• Compare secrets in constant time: Use crypto.timingSafeEqual on hashed inputs, never ===, and use a real password hash for passwords.

• Validate input as a gate: Define the shape on purpose, reject what doesn't match, and copy only the fields you asked for.

None of this is clever. That's the best thing about it. The boring stuff is what gets you, so make the boring stuff automatic, and go spend your cleverness on the parts of your product that actually need it.

At first, you only need a few prices. You send a request, get JSON back, load it into pandas, and move on. But the moment that API starts powering a backtester, dashboard, screener, valuation tool, or AI assistant, the decision becomes much more serious.

A backtester needs adjusted historical prices, splits, dividends, and stable time series. A dashboard needs fresh quotes, clean fields, and reliable responses. A stock screener needs fundamentals, ratios, and company metadata. An AI agent needs structured data that it can retrieve and use without guessing.

That's why I wouldn't start by comparing endpoint counts or pricing pages. Those matter, but they're not the first question.

The first question is: what are you building?

In this article, we’ll walk through how to choose a stock market API based on the workflow it needs to support. Then we’ll build a practical stock research workflow in Python using Alpha Vantage to see how prices, fundamentals, technical indicators, and AI-ready access can fit together in one project.

Table of Contents

• Why Stock API Choice Depends On The Workflow

  • 1. If You Are Building A Backtester

  • 2. If You Are Building A Dashboard

  • 3. If You Are Building A Stock Screener

  • 4. If You Are Building A Valuation Or Research Tool

  • 5. If You Are Building An AI Assistant Or Agent

• What A Modern Stock Market Data Workflow Actually Requires

• Building A Practical Stock Research Workflow With Alpha Vantage

  • Step 1: Fetch Adjusted Historical Prices

  • Step 2: Add Company Or Fundamental Data

  • Step 3: Add Technical Indicators

  • Step 4: Combine Everything Into A Research-Ready Table

  • Step 5: Connect The Workflow To AI Agents With MCP

• Where Each Provider Fits In The Stock API Workflow

• Provider Breakdown Through A Workflow Lens

  • 1. When The Project Needs Several Data Layers: Alpha Vantage

  • 2. When The Workflow Is Institutional: Bloomberg API

  • 3. When The Product Needs Investor Relations Widgets: QuoteMedia

  • 4. When The Workflow Is Global Historical Research: EODHD

  • 5. When The Workflow Needs US Fundamentals: Intrinio

  • 6. When The Workflow Needs Enterprise Data Delivery: Xignite

• Final Checklist Before Choosing A Stock API

• Final Thoughts

Why Stock API Choice Depends On The Workflow

A stock API should be judged by the workflow it supports, not by how long its feature list looks. The same provider can be a good fit for one project and a weak fit for another.

A clean historical dataset matters more for a backtester than a live quote endpoint. A dashboard has different problems. It needs fresh responses, predictable fields, and rate limits that don't collapse once users start refreshing the page.

Here is how I would think about it.

1. If You Are Building A Backtester

Start with historical data quality.

A backtest needs adjusted prices, splits, dividends, long history, and stable time series. If those pieces are wrong, the backtest can still run, but the results may be misleading.

For this workflow, real-time data is usually secondary. Clean historical data matters more than fast quotes.

2. If You Are Building A Dashboard

Start with freshness and reliability.

A dashboard needs quote data that updates consistently, fields that don't change unexpectedly, and rate limits that can handle repeated requests. A failed request in a notebook is annoying. A failed request in a user-facing dashboard is a product problem.

You also need to check whether the data can be displayed to users. Licensing becomes part of the workflow once the dashboard is public.

3. If You Are Building A Stock Screener

Start with fundamentals and structured fields.

A screener needs more than prices. It may need ratios, company profiles, sector data, market cap, earnings, and symbol coverage across many companies.

The hard part is comparison. If fields are inconsistent across tickers, the screener becomes a cleanup project before it becomes a useful tool.

4. If You Are Building A Valuation Or Research Tool

Start with financial statements.

A valuation workflow usually needs income statements, balance sheets, cash flow statements, earnings history, and historical fundamentals. Price data gives market context, but the business data does the heavier work.

This is where depth matters. The latest numbers are useful, but trends across multiple periods are often more important.

5. If You Are Building An AI Assistant Or Agent

Start with structure.

An AI agent shouldn't guess financial data from memory. It needs predictable API responses, clear schemas, and tool access it can use reliably.

This is where MCP-style workflows matter. If an agent can call a tool, retrieve a quote, pull fundamentals, or fetch a time series cleanly, the API becomes part of the agent’s reasoning loop.

The practical point is simple: choose the API around the system you're building. Once the workflow is clear, the rest of the decision becomes much easier.

What A Modern Stock Market Data Workflow Actually Requires

A modern stock data workflow is rarely just one API call.

You might start with market data, but most useful projects eventually need more layers. A research dashboard may need fundamentals. A screener may need technical indicators. An AI assistant may need structured responses that it can retrieve through a tool.

A simple way to think about the workflow is:

Market Data -> Fundamentals -> Indicators -> Structured Responses -> Programmatic Workflow -> AI/Agent Access

Each layer solves a different problem.

• Market data gives you prices, volume, returns, and historical movement.

• Fundamentals add business context through revenue, margins, cash flow, earnings, and company details.

• Indicators help convert raw prices into features that can support screening, research, or signal testing.

• Structured responses make the data easier to parse, join, and reuse.

• Programmatic workflows turn the raw API response into tables, charts, models, dashboards, or research outputs.

• AI or agent access lets an assistant call tools, retrieve current data, and work with structured financial context instead of relying only on static knowledge.

This is why stock API choice matters beyond the first request. The API is not only there to return data but to support the way the project grows after the prototype.

Building A Practical Stock Research Workflow With Alpha Vantage

Now let’s turn the framework into something practical.

For this section, we’ll use Alpha Vantage as the implementation API because it gives us the main layers we need for this workflow: adjusted historical prices, company data, technical indicators, and MCP-style access for AI agents.

The goal isn't to test every endpoint. The goal is to build a small research workflow that shows what a useful stock API should help us do.

We’ll build this in five steps:

1. Fetch adjusted historical prices.

2. Add company or fundamental data.

3. Add a technical indicator.

4. Combine everything into a research-ready table.

5. Connect the workflow to an AI-agent setup using MCP.

By the end, we should have a simple but practical stock research table that can support a screener, dashboard, research notebook, or AI assistant.

Step 1: Fetch Adjusted Historical Prices

Adjusted prices are the first thing I would check for any research or backtesting workflow. Raw prices can break around stock splits or dividends, while adjusted prices keep the series more useful for return calculations.

Let’s fetch daily adjusted price data for Apple.

import requests
import pandas as pd

api_key = 'YOUR ALPHA VANTAGE API KEY'

symbol = 'AAPL'

url = f'https://www.alphavantage.co/query?function=TIME_SERIES_DAILY_ADJUSTED&symbol={symbol}&outputsize=compact&apikey={api_key}'

response = requests.get(url)
data = response.json()

prices = pd.DataFrame(data['Time Series (Daily)']).T

prices.index = pd.to_datetime(prices.index)
prices = prices.sort_index()

prices = prices.rename(columns={
    '1. open': 'open',
    '2. high': 'high',
    '3. low': 'low',
    '4. close': 'close',
    '5. adjusted close': 'adjusted_close',
    '6. volume': 'volume',
    '7. dividend amount': 'dividend',
    '8. split coefficient': 'split'
})

price_cols = ['open', 'high', 'low', 'close', 'adjusted_close', 'volume', 'dividend', 'split']
prices[price_cols] = prices[price_cols].astype(float)

prices.tail()

The output gives us a clean daily price table as you can see in the image below:

For a chart, you may only need close. For research or backtesting, I would usually work with adjusted_close because it handles corporate actions more safely. Next, we can convert the time series into a few basic price features.

latest_price = prices['adjusted_close'].iloc[-1] 
return_30d = prices['adjusted_close'].pct_change(30).iloc[-1] 
volatility_30d = prices['adjusted_close'].pct_change().tail(30).std() 

price_features = {'symbol': symbol, 'latest_price': latest_price, 'return_30d': return_30d, 'volatility_30d': volatility_30d}
price_features

This returns:

{'symbol': 'AAPL',
 'latest_price': 312.06,
 'return_30d': 0.18583097277442007,
 'volatility_30d': 0.012845143800989936}

This is already more useful than a raw API response. We now have a small set of price features that can feed a dashboard, screener, research table, or AI-assisted stock analysis workflow.

Step 2: Add Company Or Fundamental Data

Price data tells us how the stock moved, but it doesn't tell us much about the company behind the ticker. For a screener, valuation tool, or research workflow, we need some business context too.

Alpha Vantage’s OVERVIEW endpoint gives company-level fields like sector, industry, market cap, PE ratio, EPS, profit margin, and other summary metrics. Let’s pull those fields and keep only the ones we need for this workflow.

overview_url = f'https://www.alphavantage.co/query?function=OVERVIEW&symbol={symbol}&apikey={api_key}'

response = requests.get(overview_url)
overview = response.json()

fundamental_features = {
    'symbol': symbol,
    'name': overview.get('Name'),
    'sector': overview.get('Sector'),
    'industry': overview.get('Industry'),
    'market_cap': overview.get('MarketCapitalization'),
    'pe_ratio': overview.get('PERatio'),
    'eps': overview.get('EPS'),
    'profit_margin': overview.get('ProfitMargin'),
    'beta': overview.get('Beta')
}

fundamental_features

This returns:

{'symbol': 'AAPL',
 'name': 'Apple Inc',
 'sector': 'TECHNOLOGY',
 'industry': 'CONSUMER ELECTRONICS',
 'market_cap': 4583336182000.0,
 'pe_ratio': 37.73,
 'eps': 8.27,
 'profit_margin': 0.272,
 'beta': 1.065}

Now we have two layers: price behavior from the time series data and business context from the company overview. The next step is to add a technical indicator so the table includes a market-derived signal as well.

Step 3: Add Technical Indicators

Fundamentals give us business context, but many research workflows also need market-derived signals. A simple example is the relative strength index, or RSI, which is often used to measure recent momentum.

Alpha Vantage has a RSI endpoint, so we can pull the indicator directly instead of calculating it from scratch.

rsi_url = f'https://www.alphavantage.co/query?function=RSI&symbol={symbol}&interval=daily&time_period=14&series_type=close&apikey={api_key}'

response = requests.get(rsi_url)
rsi_data = response.json()

rsi = pd.DataFrame(rsi_data['Technical Analysis: RSI']).T

rsi.index = pd.to_datetime(rsi.index)
rsi = rsi.sort_index()
rsi['RSI'] = rsi['RSI'].astype(float)

latest_rsi = rsi['RSI'].iloc[-1]

indicator_features = {
    'symbol': symbol,
    'rsi_14': latest_rsi
}

indicator_features

This returns:

{'symbol': 'AAPL', 'rsi_14': 79.0043}

Now the workflow has three layers:

• price behavior from adjusted historical data

• business context from company fundamentals

• momentum context from a technical indicator

None of these is enough on its own. Together, they start to look like a usable research workflow instead of a raw API test.

Step 4: Combine Everything Into A Research-Ready Table

Now we can combine the price, fundamentals, and indicator layers into one table.

This is the part that matters for most real projects. A dashboard, screener, notebook, or AI assistant usually needs a clean object it can reuse, not three separate raw API responses.

research_row = {
    **price_features,
    **fundamental_features,
    **indicator_features
}

research_table = pd.DataFrame([research_row])

research_table

This gives us a single-row research table:

This table is simple, but it already supports several use cases.

A screener can filter on pe_ratio, profit_margin, or rsi_14. A dashboard can show price, returns, sector, and market cap. A research notebook can add more tickers and compare them. An AI assistant can receive this as a compact context object instead of parsing multiple API responses on its own.

That's the real benefit of building the workflow this way. The API calls are only the beginning. The useful output is the structured table you create from them.

Step 5: Connect The Workflow To AI Agents With MCP

The table we created is useful because it has a predictable structure, which is exactly what AI workflows need.

If an agent needs stock context, it shouldn't guess from memory or parse several raw API responses every time. It should call a tool, retrieve the data, and receive something clean enough to use.

A simplified MCP workflow looks like this:

User question -> AI agent -> MCP tool call -> Stock API data -> Structured response -> Final answer

For example, a user might ask:

Is Apple looking expensive compared with its recent momentum?

An agent could retrieve price data, fundamentals, and an indicator such as RSI before answering. The important part is not that the model already “knows” the answer. It's that the model can call the right tool and work with current data.

That is where our research table helps:

research_table.to_dict(orient='records')[0]

This returns a compact dictionary:

{'symbol': 'AAPL',
 'latest_price': 312.06,
 'return_30d': 0.18583097277442007,
 'volatility_30d': 0.012845143800989936,
 'name': 'Apple Inc',
 'sector': 'TECHNOLOGY',
 'industry': 'CONSUMER ELECTRONICS',
 'market_cap': 4583336182000.0,
 'pe_ratio': 37.73,
 'eps': 8.27,
 'profit_margin': 0.272,
 'beta': 1.065,
 'rsi_14': 79.0043}

This doesn't replace proper analysis, and it shouldn't be treated as investment advice. But it gives an AI assistant a cleaner starting point than raw JSON, stale model knowledge, or a vague prompt with no data attached.

AI readiness isn't just about saying an API supports agents. The API has to return data that can be retrieved, structured, checked, and passed into a workflow without fragile glue code at every step.

Where Each Provider Fits In The Stock API Workflow

The workflow we built above is one version of a modern stock data project: prices, fundamentals, indicators, programmatic analysis, and AI-agent access working together.

Other projects may need a narrower or more specialized provider. Here's a practical way to compare the fit:

No provider fits every workflow equally well. The point of this table is to show where the fit is strongest.

Alpha Vantage works well when a project needs several layers together, especially market data, fundamentals, indicators, developer usability, and AI-agent access. EODHD is stronger when the workflow is centered on global historical research. Intrinio fits better when standardized US fundamentals are the main requirement. Bloomberg API and Xignite are more natural for institutional or enterprise environments, while QuoteMedia is more specialized around investor relations and embedded market data widgets.

This is the right way to think about stock APIs: not as one universal winner, but as different tools for different workflow shapes.

Provider Breakdown Through A Workflow Lens

The table gives a quick comparison. This section explains what that means in practice.

Instead of asking which provider is “best” in general, it is better to ask: what kind of workflow is this provider naturally built for?

1. When The Project Needs Several Data Layers: Alpha Vantage

Alpha Vantage fits well when the project needs more than one type of market data in the same workflow.

In the workflow we built earlier, we used:

• adjusted historical prices

• company data

• technical indicators

• structured output for programmatic analysis

• a format that can also support AI-agent workflows

That makes Alpha Vantage a flexible fit for stock research notebooks, screeners, dashboards, backtesting workflows, and AI assistants that need market data through tools or MCP-style access.

The main caveat is specialization. If your project needs direct exchange infrastructure, co-location, or a highly specialized institutional setup, you may need a more specialized provider. But for most research, fintech apps, and AI workflows, Alpha Vantage gives enough breadth without forcing you to combine several APIs too early.

2. When The Workflow Is Institutional: Bloomberg API

Bloomberg API makes sense when the organization already uses Bloomberg internally.

It's best suited for firms that want to connect Bloomberg data with internal tools, reports, models, and risk systems.

This isn't usually the right fit for solo developers or small teams. The cost, licensing, and ecosystem dependency make it more suitable for institutions.

3. When The Product Needs Investor Relations Widgets: QuoteMedia

QuoteMedia fits products where the main need is public-facing market data display.

That can include:

• investor relations pages

• quote widgets

• embedded charts

• company stock pages

• market data modules for public websites

This is different from building a programmatic research workflow. QuoteMedia makes more sense when presentation and embedded financial data are the core product requirement.

4. When The Workflow Is Global Historical Research: EODHD

EODHD fits well when the project needs broad historical data across global markets.

It's useful for long-horizon backtesting, global screeners, and research workflows that depend on end-of-day data from many exchanges.

The tradeoff is cleanup. Global data often brings differences in symbols, exchange calendars, currencies, and local market conventions. That's manageable, but it should be expected.

5. When The Workflow Needs US Fundamentals: Intrinio

Intrinio fits well when standardized US fundamentals are the center of the product.

It's useful for:

• valuation tools

• earnings dashboards

• fundamentals-based screeners

• professional US equity research workflows

The main thing to check is dataset fit. Before building around Intrinio, I would look closely at the specific datasets, access terms, and coverage levels the product needs.

6. When The Workflow Needs Enterprise Data Delivery: Xignite

Xignite fits larger financial applications that need formal vendor support.

This can include banks, brokerages, wealth platforms, and enterprise fintech products where support, contracts, reliability, and data relationships matter as much as the endpoint itself.

For smaller developer projects, it may feel heavier than necessary. For enterprise products, that structure can be exactly the point.

Final Checklist Before Choosing A Stock API

Before choosing a provider, I would run through this checklist.

Final Thoughts

A good stock API should reduce project risk, not just return data.

If you're building a small chart, almost any clean price endpoint can work. But once the same API starts supporting a backtester, screener, dashboard, valuation tool, or AI assistant, the decision becomes more important. The provider affects your data quality, parsing logic, refresh jobs, licensing choices, and future product direction.

This is why workflow fit matters more than endpoint count. For projects that need several layers together, such as real-time and historical market data, fundamentals, indicators, developer-friendly access, spreadsheet support, and MCP-style AI workflows, Alpha Vantage fits well. For narrower workflow needs, another provider may make more sense.

Choose the API as part of your project’s data infrastructure, not just as a list of endpoints.

AI agents don't have that context and understanding.

AI agent understand APIs through schemas, examples, randomized data and live responses. When a behavior or method is ambiguous and inconsistent, the model doesn't pause to “think” – it fills in the blanks (randomizing).

In production, those guesses could become blocks, retry storms, duplicated side effects, or broken workflows.

This is why APIs that are perfectly fine for humans frequently fail under AI agent use. The problem is rarely “the agent isn’t smart enough.” More often, the API was never designed for an agent/machine consumer that must plan, call tools, and recover from failure without a human in the loop.

In this guide, you’ll learn how to design APIs that agents can use reliably. We’ll anchor the discussion in three practical ideas:

1. Deterministic behavior: same inputs and state should yield predictable outcomes and shapes.

2. Strong schemas: contracts that are complete, descriptive, and testable.

3. Guardrails at the API boundary: authorization, validation, and safe defaults that prevent unsafe autonomy.

The aim of this article is not to build “AI-powered” APIs, but rather to build APIs that are clear, strict, and dependable, even when the caller is not an agent but a fellow developers leveraging various tools.

Table Of Contents

• Prerequisites

• Why “Good Enough for Devs” Is Not Good Enough for Agents

• Principle 1: Deterministic Behavior

• Principle 2: Strong Schemas

• Principle 3: Guardrails at the API Boundary

• Patterns That Bridge APIs and Agent Runtimes

• A Practical Before and After Example

• Checklist: Is Your API Agent-Ready?

• Conclusion

Prerequisites

Before reading this guide, it helps to have:

• A basic understanding of HTTP APIs and REST concepts

• Familiarity with JSON and API request/response patterns

• An understanding of common API concepts like authentication, pagination, and retries

Why “Good Enough for Devs” Is Not Good Enough for Agents

Human developers bring implied and contextual knowledge: they read through Slack threads, read blog posts, and recognize that “this 404 usually means you forgot the workspace ID.”

Agents mostly get whatever is in the spec, the examples, and the last response body.

That gap shows up in predictable ways:

• Ambiguous semantics: wrong endpoint or wrong parameter combination.

• Undocumented branches: the model invents fields or misreads optional behavior.

• Inconsistent error bodies: retries that shouldn't happen, or no retry when one is safe.

• Non-idempotent “do things” endpoints: duplicate charges, duplicate tickets, duplicate emails.

Industry commentary and practitioner guides converge on the same point: agents are becoming a major class of API consumer, and machine legibility matters as much as developer experience.

See for example discussions of OpenAPI as the source of truth for agents, emerging tool protocols, and traffic patterns that differ from human clients in the resources listed at the end of this article.

Principle 1: Deterministic Behavior

Determinism for agents doesn't mean “always return the same JSON forever.” It means: given the same request and the same server-side state, your API behaves in a way the agent can model and when state changes, you make that explicit.

Prefer Explicit State Over Hidden Magic

Agents struggle with “sometimes the server does X depending on internal flags.” Where humans infer intent from product copy, agents infer from patterns. If those patterns drift, autonomy breaks.

Practical habits:

• Model lifecycle explicitly (draft → submitted → approved) instead of overloading a single status field with undocumented combinations.

• Return what changed after mutations (updated resource, relevant IDs, next allowed actions).

• Avoid silent coercion (auto-correcting bad enums, silently dropping unknown fields) unless you document and signal it.

Make Writes Safe: Idempotency and Intent Keys

For any endpoint that bills, sends messages, provisions infrastructure, or otherwise does something irreversible, assume double-submission will happen.

• Support idempotency keys (header or body) for create-like operations.

• Use clear HTTP semantics: POST creates, PUT replaces where appropriate, PATCH for partial updates and document what repeats mean.

• Where duplicates are possible, offer a lookup-by-client-reference path so agents can reconcile.

Pagination and Sorting: One Pattern, Everywhere

Agents loop. If every resource paginates differently, the model will mix strategies.

To combat this, pick one pagination style (cursor vs offset) per API surface and stick to it.

Also, always return stable sort order or require sort explicitly. You should also include next links or cursors in a consistent envelope.

Timeouts, Partial Success, and Async Work

Agents hate “maybe it worked.” Long-running work should be explicitly async:

• 202 Accepted + job ID + polling or webhooks.

• Clear terminal states: succeeded, failed, canceled, with structured error details on failure.

Principle 2: Strong Schemas

If determinism is about behavior, schemas are about communication. For agents, your OpenAPI (or equivalent) isn't paperwork, it's part of the runtime interface.

Treat OpenAPI as a Contract, Not a Souvenir

A specification that lags production is worse than no spec: it trains the agent to be confidently wrong. Teams increasingly treat OpenAPI as the authoritative contract and validate requests/responses against it in CI and at the edge.

Here's the minimum bar for agent-friendly OpenAPI:

• Every operation has a summary and a description that explain when to use it, not only what it returns.

• Every request body property has description and realistic example values.

• All responses are documented including 4xx/5xx with stable JSON shapes.

Describe Intent in Natural Language, Precisely

Agents aren't offended by verbosity. They're confused by vague verbs.

Instead of:

“Gets orders.”

Prefer:

“Lists orders for the authenticated merchant. Supports filtering by status and a time window on created_at. Returns at most limit items; use cursor for the next page.”

This aligns with what multiple guides call context-aware or self-describing APIs: the schema carries semantic intent, not just types.

Examples Are Part of the Contract

You should provide a happy path example per endpoint, at least one validation error example (400) with your standard error object, and examples for optional fields when they change behavior.

Examples reduce “shape hallucination” where the model guesses field names or nesting.

JSON Schema Strictness Helps Tool-Calling Stacks

If your agent uses function calling / structured outputs, tighten schemas:

• Prefer enum for small closed sets.

• Mark fields required honestly.

• Use format (uuid, date-time) where real.

• Avoid additionalProperties: true on security-sensitive payloads if you need strict validation.

Name Things Consistently

userId in one endpoint and user_id in another is a human annoyance and an agent trap. Pick a convention and enforce it.

Principle 3: Guardrails at the API Boundary

Autonomy amplifies mistakes. Guardrails turn “oops” into blocked requests instead of incidents.

Authorization Should Be Narrow and Explicit

Agents should receive credentials scoped to least privilege. For example, use short-lived tokens, with refresh documented clearly. Use scopes that map to real actions (orders:read vs orders:write). And avoid flows that assume a human can solve (CAPTCHAs) or click (email links mid-run) or isolate those as human-in-the-loop tools.

Validate Hard, Fail Loud and Structured

Reject bad input at the edge with stable error_code values (machine-actionable), human-readable message (for logs and UI), optional field or JSON Pointer to the problem, and optional doc_url linking to documentation.

This matches guidance from several practitioner articles: opaque 500s and generic errors are where autonomous clients spiral.

RFC 7807 Problem Details (application/problem+json) is a good, widely understood pattern for HTTP APIs, a structured envelope agents can parse consistently.

Separate “Read the World” from “Change the World”

For high-impact actions (refunds, deletes, transfers), consider using a two-step pattern: first create an intent, then confirm execution.

Or you can dry-run query parameters / dedicated endpoints that validate without committing.

Also keep in mind that rate limits and quotas tuned for bursty agent behavior and autonomous loops can dwarf human traffic.

Observability is a Product Feature

Log correlation IDs, surface them in responses where safe, and monitor for retry amplification. An agent that misreads a 409 as “retry forever” becomes a denial-of-wallet attack on your own systems.

Patterns That Bridge APIs and Agent Runtimes

Workflow Documentation: Sequences, Not Just Endpoints

Agents excel when they can follow a recipe. Document common sequences (“create customer → add payment method → charge”) and consider standards meant for multi-step API flows (such as Arazzo) when your product’s complexity justifies it.

Hypermedia and “Next Steps”

Including links to plausible next actions (for example, pagination next, or related resources) reduces improvisation. This is the same spirit as HATEOAS: the response whispers what you can do next, instead of forcing the model to guess URLs.

Tool-Oriented Surfaces (For Example, MCP)

Protocols like the Model Context Protocol (MCP) are gaining traction as a way to expose curated capabilities (“tools”) with schemas agents can bind to directly.

A common pragmatic pattern is not to dump every micro-endpoint as a tool, but to expose coarse-grained tools aligned to user outcomes while keeping your underlying REST API strict and clean.

MCP isn't a substitute for good API design. It's a delivery and discovery layer. Slapping a thin wrapper on a messy API still leaves you with a messy system – it just fails faster in public.

Metadata for Discovery (llms.txt and Friends)

Some teams publish /llms.txt or similar lightweight discovery files for documentation sites. Treat these as optional signposts, not replacements for OpenAPI.

Ecosystem adoption is still evolving, but the underlying idea is sound: make the canonical machine-readable description easy to find.

A Practical Before/After

Weak Pattern (Agent-hostile)

POST /do-stuff

Response 200 OK:

{ "ok": true }

Problems: no idempotency, no structured error, no entity ID, no way to poll, the agent must guess whether “ok” means “created” or “ignored duplicate.”

Stronger Pattern (Agent-friendly)

POST /v1/invoices
Idempotency-Key: 7b3c-...

Response 201 Created:

{
  "invoice": {
    "id": "inv_9Qz",
    "status": "draft",
    "total": { "amount": "120.00", "currency": "USD" }
  },
  "links": {
    "finalize": "/v1/invoices/inv_9Qz/finalize"
  }
}

Conflict response 409 Conflict with Problem Details:

{
  "type": "https://api.example.com/problems/duplicate-idempotency-key",
  "title": "Duplicate idempotency key",
  "status": 409,
  "detail": "A different request body was sent with the same Idempotency-Key.",
  "error_code": "IDEMPOTENCY_KEY_REUSE_BODY_MISMATCH"
}

This tells the agent what happened and whether retrying is appropriate.

Checklist: Is Your API Agent-Ready?

• Contract: Published OpenAPI 3.x, validated against real traffic, with rich descriptions and examples.

• Determinism: Documented state machines, consistent pagination, explicit async for long jobs.

• Safe writes: Idempotency for side effects, reconciliation endpoints where needed.

• Errors: Stable codes, structured bodies, documented remediation paths.

• Security: Least-privilege tokens, no “mystery” side doors agents can accidentally hit.

• Operations: Rate limits, bulk endpoints where appropriate, correlation IDs, dashboards for anomalous agent traffic.

Conclusion

Designing for AI agents is, in most respects, disciplined API design — pushed to the level where machines can rely on your contract without tribal knowledge.

If you remember only three things:

1. Be predictable: in shapes, states, and side effects.

2. Be explicit: in schemas, examples, and errors.

3. Be protective: validate early, scope narrowly, and make dangerous actions hard to trigger by accident.

But sometimes, deploying your perfectly working email function to the cloud leads to unexpected and frustrating errors. You build your backend, test it locally, and it works flawlessly. Then you deploy to the cloud, and suddenly your app stops sending emails completely.

In this article, you’ll learn exactly why your email setup fails on cloud platforms like Render or Heroku, the underlying networking rules causing the issue, and how to elegantly bypass these restrictions using Brevo's HTTP API.

Let’s dive right in.

Outline

• Prerequisites

• Tools We'll Be Using

• The Problem: Nodemailer and SMTP Blocking

• The "Modern" Trap: Domain Verification

• The Ultimate Solution: Brevo and HTTP APIs

• Backend Setup

• Brevo Configuration Setup

• Creating the Email Function

• Integrating the Function into an Express Route

• Conclusion

Prerequisites

To get the absolute most out of this tutorial, it’s important to have some basic knowledge of the following:

• JavaScript and Node.js: Having a good fundamental understanding of how JS works on the server side will make it easier to follow along with the project.

• REST APIs: You should have a basic understanding of how HTTP requests (like POST and GET) work using native fetch() in Node.js.

• Express.js: A little background on creating basic server routes will be helpful, as we'll build a real-world controller.

• A basic understanding of what Nodemailer is and how cloud hosting platforms (like Render or Heroku) operate.

Tools We’ll Be Using

In one of my recent projects, I created a complex authentication flow where users needed an OTP (One Time Password) sent to their email to complete registration. I set up Nodemailer, linked my Gmail, and tested it on localhost. Within seconds, the emails arrived perfectly.

But when I deployed my backend to Render, the entire signup flow broke. After doing some deep digging, I found out why it broke and how to fix it permanently. And now that I know how it works, I wanted to share it with you all.

The Problem: Nodemailer and SMTP Blocking

So what exactly is the issue?

Nodemailer is a very popular Node.js module that lets you send emails efficiently. Usually, developers use it to connect to services like Gmail or Mailtrap using SMTP (Simple Mail Transfer Protocol). When your code tries to send an email, Nodemailer opens a connection to the mail server using Port 587 (for STARTTLS) or Port 465 (for SSL).

But cloud providers like Render, Heroku, DigitalOcean, and AWS face a massive daily battle against automated spammers. Malicious users often spin up thousands of free-tier servers specifically to blast out millions of spam emails. If a cloud provider allows this, their entire network IP address block will get blacklisted by Gmail, Outlook, and Yahoo.

To protect their network reputation, cloud providers enacted a heavy-handed, silent rule: All outbound traffic on Ports 25, 465, and 587 is strictly blocked on free and entry-level tiers.

This means your server is literally trapped behind a firewall. If you check your server logs, you won't see an "Invalid Password" error. Instead, you'll see a timeout error that looks like this:

Error: connect ETIMEDOUT 142.250.102.108:587
    at TCPConnectWrap.afterConnect [as oncomplete] (node:net:1494:16)

Your code isn't broken – it's just being blocked at the network level!

The "Modern" Trap: Domain Verification

When developers hit this wall, they often try modern API-based email services like Resend or SendGrid. These are amazing tools, but they introduce a new problem for beginners: Strict Domain Authentication.

To use Resend in production, you must own a custom domain (like yourname.com) and configure DNS records (SPF, DKIM, and DMARC). If you don't own a domain, Resend's sandbox mode strictly restricts you to sending emails only to yourself. You can't send emails to your live users.

For a developer just trying to launch a portfolio project, buying a domain just to send test emails is a huge bottleneck.

The Ultimate Solution: Brevo and HTTP APIs

We need a solution that meets two criteria:

1. It must bypass the Port 587 firewall.

2. It must let us send emails to anyone without forcing us to buy a custom domain.

This is where the architectural difference between SMTP and REST APIs comes to the rescue. While SMTP is a dedicated protocol for routing mail, a REST API operates over standard web traffic using HTTPS (Port 443). Cloud providers can't block Port 443, because doing so would prevent your server from fetching data from databases or functioning as a web server entirely.

Enter Brevo (formerly Sendinblue). Brevo is a powerful email platform that allows you to send emails via a standard REST API. Best of all, their free tier (300 emails/day) allows Single Sender Verification. You just verify your standard Gmail address, and they let you send to anyone!

By sending a JSON payload via HTTPS to Brevo's API, your server routes the traffic out of the unrestricted Port 443, bypassing the Render firewall completely.

Now that you know the theory behind the tools we’ll be using, let’s move on to writing the code.

Backend Setup

First things first, you have to set up your environment. If you don't already have Node.js installed on your computer, head to their website to download and install it.

Start by running npm init -y in your terminal. This creates the package.json file which manages your project and stores all the dependencies.

Next, run npm install express dotenv.

You might be used to installing nodemailer for your email tasks. But because we are going to use the native Node.js fetch() API to talk to the Brevo API, you actually don't need to install any heavy email libraries at all! We want to keep our backend as lightweight as possible.

Brevo Configuration Setup

Before you write the email function, you first need to configure Brevo to get access to your API key.

1. Go to Brevo.com and create a free account.

2. During setup, they will ask you to add a Sender Email. Make sure you input your standard Gmail address. They will send you an email with a link to verify you own this address.

3. Once verified and inside the dashboard, click on your profile name in the top right corner, and select SMTP & API from the dropdown menu.

4. Go to the API Keys tab and click Generate a new API key. Give it a name like "MyWebApp".

Copy this generated key and store it safely in a .env file at the root of your project:

# .env file
EMAIL_USER = yourverifiedemail@gmail.com
BREVO_API_KEY = xkeysib-your-generated-api-key-goes-here

Creating the Email Function

Now that you’ve gotten your API key and set up your environment variables, all that remains is to start putting your backend code together.

Create a file named utils/email.js.

First, start by ensuring you can load your .env file so you can easily access the credentials you generated:

require("dotenv").config();

// We'll define the function to accept dynamic options
const sendEmail = async (options) => {
  const brevoApiKey = process.env.BREVO_API_KEY;
  const senderEmail = process.env.EMAIL_USER;

  // Validate that the keys actually exist
  if (!brevoApiKey || !senderEmail) {
    throw new Error("Missing Brevo credentials in environment variables.");
  }

Next on the line, you’ll need to structure your payload. This is the JSON object that tells Brevo exactly who is sending the email, who is receiving it, and what the content is. Here’s how you can do that:

  const payload = {
    sender: {
      name: "My Awesome Web App",
      email: senderEmail, // Must match your verified Brevo email
    },
    to: [
      {
        email: options.email, // The dynamic email address of the user receiving the email
      },
    ],
    subject: options.subject,
    htmlContent: options.html,
  };

In the code above, the payload object securely packages up your information. We pass in options.email, options.subject, and options.html so that we can reuse this single function for welcome emails, password resets, and notifications.

Now, create the actual network request that sends your data to the Brevo backend. We'll use the POST method. When the data is sent, it must be stringified into a JSON format.

  try {
    const response = await fetch("https://api.brevo.com/v3/smtp/email", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "api-key": brevoApiKey,
      },
      body: JSON.stringify(payload),
    });

    const result = await response.json();

    if (!response.ok) {
      throw new Error(`Brevo API Error: ${JSON.stringify(result)}`);
    }

    console.log(`Email successfully sent to ${options.email} via Brevo HTTP API!`);
  } catch (error) {
    console.error("Error details:", error.message);
  }
};

module.exports = sendEmail;

In the code above, after the payload is submitted, if the message is sent successfully, a success log will be displayed in your terminal. But if the message wasn’t successful – maybe due to a typo in your API key – an error message will be thrown to help you debug exactly what went wrong.

Integrating the Function into an Express Route

At this point, you've successfully built a robust email function. Let's see how you would actually use this in a real Express application.

Create an index.js file and set up a simple Express server route:

const express = require("express");
const sendEmail = require("./utils/email");
const app = express();

app.use(express.json()); // Middleware to parse JSON request bodies

app.post("/api/signup", async (req, res) => {
  const { username, email } = req.body;

  // 1. Save user to database (skipped for brevity)
  
  // 2. Generate a random OTP
  const otp = Math.floor(100000 + Math.random() * 900000);

  // 3. Send the email using our new Brevo function
  try {
    await sendEmail({
      email: email,
      subject: "Welcome! Here is your Verification Code",
      html: `
        <div style="font-family: sans-serif; text-align: center;">
          <h2>Welcome to My Awesome Web App, ${username}!</h2>
          <p>Please use the verification code below to complete your registration:</p>
          <h1 style="color: #2563eb; letter-spacing: 5px;">${otp}</h1>
          <p>This code will expire in 10 minutes.</p>
        </div>
      `,
    });

    res.status(201).json({ message: "User created and email sent!" });
  } catch (error) {
    res.status(500).json({ error: "Failed to send email." });
  }
});

app.listen(8000, () => {
  console.log("Server running on port 8000");
});

And that is it! You can now hit this /api/signup endpoint from your React or Vue frontend, and it will instantly fire off a beautifully formatted email via Brevo's REST API.

Conclusion

As developers, encountering a bug that works locally but fails in production is a rite of passage. But the "Email Delivery Failed" timeout error is special. It teaches you that software engineering isn't just about writing clean syntax – it's about understanding the underlying infrastructure, network layers, and the security context of the environment your code runs in.

By swapping a protocol (SMTP) for an architectural pattern (REST API over HTTPS), you didn't just fix a bug. You successfully engineered a secure, free, and robust bypass around a cloud-level firewall without relying on heavy third-party NPM modules like Nodemailer.

If you've made it this far, I hope I've successfully shown you the importance of understanding network layers and how you can use HTTP APIs to send email messages directly from your web applications safely.

Thank you for reading!

The webhook handler is where most payment integrations silently break. Your server crashes halfway through granting access. Your email service is down when you try to send the confirmation. Your database times out during a write.

Stripe retries the entire webhook, but your handler already sent the confirmation email before it crashed. Now the customer gets two emails and no access.

This article shows you how to fix this. You'll learn how to build webhook handlers that survive failures by splitting your post-payment logic into durable, independently retried steps. The pattern works for any multi-step webhook processing, not just Stripe.

Here's what you'll learn:

• Why Stripe webhooks fail silently in production

• How a naïve inline handler breaks under real-world conditions

• The pattern: webhook receives, validates, and enqueues (nothing more)

• How to build a durable purchase flow with individually checkpointed steps

• How to handle refunds and abandoned checkouts with the same pattern

• How to test webhook handlers locally

Prerequisites

To follow along, you should be familiar with:

• Node.js and TypeScript

• Basic Stripe integration (checkout sessions, webhooks)

• SQL databases (the examples use PostgreSQL with Drizzle ORM)

• npm or any Node.js package manager

You don't need prior experience with Inngest or durable execution. This article explains both from scratch.

What You Need to Install

If you want to run the code examples, install these packages:

npm install inngest stripe drizzle-orm @react-email/components resend

You'll also need the Stripe CLI for local webhook testing. Install it via Homebrew on macOS (brew install stripe/stripe-cli/stripe) or follow the instructions in Stripe's documentation for other platforms.

Table of Contents

• Why Stripe Webhooks Fail Silently

• The Naïve Approach (and Why It Breaks)

• The Pattern: Webhook to Event to Durable Function

• How to Set Up the Webhook Endpoint

• How to Build a Durable Purchase Flow

• How to Handle Refunds with the Same Pattern

• How to Recover Abandoned Checkouts

• How to Test Webhook Handlers Locally

• Conclusion

Why Stripe Webhooks Fail Silently

The happy path is easy. A customer pays, Stripe sends a checkout.session.completed event to your server, and your handler processes it. In development, this works every time.

Production is different: Your webhook handler typically needs to do several things after a successful payment. It looks up the user in the database, records the purchase, sends a confirmation email, notifies the admin, grants access to the product (maybe via a GitHub invitation or an API key), and schedules follow-up emails. That's five or six operations involving three or four external services.

Here are the failure modes that will eventually hit your webhook handler:

1. Your server crashes mid-processing

The database write succeeded, but the email never sent. Stripe retries the webhook, and your handler runs again.

Now you have a duplicate database entry or a unique constraint error that kills the retry.

2. An external service is temporarily down

Your email provider returns a 500. Your GitHub API call gets rate-limited. Your analytics service times out.

The webhook handler throws, and Stripe retries the entire thing. But the steps that already succeeded (the database write, the first email) run again.

3. The handler times out

Stripe expects a 2xx response within about 20 seconds. If your handler does too much work, Stripe marks it as failed and retries. Your handler may have partially completed before the timeout.

4. Partial completion with no rollback

This is the worst failure mode. Steps 1 through 3 succeed. Step 4 fails. Stripe retries, and steps 1 through 3 run again.

The customer gets two confirmation emails. The database gets a duplicate record. But step 4 still fails because the underlying issue (a rate limit, a service outage) hasn't been resolved.

5. Race conditions on retry

Stripe can deliver the same event more than once even without a failure on your end. Network glitches, load balancer timeouts, and Stripe's own retry logic mean your handler must be prepared for duplicate deliveries. If your handler isn't idempotent at every step, duplicates compound the partial-completion problem.

Stripe's retry behavior is well-designed. It uses exponential backoff and retries up to dozens of times over several days. But Stripe retries the entire webhook delivery.

It has no way to know that your handler completed steps 1 through 3 and only needs to retry step 4. That distinction is your responsibility.

The core problem is that your webhook handler does too many things in a single request. Every external call is a potential failure point, and you have no checkpointing between them. When one fails, you lose track of which ones already succeeded.

The Naïve Approach (and Why It Breaks)

Here's what a typical webhook handler looks like. I've seen hundreds of variations of this pattern across codebases, tutorials, and Stack Overflow answers:

app.post("/api/payments/webhook", async (req, res) => {
  const event = stripe.webhooks.constructEvent(
    req.body,
    req.headers["stripe-signature"],
    process.env.STRIPE_WEBHOOK_SECRET
  );

  if (event.type === "checkout.session.completed") {
    const session = event.data.object;

    // Step 1: Look up the user
    const user = await db.users.findOne({ id: session.metadata.userId });

    // Step 2: Record the purchase
    await db.purchases.insert({
      userId: user.id,
      stripeSessionId: session.id,
      amount: session.amount_total,
      status: "completed",
    });

    // Step 3: Send confirmation email
    await sendEmail({
      to: user.email,
      subject: "Purchase confirmed!",
      template: "purchase-confirmation",
    });

    // Step 4: Grant product access (GitHub repo invitation)
    await addCollaborator(user.githubUsername);

    // Step 5: Send access email
    await sendEmail({
      to: user.email,
      subject: "Your repository access is ready!",
      template: "repo-access",
    });

    // Step 6: Track analytics
    await analytics.track(user.id, "purchase_completed", {
      amount: session.amount_total,
    });
  }

  res.json({ received: true });
});

This looks clean. It reads top-to-bottom. Every tutorial teaches it this way.

Now walk through what happens when step 4 fails. Maybe GitHub's API is rate-limited and the addCollaborator call throws an error. Your handler returns a 500 to Stripe.

Here is the state after the failure:

• The user exists in the database (step 1 was just a lookup, no problem).

• A purchase record was created (step 2 succeeded).

• The confirmation email was sent (step 3 succeeded).

• GitHub access was not granted (step 4 failed).

• The access email was not sent (step 5 never ran).

• Analytics were not tracked (step 6 never ran).

Stripe retries the webhook. Your handler runs again from the top:

• Step 1: Looks up the user again. Fine.

• Step 2: Tries to insert another purchase record. If you have a unique constraint on stripeSessionId, this throws. If you don't, you now have a duplicate.

• Step 3: Sends the confirmation email again. The customer gets a second "Purchase confirmed!" email.

• Step 4: Tries GitHub access again. Maybe it works this time, maybe not.

• Steps 5-6: May or may not run depending on step 4.

You can patch this with idempotency checks: "if purchase already exists, skip step 2." But now your handler is full of conditional logic for every step. And you still have the duplicate email problem, because there's no way to check "did I already send this email?" without building your own tracking system.

This approach doesn't scale. Every new step adds another failure mode, another idempotency check, and another edge case.

The Pattern: Webhook to Event to Durable Function

The fix is a separation of concerns. Your webhook handler should do exactly one thing: validate the incoming event and enqueue it for processing. Nothing else.

All the actual work (database writes, emails, API calls, analytics) moves into a durable background function where each step is individually checkpointed, retried, and tracked.

Here's the flow:

Stripe webhook
    |
    v
Webhook endpoint (validate signature, extract event, enqueue)
    |
    v
Background job system (receives event)
    |
    v
Durable function
    |-- Step 1: Look up user and purchase (checkpointed)
    |-- Step 2: Track analytics (checkpointed)
    |-- Step 3: Send confirmation email (checkpointed)
    |-- Step 4: Send admin notification (checkpointed)
    |-- Step 5: Grant GitHub access (checkpointed)
    |-- Step 6: Track GitHub access (checkpointed)
    |-- Step 7: Update purchase record (checkpointed)
    |-- Step 8: Send repo access email (checkpointed)
    |-- Step 9: Schedule follow-up sequence (checkpointed)

Each step wrapped in step.run() is a durable checkpoint. If step 5 fails:

• Steps 1 through 4 do not re-run. Their results are cached.

• Step 5 retries independently, with its own retry counter.

• Once step 5 succeeds, steps 6 through 9 continue.

This is what "durable execution" means. The function's progress survives failures. You get step-level retries instead of function-level retries. No duplicate emails. No duplicate database writes. No partial completion.

I use Inngest for this. It's an event-driven durable execution platform that provides step-level checkpointing out of the box. You define functions with step.run() blocks, and Inngest handles retry logic, state persistence, and observability. No Redis, no worker processes, no custom retry code.

Other tools can achieve similar results (Temporal, for example), but Inngest's developer experience with TypeScript is what sold me. You write normal async functions. The step.run() wrapper is the only addition.

How to Set Up the Webhook Endpoint

Your webhook endpoint should be minimal. Validate the signature, extract the event data, send it to your background job system, and return a 200 immediately.

Here's the real webhook endpoint from my production codebase:

import { constructWebhookEvent } from "@/lib/payments";
import { inngest } from "@/lib/jobs";

app.post("/api/payments/webhook", async ({ request, set }) => {
  const body = await request.text();
  const sig = request.headers.get("stripe-signature");

  if (!sig) {
    set.status = 400;
    return { error: "Missing signature" };
  }

  try {
    const event = await constructWebhookEvent(body, sig);
    console.log(`[Webhook] Received ${event.type}`);

    if (event.type === "charge.refunded") {
      const charge = event.data.object;
      await inngest.send({
        name: "stripe/charge.refunded",
        data: {
          chargeId: charge.id,
          paymentIntentId: charge.payment_intent,
          amountRefunded: charge.amount_refunded,
          originalAmount: charge.amount,
          currency: charge.currency,
        },
      });
    }

    if (event.type === "checkout.session.expired") {
      const session = event.data.object;
      await inngest.send({
        name: "stripe/checkout.session.expired",
        data: {
          sessionId: session.id,
          customerEmail: session.customer_email,
        },
      });
    }

    return { received: true };
  } catch (error) {
    console.error("[Webhook] Stripe verification failed:", error);
    set.status = 400;
    return { error: "Webhook verification failed" };
  }
});

Notice what this handler does not do: it does not look up users, write to the database, send emails, or call external APIs. It validates the Stripe signature, extracts the relevant fields, and sends a typed event to Inngest. The entire handler completes in milliseconds.

The constructWebhookEvent function wraps Stripe's signature verification:

import Stripe from "stripe";

export async function constructWebhookEvent(
  payload: string | Buffer,
  signature: string
) {
  const webhookSecret = process.env.STRIPE_WEBHOOK_SECRET;
  if (!webhookSecret) {
    throw new Error("STRIPE_WEBHOOK_SECRET is not set");
  }
  const client = new Stripe(process.env.STRIPE_SECRET_KEY);
  return client.webhooks.constructEventAsync(payload, signature, webhookSecret);
}

One critical detail: you must pass the raw request body (as a string or buffer) to Stripe's signature verification. If your framework parses the body as JSON before you can access the raw string, the signature check will fail. This is the number one cause of "webhook signature verification failed" errors.

The Inngest client setup is minimal:

import { Inngest } from "inngest";

export const inngest = new Inngest({
  id: "my-app",
});

For the purchase flow specifically, a different endpoint sends the event (the "claim" route that the frontend calls after the customer returns from Stripe checkout). But the principle is identical: validate, enqueue, return.

// After verifying payment status with Stripe
await inngest.send({
  name: "purchase/completed",
  data: {
    userId: session.user.id,
    tier,
    sessionId,
  },
});

How to Build a Durable Purchase Flow

This is the core of the article. The handlePurchaseCompleted function processes a purchase after payment using 9 individually checkpointed steps. Every step is real production code.

The example below grants access to a private GitHub repository because that's what this particular product sells.

Your product's "grant access" step will be different: upgrading a user to a Pro membership, provisioning API credits, unlocking a course, or activating a subscription. The durable step pattern is the same regardless of what you're delivering.

If step 5 fails (for example, the email provider is down), Inngest retries only step 5. Steps 1 through 4 are already checkpointed and don't re-execute. Steps 6 through 9 wait until step 5 succeeds.

import { eq } from "drizzle-orm";
import { createElement } from "react";

import { inngest } from "@/lib/jobs/client";
import { trackServerEvent } from "@/lib/analytics/server";
import { brand } from "@/lib/brand";
import { db, purchases, users } from "@/lib/db";
import {
  sendEmail,
  PurchaseConfirmationEmail,
  AdminPurchaseNotificationEmail,
  RepoAccessGrantedEmail,
} from "@/lib/email";
import { addCollaborator } from "@/lib/github";

export const handlePurchaseCompleted = inngest.createFunction(
  { id: "purchase-completed", triggers: [{ event: "purchase/completed" }] },
  async ({ event, step }) => {
    const { userId, tier, sessionId } = event.data;

    // Step 1: Look up user and purchase details
    const { user, purchase } = await step.run(
      "lookup-user-and-purchase",
      async () => {
        const userResult = await db
          .select({
            id: users.id,
            email: users.email,
            name: users.name,
            githubUsername: users.githubUsername,
          })
          .from(users)
          .where(eq(users.id, userId))
          .limit(1);

        const foundUser = userResult[0];
        if (!foundUser) {
          throw new Error(`User not found: ${userId}`);
        }

        const purchaseResult = await db
          .select({
            amount: purchases.amount,
            currency: purchases.currency,
            stripePaymentIntentId: purchases.stripePaymentIntentId,
          })
          .from(purchases)
          .where(eq(purchases.stripeCheckoutSessionId, sessionId))
          .limit(1);

        const foundPurchase = purchaseResult[0];

        return {
          user: foundUser,
          purchase: foundPurchase ?? {
            amount: 0,
            currency: "usd",
            stripePaymentIntentId: null,
          },
        };
      }
    );

    // Step 2: Track purchase completion in analytics
    await step.run("track-purchase-to-posthog", async () => {
      await trackServerEvent(userId, "purchase_completed_server", {
        tier,
        amount_cents: purchase.amount,
        currency: purchase.currency,
        stripe_session_id: sessionId,
      });
    });

    // Step 3: Send purchase confirmation to customer
    await step.run("send-purchase-confirmation", async () => {
      await sendEmail({
        to: user.email,
        subject: `Your purchase is confirmed!`,
        template: createElement(PurchaseConfirmationEmail, {
          amount: purchase.amount,
          currency: purchase.currency,
          customerEmail: user.email,
        }),
      });
    });

    // Step 4: Send admin notification
    await step.run("send-admin-notification", async () => {
      const adminEmail = process.env.ADMIN_EMAIL;
      if (!adminEmail) return;

      await sendEmail({
        to: adminEmail,
        subject: `New sale: ${user.email}`,
        template: createElement(AdminPurchaseNotificationEmail, {
          amount: purchase.amount,
          currency: purchase.currency,
          customerEmail: user.email,
          customerName: user.name,
          stripeSessionId: purchase.stripePaymentIntentId ?? sessionId,
        }),
      });
    });

    // Early return if user has no GitHub username
    if (!user.githubUsername) {
      return { success: true, userId, tier, githubAccessGranted: false };
    }

    // Step 5: Grant GitHub repository access
    const collaboratorResult = await step.run(
      "add-github-collaborator",
      async () => {
        return addCollaborator(user.githubUsername!);
      }
    );

    // Step 6: Track GitHub access granted
    await step.run("track-github-access", async () => {
      await trackServerEvent(userId, "github_access_granted", {
        tier,
        github_username: user.githubUsername,
        invitation_status: collaboratorResult.status,
      });
    });

    // Step 7: Update purchase record
    await step.run("update-purchase-record", async () => {
      await db
        .update(purchases)
        .set({
          githubAccessGranted: true,
          githubInvitationId: collaboratorResult.status,
          updatedAt: new Date(),
        })
        .where(eq(purchases.stripeCheckoutSessionId, sessionId));
    });

    // Step 8: Send repo access email
    await step.run("send-repo-access-email", async () => {
      await sendEmail({
        to: user.email,
        subject: `Your repository access is ready!`,
        template: createElement(RepoAccessGrantedEmail, {
          repoUrl: "https://github.com/your-org/your-repo",
        }),
      });
    });

    // Step 9: Schedule follow-up email sequence
    await step.run("schedule-follow-up", async () => {
      const purchaseRecord = await db
        .select({ id: purchases.id })
        .from(purchases)
        .where(eq(purchases.stripeCheckoutSessionId, sessionId))
        .limit(1);

      if (purchaseRecord[0]) {
        await inngest.send({
          name: "purchase/follow-up.scheduled",
          data: {
            userId,
            purchaseId: purchaseRecord[0].id,
            tier,
          },
        });
      }
    });

    return { success: true, userId, tier, githubAccessGranted: true };
  }
);

That's a lot of code. Let me walk through each step and explain why it's a separate checkpoint.

Step 1: Look Up User and Purchase

const { user, purchase } = await step.run(
  "lookup-user-and-purchase",
  async () => {
    // ... database queries ...
    return { user: foundUser, purchase: foundPurchase };
  }
);

This step queries the database for the user and purchase records. If the database is temporarily unreachable, this step retries on its own.

The return value (user and purchase) is cached by Inngest. Every subsequent step can use user.email, user.githubUsername, and purchase.amount without re-querying the database.

If this step fails permanently (the user doesn't exist), it throws an error that halts the entire function. This is intentional. There's no point continuing if you can't find the user.

Step 2: Track Analytics

await step.run("track-purchase-to-posthog", async () => {
  await trackServerEvent(userId, "purchase_completed_server", {
    tier,
    amount_cents: purchase.amount,
  });
});

Analytics tracking is a separate step because analytics services have their own failure modes (rate limits, outages, network timeouts). If PostHog is down, you don't want it to block the confirmation email.

In the production code, this step wraps the call in a try-catch so that a tracking failure doesn't halt the entire function. The analytics event is "nice to have," not critical.

Step 3: Send Purchase Confirmation Email

await step.run("send-purchase-confirmation", async () => {
  await sendEmail({
    to: user.email,
    subject: `Your purchase is confirmed!`,
    template: createElement(PurchaseConfirmationEmail, {
      amount: purchase.amount,
      currency: purchase.currency,
      customerEmail: user.email,
    }),
  });
});

This is the customer-facing confirmation. It's a separate step from the admin notification (step 4) because they're independent operations. If the admin email fails, the customer should still get their confirmation.

The sendEmail function uses Resend under the hood. If Resend returns a 500, this step retries. Because step 2 (analytics) already completed and is checkpointed, it won't re-run.

Step 4: Send Admin Notification

await step.run("send-admin-notification", async () => {
  const adminEmail = process.env.ADMIN_EMAIL;
  if (!adminEmail) return;

  await sendEmail({
    to: adminEmail,
    subject: `New sale: ${user.email}`,
    template: createElement(AdminPurchaseNotificationEmail, { /* ... */ }),
  });
});

Admin notifications are completely independent from customer-facing operations. Separating them means a failure in one doesn't affect the other.

Step 5: Grant GitHub Access

const collaboratorResult = await step.run(
  "add-github-collaborator",
  async () => {
    return addCollaborator(user.githubUsername!);
  }
);

This is the step most likely to fail. GitHub's API has rate limits: it can time out, and the user's GitHub username might be invalid.

By making this its own step, a GitHub API failure doesn't trigger re-sends of the confirmation email (step 3) or the admin notification (step 4). Those steps are already checkpointed.

Notice the early return before this step: if the user has no GitHub username, the function returns early after step 4. The remaining steps only run when there's a GitHub account to grant access to.

Step 6: Track GitHub Access

await step.run("track-github-access", async () => {
  await trackServerEvent(userId, "github_access_granted", {
    tier,
    github_username: user.githubUsername,
    invitation_status: collaboratorResult.status,
  });
});

This uses the collaboratorResult from step 5. Because step.run() caches return values, collaboratorResult.status is available here even if the function was interrupted and resumed between steps 5 and 6.

Step 7: Update Purchase Record

await step.run("update-purchase-record", async () => {
  await db
    .update(purchases)
    .set({
      githubAccessGranted: true,
      githubInvitationId: collaboratorResult.status,
      updatedAt: new Date(),
    })
    .where(eq(purchases.stripeCheckoutSessionId, sessionId));
});

The database update happens after GitHub access is confirmed. You only mark githubAccessGranted: true after the collaborator invitation actually succeeded.

If you updated the record before granting access and the GitHub step failed, your database would say access was granted when it was not.

Step 8: Send Repo Access Email

await step.run("send-repo-access-email", async () => {
  await sendEmail({
    to: user.email,
    subject: `Your repository access is ready!`,
    template: createElement(RepoAccessGrantedEmail, {
      repoUrl: "https://github.com/your-org/your-repo",
    }),
  });
});

This email only sends after the GitHub invitation is confirmed (step 5) and the database is updated (step 7). The ordering matters. You don't want to tell the customer "your access is ready" if the invitation hasn't been sent.

Step 9: Schedule Follow-Up Sequence

await step.run("schedule-follow-up", async () => {
  const purchaseRecord = await db
    .select({ id: purchases.id })
    .from(purchases)
    .where(eq(purchases.stripeCheckoutSessionId, sessionId))
    .limit(1);

  if (purchaseRecord[0]) {
    await inngest.send({
      name: "purchase/follow-up.scheduled",
      data: {
        userId,
        purchaseId: purchaseRecord[0].id,
        tier,
      },
    });
  }
});

The final step triggers a separate Inngest function that handles the follow-up email sequence (day 7 onboarding tips, day 14 feedback request, day 30 testimonial request). This is an event-driven chain: one function completes and triggers another.

The follow-up function uses step.sleep() to wait between emails:

export const handlePurchaseFollowUp = inngest.createFunction(
  {
    id: "purchase-follow-up",
    triggers: [{ event: "purchase/follow-up.scheduled" }],
    cancelOn: [
      {
        event: "purchase/follow-up.cancelled",
        match: "data.purchaseId",
      },
    ],
  },
  async ({ event, step }) => {
    const { userId, purchaseId } = event.data;

    await step.sleep("wait-7-days", "7d");

    await step.run("send-day-7-email", async () => {
      // Check eligibility (user exists, not unsubscribed, not refunded)
      // Send onboarding tips email
    });

    await step.sleep("wait-14-days", "7d");

    await step.run("send-day-14-email", async () => {
      // Send feedback request email
    });

    await step.sleep("wait-30-days", "16d");

    await step.run("send-day-30-email", async () => {
      // Send testimonial request email
    });
  }
);

Notice the cancelOn option. If the purchase is refunded, you can send a purchase/follow-up.cancelled event, and the entire follow-up sequence stops. No stale emails sent to customers who asked for a refund.

Why Each Step Must Be Separate

The rule is simple: any operation that calls an external service or could fail independently should be its own step.

A database query is a step because the database can be temporarily unreachable. An email send is a step because the email provider can return a 500. A GitHub API call is a step because it can be rate-limited.

If two operations always succeed or fail together (they share a single external call), they can be in the same step. But when in doubt, make it a separate step. The overhead is negligible, and the reliability gain is significant.

How to Handle Refunds with the Same Pattern

The refund flow follows the exact same durable step pattern. This function lives in the same file as handlePurchaseCompleted, so it shares the same imports (plus removeCollaborator from @/lib/github and the refund-specific email templates). Here's the handleRefund function:

export const handleRefund = inngest.createFunction(
  { id: "refund-processed", triggers: [{ event: "stripe/charge.refunded" }] },
  async ({ event, step }) => {
    const {
      chargeId,
      paymentIntentId,
      amountRefunded,
      originalAmount,
      currency,
    } = event.data;

    const isFullRefund = amountRefunded >= originalAmount;

    // Step 1: Look up the purchase and user
    const { user, purchase } = await step.run(
      "lookup-purchase-by-payment-intent",
      async () => {
        const purchaseResult = await db
          .select({
            id: purchases.id,
            userId: purchases.userId,
            stripePaymentIntentId: purchases.stripePaymentIntentId,
            githubAccessGranted: purchases.githubAccessGranted,
          })
          .from(purchases)
          .where(eq(purchases.stripePaymentIntentId, paymentIntentId))
          .limit(1);

        const foundPurchase = purchaseResult[0];
        if (!foundPurchase) {
          return { user: null, purchase: null };
        }

        const userResult = await db
          .select({
            id: users.id,
            email: users.email,
            name: users.name,
            githubUsername: users.githubUsername,
          })
          .from(users)
          .where(eq(users.id, foundPurchase.userId))
          .limit(1);

        return { user: userResult[0] ?? null, purchase: foundPurchase };
      }
    );

    if (!purchase || !user) {
      return { success: false, reason: "no_matching_purchase" };
    }

    let accessRevoked = false;

    // Step 2: Revoke GitHub access (only for full refunds)
    if (isFullRefund && user.githubUsername && purchase.githubAccessGranted) {
      const revokeResult = await step.run(
        "revoke-github-access",
        async () => {
          return removeCollaborator(user.githubUsername!);
        }
      );
      accessRevoked = revokeResult.success;
    }

    // Step 3: Update purchase status
    await step.run("update-purchase-status", async () => {
      if (isFullRefund) {
        await db
          .update(purchases)
          .set({
            status: "refunded",
            githubAccessGranted: false,
            updatedAt: new Date(),
          })
          .where(eq(purchases.id, purchase.id));
      } else {
        await db
          .update(purchases)
          .set({
            status: "partially_refunded",
            updatedAt: new Date(),
          })
          .where(eq(purchases.id, purchase.id));
      }
    });

    // Step 4: Track refund in analytics
    await step.run("track-refund-event", async () => {
      await trackServerEvent(user.id, "refund_processed", {
        charge_id: chargeId,
        amount_cents: amountRefunded,
        original_amount_cents: originalAmount,
        currency,
        is_full_refund: isFullRefund,
        github_access_revoked: accessRevoked,
      });
    });

    // Step 5: Notify customer
    await step.run("send-customer-notification", async () => {
      if (isFullRefund) {
        await sendEmail({
          to: user.email,
          subject: "Your refund has been processed",
          template: createElement(AccessRevokedEmail, {
            customerEmail: user.email,
            refundAmount: amountRefunded,
            currency,
          }),
        });
      } else {
        await sendEmail({
          to: user.email,
          subject: "Your partial refund has been processed",
          template: createElement(PartialRefundEmail, {
            customerEmail: user.email,
            refundAmount: amountRefunded,
            originalAmount,
            currency,
          }),
        });
      }
    });

    // Step 6: Notify admin
    await step.run("send-admin-notification", async () => {
      const adminEmail = process.env.ADMIN_EMAIL;
      if (!adminEmail) return;

      await sendEmail({
        to: adminEmail,
        subject: `\({isFullRefund ? "Full" : "Partial"} refund: \){user.email}`,
        template: createElement(AdminRefundNotificationEmail, {
          customerEmail: user.email,
          customerName: user.name,
          githubUsername: user.githubUsername,
          refundAmount: amountRefunded,
          originalAmount,
          currency,
          stripeChargeId: chargeId,
          accessRevoked,
          isPartialRefund: !isFullRefund,
        }),
      });
    });

    return { success: true, accessRevoked, isFullRefund, userId: user.id };
  }
);

Three things are worth calling out in the refund flow.

1. Partial versus full refunds: The function distinguishes between the two using a simple comparison: amountRefunded >= originalAmount. For a partial refund, the customer keeps access but the purchase status changes to partially_refunded. For a full refund, GitHub access is revoked and the status becomes refunded.

   This matters for your database integrity. Downstream systems (your dashboard, your analytics, your support tools) need accurate status values.

2. Conditional step execution: The "revoke GitHub access" step only runs if three conditions are true: it's a full refund, the user has a GitHub username, and access was previously granted. Inngest handles this cleanly by skipping steps that don't need to run.

   This is more readable than deeply nested if-else blocks in a monolithic handler.

3. Separate notifications for customers and admins: The customer gets a different email depending on whether the refund is full or partial. The admin always gets a detailed notification including the charge ID, the customer's GitHub username, and whether access was revoked.

These are separate steps because a failure in the admin notification shouldn't block the customer notification. The customer's email is the higher priority.

How to Recover Abandoned Checkouts

Abandoned cart recovery is where the step.sleep() method shines. When a Stripe checkout session expires, you want to send a recovery email. But not immediately.

You want to wait an hour or so, giving the customer time to return on their own.

export const handleCheckoutExpired = inngest.createFunction(
  {
    id: "checkout-expired",
    triggers: [{ event: "stripe/checkout.session.expired" }],
  },
  async ({ event, step }) => {
    const { customerEmail, sessionId } = event.data;

    if (!customerEmail) {
      return { success: false, reason: "no_email" };
    }

    // Wait 1 hour before sending recovery email
    await step.sleep("wait-before-recovery-email", "1h");

    // Send abandoned cart email
    await step.run("send-abandoned-cart-email", async () => {
      const checkoutUrl = `https://yoursite.com/pricing`;

      await sendEmail({
        to: customerEmail,
        subject: "Your checkout is waiting",
        template: createElement(AbandonedCartEmail, {
          customerEmail,
          checkoutUrl,
        }),
      });
    });

    // Track the event
    await step.run("track-abandoned-cart", async () => {
      await trackServerEvent("anonymous", "abandoned_cart_email_sent", {
        customer_email: customerEmail,
        session_id: sessionId,
      });
    });

    return { success: true, customerEmail };
  }
);

The step.sleep("wait-before-recovery-email", "1h") line is the key. This pauses the function for one hour without consuming any compute resources.

Inngest handles the scheduling internally. After one hour, the function resumes and sends the email.

Without durable execution, you would need a cron job that queries a database for expired sessions, or a delayed job queue with Redis, or a setTimeout that gets lost when your server restarts. The step.sleep() approach is simpler, more readable, and more reliable.

There's also a guard at the top of the function. If Stripe doesn't have a customer email for the session (the customer closed the checkout before entering their email), the function returns early. There's no point scheduling a recovery email with no address to send it to.

This pattern scales to more complex recovery flows. You could add a second step.sleep() and send a follow-up recovery email three days later if the customer still hasn't purchased. You could check if the customer has since completed a purchase (by querying the database in a step.run()) and skip the email if they have.

Each additional step is one more step.run() or step.sleep() call. The function reads like a script describing your business logic, not a tangle of cron jobs and database flags.

How to Test Webhook Handlers Locally

Local testing is one of the biggest pain points with Stripe webhooks. You need Stripe to send events to your local machine, and you need your background job system running to process them. Here's the setup.

How to Forward Stripe Events Locally

Install the Stripe CLI and forward webhook events to your local server:

stripe listen --forward-to localhost:3000/api/payments/webhook

The CLI prints a webhook signing secret (starting with whsec_). Set this as your STRIPE_WEBHOOK_SECRET environment variable for local development.

You can trigger test events directly:

stripe trigger checkout.session.completed
stripe trigger charge.refunded
stripe trigger checkout.session.expired

How to Run the Inngest Dev Server

Inngest provides a local dev server that shows you every function execution, every step, and every retry in real time:

npx inngest-cli@latest dev -u http://localhost:3000/api/inngest

The -u flag tells the Inngest dev server where your application is running so it can discover your functions. Open http://localhost:8288 in your browser to see the Inngest dashboard.

How to Watch Step Execution

The Inngest dev dashboard is where the durable execution pattern really clicks. When you trigger a Stripe event, you can see:

1. The event arriving in the "Events" tab.

2. The function triggering in the "Runs" tab.

3. Each step executing one by one, with its input, output, and duration.

4. If a step fails, you see the error and the retry attempt.

This visibility is something you don't get with inline webhook handlers. When a customer reports "I paid but didn't get access," you can look up the function run in the Inngest dashboard and see exactly which step failed and why. That kind of observability is invaluable in production.

How to Simulate Failures

To test the retry behavior, you can intentionally make a step fail. For example, temporarily throw an error in the "add-github-collaborator" step:

const collaboratorResult = await step.run(
  "add-github-collaborator",
  async () => {
    throw new Error("Simulated GitHub API failure");
  }
);

In the Inngest dashboard, you'll see:

• Steps 1 through 4 succeed and their results are cached.

• Step 5 fails and is retried according to the retry policy.

• Steps 6 through 9 remain pending until step 5 succeeds.

Remove the thrown error, and on the next retry, step 5 succeeds. Steps 6 through 9 then execute in sequence, while steps 1 through 4 aren't re-executed. This is the checkpoint behavior in action.

Conclusion

The pattern for reliable Stripe webhooks comes down to one principle: separate receiving from processing.

Your webhook endpoint validates the Stripe signature and sends a typed event to a background job system. That's all it does. The processing happens in a durable function where each step is individually checkpointed and retried.

Here's what this gives you:

• No duplicate emails: A step that already succeeded doesn't re-run.

• No partial state: If step 5 fails, steps 1 through 4 are preserved and step 5 retries independently.

• Full observability: You can see exactly which step failed and why, for every function run.

• Built-in delayed execution: step.sleep() handles recovery emails and follow-up sequences without cron jobs.

• Composable workflows: One function can trigger another via events, creating chains like purchase completion leading to a 30-day follow-up sequence.

This pattern isn't limited to Stripe. Any multi-step webhook processing benefits from durable execution: GitHub webhooks that trigger CI pipelines, Resend webhooks that track email delivery, or calendar webhooks that sync across services.

The principle is the same: Validate. Enqueue. Process durably.

I've used this pattern in production for Eden Stack, where the purchase flow handles everything from payment confirmation to GitHub repository access grants to multi-week email sequences. The 9-step purchase function has processed every payment without a single missed step or duplicate email.

If you're building a SaaS with Stripe, start with the webhook endpoint pattern from this article. Keep the endpoint thin and move the processing into durable steps. You'll save yourself from the 3 AM debugging session when a customer says "I paid but nothing happened."

If you want the complete Stripe webhook and Inngest integration pre-built with purchase flows, refund handling, and follow-up email sequences ready to go, Eden Stack includes everything from this article alongside 30+ additional production-tested patterns.

Magnus Rodseth builds AI-native applications and is the creator of Eden Stack, a production-ready starter kit with 30+ Claude skills encoding production patterns for AI-native SaaS development.

Rate limiting prevents this. It controls how many requests a client can make within a given timeframe, protecting your infrastructure from both intentional abuse and accidental overload.

Among the several algorithms used for rate limiting, the Token Bucket stands out for its balance of simplicity and flexibility. Unlike fixed window counters that reset abruptly, the Token Bucket allows short bursts of traffic while still enforcing a sustainable long-term rate. This makes it a practical choice for APIs where clients occasionally need to send a quick flurry of requests without being penalized.

In this guide, you'll implement a Token Bucket rate limiter in a FastAPI application. You'll build the algorithm from scratch as a Python class, wire it into FastAPI as middleware with per-user tracking, add standard rate limit headers to your responses, and test everything with a simple script. By the end, you'll have a working rate limiter you can drop into any FastAPI project.

What we'll cover:

1. Prerequisites

2. Understanding the Token Bucket Algorithm

3. Setting Up the FastAPI Project

4. Implementing the Token Bucket Class

5. Adding Per-User Rate Limiting Middleware

6. Testing the Rate Limiter

7. Where Rate Limiting Fits in Your Architecture

8. Conclusion

Prerequisites

To follow this tutorial, you'll need:

• Python 3.9 or later installed on your machine. You can verify your version by running python --version.

• Familiarity with Python and basic knowledge of how HTTP APIs work.

• A text editor such as VS Code, Vim, or any editor you prefer.

Understanding the Token Bucket Algorithm

Before writing code, it helps to understand the mechanism you'll be building.

The Token Bucket algorithm models rate limiting with two simple concepts: a bucket that holds tokens, and a refill process that adds tokens at a steady rate.

Here is how it works:

1. The bucket starts full, holding a fixed maximum number of tokens (the capacity).

2. Each incoming request costs one token. If the bucket has tokens available, the request is allowed, and one token is removed.

3. If the bucket is empty, the request is rejected with a 429 Too Many Requests response.

4. Tokens are added back to the bucket at a constant refill rate, regardless of whether requests are coming in. The bucket never exceeds its maximum capacity.

The capacity determines how large a burst the system absorbs. The refill rate defines the sustained throughput. For example, a bucket with a capacity of 10 and a refill rate of 2 tokens per second allows a client to fire 10 requests instantly, but after that, they can only make 2 requests per second until the bucket refills.

This two-parameter design gives you precise control:

Compared to other rate-limiting algorithms:

• Fixed Window counters reset at hard boundaries (for example, every minute), which can allow double the intended rate at window edges. The Token Bucket has no such boundary.

• Sliding Window counters are more accurate but more complex to implement and maintain.

• Leaky Bucket processes requests at a fixed rate and queues the rest. The Token Bucket is similar, but allows bursts instead of forcing a constant pace.

The Token Bucket is widely used in production systems. AWS API Gateway, NGINX, and Stripe all use variations of it.

Setting Up the FastAPI Project

Create a project directory and install the dependencies:

mkdir fastapi-ratelimit && cd fastapi-ratelimit

Create and activate a virtual environment:

python -m venv venv

On Linux/macOS:

source venv/bin/activate

On Windows:

venv\Scripts\activate

Install FastAPI and Uvicorn:

pip install fastapi uvicorn

Create the project file structure:

fastapi-ratelimit/
├── main.py
└── ratelimiter.py

Create main.py with a minimal FastAPI application:

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello, world!"}

Start the server to verify the setup:

uvicorn main:app --reload

You should see output similar to:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process

Open in your browser http://127.0.0.1:8000 or run curl http://127.0.0.1:8000. You should receive:

{"message": "Hello, world!"}

With the project running, you can move on to building the rate limiter.

Implementing the Token Bucket Class

Open ratelimiter.py in your editor and add the following code. This class implements the Token Bucket algorithm with thread-safe operations:

import time
import threading


class TokenBucket:
    """
    Token Bucket rate limiter.

    Each bucket starts full at `max_tokens` and refills `refill_rate`
    tokens every `interval` seconds, up to the maximum capacity.
    """

    def __init__(self, max_tokens: int, refill_rate: int, interval: float):
        """
        Initialize a new Token Bucket.

        :param max_tokens: Maximum number of tokens the bucket can hold (burst capacity).
        :param refill_rate: Number of tokens added per refill interval.
        :param interval: Time in seconds between refills.
        """
        assert max_tokens > 0, "max_tokens must be positive"
        assert refill_rate > 0, "refill_rate must be positive"
        assert interval > 0, "interval must be positive"

        self.max_tokens = max_tokens
        self.refill_rate = refill_rate
        self.interval = interval

        self.tokens = max_tokens
        self.refilled_at = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        """Add tokens based on elapsed time since the last refill."""
        now = time.time()
        elapsed = now - self.refilled_at

        if elapsed >= self.interval:
            num_refills = int(elapsed // self.interval)
            self.tokens = min(
                self.max_tokens,
                self.tokens + num_refills * self.refill_rate
            )
            # Advance the timestamp by the number of full intervals consumed,
            # not to `now`, so partial intervals aren't lost.
            self.refilled_at += num_refills * self.interval

    def allow_request(self, tokens: int = 1) -> bool:
        """
        Attempt to consume `tokens` from the bucket.

        Returns True if the request is allowed, False if the bucket
        does not have enough tokens.
        """
        with self.lock:
            self._refill()

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def get_remaining(self) -> int:
        """Return the current number of available tokens."""
        with self.lock:
            self._refill()
            return self.tokens

    def get_reset_time(self) -> float:
        """Return the Unix timestamp when the next refill occurs."""
        with self.lock:
            return self.refilled_at + self.interval

The class has three public methods:

• allow_request() is the core method. It refills tokens based on elapsed time, then tries to consume one. It returns True if the request is allowed, False if the bucket is empty.

• get_remaining() returns the number of tokens the client has left. You will use this for response headers.

• get_reset_time() returns when the next token will be added. This is also exposed in response headers.

The threading.Lock ensures that concurrent requests don't create race conditions when reading or modifying the token count. This is important because FastAPI runs request handlers concurrently.

Note: This implementation stores bucket state in memory. If you restart the server, all buckets reset. For persistence across restarts or multiple server instances, you would store token counts in Redis or a similar external store. The in-memory approach is sufficient for single-instance deployments.

Adding Per-User Rate Limiting Middleware

A single global bucket would throttle all users together. One heavy user could exhaust the limit for everyone. Instead, you'll assign a separate bucket to each user, identified by their IP address.

Add the following to ratelimiter.py, below the TokenBucket class:

from collections import defaultdict


class RateLimiterStore:
    """
    Manages per-user Token Buckets.

    Each unique client key (e.g., IP address) gets its own bucket
    with identical parameters.
    """

    def __init__(self, max_tokens: int, refill_rate: int, interval: float):
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate
        self.interval = interval
        self._buckets: dict[str, TokenBucket] = {}
        self._lock = threading.Lock()

    def get_bucket(self, key: str) -> TokenBucket:
        """
        Return the TokenBucket for a given client key.
        Creates a new bucket if one does not exist yet.
        """
        with self._lock:
            if key not in self._buckets:
                self._buckets[key] = TokenBucket(
                    max_tokens=self.max_tokens,
                    refill_rate=self.refill_rate,
                    interval=self.interval,
                )
            return self._buckets[key]

Now open main.py and replace its contents with the full application, including the rate-limiting middleware:

import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from ratelimiter import RateLimiterStore

app = FastAPI()

# Configure rate limits: 10 requests burst, 2 tokens added every 1 second.
limiter = RateLimiterStore(max_tokens=10, refill_rate=2, interval=1.0)


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """
    Middleware that enforces per-IP rate limiting on every request.
    Adds standard rate limit headers to every response.
    """
    # Identify the client by IP address.
    client_ip = request.client.host
    bucket = limiter.get_bucket(client_ip)

    # Check if the client has tokens available.
    if not bucket.allow_request():
        retry_after = bucket.get_reset_time() - time.time()
        return JSONResponse(
            status_code=429,
            content={"detail": "Too many requests. Try again later."},
            headers={
                "Retry-After": str(max(1, int(retry_after))),
                "X-RateLimit-Limit": str(bucket.max_tokens),
                "X-RateLimit-Remaining": str(bucket.get_remaining()),
                "X-RateLimit-Reset": str(int(bucket.get_reset_time())),
            },
        )

    # Request is allowed. Process it and add rate limit headers to the response.
    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = str(bucket.max_tokens)
    response.headers["X-RateLimit-Remaining"] = str(bucket.get_remaining())
    response.headers["X-RateLimit-Reset"] = str(int(bucket.get_reset_time()))
    return response


@app.get("/")
async def root():
    return {"message": "Hello, world!"}


@app.get("/data")
async def get_data():
    return {"data": "Some important information"}


@app.get("/health")
async def health():
    return {"status": "ok"}

The middleware does the following on every incoming request:

1. Extracts the client's IP address from request.client.host.

2. Retrieves (or creates) that client's Token Bucket from the store.

3. Calls allow_request(). If the bucket is empty, it returns a 429 response with a Retry-After header telling the client how long to wait.

4. If tokens are available, it processes the request normally and attaches rate limit headers to the response.

The three X-RateLimit-* headers follow a widely adopted convention:

These headers allow well-behaved clients to self-throttle before hitting the limit.

Testing the Rate Limiter

Restart the server if it's not already running:

uvicorn main:app --reload

Manual Testing with curl

Manual testing with curl is useful during development when you want to quickly verify that your middleware is working. A single request lets you confirm that the rate limit headers are present, the values are correct, and one token is consumed as expected.

This approach is fast and requires no additional setup, making it ideal for spot-checking your configuration after making changes.

Send a single request and inspect the response:

curl -i http://127.0.0.1:8000/data

You should see a 200 response with headers like:

HTTP/1.1 200 OK
x-ratelimit-limit: 10
x-ratelimit-remaining: 9
x-ratelimit-reset: 1739836801

Automated Burst Test

While curl confirms that the rate limiter is active, it can't verify that the limiter actually blocks requests when the bucket is empty. For that, you need to send requests faster than the refill rate and observe the 429 responses. An automated burst test is essential before deploying to production, after changing your bucket parameters, or when you need to verify both the blocking and refill behavior.

Create a file called test_ratelimit.py in your project directory:

import requests
import time


def test_burst():
    """Send 15 rapid requests to trigger the rate limit."""
    url = "http://127.0.0.1:8000/data"
    results = []

    for i in range(15):
        response = requests.get(url)
        remaining = response.headers.get("X-RateLimit-Remaining", "N/A")
        results.append((i + 1, response.status_code, remaining))
        print(f"Request {i+1:2d} | Status: {response.status_code} | Remaining: {remaining}")

    print()

    allowed = sum(1 for _, status, _ in results if status == 200)
    blocked = sum(1 for _, status, _ in results if status == 429)
    print(f"Allowed: {allowed}, Blocked: {blocked}")


def test_refill():
    """Exhaust tokens, wait for a refill, then confirm requests succeed again."""
    url = "http://127.0.0.1:8000/data"

    print("\n--- Exhausting tokens ---")
    for i in range(12):
        response = requests.get(url)
        print(f"Request {i+1:2d} | Status: {response.status_code}")

    print("\n--- Waiting 3 seconds for refill ---")
    time.sleep(3)

    print("\n--- Sending requests after refill ---")
    for i in range(5):
        response = requests.get(url)
        remaining = response.headers.get("X-RateLimit-Remaining", "N/A")
        print(f"Request {i+1:2d} | Status: {response.status_code} | Remaining: {remaining}")


if __name__ == "__main__":
    print("=== Burst Test ===")
    test_burst()

    # Allow bucket to refill before next test
    time.sleep(6)

    print("\n=== Refill Test ===")
    test_refill()

Install the requests library if you don't have it:

pip install requests

Run the test:

python test_ratelimit.py

You should see output similar to:

=== Burst Test ===
Request  1 | Status: 200 | Remaining: 9
Request  2 | Status: 200 | Remaining: 8
Request  3 | Status: 200 | Remaining: 7
...
Request 10 | Status: 200 | Remaining: 0
Request 11 | Status: 429 | Remaining: 0
Request 12 | Status: 429 | Remaining: 0
...
Request 15 | Status: 429 | Remaining: 0

Allowed: 10, Blocked: 5

The first 10 requests succeed (one token each from the full bucket). Requests 11 through 15 are rejected because the bucket is empty. The refill test then confirms that after waiting, tokens reappear and requests succeed again.

Note: The exact split between allowed and blocked requests may vary slightly due to timing. Tokens may refill between rapid requests. This is expected behavior.

Where Rate Limiting Fits in Your Architecture

The implementation in this tutorial runs inside your application process, which is the simplest approach and works well for single-instance deployments. In larger systems, rate limiting typically appears at multiple layers:

• API gateway level (NGINX, Kong, Traefik, Envoy): A coarse global rate limit applied to all traffic before it reaches your application. This protects against large-scale abuse and DDoS.

• Application level (this tutorial): Fine-grained per-user or per-endpoint limits inside your service. This is useful for enforcing different quotas on different API tiers.

• Both: Many production systems combine a gateway-level global limiter with an in-app per-user limiter. The gateway catches the flood and the application enforces business rules.

For multi-instance deployments (multiple server processes behind a load balancer), the in-memory RateLimiterStore won't share state across instances. In that case, replace the in-memory dictionary with Redis. The Token Bucket logic stays the same – only the storage layer changes.

Conclusion

In this guide, you built a Token Bucket rate limiter from scratch and integrated it into a FastAPI application with per-user tracking and standard rate limit response headers. You also tested the implementation to verify that burst capacity and refill behavior work as expected.

The Token Bucket algorithm gives you two straightforward controls, capacity for burst tolerance and refill rate for sustained throughput, which cover the vast majority of rate-limiting needs.

From here, you can extend this foundation by:

• Replacing the in-memory store with Redis for multi-instance deployments.

• Applying different rate limits per endpoint by creating separate RateLimiterStore instances.

• Using authenticated user IDs instead of IP addresses for more accurate client identification.

• Adding metrics and logging to track how often clients are being throttled.

This guide shows you how to build a full-stack web application that:

• Accepts audio input and transcribes the speech in it

• Prompts an AI agent with the transcription

• Displays the AI response on the UI

The application you'll build will be a simplified version of the Use Voice feature on AI chat applications highlighted in the image below:

By practising along with this article, you'll learn how to:

• Build a frontend application that uses the SpeechRecognition API to accept voice input and transcribe it

• Build a backend app that prompts an AI assistant of your choice and sends a response back to clients

• Connect both applications together to send the transcription to the backend as a prompt and display the AI response on the frontend

Optionally, you'll also learn how to host the frontend with Firebase and the backend with Google Cloud Run.

Table of Contents

• Prerequisites

• The Web Speech API

  • How to Use the Web Speech API in JavaScript for SEO

• How the Application Works

• How to Build the Application

  • Create the Backend Application with Node.js

  • Integrate an AI Assistant into the Node.js Application

  • Create the Frontend Application with Vite

• Test the Application Locally

• Deploy the Backend Application with Google Cloud Run

• Deploy the Frontend Application with Firebase

• Connect the Deployed Applications

• Conclusion

Prerequisites

This guide assumes that you have a working knowledge of HTML, CSS, and JavaScript in the browser. Basic familiarity with Node.js is beneficial but not essential.

In addition, you should have:

• Google Chrome (at least version 33 ) and a functional audio input device

• Node.js and npm installed on your computer

• An API key from any AI assistant of your choice

• A Google Cloud account and a Firebase account if you intend to deploy the applications

The Web Speech API

The Web Speech API enables applications to transcribe the speech in audio input and also synthesise audio from text. The API is made up of two components:

• The SpeechRecognition component which receives audio input, recognises speech in the input and transcribes it

• The SpeechSynthesis component which synthesises speech from text

You'll use the SpeechRecognition component in this guide.

How to Use the Web Speech API in JavaScript for SEO

The SpeechRecognition component works through a JavaScript object instantiated in code.

const recognition = new SpeechRecognition();

The recognition instance exposes several event listeners that respond to audio input. For example, the audiostart event fires when sound is first detected, logging "audio detected" to the console as shown in the snippet below.

recognition.addEventListener("audiostart", function(event){
  console.log("audio detected")
}

The first time it recognises speech in a sound bite, the speechstart event is fired.

A SpeechRecognition instance also has the ability to configure how speech recognition should work. For example, it has a property called lang which sets the language that it should recognise. The default value of the lang property is the HTML lang attribute value, or the browser's language setting. It also has a boolean property called interimResults, which when set to true, enables the instance to return transcriptions incrementally rather than waiting for the audio input to end.

Audio captured by the microphone is processed by a recognition engine which could be in a remote server (for Google Chrome) or embedded in the browser (for Firefox).

After processing, the recognition engine returns a result, which is a list of words or phrases that have been recognised in the speech.

Each transcription in the list has two properties: confidence, a numerical estimate of its accuracy ranging from 0 (low) to 1 (high), and transcript, the recognised text for all or part of the speech.

How the Application Works

In order for a SpeechRecognition instance to capture audio, it needs access to the microphone. The browser requests permission to use the microphone and, if granted, the application uses it to capture audio for the instance.

Speech captured by the instance goes through the recognition engine and produces results or transcriptions. Results with high confidence are combined and sent to the backend via an API request.

The backend uses the transcript it receives to prompt an AI assistant. The response from the AI assistant is sent back to the frontend and displayed on the UI as shown in the screenshot below:

How to Build the Application

First, you'll build a Node.js backend application that:

• Receives a text prompt from the frontend

• Sends the prompt to an AI assistant and receives a response

• Returns the response of the AI assistant to the frontend

Next, you'll build the frontend to:

• Accept your speech prompt, transcribe it, and display the transcription

• Send the transcription result to the backend

• Receive, format and display the response from the backend

Optionally, you'll deploy the frontend to Firebase and the backend to Google Cloud Run, connecting them so the application is publicly accessible.

Create the Backend Application with Node.js

The backend application you'll build in this section will receive text prompt from clients and use it to prompt an AI assistant. After receiving a response from the AI assistant, it will send the response back to the client.

We'll use Gemini in this guide, but you can use any AI assistant of your choice.

1. Create a folder for the backend app and give it a name, for example, "server".

2. In terminal, navigate to the project folder, run the npm init command, and answer the follow-up questions to generate a package.json file

3. In the root of the project, create a file named index.js.

Your project folder should have a structure like this:

├── index.js
├── package.json

The package.json file should have the following values for main , type and scripts.start:

 { 
    "main": "index.js", 
    "type": "module", 
    "scripts": { 
       "start": "node index.js" 
    }, 
}  

1. Copy and paste the code below into the index.js file to set up the server:

import http from "node:http";

async function parseRequestBody(req) { 
    return new Promise((resolve, reject) => { 
        let data = ""; 
        req.on("data", (chunk) => (data += chunk)); 
        req.on("end", () => resolve(JSON.parse(data))); 
        req.on("error", reject); 
    }); 
}

const server = http.createServer(async function (req, res) { 
    switch (req.method) { 
        case "POST":
          return res.end("POST request received");
        default:
          return res.end("non-POST request received");
    }
})

const port = Number(process.env.PORT) || 8000; 
server.listen(port, function () { 
    console.log("server running on port", port); 
});

In the code snippet above, the http module is imported from Node.js. The parseRequestBody function converts the request body stream of a HTTP request to a JavaScript object.

It responds with POST request received for POST requests and non-POST request received for all others. By default, it listens on port 8000 unless a PORT environment variable is defined.

Run npm run start to start the server. To confirm it is running, execute the following command in the terminal:

# For Linux/Mac, use:
curl -X POST -H "Content-Type: application/json" -d '{"prompt":"hello"}' http://localhost:8000

# For Windows, use:
curl.exe -X POST -H "Content-Type: application/json" -d '{"prompt":"hello"}' http://localhost:8000

You'll get the POST request received response from the server.

Integrate an AI Assistant into the Node.js Application

In this section, you'll integrate the AI assistant into the backend application, prompt it with data sent from the frontend, and return its response to the client. Again, we'll use Gemini for this here.

Visit the npm page for your chosen AI assistant to learn how to install and set it up. Here are the npm pages for the most popular AI assistants:

• Anthropic AI

• Google Gemini

• Open AI

Update the index.js file to include the setup for the AI assistant using the snippet below:

import http from "node:http";
import { GoogleGenAI } from "@google/genai"; 

const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY });

async function parseRequestBody(req) { /* minimised code */ }

const server = http.createServer(async function (req, res) {
    res.setHeader("Access-Control-Allow-Origin", "*");

    switch (req.method) { 
        case "POST":
          const body = await parseRequestBody(req);
          const response = await ai.models.generateContent({
            model: "gemini-2.5-flash", // or whatever model you have
            contents: body.prompt,
         });

         return res.end(response.text);

        default:
          return res.end("non-POST request received");
    }
}
/* previous code minimised*/

The GEMINI_API_KEY is retrieved from the environment variables and passed as the apiKey to GoogleGenAI, which initialises the AI assistant.

The POST request body is parsed into a JavaScript object, and body.prompt is passed to ai.models.generateContent to prompt the AI assistant. The text property of the response which is in Markdown format, is then returned to the client.

Restart the server and test the current setup by making an API request to it with curl using the snippet below:

# For Linux/Mac:

curl -X POST -H "Content-Type: application/json" -d '{"prompt":"hello"}' http://localhost:8000

# For Windows:

curl.exe -X POST -H "Content-Type: application/json" -d '{"prompt":"hello"}' http://localhost:8000

You'll get an AI text response in the form of Markdown.

Create the Frontend Application with Vite

Vite is a build tool that provides a faster and more seamless development experience for developing applications. You'll use Vite to create the frontend application and connect it with the backend application from the previous section.

In another folder, create a project with Vite by running the npm create vite@latest command and answer the prompts:

npm create vite@latest

Need to install the following packages:
create-vite@8.1.0
Ok to proceed? (y) y

> npx create-vite

◇  Project name:
│  [name-of-your-frontend-app] e.g prompt-ai-with-speech-frontend
│
◇  Select a framework:
│  Vanilla
│
◇  Select a variant:
│  JavaScript
│
◇  Use rolldown-vite (Experimental)?:
│  No
│
◇  Install with npm and start now?
│  Yes

Open the project created in your code editor and make the following updates:

First, replace the content of index.html with the code snippet below:

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Prompt AI with the Web Speech Recognition API</title>
  </head>
  <body>
    <main id="app">
      <section>
        <h1>Prompt AI with the Web Speech Recognition API</h1>
        <ul id="ulist_chat"></ul>
      </section>
      <div class="btn_container">
        <button id="btn_record">Record prompt</button>
      </div>
    </main>
    <script type="module" src="/src/main.js"></script>
  </body>
</html>

Then replace the content of src/style.css with the code snippet below:

:root {
  font-family: system-ui, Avenir, Helvetica, Arial, sans-serif;
  line-height: 1.5;
  font-weight: 400;

  color-scheme: light dark;
  color: rgba(255, 255, 255, 0.87);
  background-color: #242424;

  font-synthesis: none;
  text-rendering: optimizeLegibility;
  -webkit-font-smoothing: antialiased;
  -moz-osx-font-smoothing: grayscale;
}

button {
  border-radius: 8px;
  border: 1px solid transparent;
  padding: 0.6em 1.2em;
  font-size: 1em;
  font-weight: 500;
  font-family: inherit;
  background-color: #1a1a1a;
  cursor: pointer;
  transition: border-color 0.25s;
}
button:hover {
  border-color: #646cff;
}
button:focus,
button:focus-visible {
  outline: 4px auto -webkit-focus-ring-color;
}
.btn_container {
  padding: 16px 0px;
  display: flex;
  justify-content: center;
}

#ulist_chat {
  display: flex;
  flex-direction: column;
  width: 80%;
  margin: auto;
  padding: 0;
}

#ulist_chat .transcript {
  border: 1px solid tomato;
  background: #fce5e5af;
  border-radius: 4px;
  align-self: flex-end;
  list-style-type: none;
  margin: 8px;
  padding: 8px;
  max-width: 80%;
}

#ulist_chat .ai_response p { 
  margin: 2px; 
}

#ulist_chat .ai_response {
  border: 1px solid green;
  background: #e5fce8af;
  border-radius: 4px;
  align-self: flex-start;
  list-style-type: none;
  margin: 8px;
  padding: 8px;
  max-width: 80%;
}

@media (prefers-color-scheme: light) {
  :root {
    color: #000;
    background-color: #ffffff;
  }
  a:hover {
    color: #747bff;
  }
  button {
    background-color: #f9f9f9;
  }
}

Now replace the content of src/main.js with the code snippet below:

import "./style.css";
import { marked } from "marked";

const apiUrl = "http://localhost:8000";
const btnRecord = document.getElementById("btn_record");
const uListChat = document.getElementById("ulist_chat");

function ensureBrowserHasSpeechAPI() {
  if (
    !("webkitSpeechRecognition" in window) &&
    !("SpeechRecognition" in window)
  ) {
    btnRecord.style.display = "none";

    return alert(
      "This browser does not have the features required for this demo. Use Google Chrome >= v33"
    );
  }

  start();
}

function toggleRecording(config, listener) {
  if (config.isListening) {
    config.isListening = false;
    btnRecord.innerText = "Start recording";
    return listener.stop();
  }

  config.isListening = true;
  btnRecord.innerText = "Stop recording";

  return listener.start();
}

/** @param {string} transcript  */
function appendTranscriptToChatList(transcript) {
  const li = document.createElement("li");
  li.innerText = transcript;
  li.classList.add("transcript");
  uListChat.appendChild(li);
}

/** @param {string} aiResponse  */
function appendAIResponseToChatList(aiResponse) {
  const li = document.createElement("li");
  li.innerHTML = marked.parse(aiResponse);
  li.classList.add("ai_response");
  uListChat.appendChild(li);
}

/** @param {string} prompt  */
async function promptAI(prompt) {
  try {
    const response = await fetch(apiUrl, {
      body: JSON.stringify({ prompt }),
      method: "POST",
      headers: {
        "Content-Type": "application/json",
      },
    });

    if (!response.ok) {
      const err = await response.text();
      console.error(err);
      alert("An error occurred. Try again");
      return;
    }

    const text = await response.text();
    return text;
  } catch (error) {
    logError(error);
    alert("An error occurred. Try again");
    return ""
  }
}

function setUpSpeechRecognition() {
  const SpeechRecognition =
    window.SpeechRecognition || window.webkitSpeechRecognition;

  const listener = new SpeechRecognition();
  listener.continuous = true; // listen for long speech
  listener.maxAlternatives = 2; // only two transcription suggestions required
  let transcript = "";

  // automatic: onstart -> onaudiostart -> onsoundstart -> onspeechstart
  // automatic: onspeechend -> onsoundend -> onaudioend -> onresult -> onend
  // click button: onaudioend -> onresult -> onend

  listener.onend = async function () {
    if (!transcript || !transcript.trim()) return;

    btnRecord.innerText = "Thinking...";
    btnRecord.disabled = true;
    appendTranscriptToChatList(transcript);
    promptAI(transcript)
      .then(function (res) {
        appendAIResponseToChatList(res);
      })
      .finally(function () {
        btnRecord.innerText = "Record prompt";
        btnRecord.disabled = false;
        transcript = "";
      });
  };

  listener.onerror = function (err) {
    logError(err);
    alert("Error occurred while capturing speech");
  };

  listener.onresult = function (event) {
    for (const alternatives of event.results) {
      const [bestAlternative] = Array.from(alternatives).toSorted(
        (altA, altB) => altB.confidence - altA.confidence
      );

      transcript += bestAlternative.transcript;
    }
  };

  return listener;
}

async function start() {
  const config = {
    isListening: false,
  };

  const listener = setUpSpeechRecognition();

  btnRecord.addEventListener("click", function () {
    toggleRecording(config, listener);
  });
}

ensureBrowserHasSpeechAPI();

function logError(...str) {
  for (const s of str) {
    console.error("error:", s);
  }
}

marked is an npm package that helps convert Markdown text to HTML and it's a required dependency in the project. Install marked in the project by running the following command in the project's terminal:

npm install marked

The ensureBrowserHasSpeechAPI function in src/main.js checks to see if the browser in use has the WebSpeechAPI feature. If it doesn't, it prevents the application from displaying the controls for the UI. That's why you'll need a Google Chrome browser with a version greater than or equal to 33 for this guide. Those versions have the WebSpeechAPI feature.

The toggleRecording function executes when the Record prompt button is clicked. On the first click, it requests microphone permission. It also enables/disables the activity of the SpeechRecognition instance.

The setUpSpeechRecognition function sets up the SpeechRecognition instance: listener, and its configuration. It also attaches functions to be run when the end, error and result events are triggered.

• error is triggered when there is an error in capturing or processing audio

• result is triggered when the recognition engine returns transcription results

• end is triggered when the speech recognition service has disconnected from the application.

The transcript is displayed on the UI after passing it as an argument to the appendTranscriptToChatList function.

The promptAI function executes when the end event fires, accepting the speech transcript as an argument and sending it to the backend via a POST request using fetch. On success, the AI response is returned as Markdown and passed to appendAIResponseToChatList, which converts it to HTML and displays it on the UI.

Test the Application Locally

Start the backend application by running npm run start in the backend project's terminal and start the frontend application by running npm run dev in the frontend project's terminal. Visit http://localhost:5173 to view the UI of application. You should see a UI similar to the one in the image below:

Click the Record prompt button. A prompt will appear requesting microphone permission. Select "Allow while visiting the site" or "Allow this time" to grant access and begin recording. Click on the Stop recording button when you're done.

The UI will display the transcript of your speech and the application will send it to the backend as a prompt. After waiting for a short while, you'll see the response from the AI assistant displayed on the UI.

You have been able to use speech input to prompt an AI assistant, receive a response and display it. How do you make this application accessible to everyone? The next section guides you through deploying both applications.

Deploy the Backend Application with Google Cloud Run

In this section, you'll deploy the backend application with Google Cloud Run and get a URL which will be used as the apiUrl in the frontend application.

In order to host the backend application with Google Cloud Run, you need to have a:

• Google Cloud developer account

• Google Cloud project

Visit Google Cloud to create an account and create a project. You can name the project whatever you want but it's a good idea to give a descriptive name. Take note of the project's ID because you'll use it in the deployment process.

There are three ways to deploy applications on Google Cloud Run:

• Deploy a revision from an existing container image

• Deploy from a repository such as GitHub or GitLab

• Create a function using the inline editor

You can see all three options if you visit the create Cloud Run service page.

In this guide, you'll use the option to deploy from an existing container image. Follow the steps below to deploy the backend server from a container image or follow the Cloud Run documentation at build and deploy Node.js service on Cloud Run:

• Install the Google Cloud (gcloud) CLI on your computer by visiting the Install Google Cloud CLI page and following the instructions on the page for your operating system

• Initialise the gcloud CLI to connect it to your developer account by visiting the Initializing the gcloud CLI page and following the instructions on the page

• Set the project you want to deploy the backend server under by running the command below in your terminal:

# replace PROJECT_ID with your project ID

gcloud config set project PROJECT_ID

• Visit your project's IAM Admin page to enable the following roles on the service account created for this project:

  • roles/run.sourceDeveloper

  • roles/iam.serviceAccountUser

  • roles/logging.viewer

These roles are required to enable the Cloud Run Admin API and Cloud Build APIs. Take note of the service account email address.

• Enable the Cloud Run Admin API and Cloud Build APIs by running the code snippet below in your terminal:

gcloud services enable run.googleapis.com cloudbuild.googleapis.com

• Grant the Cloud Build service account access to your project by running the code snippet below in your terminal:

# replace PROJECT_ID with your project ID and 
# SERVICE_ACCOUNT_EMAIL_ADDRESS with the service account's email address

gcloud projects add-iam-policy-binding PROJECT_ID \
--member=serviceAccount:SERVICE_ACCOUNT_EMAIL_ADDRESS \
--role=roles/run.builder

Update index.js in the backend project to restrict API requests to clients specified in the ALLOWED_ORIGINS environment variable, and update the AI assistant configuration to use the API key loaded from environment variables.

// Use the API key from the environment variable
const GEMINI_API_KEY = process.env.GEMINI_API_KEY; 
const ai = new GoogleGenAI({ apiKey: GEMINI_API_KEY });

// Replace res.setHeader("Access-Control-Allow-Origin", "*"); with
res.setHeader("Access-Control-Allow-Origin", process.env.ALLOWED_ORIGINS);
res.setHeader("Access-Control-Allow-Methods", "POST,OPTIONS");
res.setHeader("Access-Control-Allow-Headers", "Content-Type");

This ensures that the application will receive POST requests from only frontend URLs specified in the ALLOWED_ORIGINS environment variable. This setup prevents the backend from being loaded with requests from frontend clients that you don't know, and also prevents the excess use of your tokens. It also keeps you from deploying the application with the AI API key hardcoded in it.

To test that the new changes work, run the backend application with the command below:

# replace YOUR_API_KEY with your Gemini API key

GEMINI_API_KEY=YOUR_API_KEY ALLOWED_ORIGINS="http://localhost:5173" npm run start

With the command in the code snippet above, the backend application will not respond to requests from frontend applications not hosted on http://localhost:5173. Try to send a prompt from the frontend application to test that it works.

To deploy the backend application to Cloud Run, run the command in the snippet below in the terminal of the backend project folder. The command sets the environment variables required for the application to run and also deploys it to Google Cloud Run.

# replace <api-key> with your Gemini API key

gcloud run deploy --source . \
--set-env-vars "ALLOWED_ORIGINS=http://localhost:5173" \
--set-env-vars "GEMINI_API_KEY=<api-key>"

Once deployment is complete, you'll receive the URL of your hosted backend. Copy it and replace the value of apiUrl in your frontend application with it. Run the frontend, record a prompt, and confirm that everything works as expected.

Deploy the Frontend Application with Firebase

In this section, you'll host the frontend application with Firebase. You need to have a Firebase account. Follow the steps below to host the frontend with Firebase:

• Create and set up a Firebase project

• Install the Firebase CLI by visiting the install Firebase CLI page and follow the instructions for your operating system

• In the terminal of the frontend project, run firebase init hosting to initialise the hosting configuration for the project. Follow the prompts and use dist as the public directory when prompted

• Run firebase deploy --only hosting to host the application with Firebase

Once deployment is complete, you will receive the URL of your hosted frontend application.

Connect the Deployed Applications

Remember that the first time you deployed your backend application, you set ALLOWED_ORIGINS to http://localhost:5173. The deployed backend application doesn't know about the URL of the deployed frontend application so it won't accept requests from it.

In the terminal of the backend application, deploy the backend application again using the command in the snippet below:

# replace <frontend-url> with your Firebase frontend URL and <api-key> 
# with your Gemini API key

gcloud run deploy --source . \
--set-env-vars "ALLOWED_ORIGINS=<frontend-url>" --set-env-vars "GEMINI_API_KEY=<api-key>"

Visit the deployed frontend application and test it. It should work without errors.

Conclusion

In this guide, you built a frontend application that captures and transcribes speech, a Node.js backend application that prompts AI, and you connected both applications together to build a simplified version of the Use Voice feature in AI chat applications.

Can you add a feature to the application that will make it read out the response from the backend when it receives it? You can use the SpeechSynthesis API to build it.

Feel free to connect with me on LinkedIn if you have any questions. Thank you for reading this far and don’t hesitate to share this article if you found it insightful. Cheers!

They answer questions they should not, they break when document retrieval is weak, they time out due to network latency, and nobody can tell exactly what happened because there are no logs and no tests.

In this tutorial, you’ll build a beginner-friendly Retrieval Augmented Generation (RAG) application designed to survive production realities. This isn’t just a script that calls an API. It’s a system featuring a FastAPI backend, a persisted FAISS vector store, and essential safety guardrails (including a retrieval gate and fallbacks).

Table of Contents

1. Why RAG Alone Does Not Equal Production-Ready

2. The Architecture You Are Building

3. Project Setup and Structure

4. How to Build the RAG Layer with FAISS

5. How to Add the LLM Call with Structured Output

6. How to Add Guardrails: Retrieval Gate and Fallbacks

7. FastAPI App: Creating the /answer Endpoint

8. How to Add Beginner-Friendly Evals

9. What to Improve Next: Realistic Upgrades

Why RAG Alone Does Not Equal Production-Ready

Retrieval Augmented Generation (RAG) is often hailed as the hallucination killer. By grounding the model in retrieved text, we provide it with the facts it needs to be accurate. But simply connecting a vector database to an LLM isn’t enough for a production environment.

Production issues usually arise from the silent failures in the system surrounding the model:

• Weak retrieval: If the app retrieves irrelevant chunks of text, the model tries to bridge the gap by inventing an answer anyway. Without a designated “I do not know” path, the model is essentially forced to hallucinate.

• Lack of visibility: Without structured outputs and basic logging, you can’t tell if bad retrieval, a confusing prompt, or a model update caused a wrong answer.

• Fragility: A simple API timeout or malformed provider response becomes a user-facing outage if you don’t implement fallbacks.

• No regression testing: In traditional software, we have unit tests. In AI, we need evals. Without them, a small tweak to your prompt might fix one issue but break ten others without you realising it.

We’ll solve each of these issues systematically in this guide.

Prerequisites

This tutorial is beginner-friendly, but it assumes you have a few basics in place so you can focus on building a robust RAG system instead of getting stuck on setup issues.

Knowledge

You should be comfortable with:

• Python fundamentals (functions, modules, virtual environments)

• Basic HTTP + JSON (requests, response payloads)

• APIs with FastAPI (what an endpoint is and how to run a server)

• High-level LLM concepts (prompting, temperature, structured outputs)

Tools + Accounts

You’ll need:

• Python 3.10+

• A working OpenAI-compatible API key (OpenAI or any provider that supports the same request/response shape)

• A local environment where you can run a FastAPI app (Mac/Linux/Windows)

What This Tutorial Covers (and What It Doesn’t)

We’ll build a production-minded baseline:

• A FAISS-backed retriever with a persisted index + metadata

• A retrieval gate to prevent “forced hallucination”

• Structured JSON outputs so your backend is stable

• Fallback behavior for timeouts and provider errors

• A small eval harness to prevent regressions

We won’t implement advanced upgrades such as rerankers, semantic chunking, auth, background jobs beyond a roadmap at the end.

The Architecture You Are Building

The flow of our application follows a disciplined path so every answer is grounded in evidence:

1. User query: The user submits a question via a FastAPI endpoint.

2. Retrieval: The system embeds the question and retrieves the top-k most similar document chunks.

3. The retrieval gate: We evaluate the similarity score. If the context is not relevant enough, we stop immediately and refuse the query.

4. Augmentation and generation: If the gate passes, we send a context-augmented prompt to the LLM.

5. Structured response: The model returns a JSON object containing the answer, sources used, and a confidence level.

Project Setup and Structure

To keep things organized and maintainable, we’ll use a modular structure. This allows you to swap out your LLM provider or your vector database without rewriting your entire core application.

Project Structure

.
├── app.py              # FastAPI entry point and API logic
├── rag.py              # FAISS index, persistence, and document retrieval
├── llm.py              # LLM API interface and JSON parsing
├── prompts.py          # Centralized prompt templates
├── data/               # Source .txt documents
├── index/              # Persisted FAISS index and metadata
└── evals/              # Evaluation dataset and runner script
    ├── eval_set.json
    └── run_evals.py

Install Dependencies

First, create a virtual environment to isolate your project:

python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
pip install fastapi uvicorn faiss-cpu numpy pydantic requests python-dotenv

Configure the Environment

Create a .env file in the root directory. We are targeting OpenAI-compatible providers:

OPENAI_API_KEY=your_actual_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

Important note on compatibility: The code below assumes an OpenAI-style API. If you use a provider that is not compatible, you must change the URL, headers (for example X-API-Key), and the way you extract embeddings and final message content in embed_texts() and call_llm().

How to Build the RAG Layer with FAISS

In rag.py, we handle the “Retriever” part of RAG. This involves turning raw text into mathematical vectors that the computer can compare.

What is FAISS (and What Does It Do)?

FAISS (Facebook AI Similarity Search) is a fast library for vector similarity search. In a RAG system, each chunk of text becomes an embedding vector (a list of floats). FAISS stores those vectors in an index so you can quickly ask:

“Given this question embedding, which document chunks are closest to it?”

In this tutorial, we use IndexFlatIP inner product and normalise vectors with faiss.normalize_L2(...). With normalised vectors, the inner product behaves like cosine similarity, giving us a stable score we can use for a retrieval gate.

Chunking Strategy With Overlap

We’ll use chunking with overlap. If we split a document at exactly 1,000 characters, we might cut a sentence in half, losing its meaning. By using an overlap, for example, 200 characters, we ensure that the end of one chunk and the beginning of the next share context.

Implementation of rag.py

import os
import faiss
import numpy as np
import requests
import json
from typing import List, Dict
from dotenv import load_dotenv

load_dotenv()

INDEX_PATH = "index/faiss.index"
META_PATH = "index/meta.json"

def chunk_text(text: str, size: int = 1000, overlap: int = 200) -> List[str]:
    chunks = []
    step = max(1, size - overlap)
    for i in range(0, len(text), step):
        chunk = text[i : i + size].strip()
        if chunk:
            chunks.append(chunk)
    return chunks

def embed_texts(texts: List[str]) -> np.ndarray:
    # Note: If your provider is not OpenAI-compatible, change this URL and headers
    url = f"{os.getenv('OPENAI_BASE_URL')}/embeddings"
    headers = {"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"}
    payload = {"input": texts, "model": "text-embedding-3-small"}

    resp = requests.post(url, headers=headers, json=payload, timeout=30)
    resp.raise_for_status()
    # If your provider uses a different response format, change the line below
    vectors = np.array([item["embedding"] for item in resp.json()["data"]], dtype="float32")
    return vectors

def build_index() -> None:
    all_chunks: List[str] = []
    metadata: List[Dict] = []

    if not os.path.exists("data"):
        os.makedirs("data")
        return

    for file in os.listdir("data"):
        if not file.endswith(".txt"):
            continue

        with open(f"data/{file}", "r", encoding="utf-8") as f:
            text = f.read()

        chunks = chunk_text(text)
        all_chunks.extend(chunks)
        for c in chunks:
            metadata.append({"source": file, "text": c})

    if not all_chunks:
        return

    embeddings = embed_texts(all_chunks)
    faiss.normalize_L2(embeddings)

    dim = embeddings.shape[1]
    index = faiss.IndexFlatIP(dim)
    index.add(embeddings)

    os.makedirs("index", exist_ok=True)
    faiss.write_index(index, INDEX_PATH)

    with open(META_PATH, "w", encoding="utf-8") as f:
        json.dump(metadata, f, ensure_ascii=False)

def load_index():
    if not (os.path.exists(INDEX_PATH) and os.path.exists(META_PATH)):
        raise FileNotFoundError(
            "FAISS index not found. Add .txt files to data/ and run build_index()."
        )

    index = faiss.read_index(INDEX_PATH)
    with open(META_PATH, "r", encoding="utf-8") as f:
        metadata = json.load(f)
    return index, metadata

def retrieve(query: str, k: int = 5) -> List[Dict]:
    index, metadata = load_index()

    q_emb = embed_texts([query])
    faiss.normalize_L2(q_emb)

    scores, ids = index.search(q_emb, k)
    results = []
    for score, idx in zip(scores[0], ids[0]):
        if idx == -1:
            continue
        m = metadata[idx]
        results.append(
            {"score": float(score), "source": m["source"], "text": m["text"], "id": int(idx)}
        )
    return results

How to Add the LLM Call with Structured Output

A major failure point in AI apps is the “chatty” nature of LLMs. If your backend expects a list of sources but the LLM returns conversational filler, your code will crash.

We solve this with structured output: instruct the model to return a strict JSON object, then parse it safely.

Implementation of llm.py

import json
import requests
import os
from typing import Dict, Any

def call_llm(system_prompt: str, user_prompt: str) -> Dict[str, Any]:
    # Note: Change URL/Headers if using a non-OpenAI compatible provider
    url = f"{os.getenv('OPENAI_BASE_URL')}/chat/completions"
    headers = {
        "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": os.getenv("OPENAI_MODEL"),
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "response_format": {"type": "json_object"},
        "temperature": 0,
    }

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=30)
        resp.raise_for_status()
        content = resp.json()["choices"][0]["message"]["content"]

        parsed = json.loads(content)
        parsed.setdefault("answer", "")
        parsed.setdefault("refusal", False)
        parsed.setdefault("confidence", "medium")
        parsed.setdefault("sources", [])
        return parsed

    except (requests.Timeout, requests.ConnectionError):
        return {
            "answer": "The system is temporarily unavailable (network issue). Please try again.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "error_type": "network_error",
        }
    except Exception:
        return {
            "answer": "A system error occurred while generating the answer.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "error_type": "unknown_error",
        }

How to Add Guardrails: Retrieval Gate and Fallbacks

Guardrails are interceptors. They sit between the user and the model to prevent predictable failures.

The Retrieval Gate: How It Works and How to Add It

In a standard RAG pipeline, the system always calls the LLM. If the user asks an irrelevant question, the retriever will still return the “closest” (but wrong) chunks.

The solution is the retrieval gate:

1. Retrieve top-k chunks and get the top similarity score

2. If the score is below a threshold (for example 0.30), refuse immediately

3. Only call the LLM when retrieval is strong enough to ground the answer

A threshold of 0.30 is a reasonable starting point when using normalised cosine similarity, but you should tune it using evals (next section).

Fallbacks and Why They Matter

Fallbacks ensure that if an API fails or times out, the user gets a helpful message instead of a crash. They also keep your API response shape consistent, which prevents frontend errors and makes logging meaningful.

In this tutorial, fallbacks are implemented inside call_llm() so your FastAPI layer stays simple.

FastAPI App: Creating the /answer Endpoint

The app.py file is the conductor. It ties retrieval, guardrails, prompting, and generation together.

Implementation of app.py

from fastapi import FastAPI
from pydantic import BaseModel
from rag import retrieve
from llm import call_llm
import prompts
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag_app")

app = FastAPI(title="Production-Ready RAG")

class QueryRequest(BaseModel):
    question: str

@app.post("/answer")
async def get_answer(req: QueryRequest):
    start_time = time.time()
    question = (req.question or "").strip()

    if not question:
        return {
            "answer": "Please provide a non-empty question.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "latency_sec": round(time.time() - start_time, 2),
        }

    # 1) Retrieval
    results = retrieve(question, k=5)
    top_score = results[0]["score"] if results else 0.0

    logger.info("query=%r top_score=%.3f num_results=%d", question, top_score, len(results))

    # 2) Retrieval Gate (Guardrail)
    if top_score < 0.30:
        return {
            "answer": "I do not have documents to answer that question.",
            "refusal": True,
            "confidence": "low",
            "sources": [],
            "latency_sec": round(time.time() - start_time, 2),
            "retrieval": {"top_score": top_score, "k": 5},
        }

    # 3) Augment
    context_text = "\n\n".join([f"Source {r['source']}: {r['text']}" for r in results])
    user_prompt = f"Context:\n{context_text}\n\nQuestion: {question}"

    # 4) Generation with Fallback
    response = call_llm(prompts.SYSTEM_PROMPT, user_prompt)

    # 5) Attach debug metadata
    response["latency_sec"] = round(time.time() - start_time, 2)
    response["retrieval"] = {"top_score": top_score, "k": 5}
    return response

Centralized Prompt – Template: prompts.py

A small but important habit: keep prompts centralised so they’re versionable and easy to evaluate.

Example prompts.py

SYSTEM_PROMPT = """You are a RAG assistant. Use ONLY the provided Context to answer.
If the context does not contain the answer, respond with refusal=true.

Return a valid JSON object with exactly these keys:
- answer: string
- refusal: boolean
- confidence: "low" | "medium" | "high"
- sources: array of strings (source filenames you used)

Do not include any extra keys. Do not include markdown. Do not include commentary."""

How to Add Beginner-Friendly Evals

In AI systems, outputs are probabilistic. This makes testing harder than traditional software. Evals (evaluations) are a set of “golden questions” and “expected behaviours” you run repeatedly to detect regressions.

Instead of “does it output exactly this string,” you test:

• Should the app refuse when the retrieval is weak?

• When it answers, does it include sources?

• Is the behaviour stable across prompt tweaks and model changes?

Step 1: Create evals/eval_set.json

This should contain both positive and negative cases.

[
  {
    "id": "in_scope_01",
    "question": "What is a retrieval gate and why is it important?",
    "expect_refusal": false,
    "notes": "Should explain gating and relate it to hallucination prevention."
  },
  {
    "id": "out_of_scope_01",
    "question": "What is the capital of France?",
    "expect_refusal": true,
    "notes": "If the knowledge base only includes our docs, the app should refuse."
  },
  {
    "id": "edge_01",
    "question": "",
    "expect_refusal": true,
    "notes": "Empty input should not call the LLM."
  }
]

Step 2: Create evals/run_evals.py

This runner calls your API endpoint (end-to-end) and checks expected behaviours.

import json
import requests

API_URL = "http://127.0.0.1:8000/answer"

def run():
    with open("evals/eval_set.json", "r", encoding="utf-8") as f:
        cases = json.load(f)

    passed = 0
    failed = 0

    for case in cases:
        resp = requests.post(API_URL, json={"question": case["question"]}, timeout=60)
        resp.raise_for_status()
        out = resp.json()

        got_refusal = bool(out.get("refusal", False))
        expect_refusal = bool(case["expect_refusal"])

        ok = (got_refusal == expect_refusal)

        # Beginner-friendly: if it answers, sources should exist and be a list
        if not got_refusal:
            ok = ok and isinstance(out.get("sources"), list)

        if ok:
            passed += 1
            print(f"PASS {case['id']}")
        else:
            failed += 1
            print(f"FAIL {case['id']} expected_refusal={expect_refusal} got_refusal={got_refusal}")
            print("Output:", json.dumps(out, indent=2))

    print(f"\nDone. Passed={passed} Failed={failed}")
    if failed:
        raise SystemExit(1)

if __name__ == "__main__":
    run()

How to Use Evals in Practice

Run your server:

uvicorn app:app --reload

In another terminal, run evals:

python evals/run_evals.py

If an eval fails, you have a concrete signal that something changed in retrieval, gating, prompting, or provider behaviour.

What to Improve Next: Realistic Upgrades

Building a reliable RAG app is iterative. Here are realistic next steps:

• Semantic chunking: Break text based on meaning instead of character count.

• Reranking: Use a cross-encoder reranker to reorder the top-k chunks for higher precision.

• Metadata filtering: Filter results by category, date, or department to reduce false positives.

• Better citations: Store chunk IDs and show exactly which chunk(s) the answer came from.

• Observability: Add request IDs, structured logs, and traces so “what happened?” is answerable.

• Async + background indexing: Move index building to a background job and keep the API responsive.

Final Thoughts: Production-Ready Is a Set of Habits

Building an AI application that survives in the real world is about building a system that is predictable, measurable, and safe.

• Retrieval quality is measurable: Use similarity scores to gate your LLM.

• Refusal is a feature: It is better to say “I do not know” than to lie.

• Fallbacks are mandatory: Design for the moment the API goes down.

• Evals prevent regressions: Never deploy a change without running your tests.

About Me

I am Chidozie Managwu, an award-winning AI Product Architect and founder focused on helping global tech talent build real, production-ready skills. I contribute to global AI initiatives as a GAFAI Delegate and lead AI Titans Network, a community for developers learning how to ship AI products.

My work has been recognized with the Global Tech Hero award and featured on platforms like HackerNoon.

In this tutorial, you’ll build a distributed rate limiter. This is the kind of system you need when your application is deployed across multiple servers or virtual machines, and you need to enforce a global limit on all incoming requests.

You’ll build a simple URL shortener application and then implement a robust rate limiter for it using a powerful and efficient combination of tools:

• Python and Flask for your web application.

• Redis as your high-speed, centralized data store for tracking requests.

• Terraform and Proxmox to define and provision your virtual machine infrastructure.

• Docker to containerize your application for easy deployment.

• Nginx as a load balancer to distribute traffic across your app servers.

• k6 to load-test your system and prove that your rate limiter actually works.

This is intended for new developers learning about various system design concepts or for experts who just want a refresher.

By the end of this guide, you'll understand not just the code, but the complete system architecture required to deploy a scalable, resilient application.

Let's get started!

Prerequisites

While not absolutely required to follow along, I’d recommend setting up a Proxmox server on an old laptop to implement the topics you learn and code along with the article. I recommend this YouTube playlist for getting started. Please note that I am in no way affiliated with this channel. I just found it helpful for me.

However, If you do not have a local Proxmox server, you can skip that part and just follow along to understand how a rate limiter is built and how it is set up to properly work with multiple servers.

Table of Contents

• Prerequisites

• The Big Picture: Our System Architecture

• Step 1: How to Define the Infrastructure with Terraform

• Step 2: How to Implement the Rate Limiter Logic in Python

• Step 3: Containerizing and Testing

• Conclusion

The Big Picture: Our System Architecture

Before we dive into the code, let's look at the architecture we're building. I will be using Proxmox Virtual Environment to setup a server cluster just like you would have in a datacenter.

How to Set Up Proxmox

Proxmox Virtual Environment is an open source platform for virtualization. It lets you manage multiple VMs, ccontainers and other clusters with ease. For instance, I turned my old gaming computer into a Proxmox server which lets me run more than 20 virtual machines on it at the same time, making it similar to my very own datacenter. This lets me experiment with distributed applications by simulating datacenter environments.

To setup your own cluster, all you need is an old computer. You can download the ISO image from here and boot from the USB drive. Once you install it, you can configure the host machine via a web browser on any other computer on the same network.

For example, my proxmox server is located at 10.0.0.108 and I can access it via the browser on my laptop.

We define all our virtual machines in our main.tf file. And run a simple command terraform apply to spin these servers up. For more reading on how to use Terraform with Proxmox, I recommend this blog post

Back to our use case, we’ll have a few virtual machines that will serve as different kinds of servers:

1. A Load balancer

2. A Rate Limiter ( A Redis Cache )

3. Two Web Servers

4. A Postgres database

5. One Virtual Machine that will test the load by simulating hundreds of calls per minute.

If all of this seems daunting, don’t worry too much about it. You don’t need to set all this up to follow along.

Centralized Rate Limiter

Since our application will run on multiple servers (or "nodes"), we can't store request counts in memory on each individual server. Why? Because each server would have its own separate count, and we wouldn't have a global rate limit.

The solution is to use a centralized data store that all our application nodes can access. This is where Redis comes in.

Here’s a diagram of our setup:

1. User requests first hit our Nginx load balancer.

2. The load balancer distributes the traffic evenly between our two web server VMs. The configuration is simple, using an upstream block to define the servers.

3. Each web server runs our Python Flask application inside a Docker container.

4. Before processing any request, the Flask app communicates with the central Redis rate limiter VM to check if the user has exceeded the rate limit.

5. If the user is within the limit, the app processes the request and interacts with the PostgreSQL Database. If they're over the limit, it sends back a “429 Too Many Requests” error.

This architecture ensures that no matter which web server handles the request, the rate limit is checked against the same, shared data source.

Step 1: How to Define the Infrastructure with Terraform

Manually setting up multiple virtual machines can be tedious and prone to errors. That's why we use Terraform, an Infrastructure as Code (IaC) tool. It lets us define our entire infrastructure in configuration files.

Note: You can skip this section if you just want to see the rate limiter in action and how it’s used.

Our main.tf file defines all the components of our system. Let's look at a key piece: the Redis VM.

# --- Redis Cache for Rate Limiter ---
resource "proxmox_vm_qemu" "redis_cache" {

    vmid        = 130
    name        = "redis-cache-rate-limiter"
    target_node = "pve"
    agent       = 1
    cores       = 1
    memory      = 1024
    # ... cloud-init config ...
    ipconfig0  = "ip=10.0.0.130/24,gw=10.0.0.1"
    # ... disk and network config ...

    # 1. Install Docker
    provisioner "remote-exec" {
        inline = [
            "sleep 30; sudo apt-get update -y",
            "sudo apt-get install -y docker.io docker-compose",
            "sudo mkdir -p /opt/redis"
        ]
    }

    # 2. Upload docker-compose file
    provisioner "file" {
         source      = "files/redis-docker-compose.yml"
         destination = "/home/${var.ssh_user}/docker-compose.yml"
    }

    # 3. Move file and run docker-compose
    provisioner "remote-exec" {
        inline = [
            "sudo mv /home/${var.ssh_user}/docker-compose.yml /opt/redis/docker-compose.yml",
            "cd /opt/redis && sudo docker-compose up -d"
        ]
    }
}

This block tells Terraform to create a Proxmox QEMU virtual machine with a specific IP address (10.0.0.130). After the VM is created, it uses provisioners to connect via SSH and run commands. Here, it installs Docker, uploads our redis-docker-compose.yml file, and starts the Redis container.

The redis-docker-compose.yml itself is very straightforward:

version: '3.8'
services:
  redis:
    image: redis:latest
    container_name: redis_cache
    restart: always
    ports:
      - "6379:6379"
    volumes:
      - redisdata:/data

volumes:
  redisdata:

This ensures we have a persistent, containerized Redis instance ready to serve our application. The Terraform configuration similarly defines our web servers, load balancer, and databases.

Step 2: How to Implement the Rate Limiter Logic in Python

Now, for the heart of our system: the Python code that implements the rate limiting logic. We're using a sophisticated and memory-efficient algorithm called the Sliding Window Log.

The idea is simple: for each user, we keep a log of the timestamps of their recent requests. We store this log in a Redis Sorted Set.

Let's break down the code from app.py.

The Flask @app.before_request Hook

Flask allows us to run code before any request is handled by its intended view function. This is the perfect place to put our rate limiter.

import psycopg2
import string
import random
import redis
import time
from flask import Flask, request, redirect, jsonify

app = Flask(__name__)

# --- Database Connection Details ---
DB_HOST = "10.0.0.200" 
DB_NAME = "urldb"
DB_USER = "myuser"
DB_PASS = "mypassword"

REDIS_HOST = "10.0.0.130" # IP of your redis-cache-lxc

# --- Rate Limiter Settings ---
RATE_LIMIT_COUNT = 10  # 10 requests
RATE_LIMIT_WINDOW = 60 # per 60 seconds

# Establish a reusable Redis connection
redis_client = redis.Redis(host=REDIS_HOST, port=6379, decode_responses=True)

@app.before_request
def rate_limiter():
    # Use the user's IP address as the key
    # In a real app, you'd handle proxies via request.environ.get('HTTP_X_FORWARDED_FOR', request.remote_addr)
    key = f"rate_limit:{request.remote_addr}"
    now = time.time()

    # Use a Redis pipeline for atomic operations
    pipe = redis_client.pipeline()
    # 1. Add current request timestamp. The score and member are the same.
    pipe.zadd(key, {str(now): now})
    # 2. Remove all timestamps older than our window
    pipe.zremrangebyscore(key, 0, now - RATE_LIMIT_WINDOW)
    # 3. Get the count of remaining timestamps
    pipe.zcard(key)
    # 4. Set an expiration on the key so it cleans itself up
    pipe.expire(key, RATE_LIMIT_WINDOW)

    # Execute the pipeline and get the results
    results = pipe.execute()
    request_count = results[2] # The result of the zcard command

    if request_count > RATE_LIMIT_COUNT:
        # Return a 429 Too Many Requests error
        return jsonify(error="Rate limit exceeded"), 429

def get_db_connection():
    conn = psycopg2.connect(host=DB_HOST, dbname=DB_NAME, user=DB_USER, password=DB_PASS)
    return conn

def init_db():
    conn = get_db_connection()
    cur = conn.cursor()
    cur.execute('''
        CREATE TABLE IF NOT EXISTS urls (
            id SERIAL PRIMARY KEY,
            short_code VARCHAR(6) UNIQUE NOT NULL,
            original_url TEXT NOT NULL
        );
    ''')
    # Check if the index exists before creating it
    cur.execute('''
        SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE c.relname = 'idx_original_url' AND n.nspname = 'public';
    ''')
    if cur.fetchone() is None:
        cur.execute('CREATE INDEX idx_original_url ON urls (original_url);')
    conn.commit()
    cur.close()
    conn.close()

def generate_short_code(length=6):
    chars = string.ascii_letters + string.digits
    return ''.join(random.choice(chars) for _ in range(length))

@app.route("/", methods=['GET'])
def index():
    return "URL Shortener is running!\n", 200

@app.route('/shorten', methods=['POST'])
def shorten_url():
    original_url = request.form['url']
    conn = get_db_connection()
    cur = conn.cursor()

    cur.execute("SELECT short_code FROM urls WHERE original_url = %s", (original_url,))
    existing_url = cur.fetchone()

    if existing_url:
        short_code = existing_url[0]
    else:
        short_code = generate_short_code()
        cur.execute("INSERT INTO urls (short_code, original_url) VALUES (%s, %s)", (short_code, original_url))
        conn.commit()

    cur.close()
    conn.close()

    return jsonify(short_url=f"/{short_code}")

@app.route('/<short_code>')
def redirect_to_url(short_code):
    conn = get_db_connection()
    cur = conn.cursor()
    cur.execute("SELECT original_url FROM urls WHERE short_code = %s", (short_code,))
    url_record = cur.fetchone()
    cur.close()
    conn.close()

    if url_record:
        return redirect(url_record[0])
    else:
        return "URL not found", 404

if __name__ == '__main__':
    init_db() 
    app.run(host='0.0.0.0', port=5000)

How It Works, Step-by-Step

1. Identify the User: We create a unique Redis key for each user based on their IP address: rate_limit:1.2.3.4.

2. Use a Pipeline: Network latency can be a bottleneck. A Redis pipeline bundles multiple commands into a single request-response cycle. This is much more efficient than sending them one by one. It also ensures the sequence of commands runs without being interrupted by commands from other clients.

3. Log the Current Request (ZADD): We add the current timestamp (as a Unix epoch) to a sorted set. We use the timestamp for both the "member" and the "score," which allows us to easily filter by time.

4. Clean Up Old Requests (ZREMRANGEBYSCORE): This is the "sliding window" part. We remove any timestamps from the set that are older than our RATE_LIMIT_WINDOW (60 seconds). This efficiently discards requests that are no longer relevant to the current rate limit period.

5. Count the Recent Requests (ZCARD): We get the cardinality (the number of items) in the set. After the previous step, this number is our count of requests within the last 60 seconds.

6. Mark the current record to expire (EXPIRE): We set an expiration on the key itself. If a user stops making requests, Redis will automatically delete their rate limit data after 60 seconds, preventing memory from filling up with old keys.

7. Execute and Check: The pipe.execute() command sends all our bundled commands to Redis. We then check the result of our ZCARD command. If the count exceeds our RATE_LIMIT_COUNT, we immediately return a 429 error.

This approach is incredibly fast and efficient. All the heavy lifting is done inside Redis, which is optimized for these kinds of operations.

Step 3: Containerizing and Testing

To deploy our application consistently across multiple VMs, we use Docker. Our Dockerfile is standard for a Python application: it starts from a Python image, installs dependencies from requirements.txt, copies the application code, and defines the command to run the app.

But how do we know it works? We test it!

We use k6, a modern load testing tool, to simulate heavy traffic. Our test script, rate-test.js, is designed specifically to verify the rate limiter.

import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
  stages: [
    // Ramp up to 20 users. This is more than the 10 req/min limit
    // and should trigger the rate limiter.
    { duration: '30s', target: 20 },
    { duration: '1m', target: 20 },
    { duration: '10s', target: 0 },
  ],
};

export default function () {
  const url = 'http://10.0.0.100/shorten'; // The Load Balancer IP
  const payload = { url: `https://www.test-ratelimit-${Math.random()}.com` };

  const res = http.post(url, payload);

  // Check if the request was successful OR if it was correctly rate-limited
  check(res, {
    'status is 200 (OK)': (r) => r.status === 200,
    'status is 429 (Too Many Requests)': (r) => r.status === 429,
  });

  sleep(1);
}

The stages array configures the test to gradually increase the number of virtual users to 20. Since our rate limit is 10 requests per minute, this load is guaranteed to trigger the limiter.

The check function is the crucial part. It verifies that the server's response code is either 200 (meaning the request was successful) or 429 (meaning our rate limiter correctly blocked the request).

We should see about 10 of our requests go through of the around 1600 requests per minute that we send from the same IP address.

We can also check the logs on our webserver to see all the requests that were sent to it.

And if we look at the Redis cache/database itself, we’ll see all the keys and the TTL at which they expire.

This is how we rate limit applications using a Redis Cache Server.

Here are the complete files used in the project.

    terraform {
    required_providers {
        proxmox = {
        source  = "telmate/proxmox"
        version = "3.0.2-rc04"
        }
    }
    }

    provider "proxmox" {
    pm_api_url          = var.proxmox_api_url
    pm_api_token_id     = var.proxmox_api_token_id
    pm_api_token_secret = var.proxmox_api_token_secret
    pm_tls_insecure     = true
    }

    # --- Shared Provisioner Connection Settings ---
    locals {
        connection_settings = {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
        }
    }

    # --- Database LXC Containers ---
    resource "proxmox_lxc" "postgres_db" {
    hostname     = "postgres-db-lxc"
    target_node  = var.target_node
    ostemplate   = var.lxc_template

    rootfs {
        storage = "local-lvm"
        size = "8G"
    }

    password     = "admin"
    unprivileged = true
    start        = true

    features {
        nesting = true
        # keyctl = true
    }

    network {
        name   = "eth0"
        bridge = "vmbr0"
        ip     = "10.0.0.200/24"
        gw     = "10.0.0.1"
    }

    provisioner "remote-exec" {
        connection {
        type        = "ssh"
        user        = var.ssh_user
        private_key = file(var.ssh_private_key_path)
        host        = split("/", self.network[0].ip)[0]
        }
        inline = [
        "sudo apt-get update",
        "sudo apt-get install -y docker.io docker-compose python3-setuptools",
        "sudo usermod -aG docker ${var.ssh_user}",
        "sudo mkdir -p /opt/postgres",
        "sudo chown ${var.ssh_user}:${var.ssh_user} /opt/postgres"
        ]
    }

    provisioner "file" {
        connection {
        type        = "ssh"
        user        = var.ssh_user
        private_key = file(var.ssh_private_key_path)
        host        = split("/", self.network[0].ip)[0]
        }
        source      = "../databases/pg-docker-compose.yml"
        destination = "/opt/postgres/docker-compose.yml"
    }

    provisioner "remote-exec" {
        inline     = ["cd /opt/postgres && sudo docker-compose up -d"]

        connection {
        type        = "ssh"
        user        = var.ssh_user
        private_key = file(var.ssh_private_key_path)
        host        = split("/", self.network[0].ip)[0]
        }
    }
    }

    resource "proxmox_lxc" "mongo_db" {
        hostname    = "mongo-db-lxc"
        target_node = var.target_node
        ostemplate  = var.lxc_template

        rootfs {
            storage = "local-lvm"
            size = "8G"
        }

        password    = "admin"
        unprivileged = true
        start       = true

        features {
            nesting = true
        # keyctl = true # Somehow this is blocking the apply command
        }

        network {
            name   = "eth0"
            bridge = "vmbr0"
            ip     = "10.0.0.210/24"
            gw     = "10.0.0.1"
        }

        # Provisioners similar to postgres_db
        provisioner "remote-exec" {
            connection {
                type        = "ssh"
                user        = var.ssh_user
                private_key = file(var.ssh_private_key_path)
                host        = split("/", self.network[0].ip)[0]
            }
            inline = [
            "sudo apt-get update",
            "sudo apt-get install -y docker.io docker-compose python3-setuptools",
            "sudo usermod -aG docker ${var.ssh_user}",
            "sudo mkdir -p /opt/mongo",
            "sudo chown ${var.ssh_user}:${var.ssh_user} /opt/mongo"
            ]
        }

        provisioner "file" {
            connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = split("/", self.network[0].ip)[0]
            }
            source      = "../databases/mongo-docker-compose.yml"
            destination = "/opt/mongo/docker-compose.yml"
        }

        provisioner "remote-exec" {
            connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = split("/", self.network[0].ip)[0]
            }
            inline     = ["cd /opt/mongo && docker-compose up -d"]
        }
    }

    # --- Redis Cache for Rate Limiter ---
    resource "proxmox_vm_qemu" "redis_cache" {

        vmid        = 130
        name        = "redis-cache-rate-limiter"
        target_node = "pve"
        agent       = 1
        cpu {
            cores       = 1
        }

        memory      = 1024
        boot        = "order=scsi0" # has to be the same as the OS disk of the template
        clone       = "debian12-cloudinit" # The name of the template
        scsihw      = "virtio-scsi-single"
        vm_state    = "running"
        automatic_reboot = true

        # Cloud-Init configuration
        cicustom   = "vendor=local:snippets/qemu-guest-agent.yml" # /var/lib/vz/snippets/qemu-guest-agent.yml
        ciupgrade  = true
        nameserver = "1.1.1.1 8.8.8.8"
        ipconfig0  = "ip=10.0.0.130/24,gw=10.0.0.1"
        skip_ipv6  = true
        ciuser     = var.ssh_user
        cipassword = var.ssh_password
        sshkeys    = var.ssh_key

        # Most cloud-init images require a serial device for their display
        serial {
            id = 0
        }

        disks {
            scsi {
            scsi0 {
                # We have to specify the disk from our template, else Terraform will think it's not supposed to be there
                disk {
                storage = "local-lvm"
                # The size of the disk should be at least as big as the disk in the template. If it's smaller, the disk will be recreated
                size    = "5G" 
                }
            }
            }
            ide {
            # Some images require a cloud-init disk on the IDE controller, others on the SCSI or SATA controller
            ide1 {
                cloudinit {
                storage = "local-lvm"
                }
            }
            }
        }

        network {
            id = 0
            bridge = "vmbr0"
            model  = "virtio"
        }

        connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = "10.0.0.130"
        }

        # 1. Install Docker and create the final app directory
        provisioner "remote-exec" {
            inline = [
                # Wait for cloud-init to finish before doing anything else
                "echo 'Waiting for cloud-init to finish...'",
                "while [ ! -f /var/lib/cloud/instance/boot-finished ]; do echo 'Still waiting...' && sleep 1; done",
                "echo 'Cloud-init finished.'",

                # Now, safely install packages
                "sudo apt-get update -y",
                "sudo apt-get install -y docker.io docker-compose",
                "sudo mkdir -p /opt/redis",
            ]
        }

        provisioner "file" {
            source      = "../caching/redis-docker-compose.yml"
            destination = "/home/${var.ssh_user}/docker-compose.yml"
        }

        provisioner "remote-exec" {
            inline = [ "sudo mv /home/${var.ssh_user}/docker-compose.yml /opt/redis/docker-compose.yml" ]
        }

        provisioner "remote-exec" {
            inline = [ "cd /opt/redis && sudo docker-compose up -d" ]
        }
    }

    resource "proxmox_vm_qemu" "web-servers" {

        count = 2

        vmid        = count.index + 150
        name        = "web-server-tf-${count.index + 1}"
        target_node = "pve"
        agent       = 1
        cpu {
            cores       = 1
        }
        memory      = 1024
        boot        = "order=scsi0" # has to be the same as the OS disk of the template
        clone       = "debian12-cloudinit" # The name of the template
        scsihw      = "virtio-scsi-single"
        vm_state    = "running"
        automatic_reboot = true

        # Cloud-Init configuration
        cicustom   = "vendor=local:snippets/qemu-guest-agent.yml" # /var/lib/vz/snippets/qemu-guest-agent.yml
        ciupgrade  = true
        nameserver = "1.1.1.1 8.8.8.8"
        ipconfig0  = "ip=10.0.0.${111 + count.index}/24,gw=10.0.0.1"
        skip_ipv6  = true
        ciuser     = var.ssh_user
        cipassword = var.ssh_password
        sshkeys    = var.ssh_key

        # Most cloud-init images require a serial device for their display
        serial {
            id = 0
        }

        disks {
            scsi {
            scsi0 {
                # We have to specify the disk from our template, else Terraform will think it's not supposed to be there
                disk {
                storage = "local-lvm"
                # The size of the disk should be at least as big as the disk in the template. If it's smaller, the disk will be recreated
                size    = "5G" 
                }
            }
            }
            ide {
            # Some images require a cloud-init disk on the IDE controller, others on the SCSI or SATA controller
            ide1 {
                cloudinit {
                storage = "local-lvm"
                }
            }
            }
        }

        network {
            id = 0
            bridge = "vmbr0"
            model  = "virtio"
        }

        connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = "10.0.0.${111 + count.index}"
        }

        # 1. Install Docker and create the final app directory
        provisioner "remote-exec" {
            inline = [
                # Wait for cloud-init to finish before doing anything else
                "echo 'Waiting for cloud-init to finish...'",
                "while [ ! -f /var/lib/cloud/instance/boot-finished ]; do echo 'Still waiting...' && sleep 1; done",
                "echo 'Cloud-init finished.'",

                # Now, safely install packages
                "sudo apt-get update -y",
                "sudo apt-get install -y docker.io",
                "sudo mkdir -p /opt/app",
            ]
        }

        # 2. Upload ONLY the necessary files to the user's home directory
        provisioner "file" {
            source      = "../web-servers/app.py"
            destination = "/home/${var.ssh_user}/app.py"
        }
        provisioner "file" {
            source      = "../web-servers/Dockerfile"
            destination = "/home/${var.ssh_user}/Dockerfile"
        }
        provisioner "file" {
            source      = "../web-servers/requirements.txt"
            destination = "/home/${var.ssh_user}/requirements.txt"
        }

        # 4. Move files from the home directory, build the image, and run the container
        provisioner "remote-exec" {
            inline = [
                # Move each file individually to be compatible with all shells
                "sudo mv /home/${var.ssh_user}/app.py /opt/app/",
                "sudo mv /home/${var.ssh_user}/Dockerfile /opt/app/",
                "sudo mv /home/${var.ssh_user}/requirements.txt /opt/app/",

                # Build the Docker image
                "sudo docker build -t my-python-app /opt/app",

                # Stop and remove any old containers to prevent conflicts
                "sudo docker stop $(sudo docker ps -q --filter ancestor=my-python-app) 2>/dev/null || true",
                "sudo docker rm $(sudo docker ps -aq --filter ancestor=my-python-app) 2>/dev/null || true",

                # Run the new container
                "sudo docker run -d --restart always -p 80:5000 my-python-app"
            ]
        }

        # In your proxmox_vm_qemu "web_servers" resource
        depends_on = [
            proxmox_lxc.postgres_db,
            proxmox_vm_qemu.redis_cache
        ]
    }

    # --- Load Balancer VM ---
    resource "proxmox_vm_qemu" "load_balancer" {
        name        = "lb-1"
        target_node = var.target_node
        clone       = var.vm_template
        agent       = 1
        cpu {
            cores       = 1
        }
        memory      = 512
        boot        = "order=scsi0" # has to be the same as the OS disk of the template
        scsihw      = "virtio-scsi-single"
        vm_state    = "running"
        automatic_reboot = true

        # --- Add these lines for Cloud Init Drive ---
                # --- Add these lines for Cloud Init Drive ---
        cicustom   = "vendor=local:snippets/qemu-guest-agent.yml" # /var/lib/vz/snippets/qemu-guest-agent.yml
        ciupgrade  = true
        nameserver = "1.1.1.1 8.8.8.8"
        ipconfig0  = "ip=10.0.0.100/24,gw=10.0.0.1"
        skip_ipv6  = true
        ciuser     = var.ssh_user
        cipassword = var.ssh_password
        sshkeys    = var.ssh_key

        # Most cloud-init images require a serial device for their display
        serial {
            id = 0
        }

        disks {
            scsi {
            scsi0 {
                # We have to specify the disk from our template, else Terraform will think it's not supposed to be there
                disk {
                storage = "local-lvm"
                # The size of the disk should be at least as big as the disk in the template. If it's smaller, the disk will be recreated
                size    = "5G" 
                }
            }
            }
            ide {
            # Some images require a cloud-init disk on the IDE controller, others on the SCSI or SATA controller
            ide1 {
                cloudinit {
                storage = "local-lvm"
                }
            }
            }
        }

        network {
            id = 0
            bridge = "vmbr0"
            model  = "virtio"
        }

        connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = "10.0.0.100"
        }

        # Step 1: Install Nginx
        provisioner "remote-exec" {
            inline = [
                # Wait for cloud-init to finish before doing anything else
                "echo 'Waiting for cloud-init to finish...'",
                "while [ ! -f /var/lib/cloud/instance/boot-finished ]; do echo 'Still waiting...' && sleep 1; done",
                "echo 'Cloud-init finished.'",

                # Now, safely install packages
                "sudo apt-get update -y",
                "sudo apt-get install -y nginx"
            ]
        }

        # Step 2: Upload config to a temporary location
        provisioner "file" {
            source      = "../web-servers/nginx.conf"
            destination = "/tmp/nginx.conf" # Use /tmp instead
        }

        # Step 3: Use sudo to move the file to its final destination and reload nginx
        provisioner "remote-exec" {
            inline = [
                "sudo mv /tmp/nginx.conf /etc/nginx/sites-available/default",
                "sudo systemctl reload nginx"
            ]
        }
    }


    # --- Load Tester VM ---
    resource "proxmox_vm_qemu" "load_tester" {
        name        = "load-tester-vm"
        target_node = var.target_node
        clone       = var.vm_template
        agent       = 1
        cpu {
            cores       = 1
        }
        memory      = 1024
        boot        = "order=scsi0" # has to be the same as the OS disk of the template
        scsihw      = "virtio-scsi-single"
        vm_state    = "running"
        automatic_reboot = true

        # --- Add these lines for Cloud Init Drive ---
        cicustom   = "vendor=local:snippets/qemu-guest-agent.yml" # /var/lib/vz/snippets/qemu-guest-agent.yml
        ciupgrade  = true
        nameserver = "1.1.1.1 8.8.8.8"
        ipconfig0  = "ip=10.0.0.160/24,gw=10.0.0.1"
        skip_ipv6  = true
        ciuser     = var.ssh_user
        cipassword = var.ssh_password
        sshkeys    = var.ssh_key

        # Most cloud-init images require a serial device for their display
        serial {
            id = 0
        }

        disks {
            scsi {
                scsi0 {
                    # We have to specify the disk from our template, else Terraform will think it's not supposed to be there
                    disk {
                    storage = "local-lvm"
                    # The size of the disk should be at least as big as the disk in the template. If it's smaller, the disk will be recreated
                    size    = "5G" 
                    }
                }
            }

            ide {
            # Some images require a cloud-init disk on the IDE controller, others on the SCSI or SATA controller
                ide1 {
                    cloudinit {
                    storage = "local-lvm"
                    }
                }
            }
        }

        network {
            id = 0
            bridge = "vmbr0"
            model  = "virtio"
        }

        provisioner "remote-exec" {
            connection {
                type        = "ssh"
                user        = var.ssh_user
                private_key = file(var.ssh_private_key_path)
                host        = "10.0.0.160"
            }
            inline = [
                # Wait for cloud-init to finish
                "echo 'Waiting for cloud-init to finish...'",
                "while [ ! -f /var/lib/cloud/instance/boot-finished ]; do echo 'Still waiting...' && sleep 1; done",
                "echo 'Cloud-init finished.'",

                # Install prerequisites
                "sudo apt-get update -y",
                "sudo apt-get install -y gnupg curl",

                # Add the k6 repository and key
                "curl -sL https://dl.k6.io/key.gpg | sudo gpg --dearmor -o /usr/share/keyrings/k6-archive-keyring.gpg",
                "echo 'deb [signed-by=/usr/share/keyrings/k6-archive-keyring.gpg] https://dl.k6.io/deb stable main' | sudo tee /etc/apt/sources.list.d/k6.list",

                # Install k6
                "sudo apt-get update",
                "sudo apt-get install -y k6"
            ]
        }

        provisioner "file" {
            connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = "10.0.0.160"
            }
            source      = "../load-testing/script.js"
            destination = "/home/${var.ssh_user}/script.js"
        }

        provisioner "file" {
            connection {
            type        = "ssh"
            user        = var.ssh_user
            private_key = file(var.ssh_private_key_path)
            host        = "10.0.0.160"
            }
            source      = "../load-testing/rate-test.js"
            destination = "/home/${var.ssh_user}/rate-test.js"
        }

    }

Conclusion

You've now seen how to build a complete, scalable, and resilient system that includes a crucial component for modern web applications: a distributed rate limiter.

We've covered the entire stack:

• Infrastructure as Code with Terraform to define our virtual machines. (check out my repo here for all the code and any updates I make).

• A centralized, high-speed cache with Redis to store our rate limiting data.

• An efficient Sliding Window Log algorithm implemented in Python with Flask.

• Containerization with Docker for consistent deployment.

• Load balancing with Nginx to distribute traffic.

• Load testing with k6 to validate our implementation.

If you’d like to learn more of the concepts that are used when building large scale systems please follow me at Sravan Karuturi.

1. The traditional SMTP setup with the built-in ‘smtplib’ module.

2. Mailtrap email API via Mailtrap’s official SDK.

If you’re unfamiliar with the tools and workflows, SMTP (Simple Mail Transfer Protocol) is the protocol commonly used for sending emails via apps and websites. Mailtrap is an email delivery platform designed for high deliverability with growth-focused features and industry-best analytics.

By the end of the article, you’ll understand how to integrate email-sending capabilities into Python projects and use Mailtrap for reliable email delivery in real-world scenarios.

Table of Contents

1. 'smtplib' Setup

2. How to Send emails with Mailtrap SMTP

3. How to Send emails with the Mailtrap Email API

4. Wrapping Up

‘smtplib’ Setup

To start sending emails with Python, I'll first use the built-in ‘smtplib’ module. This lets you connect to an SMTP server and send emails directly from your app.

So, start by importing the ‘smtplib’ module with the statement below:

import smtplib

Next, create an ‘SMTP’ object to configure the connection to your SMTP server. This object handles the email sending.

smtpObj = smtplib.SMTP(host, port)

• ‘host’ refers to the SMTP server endpoint, such as ‘live.smtp.mailtrap.io’

• ‘port’ is the communication channel used by the server. The recommended port is usually 587 for secure email sending with TLS encryption.

Pro tip: An SMTP object has a ‘sendmail’ instance object with three parameters, where each parameter is a string (‘receivers’ is a list of strings).

smtpObj.sendmail(sender, receivers, message)

If you want to ensure you’ve properly imported the ‘smtplib’ module and check the full description of arguments and classes, run the following command:

help(smtplib)

How to Send emails with Mailtrap SMTP

This method involves setting up the custom SMTP credentials you get for Mailtrap.

Important notes:

• Testing out the service with Mailtrap’s dummy domain – To try Mailtrap, you don’t need to verify your domain right away. You can use Mailtrap’s dummy domain (you get access to it when you sign up), which allows you to simulate sending emails without worrying about the DNS records. This is ideal for testing the service and getting familiar with Mailtrap’s features.

• Domain verification for production – If you plan to send real emails to recipients, you’ll need to verify your domain. This involves adding DNS records such as SPF, DKIM, and DMARC to your domain provider’s DNS settings. These records ensure your emails are delivered successfully and help protect against phishing and spoofing. In the next section, I'll show you how to set these up in your domain provider's dashboard.

Verify your sending domain (SPF, DKIM, and DMARC)

DNS records are critical to ensure your emails are delivered successfully, and mailbox providers such as Gmail and Yahoo require DNS authentication.

But before we go through a quick tutorial on how to do it, let’s review each type of record so you understand why they’re so important:

• SPF (Sender Policy Framework): The record helps mail servers determine if the sender’s IP address is authorized to send emails from your domain. Simply, adding an SPF record prevents spammers from sending emails that appear to come from your domain.

• DKIM (DomainKeys Identified Mail): DKIM uses encryption to verify the sender's domain and ensures that the email content hasn't been tampered with during transmission. This protects your emails from being spoofed.

• DMARC (Domain-based Message Authentication, Reporting & Conformance): DMARC ties SPF and DKIM together, providing a policy for handling unauthenticated emails and reporting on email activities. In a nutshell, it gives you more control over your domain’s email security.

Now, here’s how to add the records:

1. First, you need to access your domain provider's DNS settings. Usually, you can access them in the domain register or domain settings. For example, GoDaddy calls the menu Manage DNS, and it's dubbed similarly with other providers.

2. Next, add (copy-paste) the DNS records Mailtrap provides into your domain provider's DNS settings. Note that Mailtrap's records are read-made, and SPF is pre-parsed, so you don't need to create anything additional – just add the records.

3. Finally, you can check the status of your records with Mailtrap.

Below is the bare-bones script for sending emails via Mailtrap using Python. For security reasons, the script uses placeholder credentials for the username and password (except for the SMTP server endpoint and port).

When running the script, be sure to replace these placeholders with your actual Mailtrap credentials to ensure the email is sent successfully.

import smtplib
from email.mime.text import MIMEText

# Configuration
port = 587
smtp_server = "live.smtp.mailtrap.io"
login = "api"  # Your login generated by Mailtrap
password = "1a2b3c4d5e6f7g"  # Your password generated by Mailtrap

sender_email = "mailtrap@example.com"
receiver_email = "new@example.com"

# Plain text content
text = """\
Hi,
Check out the new post on the Mailtrap blog:
SMTP Server for Testing: Cloud-based or Local?
https://blog.mailtrap.io/2018/09/27/cloud-or-local-smtp-server/
Feel free to let us know what content would be useful for you!
"""

# Create MIMEText object
message = MIMEText(text, "plain")
message["Subject"] = "Plain text email"
message["From"] = sender_email
message["To"] = receiver_email

# Send the email
with smtplib.SMTP(smtp_server, port) as server:
    server.starttls()  # Secure the connection
    server.login(login, password)
    server.sendmail(sender_email, receiver_email, message.as_string())

print('Sent')

In the script:

• The ‘smtplib’ and ‘MIMEText’ modules have been imported from Python’s library.

• As mentioned, SMTP server configuration needs to be updated with your credentials. But the server endpoint and port are as is.

• Since this is a bare-bones script, I used ‘MIMEText’, which holds ‘plaintext’ only. But the script can be easily refactored to use ‘MIMEMultipart’ for both ‘plaintext’ and ‘HTML’. Jump to the quick tut below to see how it’s done.

• When sending the email, I chose to use the ‘with’ statement (context manager) to ensure the SMTP server connection gets closed right after the email gets sent.

Security tip:

Server information and the login credentials shouldn't be hardcoded into your sending script. When setting the script for production, make sure you use environment variables to store sensitive information. This makes the code more secure and more flexible, particularly when you move it between different dev stages. For example ⬇️

import os

smtp_server = os.getenv("SMTP_SERVER", "default.smtp.server")
login = os.getenv("SMTP_LOGIN")
password = os.getenv("SMTP_PASSWORD")

# Example usage in an SMTP connection setup
# smtp.login(login, password)

Note that you need to set the variables in your operating system prior to running the script.

Refactor the script to use HTML emails

HTML emails provide a better user experience. They allow you to include formatted text, images, tables, clickable links, and custom styling. This works great for marketing emails, newsletters, or any communication where design and branding matter.

So, to refactor the script, you would import ‘MIMEMultipart’ and ‘MIMEText’. This action allows you to customize the HTML emails yet keep the plain-text versions as a fallback if your recipients cannot open the HTML email.

Here’s the revised script:

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

# Configuration
smtp_server = "live.smtp.mailtrap.io"
port = 587
login = "api"  # Mailtrap login
password = "1a2b3c4d5e6f7g"  # Mailtrap password

sender_email = "mailtrap@example.com"
receiver_email = "new@example.com"

message = MIMEMultipart()
message["From"] = sender_email
message["To"] = receiver_email
message["Subject"] = "HTML Email"

# Add plain text content (optional, for email clients that don't render HTML)
message.attach(MIMEText("This is a plain text version of the email.", "plain"))

# Add HTML content
html_content = """\
<html>
  <body>
    <h1>Welcome to Mailtrap!</h1>
    <p>This is an example of an HTML email.</p>
  </body>
</html>
"""
message.attach(MIMEText(html_content, "html"))

# Send the email
with smtplib.SMTP(smtp_server, port) as server:
    server.starttls()
    server.login(login, password)
    server.sendmail(sender_email, receiver_email, message.as_string())

print('Sent')

Lastly, I’ve included video instructions for the SMTP method – so if that works better for you, feel free to check it out 🔽.

How to send email in Python using Mailtrap - Tutorial by Mailtrap

How to Send emails with the Mailtrap email API

If you're looking to move beyond using SMTP for sending emails and want to integrate Mailtrap’s email API into your Python applications, this section will walk you through how to do that.

The Mailtrap SMTP email API allows you to send emails more efficiently, with added flexibility and scalability. Before starting, make sure you have a verified sending domain on Mailtrap and the Mailtrap API token, which you’ll use to authenticate requests.

Note: I’m covering the API integration using the official Mailtrap Python SDK.

So, first you install the official SDK with the command below.

pip install mailtrap

Prerequisite: Ensure your Python package version is 3.6+ or higher.

After installing the SDK, the next step is to create a Mail object. This object will represent the email you want to send, including essential details like the sender, recipient, subject, and email content.

import mailtrap as mt

# Create the mail object
mail = mt.Mail(
    sender=mt.Address(email="mailtrap@example.com", name="Mailtrap Test"),  # Sender info
    to=[mt.Address(email="your@email.com")],  # Recipient info
    subject="You are awesome!",  # Email subject
    text="Congrats for sending a test email with Mailtrap!"  # Email content (plain text)
)

# Create a client using your API key
client = mt.MailtrapClient(token="your-api-key")

# Send the email
client.send(mail)

Quick notes:

• Sender and recipient: You need to specify the sender’s email address, which must match your verified domain. Similarly, define the recipient's email.

• Subject and text content: Set the subject and plain text content of the email. You can also add HTML content as I'll cover later.

• Client and sending: The ‘MailtrapClient’ is initialized with your Mailtrap API token, which authenticates the API request. The ‘send’ method is then called on the client, passing the ‘mail’ object.

To create the client using the Mailtrap API token, take the following path within Mailtrap:
Settings > API Tokens > Add Token

With that, you can use the following command to send emails:

# create client and send
client = mt.MailtrapClient(token="your-api-key")
client.send(mail)

Finally, here’s the SDK script for sending a bare-bones ‘plaintext’ email via Python SDK.

 from mailtrap import Mail, Address, MailtrapClient

# Create a Mail object with basic details for a plain text email
mail = Mail(
    # Specify the sender's email address and optional name
    sender=Address(email="mailtrap@example.com", name="Mailtrap Test"),
    # Specify one or more recipients; here we use a list with a single recipient
    to=[Address(email="your@email.com", name="Your Name")],
    # Subject of the email
    subject="Simple Plain Text Email",
    # The plain text content of the email
    text="This is a plain text email sent using the Mailtrap SDK. Simple and straightforward.",
    # Optional: categorize this email for easier sorting or management in the Mailtrap service
    category="Test",
    # Optional: Additional headers can be specified, but are not required for plain text emails
    headers={"X-Example-Header": "HeaderValue"}
)

# Initialize the MailtrapClient with your API token
client = MailtrapClient(token="your-api-key")

# Send the email using the client's send method
client.send(mail)

print("Plain text email sent successfully.")

In the script:

• The imported classes include ‘MailtrapClient’, ‘Mail’, and ‘Address’ because I’m sending a plain text message.

• The ‘Mail’ object contains:

  • ‘Mail’ constructor to create the object.

  • ‘Sender’ which uses ‘Address’ class to define the name and email of the sender.

  • ‘to’ which is typically an ‘Address’ objects list, but since this is a plain text email, it usually has direct recipients instead of the list.

  • ‘subject’ which is the subject of the email.

  • ‘text’ which contains the email content (in ‘plaintext’)

  • ‘headers’ and ‘category’ which are optional fields that help better manage your emails.

• The email sending flow:

  • ‘MailtrapClient’ gets created and authenticated via the API token.

  • The ‘MailtrapClient’ ‘send’ method gets called and passes the ‘mail’ object as an email-sending argument.

  • The “Plain text email sent successfully.” message gets printed to confirm the action.

Refactor the script to include HTML and attachments

Again, it’s pretty straightforward to refactor the script using the ‘MIMEMultipart’ class for more complex email structures.

Here’s the refactored code:

import mailtrap as mt
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

# Create a multipart email message
message = MIMEMultipart()
message["Subject"] = "HTML Email"

# Plain text version (for email clients that don't support HTML)
message.attach(MIMEText("This is the plain text version.", "plain"))

# HTML version
html_content = """\
<html>
  <body>
    <h1>Welcome to Mailtrap!</h1>
    <p>This is an HTML email with some <b>bold text</b> and a <a href="https://example.com">link</a>.</p>
  </body>
</html>
"""
message.attach(MIMEText(html_content, "html"))

client = mt.MailtrapClient(token="your-api-key")

# Now send the email with Mailtrap's API
mail = mt.Mail(
    sender=mt.Address(email="mailtrap@example.com", name="Mailtrap Test"),
    to=[mt.Address(email="your@email.com")],
    subject="You are awesome!",
    html=message.as_string()  # Pass the HTML content as a string
)
client.send(mail)

Environmental setup for production

Before I dive into the details, I’d like to remind you of security best practices:

1. Securely store API keys and credentials: On production, never hardcode sensitive data like API keys, email login credentials, or other secrets directly into your source code. Doing so exposes your application.

2. Use environment variables: By doing this, you can keep your credentials safe and easily switch between different configurations (like dev, staging, and production).

Now, here’s how to set it all up:

1. Use the ‘python-dotenv’ package to load environment variables from a ‘.env’ file. Install the lib with the following command:

    pip install python-dotenv
   

2. Create a ‘.env’ file in the root of your project to store your environment variables securely. This file will contain sensitive information, such as your Mailtrap API key, login credentials, and SMTP server details. Here’s an example:

    SMTP_SERVER=smtp.mailtrap.io
    SMTP_PORT=587
    SMTP_LOGIN=your_mailtrap_login
    SMTP_PASSWORD=your_mailtrap_password
    MAILTRAP_API_KEY=your_mailtrap_api_key
   

Important note: Ensure this ‘.env’ file is never pushed to version control (like Git). Add it to your ‘.gitignore’ to avoid accidental exposure.

3. Once you've created your ‘.env’ file, you need to load the variables into your Python script. At the top of your script, import the ‘dotenv’ package and call ‘load_dotenv()’ to load the environment variables.

    from dotenv import load_dotenv
    import os
   
    # Load environment variables from the .env file
    load_dotenv()
   
    # Retrieve environment variables securely
    smtp_server = os.getenv("SMTP_SERVER")
    smtp_port = os.getenv("SMTP_PORT")
    smtp_login = os.getenv("SMTP_LOGIN")
    smtp_password = os.getenv("SMTP_PASSWORD")
    mailtrap_api_key = os.getenv("MAILTRAP_API_KEY")
   

4. With the environment variables loaded, you can replace the hardcoded credentials in the script with these environment variables. Here’s an example:

    import smtplib
    from email.mime.text import MIMEText
    from dotenv import load_dotenv
    import os
   
    # Load environment variables
    load_dotenv()
   
    # Fetching SMTP credentials from environment variables
    smtp_server = os.getenv("SMTP_SERVER")
    smtp_port = os.getenv("SMTP_PORT")
    smtp_login = os.getenv("SMTP_LOGIN")
    smtp_password = os.getenv("SMTP_PASSWORD")
   
    sender_email = "mailtrap@example.com"
    receiver_email = "new@example.com"
    subject = "Plain text email"
    text = """\
    Hi,
    Check out the new post on the Mailtrap blog:
    https://blog.mailtrap.io/2018/09/27/cloud-or-local-smtp-server/
    """
   
    # Create MIMEText object
    message = MIMEText(text, "plain")
    message["Subject"] = subject
    message["From"] = sender_email
    message["To"] = receiver_email
   
    # Send email using environment variables
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()  # Secure the connection
        server.login(smtp_login, smtp_password)
        server.sendmail(sender_email, receiver_email, message.as_string())
   
    print("Email sent successfully!")
   

Pro tips:

First, ensure your environment variables are only accessible to authorized users. On a production server, this typically means only allowing access to the environment variables through the deployment configuration (for example, through Heroku’s config vars, AWS Secrets Manager, or other cloud-based secret management tools).

Second, use different environment variables for development, staging, and production. This ensures that your production environment is isolated and secured from the rest of your development process.

Once your environment variables are configured locally, deploy your application to a production environment. Make sure to set the same environment variables in your production server or service.

If you're deploying to platforms like Heroku, AWS, or Google Cloud, you can use their environment variable management tools to securely store and access your secrets without having to manage a ‘.env’ file manually.

Wrapping up

This quick tutorial provides more than enough to get started with sending emails in Python. And note that the scripts featured above can be extended to include HTML, multiple recipients, attachments, images, and so on.

If you’re interested in that and more security tips and best practices, you can check out the Mailtrap blog for more detailed tutorials.

Some of the benefits include its use by developers to restrict malicious access to websites, prevent DDoS attacks, conserve website resources, and ensure optimal web server performance.

This article covers the practical aspects of implementing rate limits in a Strapi application using several packages and techniques.

Let's get started.

Table of Contents

• Demo Project

• Koa Rate Limiter

• Custom Strapi Api Rate Limiter

• Express-rate-limiter Implementation

• Conclusion

Demo Project

We'll be building an e-commerce site using Strapi as our backend framework. We'll then set up a rate limiter in our Strapi application to help guarantee our backend security. Postman will serve as our tool for testing the API endpoints. Let's go on to create a default Strapi application.

To create a strapi application, enter npx create-strapi-app@latest {project name} on the command line and follow the commands provided. To make the installation more straightforward, stick with the quick start installation method and your app should be ready.

This installation modality automatically sets up an easy-to-use SQLite database. However, you could choose to use any other SQL database supported by Strapi.

Alternatively, you can download the starter repo for the project from here and install the necessary dependencies via npm install. Thereafter, you can execute the Strapi application by navigating to the Strapi application code folder on the command line and run npm run develop.

On successful execution, you will be provided with the link to the localhost address to customize the application.

Navigating to the link will require you to create an admin login mail and password. Successful completion of this step will give you access to the backend dashboard.

You can utilize the Strapi dashboard UI to create APIs, or you can generate an API using npm generate. The APIs created will be used in completing the setup for the rate limiting functionality. We will be creating a product store for our e-commerce site. To easily set up products, kindly navigate to the Content-Type builder tab on the sidebar.

The content-Type builder manager allows you to create various collections which will come in handy when setting up your APIs. In this case, the product and category collections will be created to enable you set up your product catalogues.

After completing the creation of the collection types, you can easily add your products seamlessly into the backend database. In my case, I created phone brand products for sale.

Also noteworthy is that the collections we created in the Strapi dashboard automatically creates an API folder for us within our codebase. We will then be working on the project codebase subsequently.

The next step in this tutorial is to set up an efficient rate limiter for our Strapi APIs created in the repo using the tools discussed above.

koa2-rate-limit

In this section, we will be using the koa2-rate-limit package to build our project rate limiter. To install the package, navigate to your project folder on the command line and execute npm i koa2-rate-limit. On successful installation, navigate to the middleware subfolder within the API folder and create a code file. For ease of integration, name it as rateLimit.js.

After that, within the rate limit file, import and initialize the koa2-rate limit package.

const RateLimit = require("koa2-ratelimit").RateLimit;

Afterwards, we can configure the koa rate limiter to a specified time interval frame and the total number of requests.

module.exports = (config, { strapi }) => {
  // Configuring the rate limiter middleware
  const limiter = RateLimit.middleware({
    interval: { min: 1 }, // Time window in minutes
    max: 3, // Maximum number of requests per interval
 });

In the code above, the rate limiter middleware was invoked and the time interval in which the rate limit gets applied was set to 1 minute. The maximum number of requests (max) was set to 3 for this tutorial. You can tweak this to suit your preference.

  return async (ctx, next) => {


    try {
      // Apply the rate limiter to the current request
      await limiter(ctx, next);
 } catch (err) {
      if (err.status === 429) {
        // Handle rate limit exceeded error
        strapi.log.warn('Rate limit exceeded.');
        ctx.status = 429;
        ctx.body = {
          statusCode: 429,
          error: 'Too Many Requests',
          message: 'You have exceeded the maximum number of requests. Please try again later.',
 };
 } else {
        // Re-throw other errors to be handled by Strapi's error-handling middleware
        throw err;
 }
 }

The code above defines a middleware which gets executed whenever a function is made on any API. If the requests exceed the given maximum, an error code is outputted. Below is the full code.

'use strict';

/**
 * `RateLimit` middleware
 */
const RateLimit = require("koa2-ratelimit").RateLimit;

module.exports = (config, { strapi }) => {
  // Configuring the rate limiter middleware
  const limiter = RateLimit.middleware({
    interval: { min: 1 }, // Time window in minutes
    max: 3, // Maximum number of requests per interval
 });

  return async (ctx, next) => {

    try {
      // Apply the rate limiter to the current request
      await limiter(ctx, next);
 } catch (err) {
      if (err.status === 429) {
        // Handle rate limit exceeded error
        strapi.log.warn('Rate limit exceeded.');
        ctx.status = 429;
        ctx.body = {
          statusCode: 429,
          error: 'Too Many Requests',
          message: 'You have exceeded the maximum number of requests. Please try again later.',
 };
 } else {
        // Re-throw other errors to be handled by Strapi's error-handling middleware
        throw err;
 }
 }

 };
};

To ensure its seamless integration to all APIs within the Strapi project, the admin middlewares must also be configured.

cconst rateLimit = require('../middlewares/rateLimit');

module.exports = [
 'strapi::logger',
 'strapi::errors',
 'strapi::security',
 'strapi::cors',
 'strapi::poweredBy',
 'strapi::query',
 'strapi::body',
 'strapi::session',
 'strapi::favicon',
 'strapi::public',

 {
   name: 'global::rateLimit',
   config: {},
 },
];

With this, we have successfully configured the rate limiter powered by koa2-ratelimiter. Here are pictures of its execution.

Custom Strapi Api Rate Limiter

Within the rateLimit file in the API/middlewares folder, create a custom rate limiter by initializing a memory store.

const requestCounts = new Map();

Thereafter, define your rate limit function and then configure the rate limiter.

module.exports = (config, { strapi }) => {

  const rateLimitConfig = strapi.config.get('admin.rateLimit', {
    interval: 60 * 1000,  
    max: 3,  
 });

The time interval above is 1 minute while the maximum number of requests that can be made within the specified time interval is 3. You can tweak it to suit your preference.

return async (ctx, next) => {

    const ip = ctx.ip; 
    const currentTime = Date.now();

    if (!requestCounts.has(ip)) {

      requestCounts.set(ip, { count: 1, startTime: currentTime });
 } else {
      const requestInfo = requestCounts.get(ip);


      if (currentTime - requestInfo.startTime > rateLimitConfig.interval) {
        requestInfo.count = 1;
        requestInfo.startTime = currentTime;
 } else {

 }


      if (requestInfo.count > rateLimitConfig.max) {
        strapi.log.warn(`Rate limit exceeded for IP: ${ip}`);

        ctx.status = 429;
        ctx.body = {
          statusCode: 429,
          error: 'Too Many Requests',
          message: 'You have exceeded the maximum number of requests. Please try again later.',
 };
        return;
 }
 }

    await next();
 };
};

Afterwards, a middleware is defined which obtains the user IP address and then stores it in the memory store. The time interval is also set from the current time the request is made and the request count gets updated with every new request made.

If the requests made exceed the maximum expected requests within the time interval of 1 minute in our case, an error is thrown. Here is the full code below.

'use strict';
const requestCounts = new Map();

module.exports = (config, { strapi }) => {

  const rateLimitConfig = strapi.config.get('admin.rateLimit', {
    interval: 60 * 1000,  
    max: 3,  
 });

  return async (ctx, next) => {

    const ip = ctx.ip; 
    const currentTime = Date.now();

    if (!requestCounts.has(ip)) {

      requestCounts.set(ip, { count: 1, startTime: currentTime });
 } else {
      const requestInfo = requestCounts.get(ip);


      if (currentTime - requestInfo.startTime > rateLimitConfig.interval) {
        requestInfo.count = 1;
        requestInfo.startTime = currentTime;
 } else {

        requestInfo.count += 1;
 }


      if (requestInfo.count > rateLimitConfig.max) {


        ctx.status = 429;
        ctx.body = {
          statusCode: 429,
          error: 'Too Many Requests',
          message: 'You have exceeded the maximum number of requests. Please try again later.',
 };
        return;
 }
 }

    await next();
 };
};

Here is a demo of the project.

Express-rate-limiter Implementation

Express rate limiter is also another important package that can be used to implement rate limiting in our project. Right now, this package will be used to implement a route-specific API rate limiting.

The next step in this tutorial is setting up an efficient rate limiter for our Strapi APIs created in the repo.

To set up rate limiters on our Strapi applications, we'll be working mainly on the routes file. This can be navigated to by accessing the src folder within the project root directory. Within the src folder, navigate to the API folder which contains all the API files for the collections created in the Strapi dashboard.

The rate limiter will be enforced in the routes section of each API. For this tutorial, I will be using the products API as a demo API in this article.

'use strict';


/**
 * product router
 */

const { createCoreRouter } = require('@strapi/strapi').factories;

module.exports = createCoreRouter('api::product.product');

This is the initial code setup in the routes.js file in our product API folder. The rate limiting tool of choice for this tutorial is express-rate-limit as it offers much simplicity and user-friendliness coupled with its efficiency. Here is a link to its documentation. To get this installed, navigate to the command line of the project directory and run

npm install express-rate-limit

On completion of its installation, we will be initializing it in the products file already created within the routes folder as follows.

const { rateLimit } = require("express-rate-limit");

Go on and configure the rate limiter to your desired specifications.

const rateLimit = require('express-rate-limit');

const limiter = rateLimit({
  windowMs: 3 * 60 * 1000, // 3 minutes
  max: 2, // limit each IP to 2 requests per windowMs
  handler: async (req, res, next) => {
    const ctx = strapi.requestContext.get();
    ctx.status = 429;
    ctx.body = {
      message: "Too many requests",
      policy: "rate limit"
    };
    // Ensure the response is ended after setting the response body and status
    ctx.res.end();
  }
});

module.exports = limiter;

The code above serves to configure the rate limiting parameters we intend to use for the file.

windowMs represents the time interval in milliseconds for the number of requests. In our case, we specified a time of 3 minutes. Also, we specified the maximum number of requests that can be made within that same time frame. In our case, we used 2 for demo purposes.

However, the limit parameter also serves as an alternative to max parameter. Also included is the handler function that gets executed whenever the requests exceed the set number. It returns an Error 429 with an error body containing “Too many requests”.

const { createCoreRouter } = require('@strapi/strapi').factories;

module.exports = createCoreRouter('api::product.product', {
  config: {
    find: {
      middlewares: [
        async (ctx, next) => {
          await new Promise((resolve, reject) => {
            limiter(ctx.req, ctx.res, (error) => {
              if (error) {
                ctx.status = 429;
                ctx.body = { error: error.message };
                reject(error);
              } else {
                resolve();
              }
            });
          });
          await next();
        }
      ]
    }
  }
});

The above code illustrates the use of the Strapi API middleware which serves to ensure that the rate limit is fulfilled before the onward execution of the API requests. It also ensures that the request is terminated when the rate limit gets exceeded. Here is the final code for the project.

'use strict';

/**
 * product router
 */

const { createCoreRouter } = require('@strapi/strapi').factories;
const rateLimit = require('express-rate-limit');

const limiter = rateLimit({
  windowMs: 3 * 60 * 1000, // 3 minutes
  max: 2, // limit each IP to 2 requests per windowMs
  handler: async (req, res, next) => {
    const ctx = strapi.requestContext.get();
    ctx.status = 429;
    ctx.body = {
      message: 'Too many requests',
      policy: 'rate limit'
    };
    // Ensure the response is ended after setting the response body and status
    ctx.res.end();
  }
});

module.exports = createCoreRouter('api::product.product', {
  config: {
    find: {
      middlewares: [
        async (ctx, next) => {
          await new Promise((resolve, reject) => {
            limiter(ctx.req, ctx.res, (error) => {

              if (error) {
                ctx.status = 429;
                ctx.body = { error: error.message };
                reject(error);
              } else {
                resolve();
              }
            });
          });
          if (ctx.status !== 429) {
            await next();
          }
        }
      ]
    }
  }
});

Here is an image showing the rate limiting functionality.

You can also download the final code for the project here. Having completed this, you can then go ahead to test the rate limiting functionality of your API. The Strapi application can be run by executing npm run develop in the command line.

Conclusion

With this, we have come to the end of the tutorial. We hope you’ve learned essentially about rate limiting, its uses, tools and best practices.

You can also design multiple rate limiters within the code and implement them in any endpoint of your choice to test it out.

Feel free to drop any questions or comments. Happy coding!