DEV Community: MongoDB Guests

How I Built an AI Assistant for My Career with Java, Spring AI, and MongoDB

MongoDB Guests — Tue, 15 Sep 2026 09:00:00 +0000

This tutorial was written by Ricardo Mello.

The best way to learn a technology is by putting it into practice in a real system.

A few months ago, I decided to build a virtual assistant that could answer questions about my career, help people learn more about my articles, videos, talks, and projects, and even schedule a call on my calendar.

At first, the idea was simple: give the assistant access to my content and let people ask questions about it. But as the project evolved, I realized that not every question should be handled in the same way. Some questions can be answered with a direct query to the database, while others benefit from semantic search. In other cases, the assistant may need to use a tool or execute multiple steps before reaching an answer.

This is how RicAI evolved from a simple RAG application into a personal AI assistant that can search, remember context, and use external services. In this article, I’ll show you how I built it and how you can build your own.

How RicAI Works

Before jumping into the implementation, let's first understand what happens when someone sends a question to RicAI:

Receive the question – The user asks anything about my career, content, projects, or talks.
Load conversation history – RicAI checks whether there is previous context for that conversation and loads it so the AI can understand what was discussed before.
Choose the right capability – Based on the question and context, RicAI decides how to handle the request:
- structured MongoDB query for dates, counts, or ordering;
- Vector Search for semantic questions;
- Calendly for scheduling requests.
Generate the response – The result from the selected capability is used to generate the final answer.
Save the conversation – The new interaction is stored so the context can be reused in the next question.
Return the response – The answer is sent back to the user.

Build With

Java 21
Spring Boot 4
Spring AI
MongoDB Vector Search

I’ll focus on the main ideas and architecture throughout the article rather than every implementation detail. The complete source code is available on GitHub.

Modeling My Career as Data

The first thing I needed was a knowledge base about my own career. Articles, videos, talks, events, and projects are stored as documents containing the information RicAI may need to retrieve later. Each document contains the actual content together with metadata such as its title, category, and date.

For example, an event can be represented like this:

new Document(
    """
    Event: DevNexus 2026.
    Talk: Real-Time Fraud Detection in Java with Kafka Streams and Vector Similarity.
    Ricardo Mello presented this session at DevNexus 2026 in Atlanta.
    The talk explored Java, Kafka Streams, Spring Boot, MongoDB, and Vector Search.
    """,
    Map.of(
        "title", "DevNexus 2026",
        "category", "event",
        "createdAt", "2026-03-01"
    )
)

An article follows the same idea:

new Document(
    """
    Article: Clean and Modular Java: A Hexagonal Architecture Approach.
    Published on Foojay.io in June 2025.
    Topics: Java, Architecture.
    """,
    Map.of(
        "title", "Clean and Modular Java: A Hexagonal Architecture Approach",
        "category", "article",
        "createdAt", "2025-06-01"
    )
)

I intentionally keep both textual content and structured metadata. The text is useful for semantic search, while metadata gives me another way to filter, sort, and query the content directly (we will see it later).

These are just two examples. The knowledge base also contains documents for articles, videos, talks, projects, and other parts of my career.

Storing the Knowledge Base with Spring AI VectorStore

Creating the documents is only the first step. Since I also want RicAI to understand questions based on meaning, I need to generate vector embeddings for this content and store them in MongoDB.

For this, I use Spring AI's VectorStore abstraction with MongoDB Atlas as the vector store and OpenAI as the embedding model. MongoDB also provides an official Spring AI integration for this use case. First, I add the Spring AI starters for OpenAI and MongoDB Atlas Vector Store to the pom.xml:

<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-starter-model-openai</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-starter-vector-store-mongodb-atlas</artifactId>
</dependency>

Then I configure the OpenAI API key and the MongoDB Vector Store to the application.yml:

spring:
  ai:
    openai:
      api-key: ${OPENAI_API_KEY}
    vectorstore:
      mongodb:
        collection-name: content_kb
        index-name: vectorstore_index
        initialize-schema: true

The OpenAI starter provides the embedding model, while the MongoDB starter configures the VectorStore backed by MongoDB Atlas. Spring AI handles most of this configuration automatically, so I don't need to manually create the embedding model or the vector store for this flow.

With everything configured, storing my career content becomes very simple:

vectorStore.add(documents);

When this method is called, Spring AI uses the configured embedding model to generate a vector representation for each document and stores the content, metadata, and embedding in the content_kb collection. The MongoDB integration supports this flow directly, including automatic Vector Search index initialization when initialize-schema is enabled.

Note: This is convenient for development and demos; in production, the Vector Search index is typically managed explicitly rather than initialized by the application.

A stored document looks conceptually like this:

// content_kb collection
{ 
  _id: "5dde34df-28ed-4837-bcc4-3a3481d2e79e",
  content: "Talk: Real-Time Fraud Detection...",
  metadata: {
    title: "Dev/nexus",
    category: "event",
    createdAt: "2026-03-01"
  },
  embedding: [
    0.0123,
    -0.0342,
    ...
  ]
}

In this project, Spring AI uses OpenAI's text-embedding-ada-002 model, which generates vectors with 1536 dimensions.

At this point, the same MongoDB document gives RicAI two different ways to retrieve information: it can use fields such as category and createdAt for direct queries, or use the embedding with MongoDB Vector Search to find content based on semantic meaning. MongoDB Vector Search is designed specifically for retrieving data based on meaning rather than exact text matches.

Adding Conversation Memory

The Problem

Memory is a very important part of an AI application because LLM calls are stateless. In other words, the model does not automatically remember what happened in a previous request.

For example, imagine that you first ask:
What articles has Ricardo written about Spring AI?

The model may answer:

Ricardo Mello has written the following articles about Spring AI: …

Then, right after that, you ask:

Which one is the most recent?

By itself, this second question is incomplete. The assistant needs the previous conversation to understand that “which one” refers to the articles returned earlier. The model may answer:

Which options are you referring to?

This happens because the second request does not contain the context from the first one. To keep a conversation, we basically need to store the previous messages and send the relevant conversation history together with the new request.

The Solution

The basic idea is simple: send the conversation history along with the user's new message, so the AI has the context it needs to understand what is happening.

There are different ways to implement this in an AI application. In RicAI, I decided to persist the conversation in MongoDB. For each interaction, I want to keep both sides of the conversation: what the user asked:

{
  id: "001", 
  message: "What articles has Ricardo written about Spring AI?",
  timestamp: "2026-09-07"
  type: "USER"
}

And the assistant response:

{
  id: "001", 
  message: "Ricardo Mello has written the following articles about Spring AI:...",
  timestamp: "2026-09-07"
  type: "ASSISTANT"
}

So, in practice, we are storing what the user asked and what the LLM answered, grouping those messages by conversation.

Of course, we could implement this ourselves: create a collection, build a service to save messages, retrieve the previous conversation, and call that service before and after every interaction. But that would add some extra work.

Spring AI already provides abstractions for this, and MongoDB can be used as the underlying ChatMemoryRepository. To enable it, I add the following dependency to the pom.xml:

<dependency>
    <groupId>org.springframework.ai</groupId>       <artifactId>spring-ai-starter-model-chat-memory-repository-mongodb</artifactId>
</dependency>

Then I configure the ChatClient with a MessageChatMemoryAdvisor:

@Bean
ChatClient chatClient(
        OpenAiChatModel chatModel,
        ChatMemory chatMemory) {
    return ChatClient.builder(chatModel)
          .defaultSystem("...")
          .defaultAdvisors(
                MessageChatMemoryAdvisor.builder(chatMemory).build()
          )
          .build();
}

The advisor is responsible for retrieving the previous messages from memory and adding them to the next request as conversation context.

The last piece is telling Spring AI which conversation we are working with. For that, I pass the conversationId when calling the ChatClient:

return chatClient.prompt(chatRequest.message())
      .advisors(advisor -> advisor
            .param(ChatMemory.CONVERSATION_ID, chatRequest.conversationId()))
      .call()
      .content();

And that’s it! From this point on, calls made through this ChatClient with a conversation ID can use the previous conversation as context. For regular chat interactions, the MessageChatMemoryAdvisor manages the conversation memory automatically. In the multi-step planning flow, the application explicitly records the final user message and assistant response using chatMemory.add(). In both cases, the messages are persisted in the ai_chat_memory collection. For example:

{
  conversationId: "daaac627-5d35-4545-b9f9-4bae4d5c53ca",
  message: {
    content: "Briefly tell me about Ricardo....",
    type: "USER",
    metadata: {
      messageType: "USER"
    }
  },
  timestamp: ISODate("2026-09-05T11:31:56.190Z"),
  sequence: 0
}

And the assistant response:

{
  conversationId: "daaac627-5d35-4545-b9f9-4bae4d5c53ca",
  message: {
    content: "Ricardo Mello is a backend engineer, international speaker...",
    type: "ASSISTANT"
  },
  timestamp: ISODate("2026-09-05T11:31:56.190Z"),
  sequence: 1
}

Defining RicAI's Role

Now that RicAI can keep the conversation context, we also need to define what kind of assistant it is and what it is allowed to do. Earlier, when configuring the ChatClient, we used:

.defaultSystem("...")

This is where I define RicAI's role, scope, and some rules about how it should behave.

.defaultSystem("""
    You are Ricardo Mello's virtual assistant, specialized in
    his professional background, employment history, articles,
    videos, events, talks, projects, and meeting scheduling.

    Answer only questions related to Ricardo Mello using
    information returned by the available capabilities.

    Prefer structured capabilities for filtering, ordering,
    counting, comparison, and exact lookup.

    Prefer semantic search for natural-language retrieval
    and discovery by meaning or topic.

    Reply in the same language as the user.
""")

The complete system prompt also includes rules for security, scheduling, and tool usage. You can check the full version in the project repository.

Choosing the Right Capability

Now that we have configured memory and populated our knowledge base, the next step is to let the assistant decide which capability should be used to answer the user's question.

For example, a user might ask:

*How many articles does Ricardo have? *

We could use semantic search to answer this question, but it is designed to retrieve relevant content based on similarity, not to calculate exact values such as counts. In this case, a structured query is a much better fit because it gives us a deterministic and exact result.

We can simply execute a count against our MongoDB collection and return the exact number of articles.

On the other hand, imagine the user asks:

Has Ricardo created any content about AI agents and memory?

Here, semantic search makes much more sense. The user is not asking for an exact field or value, but for content related to a specific meaning or topic.

And there is another type of request:

Help me schedule a call with Ricardo.

In this case, neither a regular MongoDB query nor Vector Search can answer the question. RicAI needs to interact with an external service, in this case Calendly.

This is where tools become important.

Exposing Capabilities as Tools

Tools enable the model to invoke functions or actions to accomplish a task. In RicAI, these tools are regular Java methods that the assistant can call to retrieve information or perform an action before building the final response.

Structured MongoDB Query

For example, we already have a method that retrieves my professional experience from the knowledge base:

public List<CatalogContentResult> findProfessionalExperience() {
    var pageable = PageRequest.of(
          0,
          EXPERIENCE_LIMIT,
          Sort.by(Sort.Direction.DESC, "metadata.createdAt")
    );

    return contentKbRepository
          .findByMetadataCategory("experience", pageable)
          .stream()
          .map(CatalogContentResult::from)
          .toList();
}

There is nothing AI-specific about this method. It simply queries the content_kb collection we created earlier, filters the documents by the experience category, and returns the results ordered by date.

Now we can expose this operation to the model as a tool:

@Tool(description = """
    Retrieves Ricardo Mello's professional experience,
    ordered from the most recent to the oldest.
    Use it for questions about companies, roles,
    employment periods, and career history.
    """)
public List<CatalogContentResult> findProfessionalExperience() {
    return knowledgeBaseService.findProfessionalExperience();
}

The @Tool annotation makes the method available to the model, while the description helps the model understand when this capability should be used.

RicAI has several structured operations like this inside the ContentTools class. They are useful when we need exact operations such as filtering, ordering, counting, or retrieving a complete list.

Semantic Search

Structured queries are not the best option for every question. Sometimes the user is looking for information by meaning or topic rather than an exact field or value.

For these cases, RicAI exposes a semantic search tool:

@Tool(description = """
    Searches Ricardo Mello's knowledge base using semantic similarity.
    Use it for natural-language questions, summaries,
    topic discovery, and finding content by meaning.
    """)
public List<ContentResult> searchKnowledgeBase(
      @ToolParam(description = "The information the user wants to find")
      String query,

      @ToolParam(description = "Maximum number of results")
      int limit
) {
    return knowledgeBaseService.searchKnowledgeBase(query, limit);
}

Behind this tool, we use the same VectorStore configured earlier:

return vectorStore.similaritySearch(searchRequest)
      .stream()
      .map(ContentResult::from)
      .toList();

So now RicAI has two different ways to access the same knowledge base.

For a question such as:

Which companies has Ricardo worked for?

A structured MongoDB query is a better fit.
But for something like:

Has Ricardo created content related to AI agents and memory?

Semantic search makes more sense because the goal is to find relevant content based on meaning.

External Service with Calendly

Besides answering questions about my career, RicAI can also help users schedule a call with me.

Normally, the user could simply open my Calendly page, choose an available time, and continue with the scheduling process.

But RicAI can help with this flow directly from the conversation. For example, the user can simply ask to schedule a call and then request my next available meeting times.

As the conversation continues, the user can select one of the available times and provide the information required for the meeting. RicAI then uses that context to prepare the scheduling link.

To make this possible, RicAI needs access to the Calendly API. I will not include the complete integration code here, but you can find it in the project repository.

The important part for our flow is, once again, the tool exposed to the model. One of the scheduling tools retrieves the available meeting times:

@Tool(description = """
    Lists available times for a meeting with Ricardo.
    Use it when the user needs to select a meeting time.
    """)
public List<AvailableMeetingTime> findAvailableMeetingTimes(
      @ToolParam(description = "Future range start as an ISO-8601 UTC instant")
      String startTime,

      @ToolParam(description = "Range end as an ISO-8601 UTC instant")
      String endTime,

      @ToolParam(description = "Maximum number of times to return")
      int limit
) {
    return calendlyService.findAvailableTimes(startTime, endTime, limit);
}

This is only one part of the scheduling flow. RicAI can call other scheduling tools as the conversation continues, using the selected time and user information to complete the process and prepare the final booking link.

You can check the complete set of scheduling tools here.

Finally, we need to make these tools available to the model. With Spring AI, we can register them with .defaultTools() directly in the ChatClient:

return ChatClient.builder(openAiChatModel)
      .defaultSystem("...")
      .defaultAdvisors(
            MessageChatMemoryAdvisor.builder(chatMemory).build()
      )
      .defaultTools(contentTools, schedulingTools)
      .build();

With this configuration, the model has access to both content and scheduling capabilities and can decide which tool to use based on the user's request and the conversation context.

Conclusion

In this tutorial, we explored how to turn a simple idea into a real application that can actually be useful in day-to-day interactions.

RicAI started as a way to make my career content easier to explore, but it evolved into an assistant that can search structured data, perform semantic search, remember conversations, and even interact with external services such as Calendly.

You can explore the complete project on GitHub and use it as a starting point to build your own personal AI assistant.

How to Build an AI PDF Analyzer with MongoDB, RAG, and Ollama

MongoDB Guests — Mon, 07 Sep 2026 14:36:27 +0000

This tutorial was written by Damilola Oladele.

This tutorial shows you how to build a command-line tool that reads a PDF, writes an outline and summary, and answers questions about the document. The language model runs on your local machine through Ollama.

The tool has two halves.

The first half analyzes the document, and it doesn’t use Retrieval-Augmented Generation (RAG). It reads the PDF and pulls the outline from the document’s own table of contents. The first half also reads the title and author from the PDF’s metadata and asks the local model for a section-by-section summary.

The second half answers questions, and this is the RAG pipeline. It splits the document into small overlapping passages called chunks and turns each chunk into a vector embedding. The second half then indexes those vectors in MongoDB Atlas, retrieves the chunks closest in meaning to your question, and asks the model to answer using only the retrieved text. The page and section travel with each chunk, so every answer can show its source.

The finished tool has two commands:

python cli.py analyze sample.pdf
python cli.py ask sample.pdf

You can find all the code samples for this tutorial in the GitHub repository.

Prerequisites

Make sure you have the following before you start:

Python 3.10 or later installed.
At least an intermediate understanding of the Python programming language.
A MongoDB Atlas account with an M0 Free tier cluster set up.
Your MongoDB Atlas connection string. You can find it in the MongoDB Atlas UI under Database > Connect > Drivers.
Ollama installed on your machine.
A sample PDF that has an embedded table of contents and metadata (a title and an author). This tutorial uses a Agent-Diff: Benchmarking LLM Agents on Enterprise API Tasks via Code Execution with State-Diff-Based Evaluation. The tutorial's GitHub repository also includes a copy of the PDF file.

Add your current IP address to the MongoDB Atlas access list before you run any script. Go to Security > Network Access in the MongoDB Atlas UI and add it. Your scripts won't connect to your MongoDB cluster without this configuration.

Set Up the Project and Connect to MongoDB Atlas

This section prepares your machine, project, and MongoDB Atlas connection.

Start by creating a folder for the project and move into it in your terminal. Every file in this tutorial goes into this folder:

mkdir pdf-doc-analyzer
cd pdf-doc-analyzer

Pull the language model

Ollama runs the language model on your own machine, so no request ever leaves your computer.

Confirm that Ollama is installed by running the following command:

ollama --version

If you use the Ollama desktop application, open it before continuing. The desktop application can run Ollama for you.

If you use Ollama from the command line without the desktop application, start the Ollama server:

ollama serve

Leave that terminal running, then open a second terminal and pull the model:

cd pdf-doc-analyzer
ollama pull llama3.2:3b

The `llama3.2:3b` download is about 2 GB, so pulling the model can take a few minutes.

Once the pull is successful, confirm the model is installed from that second terminal:

ollama list

You should see llama3.2:3b in the list.

Create and activate a virtual environment

Create and activate a virtual environment to keep your project's Python packages separate from the rest of your system:

python -m venv venv
source venv/bin/activate

On Windows, activate the virtual environment with:

venv\Scripts\activate

Install the packages

Create a file named requirements.txt in your pdf-doc-analyzer folder. This file lists every package the tool needs:

pymongo==4.17.0
pymupdf==1.28.2
ollama==0.6.2
sentence-transformers==6.0.0
einops==0.8.2
typer==0.27.1
rich==15.0.0
python-dotenv==1.2.3

The following are the purposes of each package:

pymongo is the Python driver for MongoDB.
pymupdf reads PDF text, metadata, and bookmarks.
ollama calls the local language model.
sentence-transformers and einops run the nomic embedding model.
typer and rich build the command-line interface and format its output.
python-dotenv loads your connection string from a .env file.

Install the packages:

python -m pip install -r requirements.txt

Add your MongoDB Atlas connection string to an environment file

Keep your connection string out of your code, so you never commit credentials by accident. Create a file named .env in your pdf-doc-analyzer folder, and add your connection string:

MONGODB_URI="mongodb+srv://<user>:<pass>@<host>/"

Replace <user>, <pass>, and <host> with the values from your MongoDB Atlas connection string. The python-dotenv package loads this value at runtime.

Create the configuration file

Create a file named config.py in your pdf-doc-analyzer folder. This file loads your connection string from .env and defines the constants that every other module imports:

import os
from dotenv import load_dotenv

# Load MONGODB_URI and any other variables from the .env file:
load_dotenv()

MONGODB_URI = os.getenv("MONGODB_URI")

DB_NAME = "doc_analyzer"
DOCUMENTS_COLLECTION = "documents"
CHUNKS_COLLECTION = "chunks"
LLM_MODEL = "llama3.2:3b"
EMBEDDING_MODEL = "nomic-ai/nomic-embed-text-v1"
EMBEDDING_DIMENSIONS = 768
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
VECTOR_INDEX_NAME = "vector_index"
APP_NAME = "devrel-tutorial-python-pdf-rag"

Create the database access file

Create a file named db.py in your pdf-doc-analyzer folder. This file owns every call to MongoDB, so the rest of the code never touches the driver directly.

import datetime
from pymongo import MongoClient
import config


def get_client() -> MongoClient:
    """Open a MongoDB client using the connection string and appName from config."""
    # Connect to MongoDB and identify the application with the configured appName.
    return MongoClient(config.MONGODB_URI, appname=config.APP_NAME)


def get_documents_collection(client: MongoClient):
    """Return the collection that stores one record per analyzed PDF."""
    return client[config.DB_NAME][config.DOCUMENTS_COLLECTION]


def get_chunks_collection(client: MongoClient):
    """Return the collection that stores one record per chunk."""
    return client[config.DB_NAME][config.CHUNKS_COLLECTION]


def save_document(client: MongoClient, record: dict) -> str:
    """Insert or replace the stored record for one PDF, keyed on its filename."""
    # Stamp the record with the time of this write:
    record["updated_at"] = datetime.datetime.now(datetime.timezone.utc)
    # Upsert by filename so re-analysis replaces the existing document record.
    get_documents_collection(client).replace_one(
        {"filename": record["filename"]}, record, upsert=True
    )
    return record["filename"]


def get_document(client: MongoClient, filename: str):
    """Return the saved record for one filename, or None when it isn't stored yet."""
    return get_documents_collection(client).find_one({"filename": filename})


def list_documents(client: MongoClient):
    """Return every saved document record, without the heavy full_text field."""
    return list(get_documents_collection(client).find({}, {"full_text": 0}))

In MongoDB, a collection is the equivalent of a table in a relational database. It's a named group of records, and each record is a document. This tool uses two collections. The documents collection holds one record per analyzed PDF, and the chunks collection holds one record per chunk. The get_documents_collection() and get_chunks_collection() helpers hand back those two collections, so the rest of the code asks for a collection by name rather than repeating the database path.

The save_document() function writes one PDF's record, and it does three things:

It stamps the record with an updated_at timestamp.
It calls replace_one() with a filter on filename and upsert=True.
It returns the filename as the record's key.

The upsert=True flag inserts a new record when no document matches the filename, and it replaces the matching record when one already exists, so a second run on the same file never creates a duplicate.

Keep in mind what replace_one() does. It swaps the entire matching record for the new one rather than merging fields, so any field left out of record disappears after the write.

You'll add two more helpers to this file, save_chunks() and get_chunks(), once chunk storage is needed.

Define the Data Structures

The tool defines its data once as a set of Python data classes, and it stores two record shapes in MongoDB.

The documents collection keeps one record per analyzed PDF. Each record uses the PDF's filename as its identifier. The record shape is as follows:

{
  "filename": "report.pdf",
  "title": "…",
  "author": "…",
  "page_count": 12,
  "full_text": "…",
  "outline": [{"title": "Authentication", "level": 1, "page": 4}, "…"],
  "summary": "…",
  "updated_at": "ISODate"
}

The chunks collection keeps one record per chunk. Each chunk is a small, overlapping slice of the document's text, sized so the language model can read it and so search can match it precisely. The record shape is as follows:

{
  "document_id": "report.pdf",
  "chunk_index": 0,
  "text": "…",
  "page_start": 4,
  "page_end": 5,
  "section": "Authentication",
  "embedding": [0.0, "…"]
}

The embedding field holds the vector for that chunk's text. $vectorSearch, the MongoDB aggregation stage you'll use later, compares your question's vector against these stored vectors and returns the closest matches by meaning. The document_id field reuses the same filename key, so chunks link back to their document, and retrieval can filter to one PDF at a time.

Now, create a file named models.py in your pdf-doc-analyzer folder, and add the following code:

from __future__ import annotations
from dataclasses import dataclass, field


@dataclass
class Line:
    """One line of text and the page it sits on."""
    text: str
    page: int


@dataclass
class Page:
    """One page: its number and the lines found on it."""
    number: int
    # default_factory gives each Page its own list, rather than one list shared
    # across every instance. Without it, all Page objects would share the same
    # list, and appending a line to one page would add it to every page:
    lines: list[Line] = field(default_factory=list)


@dataclass
class OutlineEntry:
    """One heading from the document outline: its title, nesting level, and page."""
    title: str
    level: int                  # 1 = top level, 2 = subsection, and so on
    page: int


@dataclass
class Chunk:
    """One passage of text with its page span, section, and embedding."""
    chunk_index: int
    text: str
    page_start: int
    page_end: int
    section: str | None         # nearest preceding outline heading, or None
    embedding: list[float] | None = None


@dataclass
class Document:
    """The whole analyzed PDF: its text, pages, outline, summary, and metadata."""
    filename: str
    page_count: int
    pages: list[Page] = field(default_factory=list)
    full_text: str = ""
    embedded_toc: list = field(default_factory=list)   # PyMuPDF get_toc() output
    outline: list[OutlineEntry] = field(default_factory=list)
    summary: str = ""
    title: str = ""            # from the PDF's embedded metadata
    author: str = ""           # from the PDF's embedded metadata

The preceding code defines the data structures used throughout the PDF analysis pipeline. A dataclass is a good fit here because it stores related data rather than performing complex behavior. The classes connect in one direction. PDF extraction fills a Document with Page and Line objects. The Generate the Outline and Summary section of this tutorial fills Document.outline with OutlineEntry objects. The Split the Document Into Chunks section produces Chunk objects, which receive embeddings when the code in the Generate Embeddings section runs. A later file, analyzer.py, converts a Document to and from the record stored in MongoDB.

Extract the PDF Text and Metadata

This section reads a PDF and fills in a Document object.

PyMuPDF, imported under the name pymupdf, reads a page as a nested structure. A page holds blocks, and a block holds lines. A line holds spans, where a span is a run of characters that share one font and style. You walk down to the spans to read each line's text.

The extraction pulls three things from the PDF:

The full text and the page count, which the summary and the chunk_document() function (a function defined later in the tutorial) both use.
The PDF's embedded outline through doc.get_toc(), which becomes the tool's outline.

This call reads the PDF's own bookmarks, not a printed table of contents page. A PDF can have a visible contents page without an embedded outline, and the reverse, so the two don't always match.

The title and author through doc.metadata, which make document-level questions answerable later. For example, a reader who asks "who wrote this?" gets an answer from the metadata.

Now create a file named pdf_utils.py in your pdf-doc-analyzer folder, and add the following:

import os

import pymupdf  # PyMuPDF

from models import Document, Page, Line, OutlineEntry


def extract(pdf_path: str) -> Document:
    """Read a PDF file and return a populated Document."""
    # A context manager closes the PDF automatically, even when an error is raised:
    with pymupdf.open(pdf_path) as doc:
        # 1. Things to note: the page count, and all the page text joined together:
        page_count = doc.page_count
        full_text = "\n".join(page.get_text() for page in doc)

        # 2. The PDF's embedded outline (its bookmarks), not a printed contents page.
        #    Each entry is [level, title, page]:
        embedded_toc = doc.get_toc()

        # 3. Read the PDF's embedded metadata, including its declared title and author:
        metadata = doc.metadata or {}

        # 4. Walk every page and collect each line's text, paired with its page number:
        pages: list[Page] = []
        for page_number, page in enumerate(doc, start=1):
            lines = []
            # get_text("dict") returns nested blocks, then lines, then spans:
            for block in page.get_text("dict")["blocks"]:
                for line in block.get("lines", []):
                    text = "".join(span["text"] for span in line["spans"]).strip()
                    if text:  # skip blank lines
                        lines.append(Line(text=text, page=page_number))
            pages.append(Page(number=page_number, lines=lines))

    # 5. Return the filled Document. The outline and summary stay empty until later sections:
    return Document(
        filename=os.path.basename(pdf_path),
        page_count=page_count,
        pages=pages,
        full_text=full_text,
        embedded_toc=embedded_toc,
        title=(metadata.get("title") or "").strip(),
        author=(metadata.get("author") or "").strip(),
    )


# Temporary check: Allows you to run this file directly to confirm the output:
if __name__ == "__main__":
    doc = extract("sample.pdf")
    print("pages:", doc.page_count)
    print("title:", doc.title)
    print("author(s):", doc.author)
    print("outline entries:", len(doc.embedded_toc))
    for level, title, page in doc.embedded_toc[:15]:
        print(f"  {'  ' * (level - 1)}p{page}: {title}")

The extract() function opens the PDF inside a context manager, joins its text, reads the embedded outline and the title and author, collects every line paired with its page, then returns a filled Document. The context manager closes the PDF automatically, including when an error is raised.

The file ends with a temporary block that runs only when you run the file on its own. Place your sample PDF at the root of your pdf-doc-analyzer folder and rename it to sample.pdf, then run:

python pdf_utils.py

You should see the page count, the title, the author, and the first several outline entries. The headings should appear in reading order. The extract() function that precedes it is what the rest of the tool imports.

Generate the Outline and Summary

This section turns the raw PDF into two readable results:

a nested outline
a section-by-section summary

The outline comes straight from the embedded bookmarks with no AI at all. The summary is the first model call in the tool, and it runs locally through Ollama.

Build the outline from the embedded bookmarks

The PDF's embedded outline is the author's own heading list, so it's the most reliable structure you can get. You can turn it into OutlineEntry objects directly with no model call. Open pdf_utils.py again and add this function after extract():

def outline_from_toc(doc: Document) -> list[OutlineEntry]:
    """Turn the PDF's embedded outline into a list of OutlineEntry objects."""
    return [
        OutlineEntry(title=title, level=level, page=page)
        for level, title, page in doc.embedded_toc
    ]

Each embedded outline entry arrives as [level, title, page], and this function maps it onto the OutlineEntry fields. The outline you build here serves two purposes later:

The summary uses it to process each section.
The chunk_document() function (a function defined later in the tutorial) uses it to tag each chunk with its section.

Summarize the document section by section

A small local model can't read a whole book in one request, and asking for everything at once gives a thin summary. So the summary reuses the outline. It walks each top-level section, sends that one section's text to the model, and asks for two or three sentences. It then joins the section summaries, so each section gets its own paragraph.

A minimum-length check clears any thin section the outline lists, because a near-empty structural page would waste a model call. When a section exceeds about 9,000 characters, the tool splits it into parts, summarizes each part, and then combines them into a single section summary. This avoids silently losing content and reflects how to handle sections that exceed a model’s per-request limit.

Create a file named llm_utils.py in your pdf-doc-analyzer folder, and add the following:

import ollama

import config
from models import Document


# Skip a section shorter than this. It clears thin pages, the outline still lists:
MIN_SECTION_CHARS = 700
# The most text to send in one model call. Longer sections get split into parts:
MAX_SECTION_CHARS = 9000


def _section_text(doc: Document, start_page: int, end_page: int) -> str:
    """Join the text of every line on pages in the range [start_page, end_page)."""
    parts = []
    for page in doc.pages:
        if start_page <= page.number < end_page:
            parts.extend(line.text for line in page.lines)
    return "\n".join(parts)


def _summarize(text: str, instruction: str) -> str:
    """Send one block of text to the model and return its plain-text reply."""
    response = ollama.chat(
        model=config.LLM_MODEL,
        messages=[
            {"role": "system", "content": "You write clear, concise summaries."},
            {"role": "user", "content": instruction + "\n\n" + text},
        ],
    )
    return response["message"]["content"].strip()


def _summarize_section(title: str, text: str) -> str:
    """Summarize one section, splitting it into parts when it exceeds the per-call limit."""
    # Short enough for one call: summarize the section directly.
    if len(text) <= MAX_SECTION_CHARS:
        return _summarize(
            text,
            f'In two or three sentences, summarize this section titled "{title}":',
        )
    # Too long for one call: split into parts, summarize each part, then combine those
    # summaries. This avoids silently discarding content:
    pieces = [text[i:i + MAX_SECTION_CHARS] for i in range(0, len(text), MAX_SECTION_CHARS)]
    piece_summaries = [
        _summarize(piece, f'Summarize this part of the section titled "{title}":')
        for piece in pieces
    ]
    return _summarize(
        "\n\n".join(piece_summaries),
        f'In two or three sentences, combine these notes into one summary of the section titled "{title}":',
    )


def generate_summary(doc: Document) -> str:
    """Summarize the document one top-level section at a time, then join the parts."""
    tops = [entry for entry in doc.outline if entry.level == 1]

    last_page = doc.page_count + 1
    parts = []

    for i, entry in enumerate(tops):
        start = entry.page
        end = tops[i + 1].page if i + 1 < len(tops) else last_page
        text = _section_text(doc, start, end)

        # Skip sections with too little text to summarize:
        if len(text) < MIN_SECTION_CHARS:
            continue

        paragraph = _summarize_section(entry.title, text)
        parts.append(f"{entry.title}\n{paragraph}")

    return "\n\n".join(parts)


# Temporary check: Allows you to run this file directly to confirm the output:
if __name__ == "__main__":
    import pdf_utils

    doc = pdf_utils.extract("sample.pdf")
    doc.outline = pdf_utils.outline_from_toc(doc)

    print("Outline:")
    for entry in doc.outline[:15]:
        indent = "  " * (entry.level - 1)
        print(f"{indent}p{entry.page}: {entry.title}")

    print("\nSummary:\n" + generate_summary(doc))

A model call reads a list of messages, and each message has a role. The system message sets the model's job, such as "you write clear, concise summaries." The user message carries the actual task, which here is one section's text. The _summarize() helper wraps both into a single call. The _summarize_section() helper summarizes a whole section and splits any section longer than MAX_SECTION_CHARS into parts. It then summarizes each part and combines the results into one summary. The generate_summary() function runs this for every top-level section and joins the results.

The file ends with a temporary block that runs only when you run the file on its own.

Make sure Ollama is running, llama3.2:3b is pulled, and the sample.pdf file from the prerequisites is in the root of your project folder. Then run:

python llm_utils.py

You should see a nested outline, then a summary with one paragraph per section.

The model runs on your machine and makes one call per section, so a document with many sections can take some minutes.

Build the Analyze Command

This section wires the pieces together into the analyze command.

You create two files:

analyzer.py holds the logic that extracts, builds the outline, summarizes, and saves a PDF.
cli.py turns that logic into a command you type in the terminal.

Create a file named analyzer.py in your pdf-doc-analyzer folder and add the following code:

import os

from pymongo import MongoClient

import db
import llm_utils
import pdf_utils
from models import Document, OutlineEntry


def _record(doc: Document) -> dict:
    """Turn a Document into the dictionary saved in the documents collection."""
    return {
        "filename": doc.filename,
        "page_count": doc.page_count,
        "full_text": doc.full_text,
        "outline": [{"title": e.title, "level": e.level, "page": e.page} for e in doc.outline],
        "summary": doc.summary,
        "title": doc.title,
        "author": doc.author,
    }


def _document(record: dict) -> Document:
    """Rebuild a Document from a saved database record."""
    doc = Document(filename=record["filename"], page_count=record["page_count"])
    doc.full_text = record.get("full_text", "")
    doc.summary = record.get("summary", "")
    doc.title = record.get("title", "")
    doc.author = record.get("author", "")
    doc.outline = [
        OutlineEntry(title=e["title"], level=e["level"], page=e["page"])
        for e in record.get("outline", [])
    ]
    return doc


def analyze(
    client: MongoClient,
    pdf_path: str,
    force: bool = False,
) -> Document:
    """Extract, outline, and summarize a PDF, then save it. Reuse a saved result when present."""
    filename = os.path.basename(pdf_path)

    # Reuse a saved analysis so a second run doesn't call the model again:
    if not force:
        saved = db.get_document(client, filename)
        if saved:
            return _document(saved)

    doc = pdf_utils.extract(pdf_path)
    doc.outline = pdf_utils.outline_from_toc(doc)
    doc.summary = llm_utils.generate_summary(doc)  # reads doc.outline, so set it first
    db.save_document(client, _record(doc))
    return doc

The analyze() function runs the analyze half of the tool in the following order:

extract the PDF
build the outline from the embedded bookmarks
write the summary
save the record

It checks for a saved result first, so the caching behavior is worth explaining:

The first analyze extracts, calls the model for the summary, and then saves the result.
The second analyze finds the saved document and skip the model.
The forced analyze (--force) extracts, calls the model again, and then replaces the saved result.

The cache is stored using the PDF's filename as the identifier. This is a prototype simplification, not a real document identity. A different PDF saved under an existing name could return the earlier analysis, so pass --force when you swap in a different file under an old name.

Two small libraries build the command itself. Typer turns a plain Python function into a terminal command, so analyze(pdf_path) becomes python cli.py analyze sample.pdf. Rich prints readable output, using an indented tree for the outline and a bordered panel for the summary.

Create a file named cli.py in your pdf-doc-analyzer folder, and add the following:

import typer
from rich.console import Console
from rich.panel import Panel
from rich.tree import Tree

import analyzer
import db
from models import Document

app = typer.Typer(help="Analyze a PDF and ask questions about it.")
console = Console()


# Keep the CLI ready for the ask command you'll add later:
@app.callback()
def main():
    """Analyze a PDF and ask questions about it."""


def _render(doc: Document) -> None:
    """Print the outline in a titled panel and the summary in another."""
    tree = Tree(f"[bold]{doc.filename}[/bold]")
    parents = {0: tree}
    for entry in doc.outline:
        parent = parents.get(entry.level - 1, tree)
        node = parent.add(f"p{entry.page}: {entry.title}")
        parents[entry.level] = node
    console.print(Panel(tree, title="Outline"))
    if doc.summary:
        console.print(Panel(doc.summary, title="Summary"))


@app.command()
def analyze(
    pdf_path: str,
    force: bool = typer.Option(
        False,
        help="Re-run even if a saved result exists.",
    ),
):
    """Build the outline and summary for a PDF."""
    client = db.get_client()
    try:
        doc = analyzer.analyze(client, pdf_path, force=force)
        _render(doc)
    finally:
        client.close()


if __name__ == "__main__":
    app()

The empty @app.callback() keeps your command-line interface (CLI) structured for more than one command. It holds the analyze command name in place now, and it stays in place once you add the ask command later.

Make sure Ollama is running, keep sample.pdf in your folder, then run the command:

python cli.py analyze sample.pdf

You should see the outline in a titled panel, the summary in another, and the record in your MongoDB Atlas cluster under the doc_analyzer database. Run the same command again, and it returns the saved result right away.

To run the model again, add --force to the command:

python cli.py analyze sample.pdf --force

Split the Document Into Chunks

This section starts the question-answering half of the tool. It splits the document into chunks, which are small overlapping slices of text. It also tags each chunk with the pages it covers and the section it belongs to. That tagging is what lets every answer show its source.

A model searches better over small passages than over a whole book, and small chunks make precise page and section references possible. The chunk_document() function (defined later in the tutorial) builds each piece from whole lines until it reaches about 1,000 characters. It then carries the trailing lines, up to about 150 characters, into the start of the next piece. Whole lines keep a word from being split across a boundary, and the small overlap repeats the trailing lines so text near a boundary appears in both chunks. A line boundary can still fall in the middle of a sentence, so the overlap, not the line split, is what keeps a boundary-spanning idea readable in the following chunk.

You also record the first and last page each chunk covers, and you look up its section by finding the nearest outline heading at or before the chunk's first page. These four fields, chunk_index, page_start, page_end, and section, travel with each chunk as retrieval and source metadata. The application stores and reads them directly, and the language model never produces them.

Create a file named qa.py in your pdf-doc-analyzer folder, and add the following:

from models import Chunk, Document


def chunk_document(doc: Document, target_chars: int = 1000, overlap: int = 150) -> list[Chunk]:
    """Split the document into overlapping chunks that carry their page span and section."""
    # Every line in reading order, paired with the page it sits on:
    lines = [(page.number, line.text) for page in doc.pages for line in page.lines]

    def section_for(page_number):
        # The title of the nearest outline heading at or before this page:
        current = None
        for entry in doc.outline:
            if entry.page <= page_number:
                current = entry.title
            else:
                break
        return current

    chunks = []
    window = []      # the (page, line) pairs in the current chunk, kept whole
    size = 0         # running character count for the current chunk
    index = 0

    def flush():
        nonlocal index
        text = "\n".join(line for _, line in window).strip()
        if not text:
            return
        # page_start and page_end come from the lines actually in this chunk:
        page_start = window[0][0]
        page_end = window[-1][0]
        chunks.append(Chunk(chunk_index=index, text=text,
                            page_start=page_start, page_end=page_end,
                            section=section_for(page_start)))
        index += 1

    for page_number, text in lines:
        window.append((page_number, text))
        size += len(text) + 1
        if size >= target_chars:
            flush()
            # Carry whole trailing lines into the next chunk, up to the overlap size.
            # Whole lines keep words intact, and the cap stops a long line from
            # repeating across many chunks:
            carry = []
            carried = 0
            for pair in reversed(window):
                if carried + len(pair[1]) + 1 > overlap:
                    break
                carry.insert(0, pair)
                carried += len(pair[1]) + 1
            window = carry
            size = carried
    if window:
        flush()
    return chunks


# Temporary check. Allows you to run this file directly:
if __name__ == "__main__":
    import pdf_utils

    doc = pdf_utils.extract("sample.pdf")
    doc.outline = pdf_utils.outline_from_toc(doc)

    chunks = chunk_document(doc)
    with_section = [c for c in chunks if c.section]
    print(f"chunks: {len(chunks)} ({len(with_section)} carry a section)")

    # Show the first chunk of each section, so you can see the mapping working:
    seen = set()
    for chunk in chunks:
        if chunk.section and chunk.section not in seen:
            seen.add(chunk.section)
            print(f"  p{chunk.page_start}-{chunk.page_end} | {chunk.section}")

You also need a place to store chunks. Open db.py again and add two helpers after the ones you already have. save_chunks() clears the old chunks for a document first, then inserts the new set, so a re-analysis never leaves stale chunks behind. get_chunks() reads them back:

def save_chunks(client: MongoClient, document_id: str, chunks: list) -> None:
    """Replace the stored chunks for one document with a fresh set."""
    # Clear any existing chunks for this document, then insert the new set:
    collection = get_chunks_collection(client)
    collection.delete_many({"document_id": document_id})
    if chunks:
        collection.insert_many(chunks)


def get_chunks(client: MongoClient, document_id: str) -> list:
    """Return every stored chunk for one document."""
    return list(get_chunks_collection(client).find({"document_id": document_id}))

Keep sample.pdf in your folder and run the temporary check:

python qa.py

You should see the chunk count, how many chunks carry a section, and the first chunk of each section. The sections should track the outline as the page numbers climb. A few chunks at the very front of the document show no section, which is correct, because they sit before the first heading.

Generate Embeddings

An embedding is a list of numbers that captures the meaning of a piece of text. Two pieces of text with similar meanings produce vectors that sit close together. This lets the tool match a question to relevant text based on meaning rather than exact words. This section turns each chunk into a vector with the local nomic model.

The nomic model was trained to treat stored text and search queries differently, so it uses two prefixes. The search_document: prefix marks text you store, and the search_query: prefix marks a question you search with. These prefixes are a trained behavior, not formatting, and the results degrade when you mix them.

Create a file named embedding_utils.py in your pdf-doc-analyzer folder, and add the following:

import os
import warnings

# Quiet the noisy startup logs before the model loads. The deprecation notice about
# get_extended_attention_mask comes from the nomic model's own downloaded code, not from
# anything here, so quieting the logs is the practical way to keep the terminal clean:
os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error")
warnings.filterwarnings("ignore")

from sentence_transformers import SentenceTransformer

import config

try:
    from transformers.utils import logging as hf_logging
    hf_logging.set_verbosity_error()
except Exception:
    pass

# Load the model once when this file is first imported:
model = SentenceTransformer(config.EMBEDDING_MODEL, trust_remote_code=True)


def embed_documents(texts: list[str]) -> list[list[float]]:
    """Embed many chunks in one batched call, far faster than one call per chunk."""
    # nomic wants a "search_document:" prefix for the text you store:
    prefixed = [f"search_document: {text}" for text in texts]
    vectors = model.encode(prefixed, batch_size=32, show_progress_bar=False)
    return vectors.tolist()


def embed_query(text: str) -> list[float]:
    """Embed a single question at search time."""
    # nomic wants a "search_query:" prefix for the question you search with:
    return model.encode(f"search_query: {text}").tolist()

The embed_documents() function embeds a whole list of chunks in one batched call, which is faster than embedding them one at a time. The embed_query() function embeds a single question at search time. Both return plain Python lists that are ready to store in MongoDB or pass to $vectorSearch.

`trust_remote_code=True` lets the model run Python code it downloads from Hugging Face on your machine. If that source code is ever compromised, the code runs automatically in your environment. Pin the model version for any non-prototype use, and review the source before you rely on it in production.

The first run downloads the model, about 500 MB, and caches it. Every later run loads it from that cache. You may see a one-time notice about `get_extended_attention_mask` or unauthenticated requests, both of which are harmless.

You'll test embeddings as part of the question-and-answer flow. For a quick check now, an embedding from embed_query("test") has a length of 768, which matches config.EMBEDDING_DIMENSIONS.

Create the Vector Search Index

A vector search index tells MongoDB which field holds the vectors, how many numbers each vector has, and how to measure closeness between them. You have to create it before you can run any meaning-based search. This section builds an index on the chunks.embedding field, plus a filter field so a question can target a single document.

Two details matter:

The numDimensions value has to match your embedding model exactly, so it stays at 768 for nomic-embed-text-v1. A mismatch makes the index fail.
The filter field on document_id is what lets a question search only the chunks from one PDF.

Create a file named vector_index.py in your pdf-doc-analyzer folder, and add the following code:

import time

from pymongo import MongoClient
from pymongo.operations import SearchIndexModel

import config
import db


def _index_exists(collection) -> bool:
    """True when the vector search index already exists on this collection."""
    return any(index["name"] == config.VECTOR_INDEX_NAME
              for index in collection.list_search_indexes())


def create_vector_index(client: MongoClient):
    """Create the Atlas Vector Search index over the chunks' embeddings, and wait until it's ready."""
    database = client[config.DB_NAME]
    # The collection has to exist before MongoDB Atlas can index it:
    if config.CHUNKS_COLLECTION not in database.list_collection_names():
        database.create_collection(config.CHUNKS_COLLECTION)
    collection = db.get_chunks_collection(client)

    search_index_model = SearchIndexModel(
        definition={
            "fields": [
                {"type": "vector", "path": "embedding",
                 "numDimensions": config.EMBEDDING_DIMENSIONS, "similarity": "cosine"},
                {"type": "filter", "path": "document_id"},
            ]
        },
        name=config.VECTOR_INDEX_NAME,
        type="vectorSearch",
    )
    result = collection.create_search_index(model=search_index_model)

    # The index builds in the background, so wait until MongoDB Atlas reports it queryable:
    predicate = lambda index: index.get("queryable") is True
    while True:
        indices = list(collection.list_search_indexes(result))
        if len(indices) and predicate(indices[0]):
            break
        time.sleep(5)  # check every 5 seconds

    print(result + " is ready for querying.")
    return result


def ensure_vector_index(client: MongoClient):
    """Create the index only when it's missing, so the ask command can call it safely on every run."""
    database = client[config.DB_NAME]
    if config.CHUNKS_COLLECTION not in database.list_collection_names():
        database.create_collection(config.CHUNKS_COLLECTION)
    collection = db.get_chunks_collection(client)
    if _index_exists(collection):
        return config.VECTOR_INDEX_NAME
    return create_vector_index(client)


if __name__ == "__main__":
    client = db.get_client()
    try:
        create_vector_index(client)
    finally:
        client.close()

Two things happen here, and they're separate:

The code creates the index, which is quick.
The code waits because the index becomes searchable only after MongoDB Atlas finishes building it in the background. The polling loop checks every 5 seconds until MongoDB Atlas reports the index as queryable.

The ensure_vector_index() function creates the index only when it's missing. The ask command calls it on startup, so the index gets built automatically the first time you ask a question. You can also run vector_index.py directly to build the index before asking any questions and wait for it to become ready:

python vector_index.py

The index builds in the background and can take up to two minutes. It builds only once. You can run this before any chunks exist, because the code creates the collection first.

Build the Interactive Q&A Command

The ask command starts an interactive question-and-answer session for the sample PDF.

The first question for a PDF runs the one-time preparation, then answers:

Make sure the vector index exists.
Extract, outline, and summarize the PDF if that hasn't happened yet.
Chunk the document, embed every chunk in one batch, and store the chunks.
Wait until at least one of those chunks is searchable in the index.
Embed the question, then retrieve the closest chunks for that one PDF.
Pass the retrieved chunks to the model and generate the answer.
Display the sources from the retrieved chunks.

Every later question skips steps 2 through 4 and reuses the stored chunks and their embeddings. It still embeds the new question in step 5 for searching, so a later question doesn't re-embed the document. The stages stay separate, so a weak answer is easy to trace to the stage that produced it.

Add the answer generation

The model call that writes the answer belongs with the other model calls, so you add it to llm_utils.py. It labels each retrieved chunk with its page and section, then tells the model to answer using only that context. It also puts the document's own title and author at the top of the context.

The prompt keeps the answer grounded. It tells the model to use only the supplied context, to say so plainly when the context falls short, and to keep the comparisons and relationships the context describes. It also tells the model not to invent page numbers, section names, chunk IDs, or source references, because the application builds the source list from the retrieved records. The title and author are metadata in the document rather than retrieved text, so the application places them directly into the answer context. The question still runs through the normal retrieval flow, and the model sees the retrieved chunks too, so the metadata sits alongside them and stays available even when vector search doesn't return the page that names the author.

Open llm_utils.py and add this function before the temporary check block:

def generate_answer(question: str, chunks: list[dict], title: str = "", author: str = "") -> str:
    """Answer a question using the document metadata and the retrieved chunks as context."""
    if not chunks and not (title or author):
        return "I couldn't find anything about that in the document."

    parts = []
    # Put the document's own title and author up front, so metadata questions are answerable:
    if title or author:
        meta = "Document metadata:"
        if title:
            meta += f"\nTitle: {title}"
        if author:
            meta += f"\nAuthor: {author}"
        parts.append(meta)
    # Label each chunk with its page and section so the model can ground its answer:
    for c in chunks:
        parts.append(f'[Page {c["page_start"]}, {c.get("section") or "no section"}]\n{c["text"]}')
    context = "\n\n".join(parts)
    response = ollama.chat(
        model=config.LLM_MODEL,
        messages=[
            {"role": "system", "content":
                "Answer the question using only the context provided. "
                "Do not add facts from outside the context, and do not fill gaps with general knowledge. "
                "When the context doesn't contain enough to answer, say so plainly rather than guessing. "
                "Explain terms and concepts according to how the document uses them, "
                "rather than giving generic definitions. "
                "Keep the comparisons and relationships the context describes. "
                "Do not invent page numbers, section names, chunk IDs, or source references. "
                "The application supplies the sources separately, so write your answer in prose "
                "without citing pages or sections yourself."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return response["message"]["content"].strip()

Add retrieval and the answer flow

Open qa.py and replace the import line at the top of the file with the following:

import os
import time

from pymongo import MongoClient
from pymongo.errors import OperationFailure

import analyzer
import config
import db
import embedding_utils
import llm_utils
import pdf_utils
from models import Chunk, Document, OutlineEntry

Then add these four functions under chunk_document():

def embed_and_store(client: MongoClient, doc: Document) -> int:
    """Chunk the document, embed all chunks in one batch, and save them. Returns the chunk count."""
    chunks = chunk_document(doc)
    vectors = embedding_utils.embed_documents([chunk.text for chunk in chunks])
    records = []
    for chunk, vector in zip(chunks, vectors):
        records.append({
            "document_id": doc.filename,
            "chunk_index": chunk.chunk_index,
            "text": chunk.text,
            "page_start": chunk.page_start,
            "page_end": chunk.page_end,
            "section": chunk.section,
            "embedding": vector,
        })
    db.save_chunks(client, doc.filename, records)
    return len(records)


def retrieve(client: MongoClient, document_id: str, question: str, k: int = 5) -> list[dict]:
    """Find the k chunks whose meaning is closest to the question."""
    query_vector = embedding_utils.embed_query(question)
    pipeline = [
        {"$vectorSearch": {
            "index": config.VECTOR_INDEX_NAME,
            "path": "embedding",
            "queryVector": query_vector,
            "numCandidates": k * 20,     # start near 20x the limit, then tune
            "limit": k,
            "filter": {"document_id": document_id},
        }},
        {"$project": {"_id": 0, "text": 1, "page_start": 1, "page_end": 1,
                      "section": 1, "chunk_index": 1,
                      "score": {"$meta": "vectorSearchScore"}}},
    ]
    return list(db.get_chunks_collection(client).aggregate(pipeline))


def _wait_until_searchable(client: MongoClient, document_id: str, timeout: int = 90) -> None:
    """Newly inserted chunks take a moment to reach the vector index. Poll until one is searchable."""
    probe = embedding_utils.embed_query("test")
    collection = db.get_chunks_collection(client)
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            hits = list(collection.aggregate([
                {"$vectorSearch": {
                    "index": config.VECTOR_INDEX_NAME,
                    "path": "embedding",
                    "queryVector": probe,
                    "numCandidates": 10,
                    "limit": 1,
                    "filter": {"document_id": document_id},
                }},
            ]))
        except OperationFailure:
            # MongoDB Atlas can report the index as not ready right after the insert.
            # Retry only this expected condition, and let every other error propagate:
            hits = []
        if hits:
            return
        time.sleep(3)  # check every 3 seconds
    # The chunks never became searchable, so fail loudly rather than query an unready index:
    raise TimeoutError(
        f"Chunks for {document_id} were not searchable within {timeout} seconds."
    )


def answer_question(client: MongoClient, pdf_path: str, question: str, k: int = 5) -> dict:
    """Answer a question about a PDF, and return the answer with its sources."""
    filename = os.path.basename(pdf_path)

    # Analyze the PDF the first time we see it. This builds and saves the outline and summary:
    if db.get_document(client, filename) is None:
        analyzer.analyze(client, pdf_path)

    # Build and store chunks the first time we answer for this PDF:
    if not db.get_chunks(client, filename):
        doc = pdf_utils.extract(pdf_path)            # fresh pages for chunking
        record = db.get_document(client, filename)   # reuse the saved outline
        doc.outline = [OutlineEntry(title=e["title"], level=e["level"], page=e["page"])
                       for e in record["outline"]]
        embed_and_store(client, doc)
        _wait_until_searchable(client, filename)     # let the index catch up before we query

    record = db.get_document(client, filename)   # for the document title and author
    chunks = retrieve(client, filename, question, k)
    answer_text = llm_utils.generate_answer(question, chunks,
                                            title=record.get("title", ""),
                                            author=record.get("author", ""))
    sources = [{"chunk_index": c["chunk_index"], "page": c["page_start"], "section": c.get("section")}
               for c in chunks]
    return {"answer": answer_text, "sources": sources}

The retrieve() function embeds the question, then runs $vectorSearch on the stored chunk vectors. The numCandidates value sets how many near neighbors the search considers before it returns the top limit results. A larger value can improve recall and costs some query time. It's recommended to start with at least 20 times the limit for approximate nearest-neighbor search. The code therefore, starts with numCandidates set to 20 times k and you can tune it based on your data and queries. The filter on document_id keeps the search within a single PDF, so answers stay grounded in the document you asked about.

Retrieval and answer generation are separate stages. MongoDB Vector Search selects the chunks and returns them with their stored page and section fields. The application passes those chunks to the model as context, and the model writes the answer from that context. The model doesn't run the search or decide which pages or sections count as sources. The application builds the Sources panel from the retrieved chunk records.

The answer_question() function handles the first-run setup for you. It analyzes the PDF when it's new, and it embeds and stores the chunks the first time you ask about it. Every later question reuses that stored work. A later question still embeds the one you typed, so it can search, and it reuses the stored chunk embeddings rather than re-embedding the document.

The _wait_until_searchable() poll matters. MongoDB Atlas indexes new chunks in the background, so a search that runs a split second after the insert can come back empty, which shows up as a "couldn't find anything" answer on the very first question. The poll sends a throwaway search every 3 seconds until a chunk from this document appears, so the subsequent real search returns results. If a chunk never becomes searchable within the timeout, the poll raises an error rather than letting retrieval run against an unready index. The index creation and the searchability wait are two different operations. ensure_vector_index() builds the index definition and waits for that definition to become queryable, which happens once. _wait_until_searchable() handles the separate, per-first-question delay while MongoDB Atlas adds the freshly inserted chunks to that existing index.

Your first question might take a minute to respond to, so give it up to a minute. It embeds every chunk in one batch, stores them, and waits for the MongoDB Atlas index to catch up. The spinner shows while it works. Every later question in the session reuses the stored chunks and answers quickly.

Add the ask command

Open cli.py and replace the whole file with this version, which adds the ask command next to analyze:

import typer
from rich.console import Console
from rich.panel import Panel
from rich.tree import Tree

import analyzer
import db
import qa
import vector_index
from models import Document

app = typer.Typer(help="Analyze a PDF and ask questions about it.")
console = Console()


# Keep the CLI ready for the ask command you'll add later:
@app.callback()
def main():
    """Analyze a PDF and ask questions about it."""


def _render(doc: Document) -> None:
    """Print the outline in a titled panel and the summary in another."""
    tree = Tree(f"[bold]{doc.filename}[/bold]")
    parents = {0: tree}
    for entry in doc.outline:
        parent = parents.get(entry.level - 1, tree)
        node = parent.add(f"p{entry.page}: {entry.title}")
        parents[entry.level] = node
    console.print(Panel(tree, title="Outline"))
    if doc.summary:
        console.print(Panel(doc.summary, title="Summary"))


@app.command()
def analyze(
    pdf_path: str,
    force: bool = typer.Option(
        False,
        help="Re-run even if a saved result exists.",
    ),
):
    """Build the outline and summary for a PDF."""
    client = db.get_client()
    try:
        doc = analyzer.analyze(client, pdf_path, force=force)
        _render(doc)
    finally:
        client.close()


@app.command()
def ask(pdf_path: str):
    """Start an interactive question-and-answer session about a PDF, with sources."""
    client = db.get_client()
    try:
        # Make sure the vector index exists before the first question runs:
        with console.status("Preparing the document..."):
            vector_index.ensure_vector_index(client)
        console.print(f"Ask questions about [bold]{pdf_path}[/bold]. Press Enter on an empty line to quit.")
        while True:
            question = console.input("\n[bold cyan]Question:[/bold cyan] ").strip()
            if not question or question.lower() in {"exit", "quit", "q"}:
                console.print("Goodbye.")
                break
            with console.status("Thinking..."):
                result = qa.answer_question(client, pdf_path, question)
            console.print(Panel(result["answer"], title="Answer"))
            if result["sources"]:
                lines = [f"p{s['page']}: {s['section'] or '(no section)'}" for s in result["sources"]]
                console.print(Panel("\n".join(lines), title="Sources"))
    finally:
        client.close()


if __name__ == "__main__":
    app()

Make sure Ollama is running, then start a session:

python cli.py ask sample.pdf

Type a question at the prompt, such as "What do 'ecological fidelity' and 'reproducibility' mean in the context of Agent-Diff?", and press Enter. You should get an answer drawn from the document, along with a Sources panel that lists the page and section behind it, such as p1: 2 Related Work. Open the PDF to those pages and confirm the listed section really covers the answer. Ask as many questions as you like, then press Enter on an empty line to quit.

This is an interactive question-and-answer session, not a chat. The tool prompts for a question in a loop and answers each one on its own, so the second question doesn't inherit context from the first. A question like "What is the main argument?" followed by "Who wrote the document?" produces two independent answers. Press Enter on an empty line, or type `quit`, to leave.

Key Takeaways

The tool runs almost entirely on your machine. Ollama serves the model locally; embeddings are run locally via sentence-transformers; and only MongoDB Atlas is remote, on the free M0 tier.
The ask command starts an interactive session for a PDF. The first question does the one-time chunking, embedding, and storage, and every subsequent question reuses that work.
MongoDB Atlas holds the documents and the chunks together and runs the vector search behind retrieval, so the data and the vectors live in one place.
Extraction runs without the model, and the separate RAG stages make a weak answer easy to trace to the stage that failed.
Sources come from the page and section stored with each retrieved chunk, not from the model, so every answer points back to a real passage in the document.

You can find all the code samples for this tutorial in the GitHub repository.

Building a Movie Recommendation Agent with Microsoft Agent Framework and MongoDB

MongoDB Guests — Mon, 17 Aug 2026 11:06:18 +0000

This tutorial was written by Markus Wildgruber.

In October 2025, Microsoft announced a new iteration of their AI Agent framework, called Microsoft Agent Framework. General Availability of the framework was announced in April 2026.

This framework is directed at C# and Python developers and seeks to consolidate the previous frameworks, Semantic Kernel, and AutoGen. If you have used Semantic Kernel before, some of the interfaces will be familiar to you, but especially the RAG part has changed a bit. This is what we will focus on throughout this article.

The sample: a Movie Recommendation Agent

In their sample_mflix-database, MongoDB provides a collection embedded_movies that already has embeddings created for a part of their movies collection. Equipped with this dataset, we can easily create a Movie Recommendation Agent that queries the collection using a vector search and provides answers grounded in this data.

To run this sample, you need the following components:

An instance of the MongoDB sample_mflix database with the embedded_movies collection. Please see this link for instructions on creating a free MongoDB Atlas M0 cluster. This article shows how to load the sample data into your cluster.
A vector search index named default on the embedded_movies collection. You can create this index in the MongoDB Atlas UI or have the sample code create it for you. Please note that creating the index will take a small amount of time. Use the following parameters when creating the index manually:
- Vector similarity: Cosine
- Dimensions: 1536 (the embeddings have been generated with text-embedding-ada-002 that yields vectors with 1536 dimensions)
A large language model (LLM) that is used for generating chat messages. This model must be capable of using tools. In this sample, we have used a deployment of OpenAI's gpt-4.1 model hosted on Azure OpenAI.
An embedding model that is used to vectorize the queries that the LLM generates when retrieving data for its chat messages. As the embeddings in the collection have been created using text-embedding-ada-002, we also need to use this model. In our case, this model was also hosted on Azure OpenAI.

The sample uses Blazor to create a basic user interface for the agent and shows two methods on how to put together the vector search:

Using MongoDB C# Driver directly
Using MongoDB Vector Store Connector

You can clone the repository with the sample code from Github. After downloading the sample, you can open it in Visual Studio Code or Visual Studio. To configure the secrets, you can use dotnet user-secrets and set at least the following configuration settings for the database connection string, the Azure OpenAI endpoint, and the name of the chat deployment:

dotnet user-secrets set MongoDB:ConnectionString "mongodb+srv://..."
dotnet user-secrets set Endpoint "https://..."
dotnet user-secrets set ChatDeployment "gpt-4.1"

The following two settings are optional:

dotnet user-secrets set MongoDB:Database "sample_mflix"
dotnet user-secrets set EmbeddingDeployment "text-embedding-ada-002"

A central piece is our Movie class that maps the MongoDB documents to a C# POCO:

using Microsoft.Extensions.VectorData;
using MongoDB.Bson.Serialization.Attributes;

namespace MongoDBSamples;

[BsonIgnoreExtraElements]
public record Movie(
    [property: BsonRepresentation(
        MongoDB.Bson.BsonType.ObjectId), 
        VectorStoreKey()] string Id,
    [property: BsonElement("title")] string Title,
    [property: BsonElement("plot")] string Plot,
    [property: BsonElement("fullplot")] string FullPlot,
    [property: BsonElement("cast")] List<string> Cast,
    [property: BsonElement("genres")] List<string> Genres,
    [property: BsonElement("plot_embedding"), VectorStoreVector(1536)]
        ReadOnlyMemory<float> PlotEmbedding
);

As you can see, the definition contains both attributes for declarative MongoDB mappings and also an attribute for the Vector Store implementation.

Setting up the clients

The first step when creating an agent with Microsoft Agent Framework is to set up a ChatClient instance. In our case, we will use dependency injection to set up singleton instances of AzureOpenAIClient and ChatClient:

builder.Services.AddSingleton<AzureOpenAIClient>(prov =>
    new(new Uri(cfg.Endpoint), new DefaultAzureCredential())
);
builder.Services.AddSingleton<ChatClient>(prov =>
    prov.GetRequiredService<AzureOpenAIClient>().GetChatClient(cfg.ChatDeployment)
);

In this case, it is a two-step process: first, we create an instance of AzureOpenAIClient, which is initialized with our configuration settings and uses a DefaultAzureCredential for access (in a production scenario, we would use a more fine-grained approach to explicitly use the authentication method of your choosing). Your user needs to be authenticated with Azure and have access to the models in Azure OpenAI for access to work.

Afterwards, we create the ChatClient instance that is the centerpiece of our agent.

In addition, we need an EmbeddingClient to generate the vectors for our queries:

builder.Services.AddSingleton<EmbeddingClient>(prov =>
    prov.GetRequiredService<AzureOpenAIClient>().GetEmbeddingClient(cfg.EmbeddingDeployment)
);

In addition, we register singleton instances of IMongoClient, IMongoDatabase and IMongoCollection<Movie> with the IoC-container. Please note that a vector search index named default is created when requesting the IMongoCollection<Movie> instance, in case it does not exist yet. This is to simplify running the sample, but the index should be created explicitly in a production scenario. Should the index be created, wait a bit for the index to be ready before hitting it with queries.

If you are working on a MongoDB Atlas M0 cluster, there is a limit on the number of search indexes you can create. If you already have three search indexes in your cluster, you need to remove one to run the sample.

Movie recommendation service implementation

In order to reuse the common parts of the Movie Recommendation Service, there is an interface IMovieRecommendationService that is implemented in the following class hierarchy:

Both approaches demonstrated in the following chapters share a lot of common code, e.g., the need to keep the chat history and the common steps for asking the agent for a recommendation. By creating a class hierarchy, the services can share a general approach and only have their specific logic for finding the relevant context via vector search.

Program.cs also contains the keyed registrations for the instances of the relevant services:

MongoDBVectorSearchMovieRecommendationService performs a vector search directly with the tools of the MongoDB C# Driver.
MongoDBVectorStoreMovieRecommendationService uses the MongoDB Vector Store Connector that has been created for Microsoft Semantic Kernel. So if you are already using this approach, the sample shows how to continue using a MongoDB-specific implementation of a VectorStore.

Through these keyed registrations, we can later on have a shared UI that uses the value of a route parameter to decide which implementation to use.

Creating recommendations

Let's have a closer look at the base class MovieRecommendationServiceBase. This class implements the general algorithm and contains the shared logic for the agents.

In its constructor, it injects the ChatClient instance and creates the agent with a common set of TextSearchOptions:

        _agent = chatClient.AsAIAgent(
            new ChatClientAgentOptions
            {
                ChatOptions = new()
                {
                    Instructions = """
                    You are a conversational movie recommendation agent. 
                    ...lengthy system prompt
                    """,
                },
               AIContextProviders = 
                   [new TextSearchProvider(RetrieveContextAsync, textSearchOptions)]
            }
        );

A crucial part when creating the agent is providing the instructions or system prompt that describes its capabilities:

You are a conversational and enthusiastic movie recommendation agent. Your goal is to help users find films based on their preferences.

**Core Guidelines:**
1. **Tool Usage:** You have access to the `TextSearch` tool. You must use this tool to verify plots, cast, genres, and release dates before answering. Do not rely on general training data for specific movie details.
2. **Citations:** When providing movie details, include a source link in your answer. Use ONLY the relative links found in the tool search results (e.g., [Link Text](/movies/123)). Do not invent or link to external URLs.
3. **Accuracy:** If the search tool returns no results for a specific query, do not hallucinate. State clearly that the movie or details are not found in your database.
4. **Conciseness & Tone:** Keep responses concise (maximum 2-3 sentences) but maintain an energetic, helpful tone. Use Markdown formatting (bolding, lists) to improve readability.
5. **Scope:** You specialize in movies. If a user asks about non-movie topics, politely redirect the conversation back to film recommendations.

**Response Format:**
- Use Markdown.
- Prioritize tool-verified information.
- Ensure all links are relative to the current application.

As you can see from the prompt above, it strongly encourages the agent to use context rather than rely on its own training data when providing recommendations. It also guides it to only use relative links within the current web application. This way, we can provide further details based on the data in our MongoDB database.

The AIContextProviders property is initialized with a list of context providers that - in our case - contains a TextSearchProvider instance.

The textSearchOptions are set up like this:

TextSearchProviderOptions textSearchOptions = new()
{
    SearchTime = 
        TextSearchProviderOptions.TextSearchBehavior.OnDemandFunctionCalling,
};

This defines how the agent should retrieve the context: OnDemandFunctionCalling tells the agent to use the provider like a tool and use it as it sees fit when answering the question, while BeforeAIInvoke leads to a context search being conducted before calling the LLM - the classic RAG approach. In our case, we use OnDemandFunctionCalling so that the LLM has a chance to adjust the query that is used to retrieve the context and optimize it for the query.

The most important part of the TextSearchProvider is the provided RetrieveContextAsync method. This method calls the RetrieveContextAsync method that is responsible for finding the relevant context:

private async Task<IEnumerable<TextSearchProvider.TextSearchResult>> RetrieveContextAsync(
    string query,
    CancellationToken ct
)
{
    _logger.LogInformation(
        "Retrieving context using method {Method} with query: {Query}",
        Name,
        query
    );

    List<Movie> movies = await RetrieveMoviesAsync(query, ct);

    _logger.LogInformation(
        "Retrieved {Count} movies for query {Query} with method {Method}",
        movies.Count,
        query,
        Name
    );

    return movies.Select(m => MapToSearchResult(m));
}

At the end, it converts the movies to TextSearchResult objects that the agent uses. This format contains a title, a link to details, and relevant information about the movie in text form:

private static TextSearchProvider.TextSearchResult MapToSearchResult(Movie m)
{
    var cast = m.Cast ?? new List<string>();
    var genres = m.Genres ?? new List<string>();

    return new TextSearchProvider.TextSearchResult
    {
        SourceName = m.Title,
        SourceLink = $"/movies/{m.Id}",
        Text =
            $@"""
            Title: {m.Title}
            Plot: {m.Plot}
            Full plot: {m.FullPlot}
            Cast: {string.Join(", ", cast)}
            Genres: {string.Join(", ", genres)}
            """,
    };
}

The base class then defines the abstract method that the derived classes need to implement:

protected abstract Task<List<Movie>> RetrieveMoviesAsync(string query, CancellationToken ct);

The main input is the query parameter that is used to retrieve matching documents from MongoDB.

This method is used by the implementations to convert the Movie objects returned into TextSearchResults for the agent.

The user interface

As stated above, this sample uses a Blazor application to let the user chat with the implementations of the agent. This application is kept very simple and would be much more sophisticated in terms of state management, etc.

The main component MovieRecommendator.razor receives the chosen implementation as a route parameter, creates the requested instance, and uses this to chat with the agent:

@page "/movie-recommendator/{key}"

// ...

protected override void OnParametersSet()
{
    base.OnParametersSet();
    RecommendationService = 
        ServiceProvider.GetRequiredKeyedService<IMovieRecommendationService>(key);
}

private async Task SendMessage()
{
    if (!string.IsNullOrWhiteSpace(userInput))
    {
        isLoading = true;
        StateHasChanged();
        try
        {
            AgentResponse response = 
                await RecommendationService.RecommendAsync(
                    userInput, CancellationToken.None);
            userInput = string.Empty;
        }
        finally
        {
            isLoading = false;
            StateHasChanged();
            await inputRef.FocusAsync();
        }
    }
}

Approach #1: MongoDB Vector Search with MongoDB C# Driver

The first implementation uses the methods that are provided by the MongoDB C# Driver directly, after vectorizing the query:

protected override async Task<List<Movie>> RetrieveMoviesAsync(
    string query,
    CancellationToken ct
)
{
    OpenAIEmbedding embedding = 
        await _embeddingClient.GenerateEmbeddingAsync(query);
    QueryVector queryVector = new(embedding.ToFloats());

    VectorSearchOptions<Movie>? vectorSearchOptions = new()
    {
        IndexName = "default",
        NumberOfCandidates = 100,
    };

    return await _moviesCollection
        .Aggregate()
        .VectorSearch(
            x => x.PlotEmbedding, 
            queryVector, 
            10, 
            vectorSearchOptions)
        .ToListAsync(cancellationToken: ct);
}

We limit our results to 10 movies, which should be enough for the LLM to pick a good match for the query.

Approach #2: Using the MongoDB Vector Store Connector

This implementation makes use of the MongoDB Vector Store Connector to perform the search. In the constructor, the VectorStore instance is created:

public MongoDBVectorStoreMovieRecommendationService(
    IMongoDatabase database,
    ChatClient chatClient,
    EmbeddingClient embeddingClient,
    ILogger<MongoDBVectorStoreMovieRecommendationService> logger
)
    : base(chatClient, logger)
{
    VectorStore vectorStore = new MongoVectorStore(database);
    _coll = vectorStore.GetCollection<string, Movie>("embedded_movies");
    _embeddingClient = embeddingClient;
}

When retrieving the data, the Vector Store is used to perform the vector search:

protected override async Task<List<Movie>> RetrieveMoviesAsync(
    string query,
    CancellationToken ct
)
{
    OpenAIEmbedding embedding = 
        await _embeddingClient.GenerateEmbeddingAsync(query);
    return await _coll
        .SearchAsync(new ReadOnlyMemory<float>(
            embedding.ToFloats().ToArray()), 
            10)
        .Select(m => m.Record)
        .ToListAsync();
}

As you can see, this is very similar to the first approach, but it shortens the code for the search considerably.

Wrapping it up

These two methods show how easy it is to provide an LLM with data from your MongoDB database and create a powerful agent that runs on your data. With Microsoft Agent Framework, you can integrate these functions directly into your application. It allows you to extend a wide range of .NET applications with AI and give powerful tools to your users.

Streaming MongoDB Change Events to Amazon Redshift via Atlas Stream Processing and Kinesis

MongoDB Guests — Wed, 05 Aug 2026 13:13:42 +0000

This tutorial was written by Igor Alekseev.

Real-time data pipelines that move operational data into analytical systems are a cornerstone of modern architectures. In this post, I’ll walk through how to stream MongoDB change events into Amazon Redshift using MongoDB Atlas Stream Processing and AWS Kinesis Data Streams — with no S3 staging or Firehose in between.

Architecture Overview

The flow is straightforward:

MongoDB Atlas — Changes (inserts, updates, replaces) occur in a collection.
Atlas Stream Processing — A stream processor captures these changes via a change stream, filters and reshapes them into flat JSON, and emits records to Kinesis.
AWS Kinesis Data Stream — Acts as the durable, ordered buffer between the source and the analytical store.
Amazon Redshift — A streaming materialized view reads directly from Kinesis and exposes the data as queryable columns.

The key advantage here is simplicity: there’s no intermediate S3 bucket, no Firehose delivery stream, and no ETL job to manage. Data flows from operational writes to analytical queries in near real-time.

Why This Combination?

Atlas Stream Processing handles the hardest part of change data capture — it connects natively to MongoDB change streams, handles resume tokens, and provides a pipeline language (aggregation framework) for reshaping documents on the fly. Instead of writing and hosting a custom consumer, you declare a pipeline, and Atlas manages the execution.

Kinesis provides a durable, ordered stream that decouples the producer (Atlas) from the consumer (Redshift). If Redshift has a maintenance window or you need to replay data, Kinesis retains records for up to 365 days.

Redshift Streaming Ingestion eliminates the traditional pattern of landing data in S3 first. A materialized view reads directly from the Kinesis stream, making new records queryable within seconds of arrival.

Implementation

Step 1: Create the Kinesis Data Stream

aws kinesis create-stream \
  --stream-name atlas-change-events \
  --shard-count 1 \
  --region us-east-1

A single shard handles up to 1 MB/s or 1,000 records/s ingest. Scale shards based on your change event volume.

Step 2: Set Up IAM for Atlas

Atlas Stream Processing authenticates to Kinesis via IAM AssumeRole. Create a role with a trust policy that allows the Atlas AWS account to assume it, and attach minimal Kinesis write permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:PutRecords",
        "kinesis:DescribeStreamSummary"
      ],
      "Resource": "arn:aws:kinesis:us-east-1:<YOUR_ACCOUNT>:stream/atlas-change-events"
    }
  ]
}

That’s it — just two permissions. Atlas uses the batch API (PutRecords) exclusively, so you don’t need the single-record API (PutRecord) or the full DescribeStream.

Step 3: Register the Kinesis Connection in Atlas

In the Atlas UI, navigate to Stream Processing → Connection Registry → Add Connection → Kinesis, and provide:

The stream name
The AWS region
The IAM role ARN

Step 4: Create the Stream Processor

The stream processor is an aggregation pipeline that runs continuously. Here’s the pattern:

sp.createStreamProcessor("atlasToKinesisProcessor", [
  // 1. Filter for relevant operation types
  { $match: { operationType: { $in: ["insert", "update", "replace"] } } },

  // 2. Flatten the change event — promote fullDocument fields to top level
  { $replaceRoot: {
      newRoot: { $mergeObjects: ["$fullDocument", { _operationType: "$operationType" }] }
  }},

  // 3. Project into a flat structure for Kinesis/Redshift
  { $project: {
      device_id: "$device_id",
      timestamp: "$timestamp",
      watts: "$obs.watts",
      temp: "$obs.temp",
      _operationType: 1
  }},

  // 4. Emit to Kinesis
  { $emit: {
      connectionName: "myKinesisConnection",
      stream: "atlas-change-events",
      region: "us-east-1",
      partitionKey: "$device_id"
  }}
]);

The $emit stage is Kinesis-specific. Note that stream, region, and partitionKey are all top-level fields — not nested inside a config object.

After creating the processor, start it in mongosh connected to your Stream Processing Instance (SPI):

sp.atlasToKinesisProcessor.start();

Step 5: Set Up Redshift Streaming Ingestion

On the Redshift side, create an external schema pointing to Kinesis:

CREATE EXTERNAL SCHEMA IF NOT EXISTS kinesis_schema
FROM KINESIS
IAM_ROLE 'arn:aws:iam::<YOUR_ACCOUNT>:role/<REDSHIFT_ROLE>';

Then create a streaming materialized view that parses the JSON records:

CREATE MATERIALIZED VIEW mdb_change_events AUTO REFRESH YES AS
SELECT
  approximate_arrival_timestamp,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'device_id') AS device_id,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'timestamp') AS event_timestamp,
  CAST(JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'watts') AS DOUBLE PRECISION) AS watts,
  CAST(JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'temp') AS DOUBLE PRECISION) AS temp,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), '_operationType') AS operation_type
FROM kinesis_schema."atlas-change-events";

With AUTO REFRESH YES, Redshift continuously ingests new records. You can also trigger a manual REFRESH MATERIALIZED VIEW mdb_change_events; if needed.

Step 6: Query Your Data

SELECT * FROM mdb_change_events
ORDER BY approximate_arrival_timestamp DESC
LIMIT 10;

That’s it. Operational writes in MongoDB are now queryable in Redshift within seconds.

Lessons Learned

Flatten before emitting. Redshift’s JSON_EXTRACT_PATH_TEXT is limited to 5 levels of nesting, has a 16MB record cap, and AWS recommends migrating to JSON_PARSE with SUPER type instead. By flattening in Atlas Stream Processing ($replaceRoot + $project), Kinesis receives simple flat JSON, and the Redshift materialized view stays clean.
The materialized view only captures new records. Records already in the stream when the view is created may not appear. Create the view before starting the processor, or insert fresh test data after creation.
External schema creation requires a superuser. The CREATE EXTERNAL SCHEMA statement must be run by the database owner or a user with the CREATE SCHEMA privilege. If your IAM-federated session doesn’t have these privileges, connect with the admin user and password instead.

When to Use This Pattern

This architecture works well when:

You need near-real-time analytics on operational MongoDB data
You want to avoid managing CDC infrastructure (Debezium, Kafka Connect, custom consumers)i
Your analytical queries run in Redshift, and you want columnar performance
You prefer a serverless, managed approach with minimal operational overhead

For higher throughput or more complex transformations, you might add more Kinesis shards and scale the Atlas Stream Processing instance accordingly. The architecture stays the same — just the capacity changes.

References

An agent is a graph

MongoDB Guests — Tue, 04 Aug 2026 16:04:18 +0000

This tutorial was written by Néstor Daza.

This is the fourth article in a series about building Claudius, my own Claude-based chatbot (Github). In the previous article, we scaffolded the core components of the app.

We already built the foundation: the data model, an identity the client cannot tamper with, and a health check that proves the application can reach both Atlas and Bedrock. We can now move to the streaming chat backbone, the phase where the conversation finally becomes real and Claudius turns into something you can use every day as your favorite AI tool.

We are going to build the live path, everything that happens between hitting send and tokens arriving on screen. The model runs as a small agent that can pause to search the web, stream its answer back word by word, and let you switch from a cheap model to a strong one in the middle of a thread without losing your place. The next article will be about where the conversation lives once the tokens stop, the persistence, the metering, and the guarded writing operations that keep the whole thing grounded. Here, the question is narrower and more immediate: what actually happens when you hit send?

The agent is not a prompt

The chat engine is not a single call to a model wrapped in a clever prompt; it’s a small state machine built with LangGraph, a design choice that lets the model do more than answer in one shot.

The graph has three nodes. load_context runs but does nothing yet, as a deliberate placeholder where memory retrieval will land in a later phase. agent is the model itself, bound to the tools it is allowed to call. tools is the executor that runs whatever the model asks for. A conditional edge sits after the agent; if the model requests a tool call, control passes to the tools node. If it returns a plain answer, the run ends. The tools node always loops back to the agent.

const builder = new StateGraph(MessagesAnnotation)
  .addNode("load_context", loadContext)
  .addNode("agent", agent)
  .addNode("tools", new ToolNode(tools))
  .addEdge(START, "load_context")
  .addEdge("load_context", "agent")
  .addConditionalEdges("agent", toolsCondition, ["tools", END])
  .addEdge("tools", "agent");

That loop keeps running until the model has what it needs, then returns a natural language answer instead of another tool call. A prompt answers once and stops, a graph keeps working until the task is finished.

Choose the model per run, not per graph

Here is the design decision that took the most thought, and it is invisible in the diagram above. The model is not compiled into the graph.

You might expect the agent node to close over a fixed model instance when the graph is constructed. Instead, the model is chosen per run and passed through LangGraph's configurable channel, the per-invocation configuration that travels alongside the graph state. The agent node reads the inference profile from that channel and builds its model client fresh on every turn.

async function agent(
  state: typeof MessagesAnnotation.State,
  config: RunnableConfig,
): Promise<typeof MessagesAnnotation.Update> {
  const { inferenceProfileId, maxTokens } = readConfigurable(config);
  const model = buildChatModel(
    inferenceProfileId,
    maxTokens !== undefined ? { maxTokens } : {},
  ).bindTools(tools);

  const response = await model.invoke(
    [new SystemMessage(SYSTEM_PROMPT), ...state.messages],
    config,
  );

  return { messages: [response] };
}

readConfigurable pulls the inference profile and token ceiling from the per-run config. buildChatModel constructs the ChatBedrockConverse client for that profile, and bindTools attaches the tool schema so the model knows what it is allowed to call (i.e, web_search). The node then invokes the model with the system prompt followed by the full message history, and returns the model's reply as the one update to the graph state.

Why go to this trouble? Because it is what makes switching models mid-conversation almost free. When you move from Haiku to Sonnet halfway through a thread, nothing about the stored conversation changes. The message history is untouched; on the next turn, we read a different inference profile from the config and build a different client using the same history. There is no graph to rebuild and no checkpoint to migrate. The model is a parameter of the turn, not a property of the conversation.

Model-switching appears in the request the client sends, not in the thread you see on screen. The rendered conversation carries no per-message model label, and the conversation row stores only the current model as a single field, so neither one records which model produced which turn. The request payload does, with each send carrying the conversation identifier, the chosen model, and the new text. Two consecutive sends in one conversation, sharing a conversationId and differing only in modelId, show up like this:

// Turn 3, sent while the selector was on Haiku
{
  "conversationId": "6770f1a2b3c4d5e6f7a8b9c0",
  "modelId": "claude-haiku",
  "text": "Summarize the tradeoffs we just covered."
}
// Turn 4, same conversation, selector now on Sonnet
{
  "conversationId": "6770f1a2b3c4d5e6f7a8b9c0",
  "modelId": "claude-sonnet",
  "text": "Now make the strongest case against that position."
}

The second turn still reads the full prior history that the server holds, so Sonnet answers turn four with turn three already in context, even though turn three ran on Haiku. The modelId is the public catalog identifier that the selector handed the client, paired with the display name on screen; the cross-region inference profile it resolves to remains on the server.

One tool, one predictable shape

The agent has just one tool in this phase, web_search, and it is a thin wrapper over Tavily. It returns exactly three fields per result: title, url, and snippet. Nothing else comes through.

The wrapper is hand-rolled rather than pulled from a library, to have a bit more control over the output shape. The model reads a compact JSON block, and the user interface renders it as a list of sources. Both sides depend on the shape staying fixed. Tavily returns its result text in a field called content, and the wrapper renames it to snippet at that boundary. Replace Tavily with another search service later, and the only code that changes is what's inside that wrapper.

This is a habit worth keeping past this one tool. A predictable shape at a boundary is cheap to build and prevents the rest of the system from developing a quiet dependency on a particular vendor's response format.

Bridging the event gap

The graph produces one kind of stream, the browser expects another; the route between them is a translation layer, the most intricate piece of this phase.

LangGraph emits a flat event log through its streamEvents interface. Model token chunks, tool starts, tool ends, and run boundaries all arrive as a sequence of typed events that you consume in order. The frontend uses the Vercel AI SDK, which expects something different, a typed stream of user-interface message parts. Natural text arrives as a text-start, then a run of text-delta events, then a text-end. Tool activity arrives as tool-input and tool-output parts. The two formats describe the same conversation, but they are not the same format. We need to close the gap between them.

A small state machine takes care of it. It walks the LangGraph event log and emits AI SDK parts as it goes.

for await (const ev of events) {
  switch (ev.event) {
    case "on_chat_model_stream": {
      const chunk = ev.data?.chunk as BaseMessage | undefined;
      const delta = chunk?.text ?? "";
      if (delta.length > 0) {
        if (textId === null) {
          textId = `text-${textSegment++}`;
          writer.write({ type: "text-start", id: textId });
        }
        writer.write({ type: "text-delta", id: textId, delta });
        assistantText += delta;
      }
      break;
    }
    case "on_chat_model_end": {
      endText();
      const output = ev.data?.output as
        | { usage_metadata?: UsageMetadata }
        | undefined;
      addUsage(totals, output?.usage_metadata);
      break;
    }
    case "on_tool_start": {
      writer.write({ type: "tool-input-start", toolCallId: ev.run_id, toolName: ev.name, dynamic: true });
      writer.write({ type: "tool-input-available", toolCallId: ev.run_id, toolName: ev.name, input: ev.data?.input ?? {}, dynamic: true });
      break;
    }
    case "on_tool_end": {
      writer.write({ type: "tool-output-available", toolCallId: ev.run_id, output: parseToolOutput(ev.data?.output), dynamic: true });
      break;
    }
  }
}

Read it event by event. On a model token chunk (on_chat_model_stream), if no text part is open yet, the machine opens one with a fresh identifier and a text-start, then writes the token as a text-delta and appends it to the running answer it keeps in assistantText. On the model turn ending (on_chat_model_end), endText closes the open part, emitting the text-end and resetting the identifier to null so the next token will open a new segment; the same handler pulls the token usage out of the event and adds it to the running totals, which the companion article picks up. On a tool starting (on_tool_start), the machine emits the tool-input parts. On a tool ending (on_tool_end), it parses the tool's JSON output back into a structured part that the renderer can show as sources.

Because the model finishes its turn before the tools it requires run, on_chat_model_end always fires ahead of on_tool_start. So each tool call sits cleanly between closed text segments, never inside an open one. The textSegment counter increments each time, so the answer the model writes after reading its search results opens as a fresh segment rather than reopening the one that came before the search. The renderer sees a clean sequence: natural text, then a tool call with its sources, then more text.

The tool parts are flagged dynamic. The web_search tool is not registered on the client with a schema, so as far as the AI SDK is concerned, its part is a dynamic tool, one whose shape is not known in advance. Marking it dynamic on the server lets the client assemble it into a cleanly typed part that the renderer can switch on, even though the client was never handed the tool's signature.

The conversation identifier rides back on a transient data part. When you send the first message in a brand-new conversation, the server creates the conversation row and needs to tell the client its identifier, without forcing a second request to get it. So it writes a one-off data-conversation part at the very top of the stream.

writer.write({ type: "start" });
writer.write({
  type: "data-conversation",
  data: { id: threadId, title: conversationTitle },
  transient: true,
});
writer.write({ type: "start-step" });

The client reads that part, adopts the identifier, updates the URL, and shows the new conversation in the sidebar, the part itself never becoming a message in the thread. transient marks it as out-of-band, a signal rather than content. No extra round trip, and the conversation is addressable from the first token onward.

The user sees what the agent did

The payoff of all that plumbing is visible on screen. Tokens animate in as they arrive, so the answer builds in front of you instead of appearing in one block after a wait. When the agent decides to search, a live line appears reading "Searching the web," and once results return, it collapses into an expandable list of the sources the tool found, each with its title and link.

That last behavior is a principle that runs through the whole product. Every tool call is visible and inspectable. The user should always be able to see what the agent did on their behalf, not only the polished answer that came out the other end. The event bridge is what makes this possible, because it carries tool activity through as first-class parts rather than burying it inside the model's turn.

There is one more thing to notice about the round trip, and it sets up the second half of the backbone. When you send a message, the client sends only the new text. It does not send back the transcript of everything said before. It cannot, really, because it does not hold the authoritative history. It holds only what it has rendered on screen. The full conversation lives elsewhere, and the model reads it from there on every turn.

Where it lives, and why one small identity decision makes reopening a week-old conversation as cheap as opening a fresh one, is the theme of the next article. The agent is a graph. The conversation, it turns out, is a database.

Building the foundation Claudius runs on

MongoDB Guests — Tue, 04 Aug 2026 15:21:07 +0000

This tutorial was written by Néstor Daza.

This is the third article in a series about building Claudius, my own Claude-based chatbot (Github). The previous article discussed the MongoDB data model to use for the app.

The previous article decided the shape of the data. None of it matters until the app around it is working, and getting it there is the unglamorous half of this phase. It comes down to three things: an identity system the client cannot tamper with, proof that Claudius can reach the two services it depends on, and the deployment realities that decide whether any of it runs at all. This is the boring work that quietly decides whether a project survives contact with production.

Identity: the client never gets a vote

Any Google account on Earth can sign into Claudius safely because a user's role is never something the client sends. It is decided on the server every time.

One piece of this lives outside the code. The Google provider needs an OAuth (Open Authorization) client that you register once in the Google Cloud Console, and the client identifier and secret from that registration are set in corresponding env variables. These setup steps live in the Auth.js and Google documentation, so I am not repeating them here.

Sign-in runs on Auth.js v5 with the Google provider and the MongoDB adapter. There are three roles, admin, member, and guest, and they resolve in exactly one place on the server, with a clear precedence:

export async function resolveRole(email: string | null | undefined): Promise<Role> {
  if (!email) return "guest";

  const normalized = email.toLowerCase();
  if (normalized === env.ADMIN_EMAIL.toLowerCase()) return "admin";

  const settings = await settingsCol();
  const allowlist = await settings.findOne({ _id: "allowlist" });
  if (allowlist && "emails" in allowlist) {
    const allowed = allowlist.emails.some((e) => e.toLowerCase() === normalized);
    if (allowed) return "member";
  }

  return "guest";
}

The bootstrap admin email wins outright, an allowlist of member emails in the settings collection comes next, and everyone else falls through to guest, the least-privileged default. The client supplies a Google identity and nothing more. It never names its own role, so there is no request a user can forge to promote themselves.

The session uses a JWT (JSON Web Token) rather than a database-backed session, so the resolved role is stored in the token and is available without a database read on every request. That choice has one small downside: the token caches the role until it refreshes, so adding someone to the member allowlist takes effect on their next sign-in, not the moment you edit the list. For a personal project, that is a fine trade.

The wiring is small: on first sign-in, the jwt callback calls provisionUser and stores the returned role on the token, and the session callback then copies it onto session.user for the app to read.

Provisioning is the other half of sign-in. The adapter has already written a bare user document, just the name, email, and image Google returns. It fills in the Claudius-specific fields with a single aggregation-pipeline update built around $ifNull, so each field defaults only when it is absent, and an admin's later change is never overwritten. The role is the exception, recomputed every time, since that is the field on which the whole security model rests:

export async function provisionUser(
  userId: string,
  email: string | null | undefined,
): Promise<Role> {
  const role = await resolveRole(email);
  const users = await usersCol();

  const resetsAt = new Date(Date.now() + 24 * 60 * 60 * 1000);

  await users.updateOne({ _id: new ObjectId(userId) }, [
    {
      $set: {
        role,
        allowedModels: { $ifNull: ["$allowedModels", null] },
        monthlyTokenBudget: { $ifNull: ["$monthlyTokenBudget", null] },
        status: { $ifNull: ["$status", "active"] },
        dailyMessageCount: {
          $ifNull: ["$dailyMessageCount", { count: 0, resetsAt }],
        },
      },
    },
  ]);

  return role;
}

One TypeScript detail in this layer was a bit of a pain. Augmenting the session type works through the next-auth module as you would expect, but the JWT type does not! It is only re-exported from next-auth/jwt and lives in @auth/core/jwt, so a declaration written against next-auth/jwt silently fails to merge, and the token field you were trying to type lands as unknown instead. You have to target the real module:

declare module "next-auth" {
  interface Session {
    user: {
      id: string;
      role: Role;
    } & DefaultSession["user"];
  }
}

// JWT lives in @auth/core/jwt; next-auth/jwt only re-exports it, so the
// augmentation must target the real module to merge rather than shadow.
declare module "@auth/core/jwt" {
  interface JWT {
    uid?: string;
    role?: Role;
  }
}

That silent failure is the worst kind: when there are no errors, the type just quietly goes wrong.

Connectivity: prove it reaches Atlas and Bedrock

The last piece of the skeleton is a single protected route that proves the app can talk to the things it depends on. Any signed-in role can call /api/health. It always pings Atlas, and given ?probe=bedrock it goes one step further and makes a one-token Converse call against Claude Haiku 4.5, to prove the credentials can invoke a real model. The probe is the interesting part: capped at a single output token, wrapped so no AWS internals leak back to the caller, and returning the counts read from the Converse usage metadata.

export async function bedrockHealthProbe(): Promise<BedrockHealth> {
  try {
    const llm = new ChatBedrockConverse({
      model: HAIKU_INFERENCE_PROFILE,
      region: env.AWS_REGION,
      credentials: {
        accessKeyId: env.AWS_ACCESS_KEY_ID,
        secretAccessKey: env.AWS_SECRET_ACCESS_KEY,
      },
      maxTokens: 1,
      temperature: 0,
    });

    const response = await llm.invoke([["human", "ping"]]);

    const usage = response.usage_metadata as TokenUsage | undefined;

    return {
      ok: true,
      inputTokens: usage?.input_tokens ?? null,
      outputTokens: usage?.output_tokens ?? null,
    };
  } catch {
    // Do not surface the underlying AWS/SDK error to callers.
    return { ok: false, error: "Bedrock probe failed" };
  }
}

On a live run, it comes back with the result you want:

{"ok":true,"inputTokens":8,"outputTokens":1}

Eight tokens in, one token out. That is the signal you want from a health check on an AI app, not only that the network is up, but that the credentials work and the meter is running.

The models, the probe, and the rest of the app are seeded into settings as a catalog. Each entry carries a real Bedrock cross-region inference profile ID rather than a bare model ID, which is the correct choice for cross-region routing:

[
{"id":"haiku","inferenceProfileId":"us.anthropic.claude-haiku-4-5-20251001-v1:0"},
  {"id":"sonnet","inferenceProfileId":"us.anthropic.claude-sonnet-4-6"},
  {"id":"opus","inferenceProfileId":"us.anthropic.claude-opus-4-8"}
]

Each entry also carries per-million-token input and output pricing, which a later phase uses to enforce tier costs. For now, those fields are a deliberate placeholder, to be confirmed against current Bedrock pricing before any billing math depends on them, so I am not quoting numbers here that would only mislead.

Env: validate only what this phase uses

As we work through this series, each development phase validates only the environment variables it uses, rather than declaring the project's entire surface up front. So far, we need just eight, for the database, Google auth, Bedrock, and the admin bootstrap, so the env schema gates exactly those. A missing or malformed value throws at startup, naming the offending variable and never printing its value:

const EnvSchema = z.object({
  MONGODB_URI: z.string().url(),
  AUTH_SECRET: z.string().min(1),
  AUTH_GOOGLE_ID: z.string().min(1),
  AUTH_GOOGLE_SECRET: z.string().min(1),
  AWS_ACCESS_KEY_ID: z.string().min(1),
  AWS_SECRET_ACCESS_KEY: z.string().min(1),
  AWS_REGION: z.string().min(1),
  ADMIN_EMAIL: z.string().email(),
});

const parsed = EnvSchema.safeParse(process.env);
if (!parsed.success) {
  // List the offending variable names only, never their values.
  const offenders = parsed.error.issues
    .map((issue) => issue.path.join("."))
    .join(", ");
  throw new Error(
    `Invalid environment configuration. Check these variables: ${offenders}`,
  );
}
export const env = parsed.data;

Other variables that will be introduced later, like Tavily for web search, Voyage for embeddings, Blob for file storage, and LangSmith for tracing, are not here yet.

The gotchas that cost real time

The skeleton works, but it did not the first time, and the failures are perhaps the most useful part of this build report.

The first was the most painful: the connection string has no database path, so anything that does not explicitly name the database falls back to test, which the Atlas credentials cannot touch. It bit twice, first in the index script:

MongoServerError: not authorized on test to execute command { listCollections: 1, ... $db: "test" }

Then again, after deployment, in production sign-in, because the Auth.js adapter does not go through the app's database helper and quietly defaults to test until it is told otherwise:

[auth][error] AdapterError
[auth][cause]: MongoServerError: not authorized on test to execute command
{ find: "accounts", ... $db: "test" }

The fix is to name the database explicitly in both places. The adapter gets its own databaseName:

// databaseName is set explicitly because the connection string has no default
// database; without it the adapter would target "test" (which the Atlas
// credentials cannot access).
export const { handlers, auth, signIn, signOut } = NextAuth({
  adapter: MongoDBAdapter(clientPromise, { databaseName: DB_NAME }),
  ...authConfig,
});

And the app reaches the database through one helper that names it the same way. That helper is also where the second gotcha lives. A naive new MongoClient() per request would leak a connection pool on every serverless invocation and every dev hot reload, so the client is cached on globalThis as a single instance, and DB_NAME is a constant rather than an env var because the database name is neither secret nor environment-specific:

const globalForMongo = globalThis as unknown as {
  _claudiusMongoClientPromise?: Promise<MongoClient>;
};

const client = new MongoClient(env.MONGODB_URI);

export const clientPromise: Promise<MongoClient> =
  globalForMongo._claudiusMongoClientPromise ??
  (globalForMongo._claudiusMongoClientPromise = client.connect());

export const DB_NAME = "claudius";

export async function getDb(): Promise<Db> {
  const connected = await clientPromise;
  return connected.db(DB_NAME);
}

The adapter is handed the same clientPromise, so auth, and the app shares one client and one pool rather than opening two. Making them agree on a single driver version took one line at the repo root, since the adapter and the app can otherwise resolve different ones:

"overrides": { "mongodb": "7.3.0" }

The third is a monorepo trap. The shared package is consumed as raw TypeScript and transpiled by the app's build through transpilePackages: ["@claudius/shared"], with no separate build step. That is convenient, but it means every runtime dependency the shared code imports has to resolve from the app workspace, because Vercel scopes its install to the app's root directory. The rule that keeps the build green is to declare a runtime dependency in every workspace that imports it, even when that looks like duplication.

The last is the kind of thing that only shows up on deploy. A lockfile generated on macOS omitted the Linux binaries that Tailwind's build needs, and Vercel's Linux install could not find them:

Error: Cannot find module '../lightningcss.linux-x64-gnu.node'

Declaring those binaries as optionalDependencies puts them in the lockfile for every platform, and the deploy goes green:

"optionalDependencies": {
  "@tailwindcss/oxide-linux-x64-gnu": "4.3.1",
  "lightningcss-linux-x64-gnu": "1.32.0"
}

What stands, and what a headless build cannot prove

No chat yet, but the foundation is real. Identity resolves on the server and rides safely in the token. The health route confirms Claudius can reach both Atlas and Bedrock, and the configuration fails loud and early when something is missing. The data model from the last article now has something solid underneath it.

The role-resolution logic is unit-tested, but the actual sign-in is an OAuth round trip, and that needs a real browser, not just this headless build. Confirming that an admin email lands as admin on the page and on the persisted user document is the one step I finish by hand.

Next is the streaming chat backbone, where the conversation finally gets a place to live, and I open the real checkpoint document I have been promising since the start.

Finally! MongoDB Atlas supports partial restore

MongoDB Guests — Mon, 03 Aug 2026 09:00:00 +0000

This article was written by Darshan Jayarama.

Reflecting on my tenure as a MongoDB TSE, I recall a specific incident where a DBA mistakenly dropped a production collection instead of a UAT one. At the time, we were compelled to restore the total cluster to a secondary instance, isolate the required data, and then re-import it into the live environment. The process was inefficient and expensive. We had to recover an entire 760GB cluster just to retrieve a single 50MB collection. You can imagine the massive resource drain involved.

One might wonder why we couldn't restore a single collection or select databases. That’s the new feature I am going to share with you today.

What is partial restore?

As explained in the MongoDB documentation, you can now simply restore any database or collection from the cloud backup; there is no need to restore the full cluster.

It means that if one collection is all you want, you can recover it without touching the entire dataset from the cloud backup.

Real-World Example

A production cluster will have a combination of multiple databases:

Live Production Environment

├── Customer_Profiles
├── Transaction_History
├── Stock_Management
├── Billing_Records
└── Data_Insights

If someone accidentally drops Billing_Records.invoices collection. To resolve this:

Before the Partial Restore feature:

Restore the entire cluster to a temporary Atlas cluster.
Wait for the restore to complete.
Export the required collection.
Import it back into production.
Delete the temporary cluster.

With the Partial Restore feature,simply choose:

Backup Snapshot
Database
Collection
Restore

Here is the workflow:
Click on the Backup tab, select the cluster, and select the backup snapshot as shown below.

Click the 3 dots per the selected snapshot, then click Restore.

In the next window, select the Restore Databases/Collection tab:

You will now have 2 options:

Create as new: With this option, you can choose not to remove the existing collection. If the restoring collection/DB name already exists, Atlas automatically renames the restoring collection to avoid a conflict.
Overwrite existing: Atlas replaces the existing collection with the restored version.

If you want to compare the existing and restored collection, then choose the "Create as new" option. If you want to replace the existing collection, then choose the "Overwrite existing" option. But please be aware that all the data written after the snapshot time will no longer exist.

After choosing your option, you can go ahead and select exactly the data you need. You have two ways to do this:

Using the UI: Just tick the checkboxes for the database or collection you want. Or, if you prefer, you can use the search bar to find what you need using a regex pattern.
Using List Entry: Type the database name or the database. collection name, one per line. For example, if you want everything from the Billing_records database, but only the NRI and HNI collections from the Customer_profile database, you just enter it like this: Billing_records Customer_profile.NRI Customer_profile.HNI

Keep one thing in mind —you cannot use a dot in your database name. If you use a dot, the system will treat it as a combined DB and collection name.

Once you are done with this part, just click Next. On the next screen, you will need to select the Destination details.

What Is MongoDB Atlas Vector Search

MongoDB Guests — Tue, 28 Jul 2026 10:01:53 +0000

This article was written by Moses Anumadu.

Traditional search relies on keywords and tags to recommend search results. This system works and is in use across multiple platforms. However, it has a major limitation that has become more apparent in recent years due to increased AI adoption. The major limitation of traditional search is that it returns search results based solely on keyword matches. Traditional searches, as we have them today, do not consider semantic meaning. For example:

Imagine that we built a movie website using traditional search. Now, if a user wants to find The Matrix, but they can't remember the name. They type in "dystopian AI reality simulation." A traditional keyword database will return zero results because those exact words are not in the movie title or description. This is the major challenge for traditional search. It returns results strictly from keywords, in this case, either in the movie name or description. So, if a user does not type the exact keywords stored in your database, the correct or intended movie does not show up, even though it is in the database. The platform simply does not know that it was the intended movie. This can be limiting and produce search results that do not accurately match the users' intent.

A better approach for this use case is to have the search not rely on keywords to recommend movies. It should recommend movies based on semantic similarities. This is a very good example of when vector search should be used over traditional search.

In this tutorial, we will explore what vector search is, its importance, what MongoDB Vector Search is, and why you should consider it for your next vector search implementation. With that said, let's get started.

Prerequisites

To follow along with this article, you should have a basic understanding of the following concepts:

MongoDB Atlas
Database and collection
Can perform CRUD operations
Basic understanding of web development technologies

What Is Vector Search?

Vector search is a search technology that uses vectors to identify semantic similarity and recommend results that are accurate in terms of meaning and not keywords.

With vector search, text, images, and other media file types are stored as embeddings. Embeddings are numerical values generated by machine learning models that capture the context, intent, and semantic similarity of the stored data. Items with similar meanings have embeddings (numeric values) that are closer.

When a search is done, the vector search identifies similar values by comparing the closeness of the content embeddings. The closer the embeddings, the more similar the contents. This returns a more accurate search result that relies on meaning over a keyword search that relies on keywords.

The major difference to note here is that keyword searches rely on keyword matches to recommend results, while vector searches rely on meaning by comparing embeddings to recommend results.

Keyword Search vs. Vector Search

Keyword Search	Vector Search
Search relies on exact keyword matching to return results.	Search relies on a semantic understanding of search terms.
Results are not generated based on users' intentions. They rely on keyword accuracy.	Generates results with users' intent in consideration.
The use of synonyms in keyword searches does not often return the best search results. Eg, car and vehicle won't return the same result.	Synonyms in vector search generate far more accurate results. It factors in the actual meaning of the words to generate results.
Keyword search queries are not flexible to things like typos and long search queries.	Search keywords or phrases can be sentences. It can be flexible with misspelled words and still return accurate results.

How Embeddings Enable Vector Search

What Is an Embedding?

Think of embeddings as the foundation of vector searches. Embeddings are numerical values generated by artificial intelligence. These numeric values represent the meaning of text and media files, so vector search can compare them and return more accurate, similar results.

In a keyword search query, a traditional database does not treat words as symbols with meaning; it sees them as independent flat characters that don't have links to the sentence's meaning. So it only returns results that match exactly. An example can be a search keyword of "CAT" would return a CAT. Not because it understands it, but because the keywords match. But a search keyword of VEHICLE would not return a result of CAR.

How AI generates these similarities

Let's take a moment to examine how an AI embedding model generates the numbers in an embedding. Imagine you need to describe an apple using a scale of 1-10. Is an apple sweet? Yes. On a scale of 1 - 10, how sweet is an apple? Say 8. Is an apple crunchy? Yes. On a scale of 1 - 10, how crunchy is an apple? Say 6. Is an apple red in color? Yes. On a scale of 1 to 10, how red is an apple? Say 9. This would give us an embedding of [8, 6, 9]. A fruit like a banana can be [9, 1, 0].

The AI embedding model does something very similar to the example above; when it processes a sentence, it outputs a long list of numbers (called a vector). This list serves as a precise map coordinate for that specific piece of text in a giant, digital "map of meaning."
This results in words with similar meaning having vectors that are closely related or similar.

Why Embeddings Matter

Embeddings are like the building blocks of vectors. With them, vectors are generated, and similarity in terms of meaning can be measured in a search query.

How MongoDB Vector Search Works

The Vector Search Workflow

In a MongoDB vector search implementation, data is stored together with its generated vector in the same collection. This simplifies the implementation when compared to the steps required to implement this using different technologies. Let's examine the workflow for both storing a document and retrieving it during a search query.

Storing document

Storing record

This process starts with storing our normal documents in a MongoDB collection. A document can be in text or media format.

Embeddings are generated

After storing this document, embeddings are generated using any of the AI models that have this compatibility. Some popular examples are VoyageAI, OpenAI, Cohere, and Hugging Face.

Documents are updated to contain vectors.

The data collection in Atlas is updated to contain the generated vectors. This saves the vector as a field in the collection. Keeping both the data and its embeddings in one collection.

A vector search index is created

The final step in this flow is creating the vector search index. This boosts performance during searches.

The image below provides a visual explanation of the flow;

Search workflow

The MongoDB vector search workflow consists of a few steps; it starts with a user query. The query is converted into embeddings using an AI embedding model (for best results, use the same embedding model as the one used to embed the documents in the database is advised).

After the vectors are generated, MongoDB vector search calculates the semantic proximity between the search query vector and the vector of the documents stored in the collection. It then returns the results, which are semantically matching items to the user search query. The image below provides a more visual explanation of the process.

Common Use Cases

There is no doubt about the importance of vector searches. However, this does not make it the right tool for every search implementation. Vector search implementations require AI models to generate vectors. This can make the search implementation a bit more complex and costly. With that in mind, it is important to explore some real-world use cases where it can be a good fit and also where it is not. But in general, vector searches are a great solution for search results that want semantic similarity. Some examples include:

Semantic document search
AI knowledge bases
Product recommendation systems
Similar content discovery
FAQ search
AI assistants using business data

Benefits and Considerations

Benefits

Some of the key benefits include:

More relevant search results

Using vectors to measure similarity enables vector searches to return more relevant results. Unlike keyword searches, vector search can provide relevant results even if the search query contains misspelled or slang words.

Better user experience

Vector search implementation can improve the user experience. Vector search is less rigid; it provides some flexibility for users when typing their search. This improves the users' experience of the application.

Natural language querying

Vector search implementation allows users to type search queries in natural language, rather than having to target keywords, often explaining the expected result rather than targeting the keywords.

Native integration within MongoDB

Vector Search is part of the core features of MongoDB. This means that it is already integrated with your database and collections. If you have an already existing collection in MongoDB, you can easily use your existing data to implement vector searching.

Reduced need for a dedicated vector database

With MongoDB, you do not need a separate vector database. Data can be stored alongside its vector embeddings in the MongoDB collection.

Considerations

While vector search can provide more accurate results, it is worth noting that it is not a complete replacement for traditional searches. The following key points are worth noting before deciding to implement a vector search in an application:

The need for embedding models:

Vector search implementation requires the use of AI models to generate embeddings. These models can be expensive and provide an additional configuration layer.

Additional storage requirements for vectors

Vectors can be large in size. The vector for a sentence can have up to 8,192 dimensions. Imagine when you have a hundred blog posts. This can take up a huge amount of space and can lead to higher cloud bills.

Choosing an appropriate embedding model

There are several embedding models that are great for different industries. Choosing the right model for your particular use case is a factor that should be considered. For example, a model that was designed for the fashion industry might not give the best result for a model optimized for medical research.

When keyword search may still be the better choice

Vector search is not a complete replacement for keyword search. Keyword search can still be more effective for exact match searches. An example can be searching for an exact serial number, username, or any other situation that returns an exact keyword result.

When Should You Use MongoDB Vector Search?

A vector search implementation that does not use MongoDB requires data, which can be in the form of text, images, or audio, to be stored in a traditional database; the generated vector for the items will be stored in a different dedicated vector database. This results in repetition because two databases are used, increasing the complexity of keeping the two databases in sync and fragmenting data (text and vectors).

MongoDB vector search bridges that gap. It lets data and their generated vectors be stored in the same collection.

Another major advantage that MongoDB vector search brings is that it keeps data in sync. A very common issue with vector search integrations where one database is used for data and another for storing vectors is that the data is often not in sync. If a record is deleted from the regular database and the vector is not, search results can be recommended that are no longer available in the database. MongoDB vector search stores the record and keeps it in sync. If a record is deleted from its collection, its vector is deleted as well, keeping both the record and its vector in sync.

MongoDB vector search eliminates the need for new infrastructure by storing both records and vectors in the same collection.

Conclusion

In this article, we explored what vector search is, how it works, how AI comes into the picture, and what differentiates it from traditional keyword search. We also took it a step further and explored how MongoDB vector search works and what advantages it has over other vector databases.

By the end of this article, you should have a basic understanding of how:

Vector search retrieves information based on meaning rather than exact keywords.
Embeddings make semantic search possible.
MongoDB provides native vector search capabilities that integrate with your existing data.
Understanding these fundamentals is the first step toward building intelligent search experiences.

For a deeper dive on how to implement vector search with MongoDB, you can check out the MongoDB documentation here. If you prefer a step-by-step tutorial, check out this tutorial here.

Now you have the basics; go build something great. Cheers.

A Proven Application Layer for MongoDB

MongoDB Guests — Mon, 27 Jul 2026 11:48:12 +0000

The successful alternative I used before Atlas Device SDKs. This Article was written by Raschid Jimenez.

Atlas Device SDKs reached end-of-life in September 2025, and it’s easy to understand why so many MongoDB developers liked them: direct client access, authentication, sync, and security, all tied closely to MongoDB Atlas.

Long before Atlas Device SDKs, I had already found a different way to get many of those same benefits. Using Parse Server and Back4app, I built production apps for years without the burden of maintaining custom API layers across every client and language. This article is the recipe I wish more MongoDB developers knew: why this stack works, what it gives you, and why it’s such a strong fit on top of MongoDB.

I will walk you through three things that sold me on Parse Server and Back4app: native client SDKs across every platform, row-level security for multi-tenant apps, and managed schema with index control. Along the way, I'll cover what else comes in the package: authentication, real-time updates via LiveQuery, file storage, auto-generated GraphQL APIs, and offline support. Data never leaves MongoDB Atlas. Back4app is a gateway that translates client requests into operations against your cluster.

            ┌──────────────────────────────────────┐
            │  Client (SDKs / REST / GraphQL)      │
            └────────────────────┬─────────────────┘
                                 │
            ┌────────────────────▼─────────────────┐
            │     Application Layer                │
            │       (Back4app - Stateless Gateway) │ 
            └────────────────────┬─────────────────┘
                                 │
            ┌────────────────────▼─────────────────┐
            │     MongoDB Atlas (your cluster)     │
            └──────────────────────────────────────┘

Back4app Uses MongoDB Atlas as the Datastore

Back4app is a managed backend-as-a-service (BaaS) platform. Sign up, paste in your MongoDB connection string, and you get an application backend running in front of your cluster: APIs, SDKs, authentication, real-time sync, file storage, access control, and a dashboard. No provisioning servers, no configuring authorization, no maintaining the layer yourself.

The connection model is what should land for an Atlas audience: bring your own cluster. Back4app connects to an existing Atlas deployment via a standard MongoDB connection string. No migration, no copy of your data, no proprietary storage format, no lock-in at the storage tier. If you walked away from Back4app tomorrow, your Atlas cluster would still be there, untouched, with every native MongoDB primitive intact.

What Back4app adds is the application backend layer most teams end up building anyway: auto-generated REST and GraphQL APIs, authentication, real-time sync, file storage, ACLs and CLPs, and a visual data browser.

Back4app is a layer in front of Atlas, not a replacement for it.

Parse Server Under the Hood

Under the hood, Back4app runs Parse Server, an open-source backend framework with a mature SDK ecosystem across every major platform. That's where the Parse.Object, Parse.Query, Parse.ACL, and Parse.Schema APIs in the code samples throughout this article come from. They're Parse Server primitives. Back4app is the managed platform that runs Parse Server in production (like Atlas does for MongoDB), plus the pieces around it: hosted infrastructure, the connection-string flow into Atlas, the dashboard, the spreadsheet-like data browser, role-aware visual schema editing, monitoring, scaling, and backups.

The short version: code I write with a Parse.* prefix is Parse Server. The surfaces I interact with through the Back4app dashboard are Back4app.

Client SDKs Across Every Platform

With Atlas Device SDKs gone, the question for most teams is concrete: how do clients perform operations against MongoDB Atlas (CRUD, authentication, file uploads, real-time subscriptions, and access-controlled queries with relations) without building and maintaining a custom REST layer for each?

Back4app offers a set of native SDKs that talk to the gateway, which in turn talks to your Atlas cluster. Supported platforms:

iOS (Swift, Objective-C)
Android (Java, Kotlin)
Web (JavaScript, TypeScript)
React Native, Node.js
Flutter, Dart
PHP, .NET, C#
Edge runtimes

For anything not on that list, Back4app auto-generates REST and GraphQL endpoints from the same schema.

The interesting part isn't the language list. It's that the API shape is the same across all of them. Here are the four CRUD operations in JavaScript:

// CREATE - no REST endpoint, no schema migration, just save.
const order = new Parse.Object("Order");
order.set("total", 149.99);
await order.save();

// READ - ACLs are enforced server side; this user only sees their tenant's orders.
const query = new Parse.Query("Order");
query.greaterThan("total", 100);
query.include("customer"); // hydrate the pointer in one round trip
const orders = await query.find();

// UPDATE - same object, just mutate and save.
order.set("status", "shipped");
await order.save();

// DELETE - single call, ACL-checked at the data layer.
await order.destroy();

That include("customer") line is worth looking at. On raw MongoDB, resolving a reference to fetch the related document means writing an aggregation pipeline:

// Without Parse - manual aggregation pipeline:
db.orders.aggregate([
  { $match: { total: { $gt: 100 } } },
  { $lookup: {
      from: "users",
      localField: "customer",
      foreignField: "_id",
      as: "customer"
  }},
  { $unwind: "$customer" }
]);

// With Parse - one line:
await new Parse.Query("Order")
  .greaterThan("total", 100)
  .include("customer")
  .find();

Same result, one round trip, no pipeline to maintain. The include works identically across REST, GraphQL, and every SDK.

The same read in Swift, using the modern Parse-Swift SDK:

let query = Order.query("total" > 100)
  .include("customer")
let orders = try await query.find()

A note on API symmetry: query.include("customer") reads the same in JavaScript and Swift. The naming follows each language's conventions, but the mental model is identical across platforms. Same query model, same ACL enforcement, same relational include semantics. A team shipping iOS, Android, and web doesn't re-architect the data layer per client. They translate one mental model into three idioms.

If GraphQL is the API you want, the same schema produces a typed GraphQL endpoint automatically. No resolvers to write, and the same ACLs and CLPs that govern the SDKs govern those queries too.

For reactive UIs, LiveQuery lets you subscribe to any query and receive push updates on create, update, and delete over WebSockets. No polling loops, no reconnection logic, no homegrown change-stream plumbing. Subscribe to a dashboard to live order updates, and the dashboard updates itself when an order changes upstream.

Row-Level Security and Multi-Tenancy

Multi-tenant access control is application-layer work by design: the database holds the data, the application decides who sees what. In practice, that means every team building multi-tenant SaaS on Atlas writes the same middleware: tenant filters injected into every query, role checks scattered across controllers, field redaction in serializers. That creates a lot of security-critical logic to maintain.

Back4app moves that enforcement down to the data layer. Four layers of access control work together:

Layer	Granularity	When to use
Application Keys	Per client type	Separate public, client, and master contexts
Class-Level Permissions	Per class (collection)	"Who can create or read Orders at all"
Access Control Lists	Per document	"Who can read this Order"
Protected / Field-level	Per field	isAdmin: only the server can write

Here's what each layer replaces in a hand-rolled implementation:

// Without Parse — middleware you write and never get to delete:
app.get("/api/invoices", auth, async (req, res) => {
  const invoices = await db.collection("invoices").find({
    tenantId: req.user.tenantId,        // remember this on every endpoint
    $or: [
      { ownerId: req.user._id },
      { sharedWith: req.user._id }
    ]
  }).toArray();
  // ...strip fields the role shouldn't see...
  res.json(invoices);
});

// With Parse — write zero filter middleware:
const invoices = await new Parse.Query("Invoice").find();
// ACLs and CLPs enforce the same rules across REST, GraphQL, and every SDK.

Here's how you set those ACLs up for a multi-tenant invoice scoped to a tenant role:

const invoice = new Parse.Object("Invoice");
invoice.set("amount", 1200);

const acl = new Parse.ACL();
acl.setRoleReadAccess("tenant_acme", true);          // every user in this tenant
acl.setRoleWriteAccess("tenant_acme_admins", true);  // only admins can modify
invoice.setACL(acl);

await invoice.save();

Any query from a user outside tenant_acme will not see this document. Enforcement happens server-side at the gateway, not in hopeful client code. The same filter applies whether the request comes from the iOS SDK, the GraphQL endpoint, or an inbound webhook. There is no client-side path that can bypass it.

When per-object rules aren't expressive enough ("users can update their own row, but only during business hours, and only if the invoice isn't locked"), Cloud Code provides server-side functions for custom validation:

// Cloud Code — server side function, runs in Parse Server, ACL-aware.
Parse.Cloud.beforeSave("Invoice", async (req) => {
  if (req.object.get("locked") && !req.master) {
    throw "Locked invoices cannot be modified";
  }
});

Schema and Index Management

With Parse, the schema lives in the database itself. Every client SDK, REST endpoint, and GraphQL type derives from there automatically. One source of truth, propagated to every surface your app talks to.

Parse's ODM defines classes (collections) with typed fields: strings, numbers, dates, files, GeoPoints, pointers, and relations. Parse enforces types at the gateway across REST, GraphQL, and every SDK, so a malformed payload is rejected before it reaches Atlas. The underlying MongoDB collection and the indexes Parse needs are auto-created. Schema changes are tracked through the Parse schema API or the Back4app dashboard, and they propagate to every client without anyone regenerating models by hand.

Document stores don't usually feel relational, but Back4app adds pointers for one-to-one references and relations for many-to-many associations, with include queries that hydrate them in a single round trip:

const query = new Parse.Query("Order");
query.include("customer");   // hydrates the pointer
query.include("products");   // hydrates the relation
const orders = await query.find();
// One round trip. No manual $lookup pipeline.

When you want schemas in version control rather than clicked into a dashboard, schemas can be declared programmatically, and migrations become regular code:

const schema = new Parse.Schema("Invoice");
schema
  .addString("customerEmail")
  .addNumber("amount")
  .addPointer("tenant", "Tenant")
  .addIndex("by_email", { customerEmail: 1 }); // explicit MongoDB index
await schema.save();

Parse exposes MongoDB indexes directly. It creates defaults for ACL fields and pointers, but your hand-tuned indexes stay untouched. Parse only adds the ones it needs for ACL filtering and pointer fields. Your existing indexes for aggregation pipelines, Atlas Search, or hand-tuned queries stay exactly as you set them. The addIndex() call above creates a real MongoDB index. Parse doesn't abstract that away.

If you picked MongoDB for schemaless flexibility: Parse classes can stay flexible (allowAddField: true) during prototyping and tighten as the data model stabilizes. You pick per class. Strict where it matters, flexible where it helps.

Conclusion & Next Steps

Back4app is a stateless gateway in front of your existing MongoDB Atlas cluster. Connect it with a standard MongoDB connection string, and you get the dashboard, the spreadsheet-like data browser, SDKs across every major platform, auto-generated REST and GraphQL endpoints, LiveQuery for real-time UIs, and row-level security via CLPs and ACLs. Native MongoDB primitives (aggregations, transactions, indexes, Atlas Search) remain entirely available to you.

So my advice to you is: go to Back4App and sign up for free, create an app in your preferred language, and run your first operation on your data in under 10 minutes. Back4App offers a free tier (25,000 API requests per month), which is enough to try on a real app rather than a toy. My customers loved having their app backends hosted for free and paying only when they scaled.

Build Your AI-Powered Reading List Using DigitalOcean Functions, Mistral AI, and MongoDB

MongoDB Guests — Wed, 22 Jul 2026 10:24:29 +0000

This Tutorial was written by Diego Freniche.

Since I started browsing the Internet, I've used many different services, applications, and browser plugins (even the good old bookmarks) to mark articles and posts that seemed interesting and that I wanted to read. Only not now. I'll read them later, I promise. And then you hoard a bunch of links in one place. But when you go back to actually read them, you find yourself clicking again on them just to remember what those links where about and why did you store them in the first place...

Fast forward today, and I haven't changed much in how I interact with articles I want to read later. But nowadays we can use LLMs almost for free... and they are really good at writing text summaries. So, why not create a system that, given a link, generates a summary and then stores both the link and the summary in a fast database I can use to build my own UI later?

And this is now yet another read-it-later (maybe) system was born...

Prerequisites

Before you begin this guide, you'll need the following:

a DigitalOcean account, that we'll use to write a DigitalOcean Function. This function will receive our links and run the code that will generate the LLM-powered summary and store the results in a database.
a MongoDB database, also running in DigitalOcean. We will use it to store the links and summaries of the web pages. You can learn more about deploying a managed MongoDB instance in DigitalOcean here.
access to any LLM that can write a good summary. For this example, I'll use Mistral AI, so you'll need to register a free account there and get an API token. Take a note of that API key, we'll need it later!

Step 1 - Creating an Empty DigitalOcean Function

Let's start by creating the backbone of our system. We can write a backend app using Node, Spring Boot or any other web framework. But then we need to host it in a server, write the code, correctly size the server, update the operating system and dependencies, etc. Using a serverless function we focus on the task at hand. In this case, getting a URL via a parameter, producing a summary of the page, and storing the data in the database.

So head over to Functions and click on create > Functions. First, we need to choose where our serverless function will be hosted. Although it's a serverless function, our code will need a real server to run. Only we don't have to provide it. Choose a data center close to your location (or close to where the users of your function will be). Then give it a name, choosing a label. In my case, it will be read-it-later-i-pinly-promise.

After creating the namespace, you'll be in the Overview tab. Let's add some basic code to test our function! Just click on the Create Function button.

We need to select one of the supported runtimes (in our case, Nodejs 18), and we'll give the function a name: read_later. It's important to leave the Web Function box checked, as we want to call this function directly using HTTP.

Done! Now you should see the default source code of the newly created function. This just gets an argument name from the URL and returns it. As we're not configuring a web page or a JSON response, this will be printed out as plain text by our web browser.

function main(args) {
let name = args.name || 'stranger' let greeting = 'Hello ' + name + '!' console.log(greeting)
return {"body": greeting}
}

In the main function, we receive a JSON object with all the arguments passed in the function call, we search for name, and use it. If not found, we use stranger as name. Then we compose and print out a greeting which we also return from our function.

Step 2 - testing our function

As this is accessible via HTTP calls, we have several different options to test our function.

Using cURL

If your system includes curl (Microsoft Windows 10/11 and most UNIX systems, like Linux and macOS come with cURL preinstalled), you can copy the URL of your function and add the name parameter at the end. Then call curl with:

curl https://<redacted>.doserverless.co/api/v1/web/<redacted>/default/read_later? name=Diego

In your case, instead of <redacted>, you'll see the ids of your functions. If you call the function from a Terminal, you'll see the response from our function:

curl https://<redacted>.doserverless.co/api/v1/web/<redacted>/default/read_later? name=Diego
Hello Diego!%

Try it changing the name and the code in your function and experiment with it! It's super fun!

Testing directly in DigitalOcean

We can click on Test Parameters and add a JSON with our test data, in this case:

{ "name": "Diego" }

Then, we can hit the Run button and see the results in the Output and Logs windows.

Step 3 - Adding the LLM Code

First, we will change our variable names to reflect what we're doing: producing summaries from links. Instead of name we'll use a link variable. Also, we'll return a proper HTTP response, with
statusCode 200 in case we got a link.

export async function main(args) { const link = args.link;
if (!link) { return {
statusCode: 400,
headers: { "Content-Type": "application/json" }, body: { error: "Please send a link" }
};
}
const body = { URL: link }; return {
statusCode: 200,
headers: { "Content-Type": "application/json" }, body: body
};
}

After changing the code, hit the "Save" button to avoid losing your changes. You can now re-run your function and check the results.

While we're at it, let's also change the Test Parameters (if you're testing directly from the browser) to:

(
"link": "https://en.wikipedia.org/wiki/Seville"
)

Seems like we're about to travel to Seville!

After that, we'll add the code to interact with Mistral. We need to send a POST with the prompt to their endpoint. For this, we'll need an API key from Mistral AI (see prerequisites). Save the code, but don't run the function (yet) as we need to add the secret.

const API_KEY = process.env.MISTRAL_API_KEY; // secrets are stored in the function's Settings
const API_URL = 'https://api.mistral.ai/v1/chat/completions';

async function chatWithLeChat(prompt) { try {
// call the Mistral endpoint with POST const res = await fetch(API_URL, {
method: 'POST', headers: {
'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json',
},
body: JSON.stringify({ model: 'mistral-tiny',
messages: [{ role: 'user', content: prompt }],
}),
});

const data = await res.json(); console.log(res.status, data);
return data?.choices?.[0]?.message?.content || null;
} catch (error) {
console.error('Error:', error?.message || error); return null;
}

Step 4 - Adding Secrets

As you can see, in our code, we don't embed the secrets (like API keys). Instead, we use the Functions environment to provide these:

const API_KEY = process.env.MISTRAL_API_KEY;

To add this secret MISTRAL_API_KEY we click on Settings. Then, we scroll down to Environment Variables. There, we can click on Edit

Now, click on Add and then Save. You should see your new environment variable now!

Step 5 - getting the LLM summary

It's time to call our chatWithLeChat function from main! To do that, we 1st create a system prompt that includes the link we received and instructs the LLM on what to do. Then, we'll call Mistral:

const prompt = 'I need a summary of this page: ' + link;

const summary = await chatWithLeChat(prompt);

We will extend our body to include the summary:

const body = { URL: link , summary: summary
};

This is all our code right now:

export async function main(args) { const link = args.link;
if (!link) { return {
statusCode: 400,
headers: { "Content-Type": "application/json" }, body: { error: "Please send a link" }
};
}

const prompt = 'I need a summary of this page: ' + link; const summary = await chatWithLeChat(prompt);
const body = { URL: link , summary: summary
};

return { statusCode: 200,
headers: { "Content-Type": "application/json" }, body: body
};
}

const API_KEY = process.env.MISTRAL_API_KEY; // secrets are stored in the function's Settings
const API_URL = 'https://api.mistral.ai/v1/chat/completions';

async function chatWithLeChat(prompt) { try {
// call the Mistral endpoint with POST const res = await fetch(API_URL, {
method: 'POST', headers: {
'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json',
},
body: JSON.stringify({ model: 'mistral-tiny',
messages: [{ role: 'user', content: prompt }],
}),
});

const data = await res.json(); console.log(res.status, data);
return data?.choices?.[0]?.message?.content || null;
} catch (error) {
console.error('Error:', error?.message || error); return null;
}

If you save your function and run it, most probably you'll get an error message in LOGS:

2026-02-02T12:14:17.299547912Z stderr: The function exceeded its time limits of 3000 milliseconds. Logs might be missing.

We need to give our function more time to wait for the LLM to answer!

Setting Timeouts

Head to Settings again, and click Edit on Limits. By default you have 3 secs (3000 milliseconds) to run your function before you get a timeout. We'll set that to 10 secs. Save the change, go back to Source and run the function again.

After that change, you should get a summary of the page!

Persisting data into MongoDB

We're now ready to store our body object in a MongoDB collection. One of the nicest things about MongoDB is that you work with your application objects in memory, and when ready, you send them to MongoDB using the driver (a library that will interact directly with the database). No need for ORMs or complicated frameworks in this case.

We'll start by importing the MongoDB driver:

import { MongoClient } from "mongodb";

After that, we'll write a simple function that will:

receive our object as an argument, named page.
connect to MongoDB using a MongoClient instance and the connect method.
write it into MongoDB using insertOne

async function store_in_mongodb(page) {
const mongodb_URI = process.env.MONGODB_URI; // get the MongoDB URI from an environment variable

const client = new MongoClient(mongodb_URI); try {
// connect to MongoDB await client.connect();
const database = client.db('read_it_later'); const collection = database.collection('pages');

const result = await collection.insertOne(page); return result;
} finally {
await client.close();
}
};

As we did previously with our environment variable MISTRAL_API_KEY, we'll need to add a new environment variable in Settings called MONGODB_URI with the URI of our MongoDB installation. The URI looks like this:

mongodb+srv://user:password@cluster.something-here.mongodb.net/

The last step will be to call this new function in main:

await store_in_mongodb(greeting);

Once this is done, save your function and send some links to your function and see how the collection pages gets populated!

Conclusion

In this article we've explored how to write a serverless function that can interact with an LLM and store results in MongoDB. It's incredible how far a few lines of code can take us these days. Not so long ago you needed to spend days setting up web and database servers, hosting everything, exposing it to the Internet, writing your code, etc. Now, using DigitalOcean integrated suit of services and hosted applications you focus on what matters most: your application code. Our next step could be building a new function to read from MongoDB, that will be the base for our UI, or add Semantic Search using MongoDB's Vector Search capabilities.

Here you have the complete code listing for our function:

export async function main(args) { const link = args.link;
if (!link) { return {
statusCode: 400,
headers: { "Content-Type": "application/json" }, body: { error: "Please send a link" }
};
}

const prompt = 'I need a summary of this page: ' + link; const summary = await chatWithLeChat(prompt);
const body = { URL: link , summary: summary
};

await store_in_mongodb(body);

return { statusCode: 200,
headers: { "Content-Type": "application/json" }, body: body
};
}

const API_KEY = process.env.MISTRAL_API_KEY; // secrets are stored in the function's Settings
const API_URL = 'https://api.mistral.ai/v1/chat/completions';

async function chatWithLeChat(prompt) { try {
// call the Mistral endpoint with POST const res = await fetch(API_URL, {
method: 'POST', headers: {
'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json',
},
body: JSON.stringify({ model: 'mistral-tiny',
messages: [{ role: 'user', content: prompt }],
}),
});

const data = await res.json(); console.log(res.status, data);
return data?.choices?.[0]?.message?.content || null;
} catch (error) {
console.error('Error:', error?.message || error); return null;
}
}
import { MongoClient } from "mongodb"; async function store_in_mongodb(page) {
const mongodb_URI = process.env.MONGODB_URI; // get the MongoDB URI from an environment variable

const client = new MongoClient(mongodb_URI); try {
// connect to MongoDB await client.connect();
const database = client.db('read_it_later'); const collection = database.collection('pages');

const result = await collection.insertOne(page); return result;
} finally {
await client.close();
}
};

No messages table! The data model behind my own Claude-based chatbot

MongoDB Guests — Thu, 02 Jul 2026 15:47:31 +0000

This tutorial was written by Néstor Daza.

This is the second article in a series about building Claudius, my own Claude-based chatbot (Github). The prologue made the case for building it, and for choosing MongoDB as its foundation.

Open the conversations collection in Claudius’ database and you find the usual fields of a thread header but nothing else: a userId, a title, some timestamps, and so on, but no array of messages, no messages collection sitting beside it either! The text of every conversation lives somewhere else entirely, in the LangGraph checkpointer, which I wire up later in this series. This absence is a modeling decision, and how I came up with the database schema for my chatbot is the theme of this article.

If you come from a relational background, you're used to modeling the data first when designing a database. For a project like this, you would start by finding the entities and normalizing them, and the final schema would come out of the data's structure: a conversations table and a messages table with a foreign key between them, because that is what the data looks like.

Document modeling runs the other way. You start from how the application reads and writes, and the shape of the document follows the access patterns. Claudius never reads conversation messages without the agent's full working state wrapped around them, and that state is persisted using the LangGraph checkpointer. A separate messages table would add nothing, since the app would always have to join it back to that state on every read. The access pattern says the messages belong with the agent state, so that is where they go, and conversations are left as the lightweight header the list view actually needs.

That inversion, modeling around use rather than around the data, runs through everything below.

Schema-flexible is not schemaless

This is the misconception lots of people often carry, and it is worth killing on the way in. A document database does not mean no schema; it means a flexible, use-case-based data model. A relational engine holds the schema for you through column types and constraints; MongoDB will store documents in whatever shape your application gives it, and can validate that shape with a collection validator if you want. Claudius keeps the schema in code, as one Zod schema per collection, because that single definition gives both the runtime check and the TypeScript type.

import { z } from "zod";
import { zObjectId } from "./common";

export const ConversationSchema = z.object({
  _id: zObjectId.optional(),
  userId: zObjectId,
  title: z.string(),
  modelId: z.string(),
  createdAt: z.date(),
  updatedAt: z.date(),
  archived: z.boolean(),
  expiresAt: z.date().optional(),
});

export type Conversation = z.infer<typeof ConversationSchema>;

The reason is drift. Declare a TypeScript interface for a document and a separate validator for the data coming in, and the two will disagree eventually, while the type system will lie to you with complete confidence. Deriving the type from the validator collapses them into one artifact. One definition gives you the runtime check at the boundary and the compile-time type everywhere else. For a relational reader, this is your DDL (Data Definition Language) and your ORM (Object-Relational Mapping) models together in the same structure, which is something relational stacks rarely pull off.

Another advantage is that every external input, whether an incoming HTTP body or a model's structured output, is parsed through Zod at the boundary, so by the time data reaches the logic, it is already validated and typed.

A quick tour of the model

There are eight collections in this phase: users, conversations, memories, documents, chunks, usage_events, settings, and, arriving later, jobs. Two more, checkpoints and checkpoint_writes, belong entirely to the checkpointer and are never written to directly, so I do not model them. Most of these schemas are unremarkable once you see the access-pattern framing, but two modeling moves are worth pausing on.

The first is settings. It holds a handful of global singletons, each a different shape, each identified by a string _id: the member allowlist, the model catalog, the per-role tiers, and the guest spend circuit breaker. They share one collection, so the Zod schema is a discriminated union based on _id, and parsing a settings document narrows it to the correct shape from its _id literal.

export const SettingsSchema = z.discriminatedUnion("_id", [
  AllowlistSettingsSchema,        // _id: "allowlist"
  ModelCatalogSettingsSchema,     // _id: "modelCatalog"
  TiersSettingsSchema,            // _id: "tiers"
  GuestCircuitBreakerSettingsSchema, // _id: "guestCircuitBreaker"
]);
export type Settings = z.infer<typeof SettingsSchema>;

In relational terms, this is a single typed configuration table whose rows legitimately carry different columns, which is usually awkward to model relationally without resorting to a JSON column, and here comes out clean and readable.

The second move is to use document embedding as a data modeling technique. The users document carries its dailyMessageCount as an embedded object holding the count and its reset time together, rather than as two flat fields or a side table, because the application always reads and writes them together as a unit. That is the document modeling mantra: data used together is stored together.

Let the database keep the rule

The most satisfying part of this phase is how little policy I had to write. Three product rules that I would normally enforce in application code are carried by the storage engine instead, because I shaped the data so the engine could hold them.

Retention is a TTL index

Guest data in Claudius is ephemeral. Guest-created conversations and memories carry an expiresAt date; member and admin documents leave the field off entirely. A TTL (time-to-live) index with expireAfterSeconds set to 0 tells MongoDB to delete each document at the exact moment stored in its own expiresAt field; documents without this field are never touched.

// TTL on the guest-only expiresAt field. expireAfterSeconds:0 
// means "expire exactly at the date stored in the field"; 
// documents without the field are never touched, so member/admin
// data is permanent.

await db
  .collection(COLLECTIONS.conversations)
  .createIndex({ expiresAt: 1 }, { name: "expiresAt_ttl", expireAfterSeconds: 0 });

await db
  .collection(COLLECTIONS.memories)
  .createIndex({ expiresAt: 1 }, { name: "expiresAt_ttl", expireAfterSeconds: 0 });

So, guest ephemerality, a product rule, is enforced by the engine. There is no cron job and no cleanup endpoint to remember to run. The relational equivalent is a scheduled sweep; here, the policy lives in the index.

One detail ties the TypeScript compiler to the database behavior. With exactOptionalPropertyTypes turned on in tsconfig.json, an optional field means the key is absent, not present, and set to undefined. The compiler setting and the storage rule reinforce each other.

Telemetry is a time-series collection

Every billable model call writes one usage_events document, recording the token counts, the latency, and what the call was for. That data is append-only and gets queried along a few low-cardinality dimensions over a time window, which is the textbook shape for a MongoDB time-series collection. It has a timeField, timestamp, and a single metaField that holds the grouping dimensions (userId, modelId, purpose) nested alongside the measurements.

Two things follow from this choice. A time-series collection has to be created explicitly with a timeseries spec before the first insert (they cannot be created on first write, the way an ordinary collection does), so the index script creates it behind an existence check. Once it exists, the engine's own machinery becomes visible: list the collections and a system.buckets.usage_events sits next to usage_events, which is where the bucketed storage lives:

[
  {"name":"memories","type":"collection"},
  {"name":"usage_events","type":"timeseries"},
  {"name":"system.buckets.usage_events","type":"collection"},
  {"name":"settings","type":"collection"},
  {"name":"users","type":"collection"},
  {"name":"conversations","type":"collection"},
  {"name":"accounts","type":"collection"},
  {"name":"chunks","type":"collection"},
  {"name":"system.views","type":"collection"}
]

The relational analogy is a metrics or rollup table you would otherwise hand-roll and maintain, except the bucketing is automatic and the query you write against it stays simple.

A pre-filter field is a security control

The project's hardest rule is that a vector search pre-filters by userId and can never return another user's data. That rule is not bolted onto the search in the application code. It is a property of the index.

Both memories and chunks get an Atlas Vector Search index on the embedding field, with a filter on userId (and on chunks, a second filter on documentId).

await ensureVectorIndex(db, COLLECTIONS.memories, "memories_vector", [
  { type: "vector", path: "embedding", numDimensions: 1024, similarity: "cosine" },
  { type: "filter", path: "userId" },
], created, skipped);

await ensureVectorIndex(db, COLLECTIONS.chunks, "chunks_vector", [
  { type: "vector", path: "embedding", numDimensions: 1024, similarity: "cosine" },
  { type: "filter", path: "userId" },
  { type: "filter", path: "documentId" },
], created, skipped);

Pre-filtering means the userId condition is applied during the indexed search on the same documents being scored, not afterward to whatever the search returns. Post-filtering is not only slower, since you score documents you then throw away, but it is a correctness and security risk, since the wrong documents can slip through ranking and limits before your filter ever runs.

This is also why chunks denormalizes userId. A vector search has no notion of a join, so the filter has to sit on the very documents being searched. A chunk belongs to a document that belongs to a user, but the chunk has to carry the owner itself, or the index cannot enforce the boundary. That is the embed-versus-reference tension made concrete: a chunk references its parent document by id and duplicates the one field the security boundary depends on.

Two gotchas surfaced while creating these indexes, and both are worth saying out loud. A search index can only be built on a collection that already exists, and chunks had no classic index, so the collection did not exist yet the first time the script reached for its vector index:

MongoServerError: Error retrieving collection UUID and view info ::
caused by :: Collection 'claudius.chunks' does not exist. (NamespaceNotFound)

The fix is to add an existence check and create the collection first only if it is missing. The second gotcha is that Atlas builds search indexes asynchronously, so createSearchIndex returns when the build is queued, not when it is queryable! We also need a name check that keeps reruns safe while it builds.

This is the final function to build the indexes from code using the driver's createSearchIndex, with the checks that keep the script safe to run on every deploy:

async function ensureVectorIndex(
  db: Db,
  collectionName: string,
  indexName: string,
  fields: Array<Record<string, unknown>>,
  created: string[],
  skipped: string[],
): Promise<void> {
  // A vector search index can only be built on an existing
  // collection, so create it first if nothing else has (e.g.
  // chunks has no regular index).
  const collections = await db
    .listCollections({ name: collectionName })
    .toArray();
  if (collections.length === 0) {
    await db.createCollection(collectionName);
  }

  const collection = db.collection(collectionName);
  const present = await collection.listSearchIndexes().toArray();
  if (present.some((idx) => idx.name === indexName)) {
    skipped.push(`${collectionName}.${indexName}`);
    return;
  }

  // Atlas builds the index asynchronously; createSearchIndex
  // returns once the build is queued. The name check above 
  // keeps re-runs idempotent.
  await collection.createSearchIndex({
    name: indexName,
    type: "vectorSearch",
    definition: { fields },
  });
  created.push(`${collectionName}.${indexName} (vector)`);
}

The plumbing that keeps the code honest

Two small things hold the userId rule together in practice. First, userId is an ObjectId everywhere, matching the user’s _id, so filters and lookups line up without conversion. Second, no collection is ever queried through a raw driver call. Every access goes through a typed accessor, conversationsCol(), memoriesCol(), and so on, each returning a typed collection for the right document shape. Centralizing the collection names means a rename happens in one place, and typing each accessor keeps the userId filter visible to the compiler at every call site, which is how a rule that lives in prose stays honest in code.

The whole application and the Auth.js adapter share a single, serverless-safe MongoDB client to a single fixed database name, so there is exactly one connection pool. How that database name quietly caused two separate failures is a good story, and it belongs to the next article, where I stand the foundation up.

What this article bought

No chatbot yet, but the shape of the thing is already decided. Retention, the telemetry layout, and the tenant-isolation boundary are properties of the schema and its indexes now, not code I have to remember to run later. That is the payoff of modeling around the access patterns first: the database carries more of the load because you shaped it to. The data model is the foundational piece of the whole app.

Next, we'll build this foundation up, with identity the client never gets a vote in, and a health check that proves the app can reach Atlas and Bedrock, plus a view on the unglamorous errors that cost me real time. After that comes the streaming chat backbone, which is where the conversation finally gets a place to live, and I open a real checkpoint document to show you the messages that were never in their own collection.

Advanced API Authentication Strategies with Node.js and MongoDB (Part 3)

MongoDB Guests — Thu, 02 Jul 2026 09:00:00 +0000

This article was written by Moses Anumadu.

Secure Session Management with JWT and Refresh Tokens in Node.js and MongoDB

This is the final part of this 3-part series. In parts 1 and 2, we implemented email-based authentication. Then we went further to implement TOTP-based multi-factor authentication (MFA), QR code onboarding, and half-authentication login flow.

What we have built so far is an API that allows users to prove their identity using email and password. In this section, we will go beyond that. The application should not only be able to recognize users but also protect the user through concepts like secure session management, token expiration, and session revocation. We will implement these concepts through the following:

JWT access tokens
refresh tokens
MongoDB-backed sessions
refresh token rotation
protected routes
secure logout flows

By the end of this section, we should have an authentication system that can be deployed to production and improved upon for more advanced features.

Pre-requisites

To follow along with this tutorial, ensure you have prior experience working with the following technologies:

Node.js and Express
MongoDB and Mongoose
A MongoDB Atlas account
Postman
Ngrok
Parts one and two of this series

With that said, let's get started.

Understanding Access Tokens vs Refresh Tokens

An access token is a JWT token that grants a user access to a protected route. It often has a very short-lived life cycle. A refresh token is a token with a longer expiration time solely for requesting an access token. Without the refresh token, the user would need to log out and log back in to generate a new access token. The refresh token is stored in the database, enabling the system to recognize the user without terminating the session.

You are probably wondering why you need both, right? Let's look at an example. Imagine that a hacker steals a user's access token, and with this token, he (the hacker) can do malicious things. If this access token does not have a short-lived life cycle, the attacker would have a longer window to do more harm. To prevent this, the refresh token is required to generate a new access token from time to time. So, even if an access token is stolen, within 15 minutes, that token becomes invalid. The attacker can't do any harm.

This is an extra layer of security necessary for a secure authentication system and to improve the user experience. The flow looks like: a user logs in, access token and refresh token are granted, access token lasts for 15 minutes, refresh token lasts for 7 days, access token expires after 15 minutes, user does not need to log out, user generates a new access token using the refresh token saved in the database.

Creating Token Utilities

Now, we understand why we need both an access token and a refresh token. Let's proceed to the implementation. In part 2, when we created the MFA token, we created src/utils/tokenUtils.js file. We need to update that file to generate our access token, refresh token, and MFA token. Open src/utils/tokenUtils.js and replace the code with the code below:

const jwt = require("jsonwebtoken");

const crypto = require("crypto");

const generateAccessToken = (user) => {
  return jwt.sign(
    {
      userId: user._id,
      email: user.email,
    },
    process.env.JWT_ACCESS_SECRET,
    {
      expiresIn: process.env.ACCESS_TOKEN_EXPIRES_IN,
    }
  );
};

const generateRefreshToken = () => {
  return crypto.randomBytes(64).toString("hex");
};

const hashToken = (token) => {
  return crypto
    .createHash("sha256")
    .update(token)
    .digest("hex");
};

const generateMFAToken = (user) => {
  return jwt.sign(
    {
      userId: user._id,
      type: "mfa",
    },
    process.env.JWT_ACCESS_SECRET,
    {
      expiresIn: process.env.MFA_TOKEN_EXPIRES_IN,
    }
  );
};

module.exports = {
  generateAccessToken,
  generateRefreshToken,
  hashToken,
  generateMFAToken,
};

We added three methods generateAccessToken, generateRefreshToken, hashToken to the page. The hashToken will be used later. It will be used to encrypt our refresh token into a one-way SHA-256 hash before it is stored in the database or compared when validating. The generateAccessToken and generateRefreshToken generate an access token and a refresh token, respectively.

Why Refresh Tokens Are Not JWTs

Also, did you notice that our refresh tokens were generated using secure random bytes? This line of code here: crypto.randomBytes(64).toString("hex");. This simply gives us more control over it. The refresh token should be random, unpredictable, and most importantly, database-controlled. Refresh tokens, unlike access tokens, do not need readable claims.

So, this way, we can easily revoke it, rotate it, and control the session validation or invalidation.

Creating the Session Model

Our refresh token is database-controlled. That ultimately means we need to store and retrieve it from the database. To achieve this, we need to create a MongoDB model for storing user sessions. Create src/models/Session.js and update the page with the code below:

const mongoose = require("mongoose");

const sessionSchema = new mongoose.Schema(
  {
    userId: {
      type: mongoose.Schema.Types.ObjectId,
      ref: "User",
      required: true,
    },

    refreshTokenHash: {
      type: String,
      required: true,
    },

    userAgent: {
      type: String,
      default: null,
    },

    ipAddress: {
      type: String,
      default: null,
    },

    expiresAt: {
      type: Date,
      required: true,
    },

    revokedAt: {
      type: Date,
      default: null,
    },
  },
  {
    timestamps: true,
  }
);

module.exports = mongoose.model("Session", sessionSchema);

Notice that we are storing refreshTokenHash. Always remember that sensitive data at rest should also be encrypted. We created the fields we need to save for each user session. Notice fields like ipAddress, userAgent, expiresAt and revokedAt.

Issuing Tokens After MFA Verification

Now we can generate both a refresh token and an access token, and our MongoDB model is ready to save session values in the database. Next, we need to modify the authentication flow to include these. The flow should be: a user logs in, completes MFA, the application issues an access token and a refresh token to the user, and finally, the session values (including the refresh token) are stored in our MongoDB session collection. Let's proceed to the implementation.

!!!
Returning refresh tokens in JSON bodies is unsafe for browser clients and encourages XSS-exposed storage patterns. The safer pattern is HTTP-only cookie transport for refresh credentials. This is the approach we will follow in the next section.
!!!

Updating the Login Controller

In the next section, we will send refresh tokens using HTTP-only cookies transport as this is the safer option to prevent XSS-expose attacks.
The code in src/controllers/authController.js needs to be updated. Let's start by adding the import values and HTTP-only cookies configurations.

Open src/controllers/authController.js and add the following import statement to the top part of the page

const jwt = require("jsonwebtoken");
const Session = require("../models/Session");

const {
  generateAccessToken,
  generateRefreshToken,
  hashToken,
  generateMFAToken,
} = require("../utils/tokenUtils");

After the import statements, add the following HTTP-only cookie configurations. We will use these later in the code.

const getRefreshTokenExpiryMs = () => {
  const refreshDays = Number(
  process.env.REFRESH_TOKEN_EXPIRES_DAYS
  );

  return refreshDays * 24 * 60 * 60 * 1000;
};

const getRefreshTokenCookieOptions = () => {
  return {
    httpOnly: true,
    secure: process.env.NODE_ENV === "production",
    sameSite: "strict",
    path: "/api/auth",
    maxAge: getRefreshTokenExpiryMs(),
  };
};

const setRefreshTokenCookie = (res, refreshToken) => {
  res.cookie(
  "refreshToken",
  refreshToken,
  getRefreshTokenCookieOptions()
  );
};

const clearRefreshTokenCookie = (res) => {
  res.clearCookie("refreshToken", {
    ...getRefreshTokenCookieOptions(),
    maxAge: undefined,
  });
};

Updating the MFA Login Verification Flow

Next, replace the success section inside verifyMFALogin controller method with the code below:

Revoke this section

return res.status(200).json({
success: true,
message: "MFA login successful",
});

And replace with this:

// Issue access token
const accessToken = generateAccessToken(user);

// Generate refresh token
const refreshToken =
  generateRefreshToken();

// Hash refresh token
const refreshTokenHash =
  hashToken(refreshToken);

// Session expiry
const expiresAt = new Date();

expiresAt.setDate(
  expiresAt.getDate() +
    Number(
      process.env.REFRESH_TOKEN_EXPIRES_DAYS
    )
);

// Store session
await Session.create({
  userId: user._id,
  refreshTokenHash,
  userAgent: req.headers["user-agent"],
  ipAddress: req.ip,
  expiresAt,
});
setRefreshTokenCookie(res, refreshToken);
      return res.status(200).json({
       success: true,
       message: "Login successful",
       accessToken,
     });

From the additions we made, after password verification and MFA verification, a token is issued, and a session is created for the authenticated user.

Creating JWT Authentication Middleware

Now, we can issue an access token and a refresh token. Next, we need a way to protect our routes so that only users with the right access can access them. To do this, we need to implement middlewares.

A good example can be creating a middleware that protects routes and lets only authenticated users have access to it. Let's implement that. Create the following file src/middleware/authMiddleware.js, and update it with the code below:

const jwt = require("jsonwebtoken");

const User = require("../models/User");

const requireAuth = async (
  req,
  res,
  next
) => {
  try {
    const authHeader =
      req.headers.authorization;

    if (
      !authHeader ||
      !authHeader.startsWith("Bearer ")
    ) {
      return res.status(401).json({
        success: false,
        message: "Access token required",
      });
    }

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

    const decoded = jwt.verify(
      token,
      process.env.JWT_ACCESS_SECRET
    );

    const user = await User.findById(
      decoded.userId
    );

    if (!user) {
      return res.status(401).json({
        success: false,
        message: "Invalid user",
      });
    }

    req.user = user;

    next();
  } catch (error) {
    return res.status(401).json({
      success: false,
      message:
        "Invalid or expired access token",
    });
  }
};

module.exports = requireAuth;

Creating a Protected Route

At this point, the middleware is ready. We need to attach it to our routes to restrict access to only authenticated users. Let's create a controller to get the authenticated user getMe and a route for it /api/auth/me then protect it with the middleware we created above. To create the controller method, open src/controllers/authController.js and add the following controller method to the page.

const getMe = async (req, res) => {
  return res.status(200).json({
    success: true,
    user: {
      id: req.user._id,
      email: req.user.email,
      mfaEnabled: req.user.mfaEnabled,
    },
  });
};

Also, update the export block to contain the getMe method we just added, like so

module.exports = {
  register,
  login,
  setupMFA,
  verifyMFA,
  verifyMFALogin,
  getMe,
};

Adding the Protected Route

Next, let's create the route. Open src/routes/authRoutes.js and add the following to the page.

First, import the middleware by copying and pasting the code below to the top section of the page

const { getMe } = require("../controllers/authController");
const requireAuth = require("../middleware/authMiddleware");

Then add the route below to the route on the page.

router.get("/me", requireAuth, getMe);

Testing Protected Routes

We've made progress. Let's test what we've built so far. We will test with Postman on the web and Ngrok. Ngrok will be used to tunnel our localhost address to something on the internet. Check Part One of this article for more on the setup. With that said,

Login

To generate an access token, we need to go through a series of verifications. First, we need to log in and generate an MFA token to verify MFA. Send a post request to the route below to log in.

POST /api/auth/login

Add the following to the body

{ 
  "email": "test_user@example.com",
  "password": "password123"
}

Your response should look similar to the image below if everything is done right.

MFA verification and access key generation

We are partially logged in. We need to verify MFA to be fully logged in. Send a Post request to the route below:

POST /api/auth/mfa/login/verify

Also, add the following to the body of the request:

{ "token": "TOKEN_GENERATED_BY_AUTH_APP",
  "mfaToken": "MFA_TOKEN_GENERATED_FROM_LOGIN"
}

Protected route

Next, send a GET request to the route below

GET /api/auth/me

With the following Header values

Authorization: Bearer YOUR_ACCESS_TOKEN

If the token is valid, the request should return the authenticated user information, like the image below

Also, in MongoDB Atlas, you can see that user sessions were created and stored in our session collection.

Implementing Refresh Token Rotation

Let's take our authentication a step further by implementing refresh token rotations. What we want to achieve is simple. Every refresh request should generate a new access token, generate a new refresh token, and invalidate the old refresh tokens. Let's jump right into it.

Creating the Refresh Endpoint

Add the following controller method to src/controllers/authController.js.


const refreshAccessToken = async (req, res) => {
   try {
     const refreshToken = req.cookies?.refreshToken;
      if (!refreshToken) {
       return res.status(401).json({
         success: false,
         message: "Refresh token cookie required",
       });
     }
      const refreshTokenHash = hashToken(refreshToken);
      const session = await Session.findOne({
       refreshTokenHash,
       revokedAt: null,
     });
      if (!session) {
       clearRefreshTokenCookie(res);
       return res.status(401).json({
         success: false,
         message: "Invalid session",
       });
     }
      // Expired session
     if (session.expiresAt < new Date()) {
       clearRefreshTokenCookie(res);
       return res.status(401).json({
         success: false,
         message: "Session expired",
       });
     }
      const user = await User.findById(session.userId);
      if (!user) {
       return res.status(401).json({
         success: false,
         message: "Invalid user",
       });
     }
      // ROTATE REFRESH TOKEN
     const newRefreshToken = generateRefreshToken();
      const newRefreshTokenHash = hashToken(newRefreshToken);
      session.refreshTokenHash = newRefreshTokenHash;
      await session.save();
     setRefreshTokenCookie(res, newRefreshToken);
      // Generate new access token
     const accessToken = generateAccessToken(user);
      return res.status(200).json({
       success: true,
       accessToken,
     });
   } catch (error) {
     console.error(error);
      return res.status(500).json({
       success: false,
       message: "Failed to refresh token",
     });
   }
 };

Next, let's update the exports with the freshAccesstoken method like so:

module.exports = {
  register,
  login,
  setupMFA,
  verifyMFA,
  verifyMFALogin,
  getMe,
  refreshAccessToken,
};

Adding the Refresh Route and Testing

Let's proceed by adding a route to test the refresh token feature. Open src/routes/authRoutes.js and update the page with the code below

router.post(
  "/refresh",
  refreshAccessToken
);

Let's proceed to test. Send a POST request to the route below.

POST /api/auth/refresh

Request body:

{
  "refreshToken": "YOUR_REFRESH_TOKEN"
}

If everything was done right, your response should look like the screenshot above. And the previous tokens should no longer work.

Implementing Logout and Session Revocation

A complete logout system not only removes the access token from the browser in the front end. It should also invalidate the refresh token and revoke the session from the backend. That is exactly what we will do in this section.

Creating the Logout Endpoint

Add the following controller method to src/controllers/authController.js

const logout = async (req, res) => {
   try {
     const refreshToken = req.cookies?.refreshToken;
      if (!refreshToken) {
       return res.status(400).json({
         success: false,
         message: "Refresh token cookie required",
       });
     }
      const refreshTokenHash = hashToken(refreshToken);
      await Session.findOneAndUpdate(
       {
         refreshTokenHash,
         revokedAt: null,
       },
       {
         revokedAt: new Date(),
       }
     );
     clearRefreshTokenCookie(res);
      return res.status(200).json({
       success: true,
       message: "Logged out successfully",
     });
   } catch (error) {
     console.error(error);
      return res.status(500).json({
       success: false,
       message: "Logout failed",
     });
   }
 };

From the code, we did not delete the session records. We marked them as revoked instead, as we can see below.

await Session.findOneAndUpdate(
{
refreshTokenHash,
revokedAt: null,
},
{
revokedAt: new Date(),
}
);

This is a good practice as it helps preserve session history for audit and security visibility.

Implementing Logout From All Devices

Let's take it a step further by allowing users to revoke all active sessions simultaneously. This would let them log out from all devices at once. This is a nice feature to have; it helps a user protect their account in the case of a malicious login from an unknown device.

Let's proceed, modify src/controllers/authController.js file, and add the following controller method:

const logoutAll = async (req, res) => {
   try {
     await Session.updateMany(
       {
         userId: req.user._id,
         revokedAt: null,
       },
       {
         revokedAt: new Date(),
       }
     );
     clearRefreshTokenCookie(res);
      return res.status(200).json({
       success: true,
       message: "Logged out from all devices",
     });
   } catch (error) {
     console.error(error);
      return res.status(500).json({
       success: false,
       message: "Failed to logout all sessions",
     });
   }
 };

Update the export block to contain all the methods below

module.exports = {
  register,
  login,
  setupMFA,
  verifyMFA,
  verifyMFALogin,
  getMe,
  refreshAccessToken,
  logout,
  logoutAll,
};

Adding Logout Routes

To add the routes for logout, update src/routes/authRoutes.js and add the code below. Also ensure that you have imported both logout and logoutAll methods from authController.js.

router.post("/logout", logout);

router.post(
  "/logout-all",
  requireAuth,
  logoutAll
);

Testing Logout

Send a post request to the route below. Also, pass the body value in the body of your request.

Request:

POST /api/auth/logout

Request body:

{
  "refreshToken": "YOUR_REFRESH_TOKEN"
}

Your response should look similar to the screenshot below

You can also test the logout all feature the same way. Send a Post request to the route below with a valid access token.

Request:

POST /api/auth/logout-all

Also, add a header with a user's access token.

Authorization: Bearer YOUR_ACCESS_TOKEN

Expected response:

{
  "success": true,
  "message":
    "Logged out from all devices"
}

Did you notice that the logout-all requires an access token passed in as a request header, not a refresh token like the logout route? This is because the logout-all does not have an active user session or know the user sending the request. It needs a way to identify the user, so it requires a user's access token.

All refresh token sessions should now become invalid. To be sure, you can also try using the same refresh token. It should fail because this session has been revoked.

Conclusion

A big thumbs up to you for sticking around till the end of this three-part series. We built a complete authentication API using Node.js, Express, MongoDB Atlas, and tested using Postman on the web and Ngrok for localhost routing.

In this final part of this series (part three), we discussed the difference between JWT access tokens and refresh tokens, why we need both, and walked through the programmatic implementation of the authentication by implementing the following:

JWT access token authentication
Session storage
refresh token rotation
Middleware and protected routes
secure logout and session revocation

The complete flow of the authentication with all three (3) parts put together looks like:

User registration, password authentication, MFA setup and verification, JWT access tokens and refresh token issuance, refresh token sessions, middlewares and protected routes, session.

Security is more of layers than single features. For further reading, check out the following topics for additional layers that can be added to an authentication.

Rate limiting
Email verification
HTTPS enforcement
Secure HTTP-only cookies
Brute-force protection
Device monitoring
Anomaly detection

Awesome. You can find the complete code on GitHub.