DEV Community: MongoDB

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 an Agentic Warehouse Management System — Part 3: Tools, Decisions, and Actions

Ricardo Mello — Thu, 03 Sep 2026 15:44:43 +0000

If you haven’t read the first two parts yet, I recommend starting with Part 1: Where AI Agents Add Value and Part 2: Java and Spring AI. There, we introduced the WMS scenario, discussed where an AI agent can add value, and built the first stages of the implementation with Java and Spring AI.

In this third and final part, we will build on the execution plan created in Part 2 and focus on the remaining stages of the agent workflow. We will see how each task is executed using controlled tools, how operational and business context is gathered, how the replenishment decision is made, and how the agent acts on that decision when necessary.

A live version of the Agentic WMS is available here.

From Plan to Execution

Let’s revisit the complete agent flow. The first stages: Define the Agent, Trigger the Agent, and Plan, were covered in the previous article. Now we will focus on executing the tasks, using the available tools, gathering context, and making the replenishment decision.

Execute tasks with controlled tools

With the execution plan created in Part 2, the AgentRunner starts processing the tasks one at a time. For each task, the runner reads its description and capability, executes it, stores the result, and then continues with the next task:

for (int index = 0; index < tasks.size(); index++) {
    String description = tasks.get(index).description();
    String capability = tasks.get(index).capability();
    LocalDateTime startedAt = LocalDateTime.now();
    String result = executeTask(definition, goal, description, capability, tasks);
    tasks.set(index, new AgentRun.AgentTask(
            description,
            AgentRun.TaskStatus.COMPLETED,
            capability,
            result,
            startedAt,
            LocalDateTime.now()
    ));
    agentRunService.save(withTasks(agentRun, tasks));
}

Before executing a task, the runner selects only the tools associated with its capability:

private Object[] toolsOf(
        AgentDefinition definition,
        String capability
) {
    return definition.capabilities().stream()
            .filter(candidate ->
                    candidate.name().equalsIgnoreCase(capability))
            .map(AgentCapability::tools)
            .filter(Objects::nonNull)
            .flatMap(List::stream)
            .toArray();
}

When the task is executed, the executor receives the current goal, the results produced by previous tasks, the current task, and only the tools available for its capability:

return definition.executor()
        .prompt()
        .tools(tools)
        .user("""
            %s
            Results of the previous tasks:
            %s
            Execute only this task:
            %s
            Use the available tools autonomously.
            """.formatted(
                goal,
                previousResults(tasks),
                description
        ))
        .call()
        .content();

The result of each completed task becomes part of the context available to the tasks that follow. This allows the agent to progressively gather the information it needs before making the replenishment decision.

At the same time, the capability assigned to each task defines which tools are available during its execution. This preserves the boundaries we established earlier: an ANALYSIS task, for example, can inspect information from the WMS, but it does not have access to tools that create a replenishment request.

Accessing operational data

During ANALYSIS tasks, the agent can use controlled tools to retrieve operational data from the WMS, such as inventory, invoice movements, stock movements, and pending replenishments.

For example, one of the available tools retrieves the stock movements generated by an outbound invoice:

@Tool(description = """
    Get the stock movements generated by a specific invoice.
    Use this tool to discover which products and quantities were affected
    by a completed outbound invoice.
""")
public List<StockMovement> getStockMovementByInvoiceNumber(
        @ToolParam(description = "Outbound invoice number")
        String invoiceNumber
) {
    return stockMovementService.findByInvoiceNumber(invoiceNumber);
}

With Spring AI, methods like this can be exposed to the model using annotations such as @Tool and @ToolParam. The executor can then call them when the current task requires that information.

The complete application exposes additional tools for retrieving inventory, invoice movements, pending replenishments, and other operational data. We will not cover each one here, but they follow the same idea.

Retrieving depositor policies

The depositor policies are also accessed through a dedicated tool exposed by the POLICY capability. In our application, each depositor can register business rules through the Depositor Policies page. For example, a policy may define a seven-day lead time or require at least 500 units per replenishment request:

These policies are stored as natural-language documents, which allows us to retrieve them using semantic search instead of relying soley on exact keywords. For example:

new DepositorKnowledgeEntry(
 "amz",
  "replenishment-minimum",
  KnowledgeType.REPLENISHMENT,
 "Replenishment orders for Amazon must contain at least 500 units per product."
);

When a policy is stored, its text is converted into an embedding using the configured Voyage AI embedding model and added to the vector store. During a POLICY task, the agent can ask a question based on the context collected so far:

What minimum quantities, packaging rules and lead times
apply to replenishing product BR01?

This gives us a simple Retrieval-Augmented Generation (RAG) flow: the question is used to retrieve the most relevant policies, which are then added to the execution:

The retrieval uses semantic search:

return vectorStore.similaritySearch(
        SearchRequest.builder()
                .query(question)
                .topK(3)
                .filterExpression(filter)
                .build()
);

and is filtered by depositor and knowledge type:

String filter =
        "depositorId == '" + depositorId + "'"
        + " && type in ['REPLENISHMENT', 'INBOUND', 'GENERAL']";

Together with the results from the ANALYSIS tasks, these policies give the agent the operational and business context required to decide whether replenishment is needed.

Decide

After the ANALYSIS and POLICY tasks are completed, the DECISION task uses the collected context to determine whether replenishment is required. The result ends with one of two tokens:

REPLENISHMENT_REQUIRED 
REPLENISHMENT_NOT_REQUIRED

This gives the AgentRunner a simple way to determine whether the remaining tasks should continue:

private boolean decidedToStop(String capability, String result) {
    return DECISION_CAPABILITY.equals(capability)
            && result != null
            && result.contains("REPLENISHMENT_NOT_REQUIRED");
}

If replenishment is not required, the remaining REPLENISHMENT and NOTIFICATION tasks are marked as SKIPPED, and the execution is completed:

The runner marks those tasks as SKIPPED and completes the execution:

if (decidedToStop(capability, result)) {
    skipRemaining(tasks, index + 1);

    return agentRunService.save(finish(
            agentRun,
            tasks,
            AgentRun.Status.COMPLETED,
            summarize(definition, goal, tasks)
    ));
}

If replenishment is required, the execution continues with the next tasks in the plan.

Act

Once replenishment is required, the agent can use the action tools exposed by the application.

The REPLENISHMENT task can create a replenishment request:

@Tool(description = """
    Create a replenishment request when one or more products
    need to be replenished.
""")
public String createReplenishment(
        String depositorCode,
        List<Replenishment.ReplenishmentItem> items,
        String message
) {
    // Validate the request ..
    replenishmentService.save(....)
}

Again: The tool does not replace the existing WMS logic. It gives the agent a controlled way to request the operation, while validation and persistence remain inside the application services.

Once created, the replenishment ID can be used by the NOTIFICATION task to prepare the depositor email draft:

@Tool(description = """
    Draft and store the notification email for a replenishment request.
    The email is not sent by this tool.
""")
public String draftDepositorEmail(
        @ToolParam(description = ReplenishmentIds.ID_PARAM)
        String id
) {
    ....    
}

At this point, the agent has made the decision and performed the actions required by it.

Complete

When there are no more tasks to execute, the agent run is completed and a summary is generated:

If a replenishment request was created, it is now available on the Replenishments page for review:

This completes the flow: the agent gathers context through controlled tools, decides whether replenishment is necessary, and acts only when required.

Next steps

During a single execution, the agent uses the results of previous tasks as context for the tasks that follow. A natural next step would be to give the agent memory across different executions.

For example, before creating a new replenishment request, the agent could look at previous AgentRun records and use relevant outcomes from similar situations:

The last three requests for 500 units of this product were rejected. Approved requests were usually between 200 and 300 units. Based on the current consumption, I will request 250 units.

This would allow the agent to use not only the context of the current execution, but also relevant information from previous ones. One possible evolution would be to retrieve similar past executions and add them to the agent context before making a new decision.

There are many other directions in which this agent could evolve, but this is a good example of how the same design can gradually become more context-aware over time.

Conclusion

The main goal of this series was to show how an AI agent can add value in scenarios where business decisions depend on multiple sources of context and increasingly complex rules. Our replenishment example is only a small glimpse of what this approach can support in a real Warehouse Management System.

In Part 1, we explored where an AI agent can add value in a WMS. In Part 2, we translated that design into Java and Spring AI by defining the agent, its capabilities, and its execution plan.

In this third and final part, we completed the flow: tasks were executed through controlled tools, operational data and depositor policies were added to the context, and the agent used that information to decide whether replenishment was required and act when necessary.

The main idea remains the same: the application defines what the agent is allowed to do, while the agent decides what to do within those boundaries.

Explore our MongoDB JVM Showcase repository for more Java examples and use cases. If you find it useful, consider giving it a ⭐.

Building an Agentic Warehouse Management System — Part 2: Java and Spring AI

Ricardo Mello — Thu, 03 Sep 2026 15:20:38 +0000

If you arrived here without reading the first part, I recommend starting with Part 1: Where AI Agents Add Value. There, we introduced the WMS scenario, explored the replenishment problem, and discussed where an AI agent can add value without replacing deterministic application logic.

In this second part, we will move from the design to the first steps of the implementation.

We will look at how the agent is built with Java and Spring AI, how its capabilities and boundaries are defined, how it is triggered by the WMS, and how the planner turns the agent’s goal into an execution plan.

In Part 3, we will continue from that plan and see how the agent executes tasks using controlled tools, gathers the context it needs, makes the replenishment decision, and takes action when necessary.

A live version of the Agentic WMS is available here.

Designing the agent

From this point on, we will focus on the main parts of the agent implementation using Java and Spring AI. Rather than covering Spring AI from the ground up, we will concentrate on the components directly relevant to this part of our agent: prompts, capabilities, triggering, and planning. For a broader introduction to Spring AI, see this article by my colleague Tim Kelly.

In Part 1, we established an important boundary: the WMS defines what the agent is allowed to do, while the agent decides what to do within those boundaries. Now we can translate that design into code.

Before diving into the implementation, the following diagram summarizes the complete agent flow across Parts 2 and 3. In this part, we will focus on defining the agent, triggering its execution, and creating the plan.

Define the Agent

The agent is triggered after an outbound operation is completed, with the goal of determining whether that operation created a need for replenishment.

The first step is to define how the agent should behave and what it is allowed to do.

In our implementation, we organize the agent around three main components:

planner
executor
reporter.

The planner receives the goal and determines the tasks required to achieve it:

private static final String PLANNER_PROMPT = """
   - You are the planning agent for a WMS. 
   - Create a short execution plan for the provided goal using the available capabilities. 
   - Do not execute tasks, call tools or make business decisions. 
   - Assign exactly one capability to each task. """;

The executor will receive one task at a time and perform it using the tools available for that task:

private static final String EXECUTOR_PROMPT = """
   - You are an AI agent specialized in Warehouse Management Systems.
   - Analyze inventory after outbound operations and anticipate replenishment needs.
   - Execute only the task you receive, using its available tools.
   - Apply the depositor policies and avoid unnecessary replenishments.""";

Finally, the reporter summarizes what happened during the execution:

private static final String REPORTER_PROMPT = """
    - Summarize the outcome of the agent execution.
    - Explain what was decided and why.
    - Do not repeat the task list.""";

These are simplified versions of the prompts used in the complete application, but they show the responsibility of each component. Together, these components become part of our agent definition:

public record AgentDefinition(
        String trigger,
        String goal,
        ChatClient planner,
        ChatClient executor,
        ChatClient reporter,
        List<AgentCapability> capabilities
) {}

The definition also contains the capabilities available to the agent. In our replenishment workflow, we use five:

Analysis: access invoices, inventory, and stock movements.
Policy: retrieve depositor policies.
Decision: determine whether replenishment is required.
Replenishment: create a replenishment request.
Notification: draft the related email.

Each capability defines a type of operation and the tools that can be used while performing it:

public record AgentCapability(
        String name,
        String description,
        List<Object> tools
) {}

This gives us a simple boundary around each task. An ANALYSIS task, for example, can access inventory and stock movement tools, but it cannot create a replenishment request. A DECISION task does not expose any tool and uses the results collected by the previous tasks.

Finally, we can connect these pieces when defining the replenishment agent:

return new AgentDefinition(
        "OUTBOUND_INVOICE_COMPLETED",
        "Outbound invoice %s has just been completed.",
        planner,
        executor,
        reporter,
        List.of(
                new AgentCapability(
                        "ANALYSIS",
                        "Read inventory, stock movements and invoice movements.",
                        List.of(inventoryAnalysisTool)
                ),
                new AgentCapability(
                        "DECISION",
                        "Decide whether replenishment is required.",
                        List.of()
                ),
                new AgentCapability(
                        "REPLENISHMENT",
                        "Create a replenishment request.",
                        List.of(replenishmentTool)
                )
                // Other capabilities: POLICY and NOTIFICATION
        )
);

At this point, we have defined what the agent can do and the boundaries around those operations. Before looking at how the plan is created, let's see how the agent execution starts.

Trigger the Agent

The agent starts after an outbound operation has been successfully completed. Once the transaction is committed, the application publishes the OUTBOUND_INVOICE_COMPLETED event:

public void execute(String invoiceNumber) {
   //  registerStock()
   // saveOutboundInvoice()
   eventPublisher.publishEvent(new OutboundInvoiceCompleted(number));
}

A listener receives that event and starts the agent using the outbound invoice number as the reference:

@Async
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void onOutboundInvoiceCompleted(OutboundInvoiceCompleted event) {
    agentRunner.run(
        replenishmentAgentDefinition,
        event.invoiceNumber()
    );
}

Using AFTER_COMMIT ensures that the agent starts only after the outbound operation has been persisted and the inventory already reflects the latest state. From there, the agent receives its goal and can begin planning how to accomplish it.

Plan

Once the agent is triggered, the first responsibility of the AgentRunner is to create an execution plan.

The application does not define this list of tasks in advance. Instead, the planner receives the goal together with the capabilities available to the agent and determines which tasks are required for its execution.

Each planned task contains two pieces of information: what needs to be done and which capability should be responsible for it:

public record PlannedTask(String description, String capability) {}

Based on this information, the planner generates an ordered list of tasks:

private List<AgentRun.AgentTask> plan(
        AgentDefinition definition,
        String goal
) {
    List<PlannedTask> planned = definition.planner()
            .prompt()
            .user("""
                %s
                Available capabilities:
                %s
                Create the execution plan required to accomplish this goal.
                Assign exactly one capability to each task.
                """.formatted(goal, capabilityCatalog(definition)))
            .call()
            .entity(new ParameterizedTypeReference<>() {});
  return planned.stream()
            .map(task -> new AgentRun.AgentTask(
                    task.description(),
                    AgentRun.TaskStatus.PENDING,
                    task.capability(),
                    null,
                    null,
                    null
            ))
            .toList();
}

Here, Spring AI maps the model response directly to our PlannedTask objects through structured output, giving the application a typed execution plan instead of requiring us to parse generated text manually.

Once the planner returns the tasks, we persist the initial state of the execution as an AgentRun in MongoDB.

At this point, the execution is marked as RUNNING, and every task generated by the planner starts with the PENDING status:

String goal = definition.goal().formatted(reference);
List<AgentRun.AgentTask> tasks =
        new ArrayList<>(plan(definition, goal));
AgentRun agentRun = agentRunService.save(new AgentRun(
        null,
        definition.trigger(),
        reference,
        AgentRun.Status.RUNNING,
        null,
        LocalDateTime.now(),
        null,
        List.copyOf(tasks)
));

This gives us a persistent representation of the execution from the beginning, including the tasks that were planned. At this point, the document may look like this:

{
  _id: ObjectId('6a8f5973315f4506564a6f5b'),
  trigger: 'OUTBOUND_INVOICE_COMPLETED',
  reference: 'NF-OUT-01',
  status: 'RUNNING',
  startedAt: ISODate('2026-08-27T17:04:03.350Z'),
  tasks: [
    {
      description: 'Identify the products affected by the outbound invoice.',
      status: 'PENDING',
      capability: 'ANALYSIS'
    },
    {
      description: 'Check the current inventory of the affected products.',
      status: 'PENDING',
      capability: 'ANALYSIS'
    },
    {
      description: 'Analyze recent stock movements and consumption.',
      status: 'PENDING',
      capability: 'ANALYSIS'
    },
    {
      description: 'Read the replenishment policies for the depositor.',
      status: 'PENDING',
      capability: 'POLICY'
    },
    {
      description: 'Determine whether replenishment is required.',
      status: 'PENDING',
      capability: 'DECISION'
    },
    {
      description: 'Create a replenishment request if required.',
      status: 'PENDING',
      capability: 'REPLENISHMENT'
    },
    {
      description: 'Draft the notification if a request was created.',
      status: 'PENDING',
      capability: 'NOTIFICATION'
    }
  ]
}

The plan is now ready, and the execution has been registered.

Conclusion

In this second part, we translated the agent design into the first pieces of the implementation. We defined its responsibilities and capabilities, connected it to the WMS workflow, and used the planner to turn the agent goal into a persistent execution plan.

At this point, the agent knows what needs to be done and which capability is responsible for each task. In Part 3, we will focus on how that plan is executed: using controlled tools, gathering operational and business context, making the replenishment decision, and acting when necessary.

The complete source code is available here.

Building an Agentic Warehouse Management System — Part 1: Where AI Agents Add Value

Ricardo Mello — Thu, 03 Sep 2026 14:41:06 +0000

I spent a large part of my career working in banking and logistics. In logistics, I spent more than eight years working across different stages of logistics systems, from development to implementation and production support.

That experience shaped the way I think about technology. Knowing a tool or framework is important, but what interests me most is understanding the business problem behind it and where technology can actually add value.

A few weeks ago, I returned as a speaker to a company where I had worked for many years. Being back there reminded me of several day-to-day logistical challenges and made me wonder how I would approach those problems today.

That is where the Agentic Warehouse Management System came from.

With this project, I am not trying to build a complete WMS. Instead, I want to use a logistics scenario I know well to explore how an AI agent can become part of a real business workflow and where it can add value.

In this first part, we will introduce the WMS scenario and identify where an AI agent can help. In Part 2, we will define and connect the agent using Java and Spring AI. Finally, in Part 3, we will execute the plan, gather context, make the replenishment decision, and act when necessary.

Try the application

A live version of the Agentic WMS is available here.

You can start with the Overview page, which walks through the main WMS flow and shows how to interact with the application. The application is built with:

Java and Spring Boot
Spring AI for the agent and tool integration
OpenAI for the language model
Voyage AI for embeddings
MongoDB for operational data, agent runs, and vector search

The complete source code is available here.

The Warehouse Management System

A Warehouse Management System, or simply WMS, is a software system used to manage and control the movement of products inside a warehouse. It typically handles processes such as receiving goods, tracking inventory, shipping products, replenishment, and many others. A warehouse may store products owned by the company operating it, but it may also store products for other companies. In this article, we will call these companies depositors:

A depositor owns the products stored in the warehouse, while the warehouse is responsible for receiving, storing, managing, and shipping them. A single warehouse may serve multiple depositors at the same time. For example, products from Amazon, Nike, or Walmart could be stored and managed in the same warehouse, while the inventory of each depositor remains separated. A real WMS can support many more processes, but for our application, we will focus on a smaller part of this flow:

An Inbound operation represents goods entering the warehouse and increases the available inventory for a depositor. An Outbound operation represents goods leaving the warehouse and reduces that inventory.

We will also include a simple Replenishment Request process. When additional units are needed, the warehouse can create a request indicating which products should be replenished and in what quantities. The depositor can then arrange a new inbound shipment to the warehouse. Once a replenishment request is created, a Depositor Email Draft is also prepared for notification.

Modeling the WMS flow

In the flow above, a traditional implementation would involve a few basic entities such as Inbound, Outbound, Inventory, and Replenishment Request, along with products and depositors. An Inbound operation represents products owned by a depositor entering the warehouse. For example:

// INBOUND COLLECTION
{ 
  number: 'NF-IN-01',
  depositor: { _id: ObjectId('6a88aaa60aab66cebc2c4e62'), name: 'Amazon' },
  items: [ { productCode: 'BR01', quantity: 1500 } ],
  status: 'COMPLETED' 
}

Once completed, those quantities are added to the Inventory:

// INVENTORY COLLECTION
{ 
  productCode: 'BR01', 
  depositor: { _id: ObjectId('6a88aaa60aab66cebc2c4e62'), name: 'Amazon' }, 
  quantity: 1500 
}

An Outbound operation does the opposite. Products leave the warehouse, and their quantities are reduced from the inventory:

// OUTBOUND COLLECTION
{
  number: 'NF-OUT-01',
  depositor: { _id: ObjectId('6a88aaa60aab66cebc2c4e62'), name: 'Amazon' }, 
  items: [ { productCode: 'BR01', quantity: 1480 } ],
  status: 'COMPLETED'
}

After this operation, the inventory for product BR01 belonging to the depositor Amazon would contain only 20 units. At this point, a Replenishment Request may be needed. In practical terms, the warehouse is essentially saying:

“Amazon, your product BR01 is down to only 20 units. Could you arrange a new inbound shipment before we run out of stock?”

In our WMS, this request can be represented by a document like the following:

{
  depositor: { _id: ObjectId('6a88aaa60aab66cebc2c4e62'), name: 'Amazon' }, 
  items: [ { productCode: 'BR01', quantity: 1500 } ],
  message: 'Replenishment is required for product BR01 due to low  stock.', 
  status: 'PENDING'
}

Deciding when to replenish

Creating a Replenishment Request is straightforward. The more interesting question is deciding when one should be created. Consider the following stock movements for product BR01:

On August 18, Amazon sent 1,500 units of the product to the warehouse. Over the following days, the warehouse processes outbound operations of 300 units per day.

After each outbound operation, the available inventory becomes:

1,500 → 1,200 → 900 → 600 → 300 → 0

If this consumption pattern continues, the warehouse will completely run out of BR01 on the fifth outbound operation.

A common way to automate this process is to define a minimum stock threshold. For example, whenever fewer than 500 units are available, a replenishment request could be created:

if (inventory.getQuantity() < 500) {
   replenishmentService.create(productCode, quantity, depositor);
}

This is a perfectly reasonable rule. Once the inventory reaches 300 units, the WMS detects that the stock is below the threshold (500) and requests more products.

However, consider that the depositor takes, on average, 7 days to deliver new products to the warehouse. If we wait until the stock falls below 500 units, the request may come too late, and the warehouse may run out of BR01 before the new shipment arrives.

We can account for this by adding the remaining days of stock to the decision:

int daysOfStock = inventory.getQuantity() / dailyConsumption;
if (inventory.getQuantity() < 500
        || daysOfStock < depositor.getLeadTimeDays()) {
    replenishmentService.create(productCode, quantity, depositor);
}

Now the WMS can react to low stock while also anticipating replenishment when the remaining inventory is not enough to cover the depositor's delivery lead time.

These are deterministic rules, and they work well when the decision can be clearly expressed through known conditions and calculations. As the business evolves, however, new rules and exceptions may continue to appear.

At some point, the challenge is no longer just adding another condition.

When the decision becomes contextual

The problem becomes more interesting when deciding what to do requires more than evaluating a predefined condition. Let's go back to the same example.

Amazon sent 1,500 units of BR01, and the warehouse has been shipping 300 units per day. After three days, only 600 units remain.

Using the rule we created earlier, the calculation is straightforward:

Current inventory: 600 units
Recent consumption: 300 units per day
Remaining stock: 2 days
Amazon lead time: 7 days

Since two days of stock are not enough to cover a seven-day lead time, our code will create a replenishment request.

But consider two different situations:

Scenario A: The recent consumption of 300 units per day is part of the normal demand range for BR01. There is no indication that it will slow down. In this case, creating a replenishment request makes sense.
Scenario B: The same 300-unit outbound operations came from a short promotion that has already ended. Previous movements show that BR01 typically consumes only around 50 units per day. In this case, immediately requesting the same replenishment may not be necessary.

Our deterministic rule sees the same values in both situations: 600 units in stock, 300 units of recent daily consumption, and a seven-day lead time, so it produces the same result.

But the situations are not the same.

To make a better decision, the system needs more context. It may need to inspect previous stock movements, understand whether the recent consumption represents the normal behavior of the product, retrieve relevant depositor policies, and then decide whether action is actually necessary.

This is where an AI agent starts to add value.

The agent does not replace deterministic operations such as inventory calculations, validations, persistence, or the creation of a replenishment request. Those operations remain regular application logic.

Instead, the agent can use controlled tools to look beyond the values used by the fixed rule. It can inspect previous stock movements, check pending replenishments, retrieve depositor policies, and gather additional information that helps explain the current situation.

The question is no longer only:

Is the inventory below 500 units, or is the remaining stock not enough to cover the depositor's lead time?

It becomes:

What does this situation actually mean, and do we need to act now?

That is the role of the agent in our WMS: gather the relevant information, understand the current situation, decide whether replenishment is necessary, and determine what should happen next, while keeping well-defined business operations in deterministic code.

Where does the agent fit?

In our application, the agent enters the flow after an outbound operation is successfully completed. At that point, the inventory has already been updated, and the agent receives a goal:

Determine whether this outbound operation created a need for replenishment.

Before looking at the flow, it is useful to clarify what we mean by an AI agent.

In simple terms, an AI agent receives a goal, follows a set of instructions, interacts with the application through tools, observes the information it retrieves, and decides what to do next.

In our WMS, these elements have clear responsibilities:

Goal: defines what the agent needs to accomplish. In this case, determine whether the completed outbound operation created a need for replenishment.
Instructions: define how the agent should behave, what it should consider, and the boundaries it must respect.
Tools: provide controlled access to the WMS. The agent can use them to inspect inventory and stock movements, retrieve depositor policies, check pending replenishments, or perform actions such as creating a replenishment request.

The important point is that the agent does not replace the existing WMS services or take control of the entire application. Inbound processing, outbound processing, inventory updates, validations, and persistence continue to be handled by regular Java code.

Instead, the agent acts as an orchestrator, deciding which available tools to use based on the goal and the context it gathers.

The following diagram shows where the agent becomes part of the WMS flow:

The diagram may look like a sequence of steps, but this is not a traditional hard-coded workflow where the application defines every decision in advance.

The application defines the agent's configuration, instructions, boundaries, and available tools. Within those boundaries, the agent determines what information it needs, uses the results it collects as context, decides whether action is necessary, and determines what should happen next.

For one outbound operation, the agent may conclude that replenishment is required and create a request. For another, after analyzing the available context, it may decide that no action is necessary.

This is an important distinction: the WMS controls what the agent is allowed to do, while the agent is responsible for deciding what to do within those boundaries.

Once the outbound invoice is completed, the OUTBOUND_INVOICE_COMPLETED event triggers the agent. From there, the workflow can be summarized in four steps:

Plan: Based on the goal, instructions, and available capabilities, the agent determines what information and actions may be required.
Analyze: The agent uses read-only tools to inspect information such as inventory, stock movements, pending replenishments, and depositor policies.
Decide: Using the information collected, the agent determines whether replenishment is actually necessary.
Act: If action is required, the agent can use controlled tools to create the replenishment request and prepare the depositor notification. If no action is required, the workflow ends without changing anything.

Conclusion

In this first part, we focused on where an AI agent can add value inside a Warehouse Management System and, just as importantly, where it should not replace deterministic application logic.

In Part 2, we will translate this design into Java and Spring AI by defining the agent, connecting it to the WMS workflow, and creating its execution plan.

Then, in Part 3, we will see how the agent executes those tasks using controlled tools, gathers the necessary context, makes the replenishment decision, and acts when required.

The complete source code is available here.

Building E-commerce AI Agents on MongoDB with CrewAI

Jordan Smith — Tue, 25 Aug 2026 17:04:18 +0000

AI-powered shopping agents are becoming a real differentiator for e-commerce teams. A 2026 Stord report on AI in e-commerce revealed a 34% year-over-year increase in the number of consumers who reported using generative AI for online shopping. The trend is even more pronounced among younger consumers: 37% of Gen Z and 29% of Millennials actively use AI to shop online. After seeing how the agentic commerce journey actually works, it's not hard to understand why.

This tutorial shows you exactly how to build one on MongoDB Atlas with CrewAI. You’ll create a chat-based shopping assistant that understands natural-language queries, retrieves relevant products with MongoDB Vector Search and MongoDB Search, and uses Change Streams, Atlas Triggers, and Atlas Stream Processing to keep inventory, checkout, and post-purchase experiences in sync with real-time data.

By the end, you’ll be able to:

Build a CrewAI shopping agent that calls MongoDB Vector Search tools with structured filters (price, availability).
Model an e-commerce catalog on MongoDB and generate high-quality product embeddings for semantic search.
Use MongoDB Change Streams to make checkout flows inventory-aware at the moment of purchase.
Trigger post-purchase workflows and customer notifications with MongoDB Atlas Triggers.
Detect emerging delivery issues with MongoDB Atlas Stream Processing so agents can respond proactively.

This GitHub repository wires up the code samples in this post. The code will help you build a CrewAI shopping agent and use MongoDB Vector Search on Atlas to understand the semantic meaning of customer queries, implement atomic inventory checkout with Change Streams, and (as deploy-ready artifacts) create an Atlas Trigger and an Atlas Stream Processing pipeline for post-purchase intelligence.

What is the agentic commerce journey?

Here is how a fully agentic shopping experience (a.k.a zero-click buying) works in practice:

A customer asks a question: Via voice or text, they describe what he/she needs. The agent interprets the intent behind the query, not just the keywords.
The agent finds and curates the best options: It surfaces a short list of products with detailed specifications, pricing, deals, quality ratings, and reviews.
The customer makes a purchase decision: With curated options in the chat interface, the customer asks follow-up questions and then decides to purchase. What used to take hours of browsing now takes minutes.
The agent completes the transaction: For products that support instant checkout, the agent handles the entire payment process end-to-end (provided the customer’s payment information is on file). If not, the agent can safely ask the customer to provide payment information, encrypt it, and securely store it in the database for future use.
The agent handles everything post-purchase: It tracks the order, surfaces proactive updates when status changes, and manages returns or follow-ups if something goes wrong.

Each step in this journey places real technical demands on the underlying platform. The agent needs to understand natural language, retrieve semantically relevant products, evaluate quality and price across thousands of options, confirm inventory accuracy at the moment of purchase, and stay aware of order events without waiting to be asked. That's a lot of moving parts. Most teams choose to build them on a collection of separate systems, each adding its own overhead, latency, and points of failure. But what if a single platform could support these capabilities without requiring teams to stitch together multiple specialized systems?

That is where MongoDB comes in. By combining operational data, semantic search, real-time updates, and transactional capabilities into a single platform, MongoDB can support the different stages of the agentic commerce journey.

How do MongoDB capabilities fit into each step of the agentic commerce journey?

Agent Journey Step	MongoDB Feature
Understand customer intent	Atlas Vector Search + LLM integration + Voyage AI embedding model
Produce relevant results	Atlas Vector Search + MongoDB Search
Answer follow-up questions	Atlas Vector Search + MongoDB Search
Confirm inventory at checkout	Change Streams
Proactive order status updates	Atlas Triggers
Detect delivery patterns before they escalate	Atlas Stream Processing

MongoDB is a strong candidate because it offers the native features required to power each step of the agentic commerce journey. Let's see how.

Step 1–3: How do MongoDB Vector Search on Atlas and MongoDB Search power agent discovery and decision-making?

When a customer types "lightweight running shoes under $100 with good arch support," they're describing an intent. The agent needs to understand the semantic meaning behind that query and match it to products that fit that intent. The agent needs to understand the user's intent, not just focus on keywords.

MongoDB Vector Search on Atlas handles this natively. Your product embeddings are stored alongside your product data in the same cluster. When a query comes in, the agent converts it to a vector using a best-in-class embedding model (Voyage AI), runs a search in your product catalog, and retrieves the most semantically relevant results, ranked by relevance rather than just keyword frequency. We can refer to this as a product-search tool. Let’s see exactly how to build it.

First, use the code below to create a Vector Search index that includes both the embedding field and the fields the agent will use as filters. For the query to work, the collection needs a Vector Search index that matches the fields used in the code. The embedding field is indexed as the vector field, with 2048 dimensions to match the Voyage AI embedding model. The price and in_stock fields are indexed as filter fields, which lets MongoDB apply those constraints before running the vector similarity search. That means the agent searches only for products under $100 and available to buy, rather than retrieving relevant products first and filtering them afterward.

function productToEmbeddingText(product) {
  return [
    `Name: ${product.name}`,
    `Brand: ${product.brand}`,
    `Category: ${product.category}`,
    `Description: ${product.description}`,
    `Features: ${(product.features || []).join(", ")}`,
    `Review summary: ${product.reviewSummary}`,
    `Price: ${product.price}`,
    `In stock: ${product.in_stock}`
  ].filter(Boolean).join("\n");
}

const embeddingText = productToEmbeddingText(product);
const embedding = await createEmbedding(embeddingText);

await db.collection("products").updateOne(
  { _id: product._id },
  {
    $set: {
      embedding,
      embedding_text: embeddingText
    }
  }
);

With that index in place, the agent can run a semantic search and apply price and inventory constraints before performing similarity ranking. Here is the code for that:

{
  "fields": [
    {
      "type": "vector",
      "path": "embedding",
      "numDimensions": 2048,
      "similarity": "cosine"
    },
    {
      "type": "filter",
      "path": "price"
    },
    {
      "type": "filter",
      "path": "in_stock"
    }
  ]
}

Next, use the code below to create the CrewAI shopping agent that uses the product search tool. The agent is given a role, a goal, and a backstory so it understands how to behave during the shopping experience. By attaching search_product_catalog (defined next) as a tool, we give the agent access to MongoDB Vector Search in Atlas and the live product catalog. When a customer asks for a recommendation, the agent can call that tool, retrieve relevant in-stock products, and use the results to provide a helpful, context-aware answer.

from crewai import Agent

shopping_agent = Agent(
    role="Shopping assistant",
    goal="Help customers find products that match their intent and constraints.",
    backstory=(
        "You help customers discover relevant products using MongoDB Atlas "
        "Vector Search and real-time product data."
    ),
    tools=[search_product_catalog],
    llm="gpt-4o",
)

The search_product_catalog function is the tool attached to the CrewAI shopping agent. When the agent needs to search the catalog, this tool connects to the product collection, uses Voyage AI to convert the customer’s request into a query embedding, and runs MongoDB Vector Search in Atlas against the embedding field using product_vector_index. The semantic part of the request, such as “lightweight running shoes with good arch support,” is handled by Vector Search, while the hard constraints, price <= 100 and in_stock: true, are applied as pre-filters. The result is a reusable CrewAI tool that returns products that are both semantically relevant and eligible for the customer to buy.

import os
from bson import json_util
from pymongo import MongoClient
from crewai.tools import tool

client = MongoClient(os.environ["MONGODB_URI"])
collection = client["ecommerce"]["products"]


@tool("Search product catalog")
def search_product_catalog(query: str) -> str:
    """Find in-stock products under $100 that semantically match the query.

    This version uses Atlas Vector Search auto-embedding. The application sends
    natural-language query text directly to MongoDB; Atlas generates the query
    embedding using the model configured in the Vector Search index.
    """

    pipeline = [
        {
            "$vectorSearch": {
                "index": "product_autoembed_index",
                "path": "search_text",
                "query": {
                    "text": query
                },
                "numCandidates": 100,
                "limit": 5,
                "filter": {
                    "$and": [
                        {"price": {"$lte": 100}},
                        {"in_stock": True}
                    ]
                }
            }
        },
        {
            "$project": {
                "_id": 0,
                "name": 1,
                "description": 1,
                "price": 1,
                "in_stock": 1,
                "score": {"$meta": "vectorSearchScore"}
            }
        }
    ]

    results = list(collection.aggregate(pipeline))
    return json_util.dumps(results, indent=2)

Alternative: Let Atlas generate embeddings automatically

The previous example shows manual embedding generation. If you want less application code, MongoDB Vector Search on Atlas can generate document and query embeddings automatically using an autoEmbed index. In that version, the app sends natural language query text directly to $vectorSearch. The code below explains how MongoDB Vector Search on Atlas generates document and query embeddings automatically.

{
  "fields": [
    {
      "type": "autoEmbed",
      "modality": "text",
      "path": "search_text",
      "model": "voyage-4"
    },
    {
      "type": "filter",
      "path": "price"
    },
    {
      "type": "filter",
      "path": "in_stock"
    }
  ]
}

For structured filtering, such as price ranges, brand filters, and in-stock flags, MongoDB Search layers on top of Vector Search. The agent can combine semantic similarity with exact filters in a single query, returning results that are both contextually relevant and within the customer's stated constraints.

Because your product catalog, inventory levels, reviews, and pricing (i.e., operational data) all reside in the same MongoDB database, the agent retrieves a complete, enriched product document in a single call.

Example output:

This is an example of the application's output in response to “Find in-stock products under $100 that semantically match the query.”

[
  {
    "name": "AeroFlex Runner 2",
    "description": "Lightweight daily running shoe with responsive foam and built-in arch support.",
    "price": 89.99,
    "in_stock": true,
    "score": 0.91
  },
  {
    "name": "CloudStep Support Knit",
    "description": "Breathable road running shoe designed for moderate overpronation and arch comfort.",
    "price": 94.5,
    "in_stock": true,
    "score": 0.88
  }
]

Optional: Connect your agent with the MongoDB MCP Server

The product-search tool above is one way to expose MongoDB to the agent. For more dynamic agent workflows, you can also connect the agent to MongoDB through the MongoDB MCP Server. MCP provides a standard tool interface that lets the agent discover MongoDB resources and issue structured database operations with the permissions you configure. In an e-commerce shopping flow, the agent could use MCP to look up product metadata, inspect inventory fields, retrieve customer order context, or run follow-up queries after Vector Search returns candidate products.

In the code below, the CrewAI agent is connected to MongoDB in two ways. The search_product_catalog tool handles the primary product discovery flow using Atlas Vector Search, while the MongoDB MCP Server provides the agent with a standard tool interface for follow-up database operations. The MCP server runs in read-only mode and exposes only selected tools, such as collection schema, find, and aggregate, so the agent can look up product details, inventory fields, or customer context without being given unrestricted database access.

import os
from crewai import Agent
from crewai.mcp import MCPServerStdio
from crewai.mcp.filters import create_static_tool_filter

mongo_mcp = MCPServerStdio(
    command="npx",
    args=[
        "-y",
        "mongodb-mcp-server@latest",
        "--readOnly",
    ],
    env={
        "MDB_MCP_CONNECTION_STRING": os.environ["MDB_MCP_CONNECTION_STRING"],
    },
    tool_filter=create_static_tool_filter(
        allowed_tool_names=[
            "list-databases",
            "list-collections",
            "collection-schema",
            "collection-indexes",
            "find",
            "aggregate",
        ]
    ),
    cache_tools_list=True,
)

shopping_agent = Agent(
    role="Shopping assistant",
    goal="Help customers find products that match their intent and constraints.",
    backstory=(
        "You help customers discover relevant products using MongoDB Atlas "
        "Vector Search, Atlas auto-embedding, live product data, and MongoDB MCP tools."
    ),
    tools=[search_product_catalog],
    mcps=[mongo_mcp],
    llm="gpt-4o",
)

Step 4: How do MongoDB Change Streams help with the checkout process?

This is where most teams don't anticipate the problem until they're already in production.

A customer has found the product, decided to buy it, and initiated checkout. At that exact moment, the agent needs to confirm that the item is still available.

During a high-traffic sale, inventory moves fast. The last unit of a limited-edition product can sell in the seconds between when a customer adds it to their cart and when the transaction is confirmed. If the agent is working with inventory data that doesn't reflect the final sale, the purchase goes through, the inventory system immediately rejects it, and the customer receives an "item not available" notification along with a refund. At scale, that failure pattern compounds quickly into frustrated customers, support overhead, and refund costs that add up fast.

MongoDB solves this in two parts: an atomic inventory update keeps checkout running smoothly, while Change Streams let the agent react immediately after inventory changes are committed. Change Streams is a cursor-based API that subscribes to document-level changes in a collection as soon as a write is committed to the database. There is no polling interval and therefore no sync gap. When inventory for a product drops to zero, the Change Stream fires instantly, and the agent receives the data instantly.

In the code below, checkout is protected by an atomic inventory update. The function looks for a product with the requested sku and enough remaining quantity, then decrements the quantity in the same database operation. If another customer has already purchased the last unit, the query no longer matches, and MongoDB returns None, so the checkout flow can stop before confirming the item is unavailable. Using ReturnDocument.AFTER returns the updated inventory document, giving the agent the freshest committed state after the reservation succeeds.

from pymongo import MongoClient, ReturnDocument

client = MongoClient(MONGODB_URI)
inventory = client["ecommerce"]["inventory"]


def reserve_inventory(sku: str, quantity: int = 1):
    """Atomically reserve inventory during checkout."""
    updated_product = inventory.find_one_and_update(
        {
            "sku": sku,
            "quantity": {"$gte": quantity},
        },
        {
            "$inc": {"quantity": -quantity},
        },
        return_document=ReturnDocument.AFTER,
    )

    if updated_product is None:
        raise ValueError(f"Product {sku} is no longer available.")

    return updated_product

In the code below, Change Streams detect when a product's inventory drops to zero. That event kicks off a CrewAI workflow: an inventory recovery agent finds active carts that still contain the sold-out SKU, suggests alternatives, and notifies affected customers. MongoDB handles the real-time event stream, while CrewAI handles the agent’s reasoning and tool orchestration.

from pymongo import MongoClient
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool

client = MongoClient(MONGODB_URI)
inventory = client["ecommerce"]["inventory"]


@tool("Find active sessions with SKU")
def find_active_sessions_with_sku(sku: str) -> list[str]:
    """Find active shopping sessions that currently have this SKU in cart."""
    # Replace with a query against your sessions/carts collection.
    return []


@tool("Suggest alternative product")
def suggest_alternative(sku: str) -> str:
    """Suggest a similar in-stock product for the provided SKU."""
    # Replace with MongoDB Vector Search or product recommendation logic.
    return f"Suggested replacement for {sku}"


@tool("Notify customer")
def notify_customer(session_id: str, message: str) -> str:
    """Send a message to a customer shopping session."""
    # Replace with chat, email, SMS, or app notification logic.
    return f"Queued notification for {session_id}"


stock_recovery_agent = Agent(
    role="Inventory recovery specialist",
    goal=(
        "Recover active shopping sessions when a product sells out by finding "
        "affected carts, suggesting alternatives, and notifying customers."
    ),
    backstory=(
        "You support an e-commerce shopping agent. When inventory changes, "
        "you help customers continue checkout with the least possible friction."
    ),
    tools=[
        find_active_sessions_with_sku,
        suggest_alternative,
        notify_customer,
    ],
    llm="gpt-4o",
)

stock_recovery_task = Task(
    description=(
        "Product {sku} just sold out. Find active sessions that still have "
        "this item in cart, suggest suitable alternatives, and notify each "
        "affected customer clearly."
    ),
    expected_output=(
        "A summary of affected sessions, suggested alternatives, and customer "
        "notifications that were queued."
    ),
    agent=stock_recovery_agent,
)

stock_recovery_crew = Crew(
    agents=[stock_recovery_agent],
    tasks=[stock_recovery_task],
    process=Process.sequential,
)

pipeline = [
    {
        "$match": {
            "operationType": "update",
            "fullDocument.quantity": 0,
        }
    }
]

with inventory.watch(pipeline, full_document="updateLookup") as stream:
    for event in stream:
        product = event["fullDocument"]
        sku = product["sku"]

        stock_recovery_crew.kickoff(
            inputs={
                "sku": sku,
            }
        )

Step 5: How do Atlas Triggers and Atlas Stream Processing help with Post-purchase intelligence?

The post-purchase stage is where most shopping agents stop short. A customer who has paid expects to be kept informed without having to ask. That means the agent needs to respond when order data changes, especially when a shipment enters a state that matters to the customer.

Atlas Triggers: Reacting to individual order events

When an order’s status changes to "shipped", that update is a committed database event. Atlas Triggers are a fully managed way to react to that event: Atlas opens and maintains the underlying change stream, applies your trigger filter, and invokes a serverless function when a matching order update occurs.

In the code below, the Trigger is configured on the orders collection to run only when an order status changes to "shipped", with full-document lookup enabled for update events. The function receives the updated order document through changeEvent.fullDocument builds a JSON request body with API-friendly values, then sends it to the post-purchase agent endpoint.

From there, the agent can pull the customer’s order history, anticipate likely follow-up questions such as delivery window, return policy, and product setup, and prepare a personalized response so the next customer interaction feels informed and seamless.

// Atlas Trigger Function — fires when an order status changes to "shipped"
exports = async function(changeEvent) {
  const order = changeEvent.fullDocument;

  if (!order || order.status !== "shipped") {
    return { skipped: true };
  }

  const response = await context.http.post({
    url: "https://your-agent-endpoint.com/post-purchase",
    body: {
      customerId: order.customerId,
      orderId: order._id.toString(),
      product: order.product,
      estimatedDelivery: order.estimatedDelivery
    },
    encodeBodyAsJSON: true
  });

  if (response.statusCode < 200 || response.statusCode >= 300) {
    throw new Error(
      `Post-purchase agent request failed with status ${response.statusCode}`
    );
  }

  return {
    ok: true,
    orderId: order._id.toString(),
    statusCode: response.statusCode
  };
};

Atlas Stream Processing: Detecting patterns before they become problems

Some post-purchase issues don't surface as a single event but emerge as a pattern. A carrier delay affecting multiple orders in the same region, such as due to an unexpected weather event, doesn't appear as a single document change. It shows up as a sustained pattern of delivery estimates shifting across an order cohort.

Atlas Stream Processing runs continuous aggregation pipelines over your order data and surfaces confirmed patterns before they reach customers. As illustrated in the code below, Atlas Stream Processing monitors delivery-estimate updates in the orders collection. Each update represents one order whose delivery date changed, but a single delayed order usually is not enough to involve an agent. The processor first filters for meaningful delays, such as orders whose estimated delivery time has moved by 30 minutes or more, then waits for a two-minute tumbling window to evaluate a batch of recent updates together.

Within each window, it groups delayed orders by carrier and shippingRegion. If at least 10 orders in the same carrier-region cohort are delayed, and the average delay across that group is at least 45 minutes, the processor writes one confirmed signal to agent_triggers.

A separate Change Stream can then watch that collection and invoke the demand-management agent. Then, the demand management agent can proactively draft customer communications, alert the logistics team, and flag the issue for resolution before customers start calling about missing orders. Imagine how many fewer calls your customer support team will receive if customers are proactively informed about shipping delays.

[
  {
    "$source": {
      "connectionName": "mongodb-atlas",
      "db": "ecommerce",
      "coll": "orders",
      "fullDocument": "updateLookup",
      "pipeline": [
        {
          "$match": {
            "operationType": "update",
            "updateDescription.updatedFields.estimatedDelivery": {
              "$exists": true
            }
          }
        }
      ]
    }
  },
  {
    "$replaceRoot": {
      "newRoot": "$fullDocument"
    }
  },
  {
    "$match": {
      "status": {
        "$in": ["shipped", "in_transit"]
      },
      "carrier": {
        "$exists": true
      },
      "shippingRegion": {
        "$exists": true
      },
      "originalEstimatedDelivery": {
        "$exists": true
      },
      "estimatedDelivery": {
        "$exists": true
      }
    }
  },
  {
    "$addFields": {
      "delayMinutes": {
        "$dateDiff": {
          "startDate": "$originalEstimatedDelivery",
          "endDate": "$estimatedDelivery",
          "unit": "minute"
        }
      }
    }
  },
  {
    "$match": {
      "delayMinutes": {
        "$gte": 30
      }
    }
  },
  {
    "$tumblingWindow": {
      "boundary": "processingTime",
      "interval": {
        "size": 2,
        "unit": "minute"
      },
      "pipeline": [
        {
          "$group": {
            "_id": {
              "carrier": "$carrier",
              "region": "$shippingRegion",
              "windowStart": {
                "$meta": "stream.window.start"
              }
            },
            "affectedOrderCount": {
              "$sum": 1
            },
            "averageDelayMinutes": {
              "$avg": "$delayMinutes"
            },
            "maxDelayMinutes": {
              "$max": "$delayMinutes"
            },
            "affectedOrders": {
              "$push": "$_id"
            }
          }
        }
      ]
    }
  },
  {
    "$match": {
      "affectedOrderCount": {
        "$gte": 10
      },
      "averageDelayMinutes": {
        "$gte": 45
      }
    }
  },
  {
    "$project": {
      "_id": {
        "$concat": [
          "delivery-delay:",
          "$_id.carrier",
          ":",
          "$_id.region",
          ":",
          {
            "$toString": "$_id.windowStart"
          }
        ]
      },
      "type": "delivery_delay_pattern",
      "severity": "warning",
      "carrier": "$_id.carrier",
      "region": "$_id.region",
      "windowStart": "$_id.windowStart",
      "affectedOrderCount": 1,
      "averageDelayMinutes": {
        "$round": ["$averageDelayMinutes", 0]
      },
      "maxDelayMinutes": 1,
      "affectedOrders": {
        "$slice": ["$affectedOrders", 25]
      },
      "agentAction": "draft_customer_delay_notice"
    }
  },
  {
    "$merge": {
      "into": {
        "connectionName": "mongodb-atlas",
        "db": "ecommerce",
        "coll": "agent_triggers"
      },
      "on": "_id",
      "whenMatched": "replace",
      "whenNotMatched": "insert"
    }
  }
]

Decision framework: How to choose the right event-driven service?

The decision framework below helps you choose the right event-driven service for your use case:

Figure 1. Choosing between Change Streams, Atlas Triggers, and Atlas Stream Processing based on your agent's use case.

Conclusion

Building a fully agentic shopping experience requires more than a good model and a well-tuned prompt. Every step of the journey, from product discovery to post-purchase follow-up, places real demands on the underlying data platform. The agent needs to retrieve semantically relevant products, confirm inventory at the moment of purchase, and react to order events without waiting to be asked.

Most teams solve this by assembling a stack of specialized systems: a search service, a vector database, a message broker, event processing infrastructure, and an operational database underneath it all. That architecture can work, but every handoff adds latency, synchronization headaches, operational overhead, and another point where data can drift from reality. For an agent that needs to reason and act in the moment, those gaps mean the agent fails to provide customers with valuable information when needed. Not because there is anything wrong with the agent, but because the data it relied on was outdated.

MongoDB consolidates these into a single platform. Vector Search, MongoDB Search, Change Streams, Atlas Triggers, and Atlas Stream Processing are all native features of the same cluster that holds your operational data. The agent works with data that reflects the world as it actually is, and the architecture stays simple enough to debug when something inevitably needs attention.

Build your first agent with MongoDB and CrewAI

If you want to put this into practice, MongoDB has an official integration with CrewAI. Use this GitHub repository to start building your E-commerce AI agents on MongoDB.

If you are already running on MongoDB Atlas, you have everything you need to get started. Sign up for MongoDB Atlas and follow the CrewAI integration tutorial to deploy your first e-commerce agent today.

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.

Developing a World Cup Stats application with Django MongoDB Extensions' MQL Panel

Abigail Afi Gbadago — Fri, 31 Jul 2026 13:32:12 +0000

Django MongoDB Extensions is a package that contains additional developer tools for the Django MongoDB Backend. One of these tools is the MongoDB Query Language (MQL) panel, and it shows MQL queries that your application runs during a request, including execution times and query plans. This tutorial shows you how to configure and use the panel.

We will highlight the functionality of Django Debug Toolbar by building a World Cup 2026 analytics application that queries match data using the Django ORM and raw aggregation pipelines. We'll use the Django Debug Toolbar to inspect the pipelines these queries generate as MQL.

There are two ways to follow this tutorial.
Option A: Clone the GitHub repo containing the finished code from this tutorial, set it up using the repo's README, and explore it on your own. Link to the GitHub project:django-mongodb-world-cup

Option B: Start the project from scratch by following the steps below.

Start the project from scratch

Prerequisites
NB: This tutorial uses config as the Django project name and world_cup as the app name throughout. If you use different names, you'll need to update every reference to them in the code examples that follow (imports, INSTALLED_APPS, and urls.py).

Before starting this tutorial, you will need to complete the following steps:

Download and install Django MongoDB Backend
Create a MongoDB deployment
Create a connection string
Create a Django project As a quick reference, you can create the Django project with the following command:

django-admin startproject config --template https://github.com/mongodb-labs/django-mongodb-project/archive/refs/heads/6.0.x.zip

(NB: I used config here, but you can also specify a Django project name of your choice, for example, world_cup_project)

Configure your MongoDB connection

These steps are explained in detail in the Get Started with Django MongoDB Backend tutorial.

Once these prerequisites are complete, the next step is installing Django MongoDB Extensions.

Install Django MongoDB Extensions

Navigate to your project's root directory; either the folder created by startproject, or the folder created when you clone the repository (i.e., django-mongodb-world-cup). This is also where you should create and activate your virtual environment, if you haven't already:

python3 -m venv venv
source venv/bin/activate

Create a requirements.txt file at the root of the project with the following contents:

#requirements.txt
django-mongodb-extensions
django-debug-toolbar
python-dotenv

Run the following command in your terminal to install the packages:
pip install -r requirements.txt
Create a .env file at the root of the project with the following contents and replace the placeholders with your own values.

For the SECRET_KEY, copy the value Django already generated in your settings.py when you ran startproject; it's normally hardcoded there by default, but we move it into .env to keep it out of version control, since it's used to sign session cookies, CSRF tokens, and other security-sensitive data.

#.env 
SECRET_KEY=[REPLACE_WITH_YOUR_SECRET_KEY]
MONGODB_URI="mongodb+srv://<username>:<password>@<cluster>.mongodb.net/?retryWrites=true&w=majority&appName=world_cup"
DB_NAME=[REPLACE_WITH_YOUR_DB_NAME]

Then update DATABASES and SECRET_KEY variables in settings.py to read from the environment instead of being hardcoded. Your settings.py should now include the following:

import os
from dotenv import load_dotenv
load_dotenv()
SECRET_KEY = os.getenv('SECRET_KEY')
DATABASES = {
    'default': {
        'ENGINE': 'django_mongodb_backend',
        'HOST': os.getenv('MONGODB_URI'),
        'NAME': os.getenv('DB_NAME'),
    },
}

Add Django MongoDB Extensions and Django Debug Toolbar to your installed apps: Navigate to your project's(or cloned repo) settings.py file and add "django_mongodb_extensions" to the INSTALLED_APPS setting, as shown in the following example:

INSTALLED_APPS = [
    'django_mongodb_extensions',
    'debug_toolbar',
]

Register the MQL panel

Replace your existing DEBUG_TOOLBAR_PANELS setting with the list below. This is Django Debug Toolbar's default panel list with one change: SQLPanel is swapped out for MQLPanel.

DEBUG_TOOLBAR_PANELS = [
    'debug_toolbar.panels.history.HistoryPanel',
    'debug_toolbar.panels.versions.VersionsPanel',
    'debug_toolbar.panels.timer.TimerPanel',
    'debug_toolbar.panels.settings.SettingsPanel',
    'debug_toolbar.panels.headers.HeadersPanel',
    'debug_toolbar.panels.request.RequestPanel',
    'django_mongodb_extensions.mql_panel.MQLPanel',
    'debug_toolbar.panels.staticfiles.StaticFilesPanel',
    'debug_toolbar.panels.templates.TemplatesPanel',
    'debug_toolbar.panels.alerts.AlertsPanel',
    'debug_toolbar.panels.cache.CachePanel',
    'debug_toolbar.panels.signals.SignalsPanel',
    'debug_toolbar.panels.community.CommunityPanel',
    'debug_toolbar.panels.redirects.RedirectsPanel',
    'debug_toolbar.panels.profiling.ProfilingPanel',
]

Configure Django Debug Toolbar Middleware

Add DebugToolbarMiddleware as the first item in MIDDLEWARE, and set INTERNAL_IPS:

MIDDLEWARE = [
    'debug_toolbar.middleware.DebugToolbarMiddleware',
    'django.middleware.security.SecurityMiddleware',
    # ... rest of middleware
]
INTERNAL_IPS = ['127.0.0.1']

Add Django Debug Toolbar’s URLs

In urls.py, add the following:

from django.conf import settings
from django.urls import include, path
urlpatterns = [
    # ... your app urls
]
if settings.DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

Create an Application

From your project's root directory, run the following command to create a new Django app. This tutorial uses the name world_cup throughout, so if you use a different name, you'll need to update it in every file this tutorial references (INSTALLED_APPS, imports, urls.py):
python manage.py startapp world_cup

Next, include your app in your project by opening the settings.py file in config and editing your INSTALLED_APPS setting to resemble the following code:

INSTALLED_APPS = [
    'config.apps.MongoAdminConfig',
    'config.apps.MongoAuthConfig',
    'config.apps.MongoContentTypesConfig',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'django_mongodb_backend',
    'django_mongodb_extensions',
    'debug_toolbar',
    'world_cup',
]

Create models for Goal and Match data

The world_cup application stores the World Cup match data using two models: Match and Goal. The Goal model is embedded inside Match as an array of goal events because it is usually queried together with the corresponding matches.

Open the models.py file in the world_cup directory and replace its contents with the following code:

from django.db import models
from django_mongodb_backend.fields import EmbeddedModelArrayField
from django_mongodb_backend.managers import MongoManager
from django_mongodb_backend.models import EmbeddedModel

class Goal(EmbeddedModel):
    team = models.CharField(max_length=100)
    name = models.CharField(max_length=100)
    minute = models.CharField(max_length=10)     
    penalty = models.BooleanField(default=False)
    own_goal = models.BooleanField(default=False)

    def __str__(self):
        return f"{self.name} (({self.team}, {self.minute}')"
class Match(models.Model):
    num = models.IntegerField(null=True, blank=True)        
    round = models.CharField(max_length=50)
    date = models.DateField()
    kickoff_time = models.CharField(max_length=20)           
    group = models.CharField(max_length=20, blank=True)      
    ground = models.CharField(max_length=100)
    team1 = models.CharField(max_length=100)
    team2 = models.CharField(max_length=100)
#Fields ending in "1" refer to team1, fields ending in "2" refer to team2
    ft_score1 = models.IntegerField(null=True, blank=True)      #ft = full-time
    ft_score2 = models.IntegerField(null=True, blank=True)
    ht_score1 = models.IntegerField(null=True, blank=True)  #ht = half-time
    ht_score2 = models.IntegerField(null=True, blank=True)
    et_score1 = models.IntegerField(null=True, blank=True)  #et = extra-time
    et_score2 = models.IntegerField(null=True, blank=True)
    pen_score1 = models.IntegerField(null=True, blank=True) #penalty score
    pen_score2 = models.IntegerField(null=True, blank=True)
   goals = EmbeddedModelArrayField(Goal, default=list, blank=True)
objects = MongoManager()

    def __str__(self):
        return f"{self.team1} vs {self.team2} ({self.round})"

The Goal model represents the world_cup.world_cup_match collection and stores information about the goals. This model contains fields such as name, minute, penalty, and own goal. The Goal model contains the embedded document values stored in the Match model.

Run Migrations

Generate and apply the migration for the world_cup app:

python manage.py makemigrations world_cup
python manage.py migrate

Loading Data

mkdir -p world_cup/management/commands
mkdir -p world_cup/data
touch world_cup/management/__init__.py
touch world_cup/management/commands/__init__.py
touch world_cup/management/commands/load_matches.py

For the World Cup data, we use the JSON file from Open Football GitHub source and save it to the data folder under the world_cup folder (/world_cup/data/worldcup_2026.json).

After obtaining the data (JSON file), we create a script (world_cup/management/commands/load_matches.py) to load the matches. Replace the following content in the load_matches.py file.

# world_cup/management/commands/load_matches.py
import json
from datetime import datetime
from pathlib import Path

from django.core.management.base import BaseCommand

from world_cup.models import Goal, Match

WORLD_CUP_DATA = Path(__file__).resolve().parent.parent.parent / "data" / "worldcup_2026.json"


def build_goals(raw_goals, team_name):
    return [
        Goal(
            team=team_name,
            name=g["name"],
            minute=g["minute"],
            penalty=g.get("penalty", False),
            own_goal=g.get("owngoal", False),
        )
        for g in raw_goals
    ]
def to_match_kwargs(raw_match):
    score = raw_match.get("score", {})
    ft = score.get("ft", [None, None])
    ht = score.get("ht", [None, None])
    et = score.get("et", [None, None])
    p = score.get("p", [None, None])

    return dict(
        num=raw_match.get("num"),
        round=raw_match["round"],
        date=datetime.strptime(raw_match["date"], "%Y-%m-%d").date(),
        kickoff_time=raw_match["time"],
        group=raw_match.get("group", ""),
        ground=raw_match["ground"],
        team1=raw_match["team1"],
        team2=raw_match["team2"],
        ft_score1=ft[0],
        ft_score2=ft[1],
        ht_score1=ht[0],
        ht_score2=ht[1],
        et_score1=et[0],
        et_score2=et[1],
        pen_score1=p[0],
        pen_score2=p[1],
        goals=build_goals(raw_match.get("goals1", []), raw_match["team1"]) + build_goals(raw_match.get("goals2", []), raw_match["team2"]),
    )
class Command(BaseCommand):
    help = (
        "Load the curated World Cup 2026 matches (Groups A, B, I, J) into MongoDB. "
        "Source: worldcup_data/worldcup_2026.json — 24 matches total, "
        "16 completed with real goals, 8 unplayed (Matchday 14/17)."
    )

    def handle(self, *args, **options):
        with open(WORLD_CUP_DATA) as f:
            raw_matches = json.load(f)["matches"]

        self.stdout.write(f"Found {len(raw_matches)} matches to load.")

        batch_size = 5
        created = 0
        for i in range(0, len(raw_matches), batch_size):
            batch = raw_matches[i : i + batch_size]
            Match.objects.bulk_create([Match(**to_match_kwargs(m)) for m in batch])
            created += len(batch)
            self.stdout.write(f"Loaded {created}/{len(raw_matches)} matches...")

        self.stdout.write(self.style.SUCCESS(f"Done: {created} matches loaded."))

Next, we load the data by running:
python manage.py load_matches

Once this is successful, we can see the results either by querying the database via the Python shell or via MongoDB Atlas’s Data Explorer. For example, to confirm the data loaded correctly and inspect an embedded Goal array, run:

python manage.py shell

In this shell:

>>> from world_cup.models import Match
>>> Match.objects.count()
100
>>> Match.objects.get(team1="France", team2="Senegal").goals1

Match results queried via Python shell

World Cup match stats after upload in MDB Atlas

Create Views

These views provide three pages: a total goals counter using the Django ORM, a top scorers leaderboard using a raw aggregation pipeline, and an interactive query shell that allows you to run five different query types.

Open the views.py file in your world_cup directory and replace its contents with the following code:

# world_cup/views.py
from django.db import connections
from django.db.models import F, Sum
from django.shortcuts import render

from .models import Match


def tournament_goals(request):
    """Pipeline 1 — built-in aggregate: total goals across completed matches."""
    result = (
        Match.objects.filter(ft_score1__isnull=False)
        .annotate(total_goals=F("ft_score1") + F("ft_score2"))
        .aggregate(tournament_goals=Sum("total_goals"))
    )
    return render(request, "world_cup/goals.html", {
        "total_goals": result["tournament_goals"] or 0,
    })


def top_scorers(request):
    """Pipeline 2 — raw_aggregate: leaderboard via $unwind + $group."""
    pipeline = [
        {"$unwind": "$goals"},
        {"$match": {"goals.own_goal": {"$ne": True}}},
        {"$group": {"_id": "$goals.name", "total_goals": {"$sum": 1}}},
        {"$sort": {"total_goals": -1}},
        {"$limit": 10},
    ]
    results = Match.objects.raw_aggregate(pipeline)
    scorers = [{"name": r.pk, "goals": r.total_goals} for r in results]
    return render(request, "world_cup/scorers.html", {"scorers": scorers})


ALLOWED_QUERIES = {
    "team": {
        "description": "All matches for a team",
        "example": "team France",
    },
    "round": {
        "description": "Matches in a round (e.g. round Quarter-final)",
        "example": "round Quarter-final",
    },
    "top scorers": {
        "description": "Top 10 scorers via $unwind + $group pipeline",
        "example": "top scorers",
    },
    "goals by round": {
        "description": "Goals per round via $group pipeline",
        "example": "goals by round",
    },
    "goals": {
        "description": "Total tournament goals",
        "example": "goals",
    },
}


def _execute_query(query_text):
    """Parse and execute a user query, return result dict."""
    q = query_text.strip().lower()

    if q == "goals":
        result = (
            Match.objects.filter(ft_score1__isnull=False)
            .annotate(total_goals=F("ft_score1") + F("ft_score2"))
            .aggregate(tournament_goals=Sum("total_goals"))
        )
        return {
            "results": [{"tournament_goals": result["tournament_goals"] or 0}],
        }

    if q in ("top scorers", "scorers", "top"):
        pipeline = [
            {"$unwind": "$goals"},
            {"$match": {"goals.own_goal": {"$ne": True}}},
            {"$group": {"_id": "$goals.name", "total_goals": {"$sum": 1}}},
            {"$sort": {"total_goals": -1}},
            {"$limit": 10},
        ]
        results = Match.objects.raw_aggregate(pipeline)
        return {
            "results": [{"scorer": r.pk, "goals": r.total_goals} for r in results],
        }

    if q.startswith("team "):
        team = query_text.strip()[5:].strip()
        from django.db.models import Q
        matches = Match.objects.filter(Q(team1__iexact=team) | Q(team2__iexact=team)).order_by("date")
        return {
            "query": f'db.world_cup_match.aggregate([{{$match: {{$or: [{{team1: {{$regex: "^{team}$", $options: "i"}}}}, {{team2: {{$regex: "^{team}$", $options: "i"}}}}]}}}}, {{$sort: {{date: 1}}}}])',
            "results": [
                {
                    "date": str(m.date),
                    "round": m.round,
                    "team1": m.team1,
                    "team2": m.team2,
                    "score": f"{m.ft_score1}-{m.ft_score2}" if m.ft_score1 is not None else "TBD",
                }
                for m in matches
            ],
        }

    if q.startswith("round "):
        round_name = query_text.strip()[6:].strip()
        matches = Match.objects.filter(round__icontains=round_name).order_by("date")
        return {
            "query": f'db.world_cup_match.aggregate([{{$match: {{round: {{$regex: "{round_name}", $options: "i"}}}}}}, {{$sort: {{date: 1}}}}])',
            "results": [
                {
                    "date": str(m.date),
                    "team1": m.team1,
                    "team2": m.team2,
                    "score": f"{m.ft_score1}-{m.ft_score2}" if m.ft_score1 is not None else "TBD",
                }
                for m in matches
            ],
        }

    if q in ("goals by round", "goals per round"):
        pipeline = [
            {"$match": {"ft_score1": {"$ne": None}}},
            {"$group": {
                "_id": "$round",
                "total_goals": {"$sum": {"$add": ["$ft_score1", "$ft_score2"]}},
                "matches": {"$sum": 1},
            }},
            {"$sort": {"total_goals": -1}},
        ]
        results = Match.objects.raw_aggregate(pipeline)
        return {
            "results": [{"round": r.pk, "goals": r.total_goals, "matches": r.matches} for r in results],
        }

    return None


def query_shell(request):
    """Interactive query shell — form POST triggers full page reload for MQL panel."""
    context: dict = {"queries": ALLOWED_QUERIES}

    if request.method == "POST":
        query_text = request.POST.get("query", "").strip()
        context["last_query"] = query_text

        if query_text:
            result = _execute_query(query_text)
            if result is None:
                context["error"] = f"Unknown query. Try: {', '.join(ALLOWED_QUERIES.keys())}"
            else:
                context["mongo_query"] = result.get("query", "")
                context["results"] = result.get("results", [])

    return render(request, "world_cup/shell.html", context)

Configure URLs for your views

Create a new file called urls.py file in your world_cup directory. To map the views defined in the preceding step to URLs, paste the following code into urls.py:

from django.urls import path

from . import views

urlpatterns = [
    path("goals/total/", views.tournament_goals, name="tournament_goals"),
    path("goals/top-scorers/", views.top_scorers, name="top_scorers"),
    path("shell/", views.query_shell, name="query_shell"),
]
`

Then, navigate to the config/urls.py file and replace its contents with the following code:
`
from django.conf import settings
from django.shortcuts import render
from django.urls import include, path


def landing(request):
    return render(request, "world_cup/landing.html")


urlpatterns = [
    path("", landing, name="landing"),
    path("world_cup/", include("world_cup.urls")),
]

if settings.DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

Create Templates

In your world_cup directory, create a subdirectory called templates.

Under templates, create five HTML files to display the base page, goals page(showing total goals by aggregation), landing page, scorers page(top scorers), and shell page(interactive shell to run queries and aggregation pipelines interactively).

base.html: This is the shared layout (nav, styles) that the other four templates extend.

#world_cup/templates/base.html
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{% block title %}World Cup 2026{% endblock %}</title>
    <style>
        :root {
            --nav-bg: #1C2D38;
            --page-bg: #f0f2f5;
            --card-bg: #ffffff;
            --text-primary: #1C2D38;
            --text-secondary: #5c6970;
            --border: #e2e5e9;
            --accent: #1a6b4b;
            --accent-light: #e8f5ef;
            --white: #ffffff;
        }

        * { margin: 0; padding: 0; box-sizing: border-box; }

        body {
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
            background: var(--page-bg);
            color: var(--text-primary);
            min-height: 100vh;
        }

        nav {
            background: var(--nav-bg);
            padding: 0.9rem 2rem;
            display: flex;
            align-items: center;
            justify-content: space-between;
            border-bottom: 3px solid var(--accent);
        }

        nav .brand {
            color: var(--white);
            font-size: 1.1rem;
            font-weight: 700;
            text-decoration: none;
        }

        nav .nav-links a {
            color: rgba(255,255,255,0.85);
            text-decoration: none;
            margin-left: 2rem;
            font-size: 0.9rem;
        }

        nav .nav-links a:hover {
            color: var(--white);
        }

        .container {
            max-width: 900px;
            margin: 2rem auto;
            padding: 0 1.5rem;
        }

        header {
            margin-bottom: 1.5rem;
        }

        header h1 {
            font-size: 1.6rem;
            font-weight: 700;
            color: var(--text-primary);
        }

        header p {
            color: var(--text-secondary);
            font-size: 0.9rem;
            margin-top: 0.25rem;
        }

        .card {
            background: var(--card-bg);
            border-radius: 8px;
            padding: 1.5rem;
            margin-bottom: 1rem;
            border: 1px solid var(--border);
        }

        .card:hover {
            border-color: var(--accent);
        }

        .card a {
            text-decoration: none;
            color: var(--text-primary);
            display: block;
        }

        .card h3 {
            font-size: 1rem;
            font-weight: 600;
            margin-bottom: 0.3rem;
        }

        .card p {
            color: var(--text-secondary);
            font-size: 0.85rem;
        }

        .badge {
            display: inline-block;
            background: var(--accent);
            color: var(--white);
            font-size: 0.7rem;
            font-weight: 600;
            padding: 0.2rem 0.6rem;
            border-radius: 4px;
            margin-bottom: 0.5rem;
            text-transform: uppercase;
        }

        table {
            width: 100%;
            border-collapse: collapse;
            margin-top: 0.5rem;
        }

        th, td {
            text-align: left;
            padding: 0.75rem 1rem;
            border-bottom: 1px solid var(--border);
        }

        th {
            color: var(--text-secondary);
            font-size: 0.75rem;
            font-weight: 600;
            text-transform: uppercase;
            letter-spacing: 0.03em;
        }

        td {
            font-size: 0.9rem;
        }

        .back-link {
            display: inline-block;
            margin-bottom: 1rem;
            color: var(--accent);
            text-decoration: none;
            font-size: 0.85rem;
        }

        .back-link:hover {
            text-decoration: underline;
        }

        .stat-value {
            font-size: 3rem;
            font-weight: 700;
            color: var(--accent);
        }

        .stat-label {
            color: var(--text-secondary);
            font-size: 0.9rem;
            margin-top: 0.25rem;
        }
    </style>
</head>
<body>
    <nav>
        <a href="/" class="brand">World Cup 2026 Stats</a>
        <div class="nav-links">
            <a href="{% url 'tournament_goals' %}">Goals</a>
            <a href="{% url 'top_scorers' %}">Scorers</a>
            <a href="{% url 'query_shell' %}">Shell</a>
        </div>
    </nav>
    <div class="container">
        {% block content %}{% endblock %}
    </div>
</body>
</html>

landing.html: This is the app's home page, linking out to the goals, scorers, and shell pages.
Endpoint: http://127.0.0.1:8000/

#world_cup/templates/landing.html
{% extends "world_cup/base.html" %}

{% block title %}World Cup 2026 — Home{% endblock %}

{% block content %}
<header>
    <h1>⚽ World Cup 2026 Stats</h1>
</header>

<div class="card">
    <a href="{% url 'tournament_goals' %}">
        <span class="badge">Aggregation</span>
        <h3>Total Goals</h3>
        <p>Sum of all goals scored — view the MQL query in the Debug Toolbar</p>
    </a>
</div>

<div class="card">
    <a href="{% url 'top_scorers' %}">
        <span class="badge">Pipeline</span>
        <h3>Top Scorers</h3>
        <p>$unwind + $group pipeline — view execution time in the MQL panel</p>
    </a>
</div>

<div class="card">
    <a href="{% url 'query_shell' %}">
        <span class="badge">Interactive</span>
        <h3>Query Shell</h3>
        <p>Run find queries and aggregation pipelines </p>
    </a>
</div>
{% endblock %}

goals.html: This displays the total goals count from the tournament_goals view's aggregation.
Endpoint: http://127.0.0.1:8000/world_cup/goals/total/

#world_cup/templates/goals.html
{% extends "world_cup/base.html" %}

{% block title %}Total Goals — World Cup 2026{% endblock %}

{% block content %}
<a href="/" class="back-link">← Back to Home</a>
<header>
    <h1>Tournament Goals</h1>
    <p>Total goals across all completed matches</p>
</header>

<div class="card" style="text-align: center; padding: 3rem;">
    <div class="stat-value">{{ total_goals }}</div>
    <div class="stat-label">goals scored</div>
</div>
{% endblock %}

scorers.html: This displays the top scorers leaderboard from the top_scorers view's raw pipeline.
Endpoint: http://127.0.0.1:8000/world_cup/goals/top-scorers/

#world_cup/templates/scorers.html
{% extends "world_cup/base.html" %}

{% block title %}Top Scorers — World Cup 2026{% endblock %}

{% block content %}
<a href="/" class="back-link">← Back to Home</a>
<header>
    <h1>Top Scorers</h1>
    <p>Leading goal scorers in the tournament</p>
</header>

<div class="card">
    <table>
        <thead>
            <tr>
                <th>#</th>
                <th>Player</th>
                <th>Goals</th>
            </tr>
        </thead>
        <tbody>
            {% for scorer in scorers %}
            <tr>
                <td>{{ forloop.counter }}</td>
                <td>{{ scorer.name }}</td>
                <td>{{ scorer.goals }}</td>
            </tr>
            {% empty %}
            <tr><td colspan="3">No scoring data available yet.</td></tr>
            {% endfor %}
        </tbody>
    </table>
</div>
{% endblock %}

shell.html: This is the interactive query shell which takes an input form plus a results table, for running any of the commands from ALLOWED_QUERIES and compares them in the MQL panel.

Endpoint: http://127.0.0.1:8000/world_cup/shell/

#/templates/shell.html
{% extends "world_cup/base.html" %}

{% block title %}Query Shell — World Cup 2026{% endblock %}

{% block content %}
<a href="/" class="back-link">← Back to Home</a>
<header>
    <h1>Query Shell</h1>
    <p>Each query reloads the page — check the MQL panel in the Debug Toolbar</p>
</header>

<div class="shell">
    <div class="shell-output" id="output">
        {% if not last_query %}
        <div class="shell-line system">Click an example or type a query:</div>
        <div class="shell-line system"></div>
        <div class="shell-line system" style="color: var(--mdb-gray); opacity: 0.6;">— Queries (find) —</div>
        <div class="shell-line example" data-query="team France">team France</div>
        <div class="shell-line example" data-query="team England">team England</div>
        <div class="shell-line example" data-query="round Quarter-final">round Quarter-final</div>
        <div class="shell-line system"></div>
        <div class="shell-line system" style="color: var(--mdb-gray); opacity: 0.6;">— Aggregation Pipelines —</div>
        <div class="shell-line example" data-query="top scorers">top scorers</div>
        <div class="shell-line example" data-query="goals by round">goals by round</div>
        <div class="shell-line example" data-query="goals">goals</div>
        {% endif %}

        {% if last_query %}
        <div class="shell-line query">{{ last_query }}</div>
        {% endif %}

        {% if error %}
        <div class="shell-line error">{{ error }}</div>
        {% endif %}

        {% if mongo_query %}
        <div class="shell-line mongo-query">{{ mongo_query }}</div>
        {% endif %}

        {% if results %}
        <div class="shell-line result">
            <table class="result-table">
                <thead>
                    <tr>
                        {% for key in results.0.keys %}
                        <th>{{ key }}</th>
                        {% endfor %}
                    </tr>
                </thead>
                <tbody>
                    {% for row in results %}
                    <tr>
                        {% for val in row.values %}
                        <td>{{ val|default:"—" }}</td>
                        {% endfor %}
                    </tr>
                    {% endfor %}
                </tbody>
            </table>
        </div>
        {% endif %}
    </div>

    <form method="post" action="{% url 'query_shell' %}" class="shell-input-row">
        {% csrf_token %}
        <span class="prompt">▶</span>
        <input type="text" name="query" id="query-input" value="" placeholder="e.g. team France, top scorers, round Quarter-final" autofocus>
    </form>
</div>

<style>
    .shell {
        background: var(--card-bg);
        border: 1px solid var(--border);
        border-radius: 8px;
        overflow: hidden;
        font-family: 'SF Mono', 'Fira Code', 'JetBrains Mono', monospace;
        font-size: 0.85rem;
    }

    .shell-output {
        padding: 1.5rem;
        max-height: 500px;
        overflow-y: auto;
    }

    .shell-line {
        margin-bottom: 0.4rem;
        line-height: 1.5;
    }

    .shell-line.system {
        color: var(--text-secondary);
    }

    .shell-line.query {
        color: var(--accent);
        font-weight: 600;
    }

    .shell-line.query::before {
        content: "▶ ";
        opacity: 0.5;
    }

    .shell-line.mongo-query {
        color: var(--text-secondary);
        font-size: 0.78rem;
        padding-left: 1rem;
        background: var(--page-bg);
        padding: 0.5rem 1rem;
        border-radius: 4px;
        margin: 0.3rem 0;
        overflow-x: auto;
    }

    .shell-line.result {
        color: var(--text-primary);
        padding-left: 0;
    }

    .shell-line.error {
        color: #c0392b;
    }

    .shell-line.example {
        color: var(--accent);
        cursor: pointer;
        padding-left: 1rem;
        opacity: 0.8;
    }

    .shell-line.example::before {
        content: "▶ ";
        opacity: 0.4;
    }

    .shell-line.example:hover {
        opacity: 1;
        text-decoration: underline;
    }

    .shell-input-row {
        display: flex;
        align-items: center;
        border-top: 1px solid var(--border);
        padding: 0.75rem 1.5rem;
        background: var(--page-bg);
    }

    .prompt {
        color: var(--accent);
        margin-right: 0.75rem;
        font-size: 0.8rem;
    }

    .shell-input-row input {
        flex: 1;
        background: transparent;
        border: none;
        outline: none;
        color: var(--text-primary);
        font-family: inherit;
        font-size: inherit;
    }

    .shell-input-row input::placeholder {
        color: var(--text-secondary);
        opacity: 0.5;
    }

    .result-table {
        width: 100%;
        border-collapse: collapse;
        margin: 0.5rem 0;
    }

    .result-table th {
        color: var(--text-secondary);
        font-size: 0.7rem;
        text-transform: uppercase;
        text-align: left;
        padding: 0.3rem 0.75rem;
        border-bottom: 1px solid var(--border);
    }

    .result-table td {
        padding: 0.3rem 0.75rem;
        font-size: 0.8rem;
        color: var(--text-primary);
    }
</style>

<script>
document.querySelectorAll('.shell-line.example').forEach(el => {
    el.addEventListener('click', () => {
        document.getElementById('query-input').value = el.dataset.query;
        el.closest('.shell').querySelector('form').submit();
    });
});
</script>
{% endblock %}

Build an Aggregation Pipeline

The tournament_goals view uses the Django ORM's aggregate() and Sum() to total up goals, and the MQL panel captures that query automatically. For more advanced analysis, like ranking top scorers, you can use MongoDB's aggregation pipelines. The top_scorers view in views.py does exactly this: it uses $concatArrays to merge embedded goal arrays, $unwind to flatten them into individual documents, and $group to count goals per player.

The top_scorers view uses raw_aggregate(), a method which accepts a raw MongoDB aggregation pipeline directly, including stages like $unwind, and returns Django model instances. See the QuerySet API reference for more details.

When you navigate to the /world_cup/goals/top-scorers/ and open the MQL panel in the Django Debug Toolbar. You'll see the full aggregation pipeline that was executed, along with its timing. The MQL panel captures both ORM-generated queries and raw pipelines, so you can inspect and optimize either approach from the same interface.

Verify the MQL panel

Start your development server by running the following command:
python manage.py runserver

Navigate to any page in your application that queries MongoDB. Open the Django Debug Toolbar and select the MQL panel to view the MongoDB queries executed during the request, including their execution times and explain plans.

View worldcup.goals query information

Now, navigate to http://127.0.0.1:8000/world_cup/goals/total/ to view MongoDB queries executed during the request.
The Django Debug Toolbar appears on the right side of the page. Select the MQL label to view the MongoDB query used to calculate the total goals, which resembles the following code:

After, you can explore the query information by clicking on the following buttons:
Select the + button to the left of the query to view the call stack, which resembles the following call trace:

Next, select the Query button to the top-right of the query, in the ACTION column, to view the pretty-printed MQL. This MQL resembles the following code:

Now, select the Explain button to the right of the query, in the ACTION column, to view the explain plan for the query. This plan resembles the following code:

View the top scorers aggregation pipeline

Next, navigate to http://127.0.0.1:8000/world_cup/goals/top-scorers/ to see the aggregation pipeline in the MQL panel. We see that this page runs a raw multi-stage pipeline ($concatArrays, $unwind, $group, $sort, $limit). Select the MQL panel to view the pipeline stages, execution time, and explain plans.

View query shell queries

Finally, access the endpoint http://127.0.0.1:8000/world_cup/shell/ and submit a query such as team France. The page reloads with the results, and the MQL panel shows the aggregation pipeline that was executed. Even though the view uses the Django ORM's filter() and order_by(), the MongoDB backend translates these into an aggregation pipeline.

Next Steps

Congratulations on getting this far! You now have a Django application that uses Django MongoDB Extensions to view and analyze MongoDB queries in the Django Debug Toolbar.

Here are some debugging exercises you can experiment with:

Run the "round" search in the query shell (e.g., round Group A) and check its query plan in the MQL panel. You'll see a COLLSCAN which is a full collection scan because round__icontains compiles to an unanchored regex, which can't use an index no matter what you add.
Compare it against round__iexact on the same field, and the panel shows IXSCAN instead. It is similar code but different queries underneath, and that is exactly what the Django Debug Toolbar makes evident.

To learn more about the Django MongoDB Extensions package, see django-mongodb-extensions and the django-mongodb-extensions repository on GitHub.
To learn how to query your data, see the Specify a Query guide.

Link to this GitHub Project: Django MongoDB World Cup project

Link to the Data: Open Football Dataset