DEV Community: MongoDB

Leveraging Skills in a MongoDB MCP Agent

MongoDB Guests — Fri, 16 Oct 2026 09:00:00 +0000

This article was written by Jack Woehr.

SoftWoehr LLC - Seiden Group - ProCern Technology Solutions.

Skills in a ReAct (Reason - Act) Agent for MongoDB

In this article, we describe our open-source mymongodbmcpagent implementing a ReAct (Reason - Act) agent leveraging the MongoDB MCP Server and the MongoDB agent-skills package. Our code also provides a chat interface so that you can talk to your MongoDB instance and offer queries whose answers require strategies derived from the skills package, leading to multiple exchanges between the agent and the LLM.

Evolving from a one-pass MongoDB Server Agent

If you read my previous article, A Simple MongoDB Server Agent, you saw an agent/chat pair using an LLM provider to answer one-pass queries to MongoDB via the MongoDB MCP Server. I used that architecture to add MongoDB capabilities to a multi-agent dashboard that connects to many different projects I work on.

Reuse the scripting from the earlier article

Keep a link to that earlier article because its associated code on GitHub includes a nice script and .env.example to start the MongoDB MCP server, which can be used with the code we're presenting in this article.

Prerequisite

As in the earlier article, we assume you have a MongoDB Atlas Cluster with some sample data to query.

Getting Started

A full Getting Started document is available in the repository to help you set up.

Add Skills to MCP

MCP tools are … tools! An LLM has no arms and legs. All it can do is talk, but once informed that there are tools (provided by the MCP server) and the skills to use them, it can call for the tools to be used based on its decisions.
Skills are instructions for using tools to accomplish tasks that require more than one tool call. They're in text, in Markdown format, and are prompts, plain English instructions for the language model to follow. Take a look at the skills directory of the repo just mentioned.
The Agent mediates between
- the User
- the Inference Engine (the LLM)
- the tools and skills
  - loading the tools list for the LLM
  - loading the skills list for the LLM
  - calling the tools on behalf of the LLM

Skills add complexity to the Agent

Using skills requires the agent to execute a ReAct (Reasoning + Acting) loop that supports multiple LLM tool calls until it arrives at a response.

The enhanced ReAct loop agent

The code for the enhanced agent and its associated chat REPL is found at mymongodbmcpagent. Pay close attention to the .env.example file and create a .env file so that the agent and chat will both load and function as expected.

The diagram below shows the flow of the agent.

The payoff: more sophisticated queries

In the screenshot below, with the agent (and, of course, the MongoDB MCP server) running, I have asked "find me pottery mugs".

Answering this query requires that the LLM use its skills to exercise the ReAct loop to:

list the databases in the Atlas installation we've pointed it to
list collections in those databases looking for something useful ("Ha! There's a collection called pottery in the herwheel3_test database!")
work out how to find "mugs" in that collection ("Hmm, there are arrays of categories and tags in these documents …")

As the Agent itself prints some of its processing to standard out, we can see that it took 6 iterations:

[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 1...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Tool selected: list-databases
[req=72b91571-d634-4597-a54a-6b992939335a] Session may have expired, attempting to reconnect...
[req=72b91571-d634-4597-a54a-6b992939335a] ✓ Successfully initialized session with MongoDB MCP server
[req=72b91571-d634-4597-a54a-6b992939335a] Reconnected, retrying tool call...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 2...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Tool selected: list-collections
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 3...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Tool selected: find
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 4...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Tool selected: collection-schema
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 5...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Tool selected: find
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Iteration 6...
[req=72b91571-d634-4597-a54a-6b992939335a] [ReAct] Final answer reached.

The secret sauce in the `agent-server.js` code

The prompt reveals how the LLM interacts with the agent to make multiple tool calls and then signal when it has arrived at a final response:

  let currentPrompt = `You are an expert MongoDB autonomous agent.
Your goal is to answer the user's question by taking actions and analyzing their results.


Here is a list of available tools:
${formattedTools}


${loadedSkills ? `You have the following skills available to guide your actions:\n${loadedSkills}\n` : ''}
User's request: "${question}"


Decide what to do next based on the request.
- If you need to use a tool to gather more information or execute a step, return a JSON object with "action": "tool", "toolName": "<name>", and "toolArguments": {<args>}.
- If you have gathered enough information and can answer the user's request (or if no tools are needed), return a JSON object with "action": "answer", and "finalAnswer": "<your natural language answer>".


Your response MUST be exactly ONE valid JSON object and nothing else. Do not use markdown blocks like \`\`\`json.`;

json
You'll see in the source code that agent-server.js checks the incoming JSON on each pass through the LLM's responses, where it executes tool calls and watches for that finalAnswer indicator. Naturally, there is a configurable '.env ' limit to the number of iterations so you don't spend your rent money on a wedged inference.

Agentic code assistance

This project is extracted from a component of a larger project (my Dashboard, not yet released) and was coded over several months in spare time using the following agentic code assistants (listed in alphabetical order):

Variety may well be the spice of life, but it's also an essential oil for lubricating one's experience with agentic code assistance as this field rapidly evolves.

Summary

MongoDB MCP server is not just a plugin to code assist; it's an application building block.
MongoDB MCP server is best leveraged using skills.
Skills require the agent to present a ReAct (Reasoning + Acting) architecture.
This article is accompanied by code.
- The code supports several popular LLM providers.
- The code is capable of servicing multiple concurrent requests.

What agent memory actually asks of a database

Pash10g — Mon, 12 Oct 2026 13:08:39 +0000

Most agent-memory posts are about the memory algorithm. This one is about the storage layer underneath it, because that's where I've watched these systems actually fall over.

The pattern is familiar by now. An agent accumulates conversation history, you can't replay all of it into every call, so you compress it into some smaller representation and retrieve the relevant parts on demand. Plenty of libraries do the compression well. What they mostly hand-wave is what happens next: the compressed state has to live somewhere, and that somewhere has to survive a crash, resist tampering, serve hybrid queries fast, and delete on request in a way you can prove to an auditor.

Those four requirements turn out to be a fairly demanding database spec. I spent a couple of days testing them against MongoDB using Semvec, a memory engine that keeps a fixed-size state vector plus bounded memory tiers, with an Atlas persistence layer on top. Everything below is measured on a laptop against mongodb/mongodb-atlas-local 8.3.8, and you can reproduce it.

Requirement 1: The write has to be atomic across three shapes of data

Every turn produces three writes that have to agree with each other:

What	Collection	Role
The turn itself, sequence-numbered	`compliance_events`	source of truth
A serialized engine state blob	`state_snapshots`	recovery fast path
Extracted facts plus their embeddings	`memory_items`	what queries hit

If those three drift apart, the memory is corrupt in ways that are hard to detect. An event logged without its items means a fact silently vanishes from retrieval. A snapshot written from a state that doesn't match the log means recovery restores the wrong thing.

This requirement quietly rules out most of the standard stack. If your log is in Postgres and your vectors are in a dedicated vector store, there is no transaction boundary around those three writes, so you reach for an outbox table or a saga and accept a window of inconsistency. In MongoDB, there are three collections in one database, and the commit is one multi-document transaction:

committer = TurnCommitter(client, db, scope, embedder=embedder, snapshot_every=10)
committer.run_turn(state, session="s1", turn=1, text="customer prefers index funds")

turn 04 committed | event_seq=4 | items +1 | dedup.is_update=False | snapshot ✓

Measured (on a laptop) cost of that atomicity: p50 21.7 ms, p95 47.2 ms. For context, generating the embedding for the same turn took p50 81.8 ms on CPU. The transaction is not what you'll be optimizing.

event_seq is monotonic per tenant and agent, so gaps are detectable rather than silent. That matters for the next requirement.

Requirement 2: You have to be able to rebuild state and prove it's the same state

"Durable" usually means "we back it up." For agent memory that's too weak. If you restore a snapshot and the state is subtly different from what the agent had before, the agent's behavior changes, and nothing tells you.

Because every turn is in an append-only log, recovery can replay the log from scratch and compare:

recovered: replayed_events=10  (= the 10 memory turns of the event log;mode=full replay, log is the source of truth)
engine checksum (recovered): a8d9d3b4ffa8adf3… == live ✓

Byte-identical, not approximately equal. That's what lets you treat the snapshots collection as a disposable cache: if a snapshot is missing or suspect, throw it away and replay. The same property held across the LOCOMO benchmark, where checksums matched on 10 of 10 conversations and the run is set to abort on the first mismatch.

Two MongoDB details make this practical rather than theoretical. Snapshots are compare-and-swap versioned, so two processes racing to checkpoint the same session can't silently overwrite each other. And integrity is checked twice on read, once against a SHA-256 of the blob and once against the engine's own embedded checksum, which distinguishes storage corruption from a version mismatch.

Requirement 3: Retrieval has to be hybrid, and it should happen server-side

Semantic search alone doesn't work for this workload. Ask an agent, "What's my policy number?" and vector similarity gives you insurance-adjacent chunks. You need the lexical match too, plus a recency bias, because in a conversation, the newer statement usually supersedes the older one.

MongoDB 8.1 added $rankFusion, which does that fusion inside a single aggregation:

retrieval path: $rankFusion (server-side, MongoDB 8.3)

One round trip, p50 15.2 ms, with per-tenant weights over the semantic, lexical, and recency rankings. The alternative is three queries and a reciprocal-rank-fusion loop in your application, which is more code, more latency, and a reranking implementation you now own.

Worth checking your server version before you count on it. My cloud test cluster was on 8.0.30, which predates $rankFusion; the library detects this and falls back to client-side fusion, so it keeps working. Nothing raises: the fallback is logged at INFO level, RetrievalEngine.uses_rank_fusion turns false, and python -m semvec_atlas.deployment <uri> reports $rankFusion: no. Check one of those rather than assuming you are on the server-side path.

In the benchmark, hybrid retrieval through Atlas scored 0.5773 on LLM-judge accuracy, compared to 0.5322 for the same engine retrieving in-process with no database. LLM-judge accuracy here means an LLM grades each answer against ground truth and you take the pass rate using the mem0 evaluation prompts verbatim. Persisting to a database and querying it did not cost accuracy; it gained about 4.5 points, because fused ranking beats the in-process scorer. Retrieval latency went from 165 ms to 187 ms at p50 to get that.

Reader context also stayed flat while this happened, which I'll come back to at the end. Fair warning on the benchmark: those accuracy figures aren't comparable to published mem0 or Zep numbers, because this run doesn't replicate their prompt routing or entity extraction. The comparison that's valid is Atlas-backed against in-process under identical conditions.

Requirement 4: Deletion has to be provable, and it reaches further than you think

This is the requirement I'd underestimated, and it's the one I'd now lead with when talking to anyone in a regulated industry.

A GDPR Article 17 request means a user's data must be gone, and "gone" has to be demonstrable. In this implementation, the erasure produces a signed certificate:

certificate: user=<tenant>/<agent> events_wiped=16
signature (RS256): jRKdVEuIbnDfuT5KZwMSE6Hy07RzASc0…
signature verifies: ✓
memory_items left: 0 | snapshots left: 0
quarantine vault left: 0 | learned anchors left: 0
retrieval after forget: 0 hits (expected 0)

Note the two lines before the retrieval check, because they're the interesting failure mode. This system has an ingest filter that blocks suspicious text before it reaches memory, and blocked text is held in a quarantine collection for human review. It also derives embeddings from blocked text to recognize future variants.

Both of those contain user data. The quarantine vault holds the full original text. The learned anchors are vectors derived from it. An implementation that deletes memory_items and calls the job done has left personal data sitting in its own safety subsystem, which, under Article 17, is not a technicality. Getting this right requires knowing every derived copy, and it's much easier to enumerate those when they're collections in one database rather than artifacts spread across four services.

A word on what the signature does and doesn't prove. The RS256 key is yours: the operator supplies the private key, and the certificate is signed after every wipe step has completed and been counted. So the signature proves the deletion ran to completion under your key and hasn't been altered since. It is not third-party attestation; if you need an external party to vouch for it, you hand them the certificate and your public key. That's the same trust model as any self-signed audit record, and it's worth being upfront about before an auditor asks.

Same argument in general form: the moment your source of truth and your vector index live in separate systems, "delete this user" becomes a distributed systems problem with no transaction boundary and no single place to verify the result.

The security layer, briefly

The ingest filter deserves a note because it's the part I expected to be hand-wavy, but it wasn't.

The threat is specific to persistent memory. A one-shot prompt injection is a bad answer, and then it's over. An injection that gets written into memory is retrieved and re-injected on every future turn, and it will be retrieved precisely because it's semantically relevant to whatever the attacker aimed it at.

The filter is cosine similarity against a handful of anchor sentences describing what must never enter memory, with a 0.45 threshold. A direct injection attempt scored 0.696 and was blocked. A leetspeak version of the same attack (1gn0r3 4ll pr3v10u5 1n57ruc710n5) scored 0.298 and passed, because the embedder sees gibberish tokens. So there's a de-obfuscation pass that normalizes character substitutions and homoglyphs and re-checks: 0.633, blocked. I appreciated that the demo shows its first line of defense failing rather than only the version that works.

The part with operational teeth: blocked text is quarantined, not dropped, and the review queue shows only a hash, a 16-character preview, and the score, so an operator can triage without reading customer data. Releases require a reason. Legitimate text that trips the filter (one test case scored 0.632, phrased like an injection but entirely benign) can be released and recommitted. Cosine similarity can't read intent, so false positives are a certainty, and the workflow assumes them.

Every block also teaches. The embedding of a blocked attempt is stored as a learned anchor, and those anchors live in a collection shared across the whole tenant rather than per agent. In the demo, the attack is blocked at advisor-01. A reworded variant is then sent to advisor-02, an agent with no seed catalog of its own and no history with this attacker, and it's blocked at 0.938 against the anchor advisor-01 learned. One agent gets hit, the fleet is immune. Releasing a false positive un-learns its anchor, so the operator workflow above doesn't poison the shared list. I haven't seen another memory layer demonstrate this, and it's only trivial because the anchors are documents in the same database every agent already reads: no sync job, no cache invalidation, no second system.

To be clear about scope: this catches the obvious class of attack, the kind that reads like an instruction to the model. It is not a defense against an adversarially optimized payload. An injection phrased to sound like a benign customer statement won't resemble any anchor and will pass, for the same reason the leetspeak variant passed before de-obfuscation. It raises the cost of the lazy attack and leaves an audit trail, and that's what it claims to do.

Blocked attempts leave the engine state bit-identical to a run where the attack never happened, while still writing content-free audit markers recording that something was blocked. Audit trail without storing the payload.

Reproducing it

git clone [https://github.com/VersinoPsiOmega/semvec-mongodb-atlas.git](https://github.com/VersinoPsiOmega/semvec-mongodb-atlas.git)
cd semvec-mongodb-atlas
docker run -d --name semvec-atlas-mongo -p 27017:27017 \
    --ulimit nofile=64000:64000 mongodb/mongodb-atlas-local:8.3.8

python3 -m venv ~/.venvs/semvec-atlas
~/.venvs/semvec-atlas/bin/pip install -e ".[langgraph,compliance,mcp,dev]" \
    sentence-transformers openai tiktoken

demo/run.sh --no-llm

The --ulimit flag is not decoration. Each test provisions its own database, and a few hundred of those exhausted the default file-descriptor limit, at which point WiredTiger takes a fatal assertion and mongod dies mid-run. What you see is a wall of OSError: connection closed failures that look like application bugs. With the limit raised, the suite goes from a 25-minute partial collapse to 41 seconds.

Everything except the answer text is deterministic: checksums, sequence numbers and counts reproduce exactly; the latencies depend on your hardware.

Why you'd want this

If you're building agents, you've already hit the wall, or you're about to: conversation history grows, context costs grow with it, and every turn gets more expensive than the last. The usual fix is to compress. The part nobody tells you is that compression only pays off if the compressed state is somewhere you can trust.

Here's what that looked like in practice. Across 5,882 turns of real conversation, the context sent to the model per call stayed at 987 tokens. That held on the 689-turn conversation just like the 369-turn one. Replaying the full history would have cost 27,353. So: 96% less context, and it doesn't drift upward as the conversation gets longer. Turn 700 costs what turn 7 did.

And it didn't cost quality. LLM-judge accuracy was 0.5773 going through Atlas against 0.5322 for the same engine holding everything in memory with no database at all. Persisting to a database didn't tax accuracy, it improved it, because fused ranking beats an in-process scorer.

Fixed cost per turn, and accuracy that holds. That's what makes an agent affordable to run at turn 1,000 rather than just impressive in a demo.

What MongoDB is doing for this

Fixed-size memory isn't a new idea. What's different here is not having to assemble the storage underneath it from four separate services.

Everything in this post ran on one database:

The turn log, the state snapshot, and the vectors commit together in one transaction, 21.7 ms at p50. No outbox, no saga, no window where a fact exists in your log but not your index.
Hybrid ranking over semantic, lexical, and recency signals happens server-side in one aggregation via $rankFusion, 15.2 ms at p50. Not three queries and a fusion loop you have to write and own.
"Delete this user" is one operation you can verify, with a signed certificate, because there is one place the data lives.

Try that on Postgres plus a dedicated vector store, and every one of those becomes distributed-systems work: consistency code you maintain, reranking you implement, and a GDPR request that turns into an archaeology project across four services. The alternative to one database isn't four databases, it's four databases plus the glue between them and the compliance gaps in the seams.

The whole thing runs on your laptop today: mongodb/mongodb-atlas-local, one docker run, same code path as Atlas in production.

One last data point for anyone about to optimize the wrong layer. The embedder was 4x the cost of the database write and 5x the cost of retrieval. The database was never the bottleneck.

MongoDB as a Vector Database for AI Agents

MongoDB Guests — Fri, 02 Oct 2026 09:00:00 +0000

This article was written by Néstor Daza.

Introduction

MongoDB is a general-purpose document database with several modules embedded in its core that turn it into a complete data platform for modern application development. One of them is native Vector Search, allowing MongoDB to manage your data while also powering semantic retrieval and agent memory for your application. You don't need a separate vector store.

In this tutorial, you will build a Recipe Assistant Agent called Chez Robot using MongoDB, VoyageAI, Claude, and LangChain. The agent will answer natural language questions like "what can I make with rice and saffron?" by performing a semantic search over a collection of recipes fetched from TheMealDB, a free, public recipe API that requires no key. It will also remember your personal preferences across sessions, storing and retrieving that memory from MongoDB using the same vector index.

By the end, you will understand how MongoDB handles embeddings and Vector Search, how LangChain wires together retrieval and memory, and how Voyage AI models fit into the pipeline.

What you will need

Before you begin, make sure you have the following:

Python 3.10 or higher. You can verify with python --version. If you need to install it, follow the official Python installation guide.
A MongoDB Atlas account where you will set up a free M0 cluster later. You can create the account at mongodb.com/cloud/atlas (no credit card required for M0). It is also possible to use a local Instance of MongoDB Community Edition.
mongosh (MongoDB Shell). You will use it to create the vector search indexes and query the database. Follow the official installation guide to set it up.
A Voyage AI API key. Sign up at voyageai.com and generate a key from your dashboard. Voyage AI offers a free tier generous enough to complete this tutorial.
An Anthropic API key for Claude, which will drive the agent's reasoning. You can create one at console.anthropic.com.
Basic familiarity with what embeddings and Vector Search are, what a language model agent does, and how MongoDB stores documents. If any of those feel shaky, the MongoDB Vector Search docs and Anthropic's intro to Claude are good starting points.

Understanding the Architecture

Before writing any code, let's cover some basic concepts about Vector Search, the role MongoDB plays, and why it makes sense to use MongoDB as the data backbone of your AI Agentic applications.

A MongoDB document stores your application data: fields, arrays, nested objects, etc. A vector-enabled document stores all of that, plus an additional field containing a high-dimensional float array: the embedding. When you run a Vector Search query, MongoDB compares your query embedding against the stored embeddings of all your documents in a collection; the result is a ranked list of documents ordered by semantic similarity, effectively finding elements that have a meaningful relationship with the original query.

Vector Search is the backbone of two capabilities that make AI agents genuinely useful in production: knowledge retrieval and persistent memory.

RAG (Retrieval-Augmented Generation): The agent retrieves relevant information from external sources before generating an answer, grounding its output in real data rather than the model's training knowledge alone.
Persistent memory: Between sessions, the agent stores what it learns about your preferences and previous interactions, and retrieves the most contextually relevant memories at the start of each new conversation.

Putting everything together, we can build out the Chez Robot app with a full stack that looks like this:

TheMealDB -> VoyageAI embeddings -> MongoDB Atlas (vector index)
                                                  |
                                        LangChain retriever
                                                  |
                                       Claude (via LangChain)
                                                  |
                                               Your app

All data concerns in an AI agent, knowledge retrieval, persistent memory, and conversation history, can be stored in MongoDB. One connection string, one query language, one place to manage your data. That is what makes MongoDB a natural fit for agentic applications: it handles both structured and semantic workloads without splitting your infrastructure or adding complexity to your software.

Step 1 — Setting Up Your Environment

Create a project directory and a virtual environment to keep dependencies isolated:

mkdir chez-robot-agent
cd chez-robot-agent
python -m venv .venv
source .venv/bin/activate   # On Windows: .venv\Scripts\activate

Install the required packages:

pip install langchain langchain-anthropic langchain-mongodb \
            langchain-voyageai pymongo requests python-dotenv

Here is what each package does:

langchain provides the agent framework and the retrieval abstractions.
langchain-anthropic is the LangChain integration for Claude.
langchain-mongodb provides MongoDBAtlasVectorSearch, which stores and queries embeddings in MongoDB directly from LangChain.
langchain-voyageai wraps the Voyage AI embedding API in a LangChain-compatible interface.
pymongo is the underlying MongoDB driver.
requests is used to fetch recipes from TheMealDB.
python-dotenv loads your secrets from a .env file.

Add a .gitignore so you never accidentally commit secrets:

.env
.venv/
__pycache__/

Step 2 — Setting Up MongoDB Atlas

Head to MongoDB Atlas and log in to your account.

To create your cluster, go to Clusters, click Create, and select the M0 free tier. Choose your preferred cloud provider and region, in Cluster Details, name your cluster chez-robot, and click Create Cluster. It'll take a couple of minutes to get everything up and running.

Now you need a database user. In the Security section of the left menu, go to Database & Network Access and click Add New Database User. Choose Password as the authentication method, set a username and a strong password, and assign the Read and Write to any database built-in role. Click Add User.

Finally, whitelist your IP address to allow secure access to your cluster. In Network Access, select IP Access List and click Add IP Address. Click Add Current IP Address to whitelist your machine, then click Confirm. You might already have 0.0.0.0/0 configured, which allows access from anywhere; this is useful if you are working from a dynamic IP or want to keep things simple for a local tutorial, but not recommended for production deployments.

Create a .env file in the project root:

MONGODB_URI=<your-atlas-connection-string>
VOYAGE_API_KEY=<your-voyage-api-key>
ANTHROPIC_API_KEY=<your-anthropic-api-key>

You can find your database connection string in the Atlas dashboard under Cluster > Connect > Drivers. Make sure to replace <db_username> and <db_password> in the URI with the credentials you just created.

Step 3 — Creating the Vector Search Indexes

With the cluster ready, you can create both vector search indexes before ingesting any data. You will create the collections using db.createCollection(), then attach the vector search indexes to them. This way, everything is set up before any data arrives.

Create a file called scripts/create_indexes.js in your project root (this is a JavaScript file executed directly by mongosh, no additional dependencies needed):

// create_indexes.js

const DB_NAME = "chez_robot";

db = db.getSiblingDB(DB_NAME);

// Create collections explicitly before creating vector search indexes
db.createCollection("recipes");
db.createCollection("memories");

// Vector search index for recipes
db.recipes.createSearchIndex({
  name: "recipe_vector_index",
  type: "vectorSearch",
  definition: {
    fields: [
      {
        type: "vector",
        path: "embedding",
        numDimensions: 1024,
        similarity: "cosine"
      },
      {
        type: "filter",
        path: "tags"
      }
    ]
  }
});

// Vector search index for agent memory
db.memories.createSearchIndex({
  name: "memory_vector_index",
  type: "vectorSearch",
  definition: {
    fields: [
      {
        type: "vector",
        path: "embedding",
        numDimensions: 1024,
        similarity: "cosine"
      }
    ]
  }
});

print("Index creation finished for recipes and memories collections.");

A few things worth noting here. The numDimensions value of 1024 matches the default output size of voyage-4, the VoyageAI model you will use to generate embeddings. If you ever switch to a different VoyageAI model, verify its output dimensions in the VoyageAI model documentation and update this value accordingly.

The cosine similarity metric measures the angle between vectors rather than their absolute distance, which works well for semantic similarity tasks where the direction of the embedding matters more than its magnitude.

The filter field on tags in the recipes index allows you to pre-filter results by dietary category before the vector search runs, which you will use later. For a complete reference on vector search index definitions, check the official documentation.

Make sure you have mongosh installed; then export your connection string as an environment variable (the same URI you added to your .env file in Step 2) and run the index creation script.

export MONGODB_URI=<your-atlas-connection-string>
mongosh "$MONGODB_URI" --file scripts/create_indexes.js

Step 4 — Ingesting Recipes from TheMealDB

We're ready to fetch recipes and store them as vector-enabled documents in your cluster. TheMealDB's API allows you to retrieve recipes by their first letter; you will loop through the alphabet, load each recipe, generate embeddings for it, and store them as vector-enabled documents in MongoDB.

Create a file called ingest.py:

import os
import requests
from pymongo import MongoClient
from langchain_voyageai import VoyageAIEmbeddings
from dotenv import load_dotenv

load_dotenv()

# --- Clients ---
mongoDBClient = MongoClient(os.environ["MONGODB_URI"], appname="devrel-tutorial-vectorsearch-agents")
db = mongoDBClient["chez_robot"]
collection = db["recipes"]

voyageAIClient = VoyageAIEmbeddings(
    voyage_api_key=os.environ["VOYAGE_API_KEY"],
    model="voyage-4",
)

# --- Fetch recipes from TheMealDB ---
def fetch_recipes_by_letter(letter: str) -> list[dict]:
    url = f"https://www.themealdb.com/api/json/v1/1/search.php?f={letter}"
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    meals = response.json().get("meals") or []
    return meals

def build_recipe_text(meal: dict) -> str:
    """Build a single string representation of a recipe for embedding."""
    ingredients = []
    for i in range(1, 21):
        ingredient = (meal.get(f"strIngredient{i}") or "").strip()
        measure = (meal.get(f"strMeasure{i}") or "").strip()
        if ingredient:
            ingredients.append(f"{measure} {ingredient}".strip())

    return (
        f"Recipe: {meal['strMeal']}\n"
        f"Category: {meal.get('strCategory', '')}\n"
        f"Cuisine: {meal.get('strArea', '')}\n"
        f"Ingredients: {', '.join(ingredients)}\n"
        f"Instructions: {meal.get('strInstructions', '')[:500]}"
    )

def build_tags(meal: dict) -> list[str]:
    tags = []
    if meal.get("strCategory"):
        tags.append(meal["strCategory"].lower())
    if meal.get("strArea"):
        tags.append(meal["strArea"].lower())
    raw_tags = meal.get("strTags") or ""
    tags += [t.strip().lower() for t in raw_tags.split(",") if t.strip()]
    return list(set(tags))

# --- Ingest ---
def ingest(letters: str = "abcdefghijklmnopqrstuvwxyz"):
    print("Fetching recipes from TheMealDB...")
    all_meals = []
    for letter in letters:
        meals = fetch_recipes_by_letter(letter)
        all_meals.extend(meals)
        print(f"  {letter}: {len(meals)} recipes")

    print(f"\nTotal recipes fetched: {len(all_meals)}")
    print("Generating embeddings with Voyage AI...")

    texts = [build_recipe_text(meal) for meal in all_meals]

    # Voyage AI supports batching - send up to 128 texts at a time
    batch_size = 64
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        batch_embeddings = voyageAIClient.embed_documents(batch)
        all_embeddings.extend(batch_embeddings)
        print(f"  Embedded {min(i + batch_size, len(texts))}/{len(texts)}")

    print("Writing documents to MongoDB ...")
    documents = []
    for meal, text, embedding in zip(all_meals, texts, all_embeddings):
        documents.append({
            "meal_id": meal["idMeal"],
            "name": meal["strMeal"],
            "category": meal.get("strCategory"),
            "cuisine": meal.get("strArea"),
            "tags": build_tags(meal),
            "source_text": text,
            "thumbnail": meal.get("strMealThumb"),
            "embedding": embedding,
        })

    # Use replace to avoid duplicates if you re-run ingest
    for doc in documents:
        collection.replace_one(
            {"meal_id": doc["meal_id"]},
            doc,
            upsert=True,
        )

    print(f"Done. {len(documents)} recipes stored in MongoDB.")

if __name__ == "__main__":
    ingest()

Run the ingestion script:

python ingest.py

You will see output similar to:

Fetching recipes from TheMealDB...
  a: 9 recipes
  b: 14 recipes
  ...
Total recipes fetched: 303
Generating embeddings with Voyage AI...
  Embedded 64/303
  Embedded 128/303
  ...
Done. 303 recipes stored in MongoDB.

Each document in your recipes collection now has an embedding field alongside the recipe's structured data. The embedding is an array of 1024 (the dimensions) floating-point numbers, generated by passing the full recipe text (name, category, cuisine, ingredients, and a truncated version of the instructions) through voyage-4. It captures the meaning of the recipe, not just its keywords, which is what allows a query like "light Japanese noodle dish" to match a recipe for Miso Ramen even if those exact words don't appear together in the document.

Step 5 — Building the Agent

We're ready to build our agent. Create a file called agent.py:

import os
import warnings
from datetime import datetime, UTC
from pymongo import MongoClient
from langchain_anthropic import ChatAnthropic
from langchain_mongodb import MongoDBAtlasVectorSearch
from langchain_voyageai import VoyageAIEmbeddings
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.prebuilt import create_react_agent
from dotenv import load_dotenv

warnings.filterwarnings("ignore", category=DeprecationWarning)

load_dotenv()

# --- Clients ---
mongoDBClient = MongoClient(os.environ["MONGODB_URI"], appname="devrel-tutorial-vectorsearch-agents")
db = mongoDBClient["chez_robot"]

voyageAIClient = VoyageAIEmbeddings(
    voyage_api_key=os.environ["VOYAGE_API_KEY"],
    model="voyage-4",
)

llm = ChatAnthropic(
    model="claude-haiku-4-5",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
    temperature=0,
)

# --- Vector stores ---
recipe_store = MongoDBAtlasVectorSearch(
    collection=db["recipes"],
    embedding=voyageAIClient,
    index_name="recipe_vector_index",
    text_key="source_text",
    embedding_key="embedding",
)

memory_store = MongoDBAtlasVectorSearch(
    collection=db["memories"],
    embedding=voyageAIClient,
    index_name="memory_vector_index",
    text_key="content",
    embedding_key="embedding",
)


# --- Tools ---

@tool
def search_recipes(query: str, dietary_filter: str = "") -> str:
    """
    Search for recipes by ingredient, cuisine, dish type, or cooking style.
    Optionally filter by dietary category (e.g. 'vegetarian', 'seafood', 'chicken').
    Returns the top 5 most relevant recipes.
    """
    search_kwargs = {"k": 5}

    if dietary_filter:
        search_kwargs["pre_filter"] = {
            "tags": {"$in": [dietary_filter.lower()]}
        }

    results = recipe_store.similarity_search(query, **search_kwargs)

    if not results:
        return "No recipes found matching that query."

    output = []
    for i, doc in enumerate(results, 1):
        output.append(f"**Recipe {i}:** {doc.metadata.get('name', 'Unknown')}\n{doc.page_content[:400]}\n")

    return "\n".join(output)


@tool
def save_preference(preference: str) -> str:
    """
    Save a dietary preference or food restriction that the user has mentioned.
    Examples: 'user is vegetarian', 'user dislikes cilantro', 'user is allergic to shellfish'.
    """
    doc = {
        "content": preference,
        "created_at": datetime.now(UTC).isoformat(),
        "type": "preference",
    }
    memory_store.add_texts(
        texts=[preference],
        metadatas=[doc],
    )
    return f"Got it, I've saved that preference: '{preference}'"


@tool
def recall_preferences(context: str) -> str:
    """
    Recall stored user preferences relevant to the current cooking context.
    Use this at the start of a session or when the user asks for personalized recommendations.
    """
    results = memory_store.similarity_search(context, k=5)
    if not results:
        return "No preferences stored yet."

    preferences = [doc.page_content for doc in results]
    return "Remembered preferences:\n" + "\n".join(f"- {p}" for p in preferences)


# --- Load session memory from MongoDB ---
def load_session_memory(session_id: str, k: int = 10) -> list:
    """Retrieve the most recent conversation turns for this session."""
    turns = list(
        db["sessions"].find(
            {"session_id": session_id},
            sort=[("timestamp", -1)],
            limit=k,
        )
    )
    turns.reverse()

    history = []
    for turn in turns:
        if turn["role"] == "human":
            history.append(HumanMessage(content=turn["content"]))
        else:
            history.append(AIMessage(content=turn["content"]))
    return history


def save_turn(session_id: str, role: str, content: str):
    """Persist a single conversation turn to MongoDB."""
    db["sessions"].insert_one({
        "session_id": session_id,
        "role": role,
        "content": content,
        "timestamp": datetime.now(UTC),
    })


# --- Agent setup ---
tools = [search_recipes, save_preference, recall_preferences]

system_prompt = """You are a knowledgeable and enthusiastic recipe assistant. You help users 
find recipes based on ingredients they have, cuisines they love, or dietary needs they follow.

At the start of each conversation, use the recall_preferences tool to check if the user 
has any stored dietary preferences, allergies, or food restrictions, and factor those 
into every recommendation you make.

When a user mentions a preference, restriction, or allergy, always save it using the 
save_preference tool so you can remember it next time.

When recommending recipes, be specific: mention the cuisine, key ingredients, and why 
it matches what the user is looking for. Keep your tone warm and conversational."""

agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt=system_prompt,
)


# --- Main loop ---
def run(session_id: str = "default"):
    print(f"\nChez Robot ready (session: {session_id})")
    print("Ask me anything about recipes. Type 'quit' or 'exit' to leave.\n")

    chat_history = load_session_memory(session_id)

    if chat_history:
        print(f"  (Loaded {len(chat_history)} messages from your last session)\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit", "q"):
            print("Goodbye! Your preferences have been saved for next time.")
            break
        if not user_input:
            continue

        result = agent.invoke({
            "messages": chat_history + [HumanMessage(content=user_input)]
        })

        response = result["messages"][-1].content
        print(f"\nChez Robot: {response}\n")

        chat_history.append(HumanMessage(content=user_input))
        chat_history.append(AIMessage(content=response))

        save_turn(session_id, "human", user_input)
        save_turn(session_id, "ai", response)


if __name__ == "__main__":
    import sys
    session = sys.argv[1] if len(sys.argv) > 1 else "default"
    run(session_id=session)

There are a few things here worth unpacking.

The agent has separate tools for searching recipes, saving preferences, and recalling preferences. This gives the LLM clear, distinct responsibilities to reason about rather than a single overloaded function. The agent uses each tool's function signature and docstring to decide when to invoke it; writing descriptive docstrings is a good practice and a direct input to the agent's reasoning.

Recipe retrieval and preference management are deterministic tasks; you want the agent to reliably call the right tools in the right order, not to get creative. Temperature 0 keeps the LLM's behavior consistent and predictable for tool-use scenarios. If you were building a more open-ended creative assistant, you could consider raising this value.

The pre_filter parameter in similarity_search applies a MongoDB query filter before the vector search runs, which is more efficient than post-filtering results. When a user specifies a dietary category, filtering by the tags field narrows the candidate set significantly, helping the vector search find the most semantically similar recipe within that narrowed set.

Session and preference memory serve different purposes and live in different collections. Session memory in sessions is simple document storage: raw text, ordered by timestamp, no embeddings needed. Preference memory in memories uses vector embeddings, so when the agent calls recall_preferences("Japanese noodle dishes"), it retrieves preferences such as "user avoids shellfish" or "user loves umami-forward flavors" based on semantic relevance to the current context, not just keyword matching. This is what makes the memory feel "intelligent" rather than mechanical.

Step 6 — Running the Agent

Start the agent with a session name. Using named sessions lets you maintain separate conversation histories for different users or contexts:

python agent.py my-session

You will see the agent initialize and load any prior history:

Chez Robot ready (session: my-session)
Ask me anything about recipes. Type 'quit' or 'exit' to leave.

Try a sequence of interactions to see the full pipeline in action; keep in mind that the agent's responses might vary compared to the examples here:

You: I'm vegetarian and I really don't like anything too spicy.

Chez Robot: Got it! I've saved that for you. I'll keep your vegetarian diet and preference for mild flavors in mind for every recommendation. What are you in the mood to cook today?

You: I have some chickpeas, tomatoes, and spinach. What can I make?

Chez Robot: With chickpeas, tomatoes, and spinach, you'd love a Chana Masala. It's a classic Indian chickpea curry that's hearty and flavourful without being overly spicy (you can control the chili level). Another great option is a Shakshuka variation with chickpeas added for extra protein...

You: quit

Goodbye! Your preferences have been saved for next time.

Now restart with the same session name and notice how the agent remembers:

python agent.py my-session

  (Loaded 4 messages from your last session)

You: Suggest something for dinner tonight.

Chez Robot: Based on what I know about you, vegetarian and mild flavors, let me pull up some ideas...

The my-session conversation history is loaded from the sessions collection, retrieving your dietary preferences from the memories collection using vector similarity against the current conversational context. This is the full memory loop in action: MongoDB stores it, VoyageAI encodes it, and the agent retrieves only what's relevant using Vector Search.

Step 7 — Inspecting What MongoDB is Storing

Let's take a look at the actual documents stored in MongoDB to understand their structure. With mongosh already set up, you can query the database directly:

mongosh "$MONGODB_URI"

Once connected, switch to the chez_robot database and inspect a recipe document:

use chez_robot

db.recipes.findOne(
  {},
  { _id: 0, name: 1, tags: 1, embedding: { $slice: 5 } }
)

Output:

{
  name: 'Chana Masala',
  tags: [ 'vegetarian', 'indian', 'curry' ],
  embedding: [ 0.0234, -0.1205, 0.0871, 0.0543, -0.0312 ] // only showing the first 5 dimensions out of 1024
}

The embedding field is an array of 1,024 floating-point numbers. The specific values are not human-interpretable in isolation; their meaning only emerges when you compute the similarity between two of them. A query embedding for "chickpea curry" will have a high cosine similarity to this document's embedding because they encode similar semantic content, even if the exact words differ.

You could also inspect a stored preference memory:

db.memories.findOne(
  { type: "preference" },
  { _id: 0, content: 1, embedding: { $slice: 5 } }
)

{
  content: 'user likes vegetarian dishes',
  embedding: [ 0.0234, -0.1205, 0.0871, 0.0543, -0.0312 ] // only showing the first 5 dimensions out of 1024
}

Or a session turn:

db.sessions.findOne(
  { session_id: "my-session" },
  { _id: 0 }
)

{
  session_id: 'my-session',
  role: 'human',
  content: 'I am vegetarian and I really do not like anything too spicy',
  timestamp: ISODate('2026-04-24T13:11:05.898Z')
}

Notice that documents in the memories collection have an embedding field, just like the recipes. This is what makes preference retrieval semantic rather than literal: when the agent calls recall_preferences, it doesn't look up your preferences by keyword; it converts the current conversational context into an embedding and finds the stored preferences that are most semantically similar to it. A preference saved as "user loves bold Mediterranean flavors" will surface when the user asks about Greek or Spanish food, even if those words never appeared in the original preference.

Session turns in the sessions collection, on the other hand, have no embeddings. They are retrieved by session ID and timestamp because conversation history needs to be ordered and complete, not semantically ranked.

Keep in mind that this is a deliberately simple approach to memory management. Agent memory is actually a rich and complex topic with many strategies for storage, retrieval, forgetting, and summarization. If you want to go deeper, this MongoDB Agent Memory article is a great starting point.

Going Further

You have a working recipe agent with semantic search and persistent memory. These ideas could expand its capabilities and help you explore other features:

Add a pre-filter. Expose the dietary_filter parameter in the conversation so users can say "only show me Italian recipes," and the tag filter is applied automatically. You already have the tags field indexed for this.

Score and weight memories. Right now, all memories are treated equally. You could add a relevance_score field and use MongoDB's $addFields aggregation stage during retrieval to boost more recent or more frequently confirmed preferences.

Use Hybrid Search. MongoDB supports hybrid search, combining vector similarity scores with keyword relevance using Reciprocal Rank Fusion and Relative Score Fusion. This is useful when users search for exact recipe names rather than by semantic descriptions.

Conclusion

In this tutorial, you built a recipe assistant agent that uses MongoDB Vector Search for two purposes: semantic retrieval over a recipes collection, and persistent vector-based memory across sessions. You fetched real recipe data from a public API, embedded it with Voyage AI's voyage-4 model, and stored the results as vector-enabled documents in MongoDB. You then wired a LangChain tool-calling agent to an LLM to give the app reasoning and natural language understanding.

The key insight is that a document database and a vector database are not separate infrastructure concerns when you use MongoDB. The same cluster, the same connection string, and the same query patterns handle both structured filtering and semantic search. That consolidation reduces operational complexity and keeps your data model coherent; your recipe documents don't have to live in two places. To learn more, go to Build AI Agents with MongoDB and start coding!

For more on the tools used in this tutorial:

MongoDB as a Vector Database for AI Agents

Nestor Daza — Thu, 01 Oct 2026 16:23:24 +0000

This article was written by Néstor Daza

Introduction

MongoDB is a general-purpose document database with several modules embedded in its core that turn it into a complete data platform for modern application development. One of them is native Vector Search, allowing MongoDB to manage your data while also powering semantic retrieval and agent memory for your application. You don't need a separate vector store.

In this tutorial, you will build a Recipe Assistant Agent called Chez Robot using MongoDB, VoyageAI, Claude, and LangChain. The agent will answer natural language questions like "what can I make with rice and saffron?" by performing a semantic search over a collection of recipes fetched from TheMealDB, a free, public recipe API. It will also remember your personal preferences across sessions, storing and retrieving that memory from the database using the same vector index.

By the end, you will understand how MongoDB handles embeddings and Vector Search, how LangChain wires together retrieval and memory, and how Voyage AI models fit into the pipeline.

What you will need

Before you begin, make sure you have the following:

Python 3.10 or higher. You can verify with python --version. If you need to install it, follow the official Python installation guide.
A MongoDB Atlas account where you will set up a free M0 cluster later. You can create the account at mongodb.com/cloud/atlas (no credit card required for M0). It is also possible to use a local Instance of MongoDB Community Edition.
mongosh (MongoDB Shell). You will use it to create the vector search indexes and query the database. Follow the official installation guide to set it up.
A Voyage AI API key. Sign up at voyageai.com and generate a key from your dashboard. Voyage AI offers a free tier generous enough to complete this tutorial.
An Anthropic API key for Claude, which will drive the agent's reasoning. You can create one at console.anthropic.com.
Basic familiarity with what embeddings and Vector Search are, what a language model agent does, and how MongoDB stores documents. If any of those feel shaky, the MongoDB Vector Search docs and Anthropic's intro to Claude are good starting points.

Understanding the Architecture

A MongoDB document stores your application data: fields, arrays, nested objects, etc. A vector-enabled document stores all of that, plus an additional field containing a high-dimensional float array: the embedding. When you run a Vector Search query, MongoDB compares your query embedding against the stored embeddings of all your documents in a collection; the result is a ranked list of documents ordered by semantic similarity, effectively finding elements that have a meaningful relationship with the original query.

Vector Search is the backbone of two capabilities that make AI agents genuinely useful in production: knowledge retrieval and persistent memory.

RAG (Retrieval-Augmented Generation): The agent retrieves relevant information from external sources before generating an answer, grounding its output in real data rather than the model's training knowledge alone.
Persistent memory: Between sessions, the agent stores what it learns about your preferences and previous interactions, and retrieves the most contextually relevant memories at the start of each new conversation.

Putting everything together, we can build out the Chez Robot app with a full stack that looks like this:

TheMealDB -> VoyageAI embeddings -> MongoDB Atlas (vector index)
                                                  |
                                        LangChain retriever
                                                  |
                                       Claude (via LangChain)
                                                  |
                                               Your app

Step 1 — Setting Up Your Environment

Create a project directory and a virtual environment to keep dependencies isolated:

mkdir chez-robot-agent
cd chez-robot-agent
python -m venv .venv
source .venv/bin/activate   # On Windows: .venv\Scripts\activate

Install the required packages:

pip install langchain langchain-anthropic langchain-mongodb \
            langchain-voyageai pymongo requests python-dotenv

Here is what each package does:

langchain provides the agent framework and the retrieval abstractions.
langchain-anthropic is the LangChain integration for Claude.
langchain-mongodb provides MongoDBAtlasVectorSearch, which stores and queries embeddings in MongoDB directly from LangChain.
langchain-voyageai wraps the Voyage AI embedding API in a LangChain-compatible interface.
pymongo is the underlying MongoDB driver.
requests is used to fetch recipes from TheMealDB.
python-dotenv loads your secrets from a .env file.

Add a .gitignore so you never accidentally commit secrets:

.env
.venv/
__pycache__/

Step 2 — Setting Up MongoDB Atlas

Head to MongoDB Atlas and log in to your account.

To create your cluster, go to Clusters, click Create, and select the M0 free tier. Choose your preferred cloud provider and region, in Cluster Details, name your cluster chez-robot, and click Create Cluster. It'll take a couple of minutes to get everything up and running.

Finally, whitelist your IP address to allow secure access to your cluster. In Network Access, select IP Access List and click Add IP Address. Click Add Current IP Address to whitelist your machine, then click Confirm. You might already have 0.0.0.0/0 configured, which allows access from anywhere; this is useful if you are working from a dynamic IP or want to keep things simple for a local tutorial, but not recommended for production deployments.

Create a .env file in the project root:

MONGODB_URI=<your-atlas-connection-string>
VOYAGE_API_KEY=<your-voyage-api-key>
ANTHROPIC_API_KEY=<your-anthropic-api-key>

Step 3 — Creating the Vector Search Indexes

Create a file called scripts/create_indexes.js in your project root (this is a JavaScript file executed directly by mongosh, no additional dependencies needed):

// create_indexes.js

const DB_NAME = "chez_robot";

db = db.getSiblingDB(DB_NAME);

// Create collections explicitly before creating vector search indexes
db.createCollection("recipes");
db.createCollection("memories");

// Vector search index for recipes
db.recipes.createSearchIndex({
  name: "recipe_vector_index",
  type: "vectorSearch",
  definition: {
    fields: [
      {
        type: "vector",
        path: "embedding",
        numDimensions: 1024,
        similarity: "cosine"
      },
      {
        type: "filter",
        path: "tags"
      }
    ]
  }
});

// Vector search index for agent memory
db.memories.createSearchIndex({
  name: "memory_vector_index",
  type: "vectorSearch",
  definition: {
    fields: [
      {
        type: "vector",
        path: "embedding",
        numDimensions: 1024,
        similarity: "cosine"
      }
    ]
  }
});

print("Index creation finished for recipes and memories collections.");

The cosine similarity metric measures the angle between vectors rather than their absolute distance, which works well for semantic similarity tasks where the direction of the embedding matters more than its magnitude.

Make sure you have mongosh installed; then export your connection string as an environment variable (the same URI you added to your .env file in Step 2) and run the index creation script.

export MONGODB_URI=<your-atlas-connection-string>
mongosh "$MONGODB_URI" --file scripts/create_indexes.js

Step 4 — Ingesting Recipes from TheMealDB

Create a file called ingest.py:

import os
import requests
from pymongo import MongoClient
from langchain_voyageai import VoyageAIEmbeddings
from dotenv import load_dotenv

load_dotenv()

# --- Clients ---
mongoDBClient = MongoClient(os.environ["MONGODB_URI"], appname="devrel-tutorial-vectorsearch-agents")
db = mongoDBClient["chez_robot"]
collection = db["recipes"]

voyageAIClient = VoyageAIEmbeddings(
    voyage_api_key=os.environ["VOYAGE_API_KEY"],
    model="voyage-4",
)

# --- Fetch recipes from TheMealDB ---
def fetch_recipes_by_letter(letter: str) -> list[dict]:
    url = f"https://www.themealdb.com/api/json/v1/1/search.php?f={letter}"
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    meals = response.json().get("meals") or []
    return meals

def build_recipe_text(meal: dict) -> str:
    """Build a single string representation of a recipe for embedding."""
    ingredients = []
    for i in range(1, 21):
        ingredient = (meal.get(f"strIngredient{i}") or "").strip()
        measure = (meal.get(f"strMeasure{i}") or "").strip()
        if ingredient:
            ingredients.append(f"{measure} {ingredient}".strip())

    return (
        f"Recipe: {meal['strMeal']}\n"
        f"Category: {meal.get('strCategory', '')}\n"
        f"Cuisine: {meal.get('strArea', '')}\n"
        f"Ingredients: {', '.join(ingredients)}\n"
        f"Instructions: {meal.get('strInstructions', '')[:500]}"
    )

def build_tags(meal: dict) -> list[str]:
    tags = []
    if meal.get("strCategory"):
        tags.append(meal["strCategory"].lower())
    if meal.get("strArea"):
        tags.append(meal["strArea"].lower())
    raw_tags = meal.get("strTags") or ""
    tags += [t.strip().lower() for t in raw_tags.split(",") if t.strip()]
    return list(set(tags))

# --- Ingest ---
def ingest(letters: str = "abcdefghijklmnopqrstuvwxyz"):
    print("Fetching recipes from TheMealDB...")
    all_meals = []
    for letter in letters:
        meals = fetch_recipes_by_letter(letter)
        all_meals.extend(meals)
        print(f"  {letter}: {len(meals)} recipes")

    print(f"\nTotal recipes fetched: {len(all_meals)}")
    print("Generating embeddings with Voyage AI...")

    texts = [build_recipe_text(meal) for meal in all_meals]

    # Voyage AI supports batching - send up to 128 texts at a time
    batch_size = 64
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        batch_embeddings = voyageAIClient.embed_documents(batch)
        all_embeddings.extend(batch_embeddings)
        print(f"  Embedded {min(i + batch_size, len(texts))}/{len(texts)}")

    print("Writing documents to MongoDB ...")
    documents = []
    for meal, text, embedding in zip(all_meals, texts, all_embeddings):
        documents.append({
            "meal_id": meal["idMeal"],
            "name": meal["strMeal"],
            "category": meal.get("strCategory"),
            "cuisine": meal.get("strArea"),
            "tags": build_tags(meal),
            "source_text": text,
            "thumbnail": meal.get("strMealThumb"),
            "embedding": embedding,
        })

    # Use replace to avoid duplicates if you re-run ingest
    for doc in documents:
        collection.replace_one(
            {"meal_id": doc["meal_id"]},
            doc,
            upsert=True,
        )

    print(f"Done. {len(documents)} recipes stored in MongoDB.")

if __name__ == "__main__":
    ingest()

Run the ingestion script:

python ingest.py

You will see output similar to:

Fetching recipes from TheMealDB...
  a: 9 recipes
  b: 14 recipes
  ...
Total recipes fetched: 303
Generating embeddings with Voyage AI...
  Embedded 64/303
  Embedded 128/303
  ...
Done. 303 recipes stored in MongoDB.

Each document in your recipes collection now has an embedding field alongside the recipe's structured data. The embedding is an array of 1024 (the dimensions) floating-point numbers, generated by passing the full recipe text (name, category, cuisine, ingredients, and a truncated version of the instructions) through voyage-4. It captures the meaning of the recipe, not just its keywords, which is what allows a query like "light Japanese noodle dish" to match a recipe for Miso Ramen even if those exact words don't appear together in the document.

Step 5 — Building the Agent

We're ready to build our agent. Create a file called agent.py:

import os
import warnings
from datetime import datetime, UTC
from pymongo import MongoClient
from langchain_anthropic import ChatAnthropic
from langchain_mongodb import MongoDBAtlasVectorSearch
from langchain_voyageai import VoyageAIEmbeddings
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.prebuilt import create_react_agent
from dotenv import load_dotenv

warnings.filterwarnings("ignore", category=DeprecationWarning)

load_dotenv()

# --- Clients ---
mongoDBClient = MongoClient(os.environ["MONGODB_URI"], appname="devrel-tutorial-vectorsearch-agents")
db = mongoDBClient["chez_robot"]

voyageAIClient = VoyageAIEmbeddings(
    voyage_api_key=os.environ["VOYAGE_API_KEY"],
    model="voyage-4",
)

llm = ChatAnthropic(
    model="claude-haiku-4-5",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
    temperature=0,
)

# --- Vector stores ---
recipe_store = MongoDBAtlasVectorSearch(
    collection=db["recipes"],
    embedding=voyageAIClient,
    index_name="recipe_vector_index",
    text_key="source_text",
    embedding_key="embedding",
)

memory_store = MongoDBAtlasVectorSearch(
    collection=db["memories"],
    embedding=voyageAIClient,
    index_name="memory_vector_index",
    text_key="content",
    embedding_key="embedding",
)


# --- Tools ---

@tool
def search_recipes(query: str, dietary_filter: str = "") -> str:
    """
    Search for recipes by ingredient, cuisine, dish type, or cooking style.
    Optionally filter by dietary category (e.g. 'vegetarian', 'seafood', 'chicken').
    Returns the top 5 most relevant recipes.
    """
    search_kwargs = {"k": 5}

    if dietary_filter:
        search_kwargs["pre_filter"] = {
            "tags": {"$in": [dietary_filter.lower()]}
        }

    results = recipe_store.similarity_search(query, **search_kwargs)

    if not results:
        return "No recipes found matching that query."

    output = []
    for i, doc in enumerate(results, 1):
        output.append(f"**Recipe {i}:** {doc.metadata.get('name', 'Unknown')}\n{doc.page_content[:400]}\n")

    return "\n".join(output)


@tool
def save_preference(preference: str) -> str:
    """
    Save a dietary preference or food restriction that the user has mentioned.
    Examples: 'user is vegetarian', 'user dislikes cilantro', 'user is allergic to shellfish'.
    """
    doc = {
        "content": preference,
        "created_at": datetime.now(UTC).isoformat(),
        "type": "preference",
    }
    memory_store.add_texts(
        texts=[preference],
        metadatas=[doc],
    )
    return f"Got it, I've saved that preference: '{preference}'"


@tool
def recall_preferences(context: str) -> str:
    """
    Recall stored user preferences relevant to the current cooking context.
    Use this at the start of a session or when the user asks for personalized recommendations.
    """
    results = memory_store.similarity_search(context, k=5)
    if not results:
        return "No preferences stored yet."

    preferences = [doc.page_content for doc in results]
    return "Remembered preferences:\n" + "\n".join(f"- {p}" for p in preferences)


# --- Load session memory from MongoDB ---
def load_session_memory(session_id: str, k: int = 10) -> list:
    """Retrieve the most recent conversation turns for this session."""
    turns = list(
        db["sessions"].find(
            {"session_id": session_id},
            sort=[("timestamp", -1)],
            limit=k,
        )
    )
    turns.reverse()

    history = []
    for turn in turns:
        if turn["role"] == "human":
            history.append(HumanMessage(content=turn["content"]))
        else:
            history.append(AIMessage(content=turn["content"]))
    return history


def save_turn(session_id: str, role: str, content: str):
    """Persist a single conversation turn to MongoDB."""
    db["sessions"].insert_one({
        "session_id": session_id,
        "role": role,
        "content": content,
        "timestamp": datetime.now(UTC),
    })


# --- Agent setup ---
tools = [search_recipes, save_preference, recall_preferences]

system_prompt = """You are a knowledgeable and enthusiastic recipe assistant. You help users 
find recipes based on ingredients they have, cuisines they love, or dietary needs they follow.

At the start of each conversation, use the recall_preferences tool to check if the user 
has any stored dietary preferences, allergies, or food restrictions, and factor those 
into every recommendation you make.

When a user mentions a preference, restriction, or allergy, always save it using the 
save_preference tool so you can remember it next time.

When recommending recipes, be specific: mention the cuisine, key ingredients, and why 
it matches what the user is looking for. Keep your tone warm and conversational."""

agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt=system_prompt,
)


# --- Main loop ---
def run(session_id: str = "default"):
    print(f"\nChez Robot ready (session: {session_id})")
    print("Ask me anything about recipes. Type 'quit' or 'exit' to leave.\n")

    chat_history = load_session_memory(session_id)

    if chat_history:
        print(f"  (Loaded {len(chat_history)} messages from your last session)\n")

    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit", "q"):
            print("Goodbye! Your preferences have been saved for next time.")
            break
        if not user_input:
            continue

        result = agent.invoke({
            "messages": chat_history + [HumanMessage(content=user_input)]
        })

        response = result["messages"][-1].content
        print(f"\nChez Robot: {response}\n")

        chat_history.append(HumanMessage(content=user_input))
        chat_history.append(AIMessage(content=response))

        save_turn(session_id, "human", user_input)
        save_turn(session_id, "ai", response)


if __name__ == "__main__":
    import sys
    session = sys.argv[1] if len(sys.argv) > 1 else "default"
    run(session_id=session)

There are a few things here worth unpacking.

The pre_filter parameter in similarity_search applies a MongoDB query filter before the vector search runs, which is more efficient than post-filtering results. When a user specifies a dietary category, filtering by the tags field narrows the candidate set significantly, helping the vector search find the most semantically similar recipe within that narrowed set.

Session and preference memory serve different purposes and live in different collections. Session memory in sessions is simple document storage: raw text, ordered by timestamp, no embeddings needed. Preference memory in memories uses vector embeddings, so when the agent calls recall_preferences("Japanese noodle dishes"), it retrieves preferences such as "user avoids shellfish" or "user loves umami-forward flavors" based on semantic relevance to the current context, not just keyword matching. This is what makes the memory feel "intelligent" rather than mechanical.

Step 6 — Running the Agent

Start the agent with a session name. Using named sessions lets you maintain separate conversation histories for different users or contexts:

python agent.py my-session

You will see the agent initialize and load any prior history:

Chez Robot ready (session: my-session)
Ask me anything about recipes. Type 'quit' or 'exit' to leave.

Try a sequence of interactions to see the full pipeline in action; keep in mind that the agent's responses might vary compared to the examples here:

You: I'm vegetarian and I really don't like anything too spicy.

Chez Robot: Got it! I've saved that for you. I'll keep your vegetarian diet and preference for mild flavors in mind for every recommendation. What are you in the mood to cook today?

You: I have some chickpeas, tomatoes, and spinach. What can I make?

Chez Robot: With chickpeas, tomatoes, and spinach, you'd love a Chana Masala. It's a classic Indian chickpea curry that's hearty and flavourful without being overly spicy (you can control the chili level). Another great option is a Shakshuka variation with chickpeas added for extra protein...

You: quit

Goodbye! Your preferences have been saved for next time.

Now restart with the same session name and notice how the agent remembers:

python agent.py my-session

  (Loaded 4 messages from your last session)

You: Suggest something for dinner tonight.

Chez Robot: Based on what I know about you, vegetarian and mild flavors, let me pull up some ideas...

Step 7 — Inspecting What MongoDB is Storing

Let's take a look at the actual documents stored in MongoDB to understand their structure. With mongosh already set up, you can query the database directly:

mongosh "$MONGODB_URI"

Once connected, switch to the chez_robot database and inspect a recipe document:

use chez_robot

db.recipes.findOne(
  {},
  { _id: 0, name: 1, tags: 1, embedding: { $slice: 5 } }
)

Output:

{
  name: 'Chana Masala',
  tags: [ 'vegetarian', 'indian', 'curry' ],
  embedding: [ 0.0234, -0.1205, 0.0871, 0.0543, -0.0312 ] // only showing the first 5 dimensions out of 1024
}

You could also inspect a stored preference memory:

db.memories.findOne(
  { type: "preference" },
  { _id: 0, content: 1, embedding: { $slice: 5 } }
)

{
  content: 'user likes vegetarian dishes',
  embedding: [ 0.0234, -0.1205, 0.0871, 0.0543, -0.0312 ] // only showing the first 5 dimensions out of 1024
}

Or a session turn:

db.sessions.findOne(
  { session_id: "my-session" },
  { _id: 0 }
)

{
  session_id: 'my-session',
  role: 'human',
  content: 'I am vegetarian and I really do not like anything too spicy',
  timestamp: ISODate('2026-04-24T13:11:05.898Z')
}

Going Further

You have a working recipe agent with semantic search and persistent memory. These ideas could expand its capabilities and help you explore other features:

Add a pre-filter. Expose the dietary_filter parameter in the conversation so users can say "only show me Italian recipes," and the tag filter is applied automatically. You already have the tags field indexed for this.

Score and weight memories. Right now, all memories are treated equally. You could add a relevance_score field and use MongoDB's $addFields aggregation stage during retrieval to boost more recent or more frequently confirmed preferences.

Use Hybrid Search. MongoDB supports hybrid search, combining vector similarity scores with keyword relevance using Reciprocal Rank Fusion and Relative Score Fusion. This is useful when users search for exact recipe names rather than by semantic descriptions.

Conclusion

In this tutorial, you built a recipe assistant agent that uses MongoDB Vector Search for two purposes: semantic retrieval over a recipes collection, and persistent vector-based memory across sessions. You fetched real recipe data from a public API, embedded it with Voyage AI's voyage-4 model, and stored the results as vector-enabled documents in MongoDB. You then wired a LangChain tool-calling agent to an LLM to give the app reasoning and natural language understanding.

The key insight is that a document database and a vector database are not separate infrastructure concerns when you use MongoDB. The same cluster, the same connection string, and the same query patterns handle both structured filtering and semantic search. That consolidation reduces operational complexity and keeps your data model coherent; your recipe documents don't have to live in two places. To learn more, go to Build AI Agents with MongoDB and start coding!

For more on the tools used in this tutorial:

Building an Intelligent Document Processing Pipeline with MongoDB Atlas

MongoDB Guests — Thu, 01 Oct 2026 13:45:50 +0000

This tutorial was written by Matteo Rossi.

Imagine a supplier changes the layout of an invoice. The filename still says "invoice," so the file goes to the right parser, but the total has moved to another part of the page. The parser puts a number in the wrong field. Later, someone looking at the saved result has no easy way to see where that number came from.

I want the next system to see where each answer came from, not just the answer itself. The pipeline reads the document to decide what type it is and which values to extract. When a model fills in a field, we keep the original text and its position on the page, along with the model used and the source of any score.
MongoDB Atlas lets us store that information together. For files with a fixed, machine-readable format, a rule-based parser is still the better choice.

Why Keep the Evidence?

When a model says, "This is an invoice," it might be wrong. We can compare it with known invoices and check how close the next possible type is to it. The same goes for "the total is 1525.00": we need to see the text it came from and where it appears on the page. An OCR score can help, but only if we know what it measures.

If we save only the value, we can't tell a clear scan from a poor one. The person reviewing the result doesn't know where to look, and the team can't easily choose which documents to check first.

The filename, folder, and sender address can still help, but they don't tell us what is on the page. If people change the way they name or send files, we need to check the content.

Templates still make sense when you have a few formats that rarely change. A rule-based parser can be cheaper and easier to maintain, although it still needs checks: a layout may match a rule and put a value in the wrong field.

Where is The Risk

Think of a common setup. Optical character recognition (OCR) reads the scan, a rule based on the filename picks a parser, and a template finds the fields on the page. The result goes into database tables. This works well for layouts that stay the same, but it gets harder when you have many formats or when they change without warning.

The router trusts the filename. A supplier can change the way it names files without telling you. A filename rule can't check whether the file contains the right type of document, so a bad match may reach the parser without an error. You can measure how often the rule is right if you have documents with known types, but the rule itself cannot say, "I'm not sure." A classifier that reads the content can send uncertain cases to review.
Different types need different fields. Invoices, delivery notes, and contracts have little in common beyond basic file information. One wide table leaves many empty columns. Separate tables can work, but they make queries across document types harder. A relational database can still store the source and score for each value. A document database simply lets us keep them next to the value, while each type has its own set of fields.
The write step loses useful details. If extraction returns a score or a place on the page, but we save only the value, we lose those details. Still, a score given by a generative model is not always a reliable measure of how often it is right. Saving it doesn't make it trustworthy.

For each field filled in by a model, keep the value and the place it came from. If there is a score, record how it was made, along with the model version. If you don't have a score you can trust, leave it empty and use other checks to decide whether someone should review the field.

Keeping the Evidence with the Document

The page images can stay in object storage. In Atlas, I'd keep the OCR text, document type, extracted values, and review status together, with a link to the original file.

Start with the document. One documents collection holds a record for each file. The example leaves out the OCR text for each part of the page to keep it short. In a real system, you need that text and its position so a reviewer can check each value against the scan. The fields in extraction change with the document type.

{
  "_id": "6f1c9a2e",
  "source": { "uri": "s3://ingest/2026-09-14/INV-88213.pdf", "receivedAt": "2026-09-14T02:11:04Z" },
  "status": "needs_review",
  "classification": { "type": "invoice", "score": 0.81, "margin": 0.34, "exemplarSetVersion": 7 },
  "extraction": {
    "invoiceTotal": {
      "value": "1525.00",
      "confidence": { "score": 0.93, "source": "ocr", "method": "token-aggregation" },
      "provenance": { "page": 1, "boundingBox": [0.61, 0.72, 0.84, 0.78], "text": "Total EUR 1.525,00" }
    },
    "vatAmount": {
      "value": "275.00",
      "confidence": { "score": 0.46, "source": "model", "method": "self-reported-uncalibrated" },
      "provenance": { "page": 1, "boundingBox": [0.61, 0.66, 0.84, 0.71], "text": "VAT 22% 275,00" }
    }
  },
  "review": {
    "required": true,
    "priority": 87,
    "reasons": [
      { "code": "LOW_FIELD_CONFIDENCE", "field": "vatAmount" },
      { "code": "HIGH_VALUE_DOCUMENT" }
    ]
  },
  "pipeline": {
    "embeddingModel": "voyage-4",
    "extractionModel": "<model-id>",
    "contractVersion": 3,
    "classifiedAt": "2026-09-14T02:11:09Z"
  }
}

The two scores in the example mean different things. The OCR score of 0.93 tells us how clearly the system reads the text; it doesn't tell us if the value belongs in the right field. The 0.46 comes from the model itself and hasn't been tested against real results. Keeping the source and method with each score helps us use them in different ways. Sometimes, a check against a business rule and a link to the page tell us more than either score.

The page number, the box around the text, and the original words let a reviewer find the value on the scan. We have to keep that link from the OCR step onward. If we pass only plain text to the extraction step, we can't recover the page position later.

One important thing is to decide what needs to be reviewed when you save the result. One score limit for the whole document treats every field the same. A doubtful note may matter less than a total of 10,000 that looks right but isn't. The worker checks the fields and saves their decision in review, so the queue can show the most important cases first.

It’s important to try finding the type by example. The worker turns the first page into a vector, then uses Atlas Vector Search to find similar files in exemplars, a collection labelled by people. It groups the closest results by type instead of relying on the filename. You still need to test this with your own files: the first page won't always be enough.

db.exemplars.aggregate([
  { $vectorSearch: {
      index: "exemplar_vectors",
      path: "embedding",
      queryVector: firstPageEmbedding,
      numCandidates: 200,
      limit: 25
  }},
  { $project: { label: 1, score: { $meta: "vectorSearchScore" } } },
  { $group: {
      _id: "$label",
      neighbours: { $sum: 1 },
      maxScore: { $max: "$score" },
      avgScore: { $avg: "$score" }
  }},
  { $addFields: { rank: { $add: [
      { $multiply: ["$maxScore", 0.7] },
      { $multiply: ["$avgScore", 0.3] }
  ]}}},
  { $sort: { rank: -1 } },
  { $limit: 2 }
])

Adding up the scores for each type can favour the type with more examples among the top results. The query uses the best score and the average instead, considering that the 0.7 and 0.3 weights are just an example. The query returns the top two types so the worker can compare them. If the best score is too low or the gap is too small, it sends the file to review. These scores are not probabilities, but use a separate set of documents labelled by people to choose the weights and limits, and check what happens when one type has more examples than another.

For this example, use Voyage AI's voyage-4 with its default vector size of 1024 and a cosine vector index. The model can read up to 32,000 tokens, but check the length of each page so no text gets cut off. An invoice and a credit note may still look almost the same. Test such cases with documents kept aside for testing before you try voyage-4-large or add another model to decide between the two.

Atlas can also create the vectors for you. Set the text field and model in an autoEmbed index, then pass query: { text: ... } to $vectorSearch instead of queryVector. Here, the worker creates the vector itself, so it can record the model used for each run. It’s important to highlight that MongoDB currently lists Automated Embedding as a Preview feature.

To add a type, provide examples labelled by people and a list of fields to extract. If the match is weak, send the file to review before extraction; once a person confirms its type, it can join exemplars. Limit the number of examples per type and give each set a version number so you can track the effect of changes. Changing the vector model may mean creating new vectors and a new index. Models in the voyage-4 family produce compatible vectors, but you still need to test their results.

Lastly, define what to extract. The document_types collection holds the expected fields and format for each type. The worker asks the model to follow that format and checks the answer before saving it. When you deploy the system, you also use these definitions to update the database rules that check new records.

// document_types: one entry per type
{
  _id: "invoice",
  version: 3,
  reviewPolicy: { minFieldConfidence: 0.75, highValueThreshold: 10000 },
  extractionSchema: {
    bsonType: "object",
    required: ["invoiceNumber", "invoiceTotal", "issueDate"],
    properties: { /* one entry per field, each a value/confidence/provenance object */ }
  }
}

// deployment step: compile the registered schemas into the collection validator
db.runCommand({
  collMod: "documents",
  validator: { $jsonSchema: {
    bsonType: "object",
    required: ["source", "status", "classification", "pipeline"],
    oneOf: [
      { properties: { classification: { properties: { type: { enum: ["invoice"] } } },
                      extraction: invoiceExtractionSchema } },
      { properties: { classification: { properties: { type: { enum: ["delivery_note"] } } },
                      extraction: deliveryNoteExtractionSchema } }
    ]
  }}
})

This snippet shows the idea, not a command you can run as it is: invoiceExtractionSchema and deliveryNoteExtractionSchema stand for schema objects created during deployment. The full database rule also needs to accept files sent to review before extraction, and to require extraction and classification.type for files that were extracted. MongoDB does not read document_types every time it saves a file. These rules can catch missing fields or the wrong format, but not a total that looks valid and is wrong.

Keeping a Review Queue

The query below lists files that need a person to check them, starting with the highest priority. If the classifier cannot choose a type, or a result fails a format check, the worker also sets status: "needs_review" and saves the reason in review.reasons.

db.documents.createIndex({ status: 1, "review.required": 1, "review.priority": -1 })

db.documents.aggregate([
  { $match: { status: "needs_review", "review.required": true } },
  { $sort: { "review.priority": -1 } },
  { $limit: 50 },
  { $project: { "source.uri": 1, "classification.type": 1, "review.reasons": 1 } }
])

A worker watches Change Streams and moves each file to the next step. It may receive the same event twice, so it must check the file's current status and pipeline version before making a change. Atlas Triggers can handle simple steps. A separate worker gives you more control when you need to retry failed calls, limit how many files run at once, or keep a queue of failures.

A file usually moves from received to classified, then extracted, and finally to confirmed or needs_review. A file also goes to review if the classifier cannot choose a type or if the result fails a check. In the first case, the reviewer starts by choosing the type.

Sometimes a Parser is Enough

An electronic invoice in a required XML format already names its fields. Read it with a parser and check that it follows the format; a model would add cost and another way to fail. Keep the original file and handle files that don't follow the rules, but you don't need a model to find the total.

The same applies when most files use one familiar layout. Try the parser first and send a file to the model if it fails a check or uses a new layout. Even a parser that reports success can put a value in the wrong field, so keep track of later corrections.

The model is most useful for changing layouts and document types that have no good template. For the rest, start with the parser.

Migration Path

You can add these parts while the old pipeline keeps running. You may still need to change how OCR text, page positions, and review decisions are saved.

Keep the original and the OCR text. Save text and page positions, but leave the old routing in place for now. Reviewers will need both later.
Run the new classifier without using its answers yet. Compare its choices with the old router and check the files where they disagree. Agreement does not mean either one is right.
Collect examples with checked labels. Ask people to confirm the types of files from the old pipeline. Use some as examples and keep the rest separate to test error rates and choose score limits.
Change routing for one type at a time. Start where the old router makes expensive mistakes. Send uncertain files to review unless the old route has passed a check based on the file's content.
Try extraction for types without templates. Compare the results with the original pages before using them for payments or approvals.
Set review rules for each field. A wrong total costs more than a missing optional note. Use real corrections to test the rules, not just model scores.
Check results after launch. Track how often the classifier is right on checked files, how often it asks for review, how many files go through without review, and how many errors people find later. Late corrections show mistakes that the other checks missed.

What Survives is the Pipeline

Go back to the invoice with the moved total. A classifier that reads the page may still call it an invoice. But did the extraction step put the amount in the right field? If we keep the original text and its place on the page, a reviewer can check it when a rule finds something odd. The model can still make mistakes; the difference is that we have a way to find them before another system uses the value.

Multiplayer Game Development with TypeScript, Phaser, Socket.io, and MongoDB

Nic Raboy — Mon, 28 Sep 2026 20:12:41 +0000

Game data rarely looks the same for very long. A player document might start out with just a username and a score, and a few sprints later, it needs an inventory, a set of unlocked achievements, and a friends list, each shaped differently from the last. MongoDB's document model is a natural fit for that kind of growth, since a new field is just a new field, with no migration to write and no join to add. A match record can embed both players directly, scores and all, so reading back a finished match is a single document lookup rather than a query spread across multiple tables. Atomic operators like $inc and $max also mean stats and high scores can be updated safely by many concurrent players at once, without the application code having to coordinate a read-then-write itself.

In this tutorial, we're going to build a simple multiplayer game. Each player controls a white square on a canvas, and for thirty seconds, yellow coin squares spawn at random positions. Whoever collects the most coins when the timer runs out wins. Movement, coin spawning, and scoring all happen over WebSockets through Socket.io, with the server acting as the single source of truth for everything that happens during a match. MongoDB sits underneath that, storing player accounts, match records, and the leaderboard.

We'll use TypeScript, Zod, Socket.io, and Express on the backend, and Phaser, TypeScript, and Socket.io on the frontend. The MongoDB Node.js driver is used directly, without an ODM like Mongoose, so every query in this tutorial is plain driver code. If you'd rather skip ahead or just want to run the finished project, the full source is available on GitHub.

The Prerequisites

Prior to starting this tutorial, you'll need a few things in place:

A MongoDB instance, either a local server or an Atlas cluster on the free tier
Node.js 18+
Two browser tabs or windows, since a match needs two players to start

The expectation is that you can already connect to MongoDB with a connection string. If you need help getting an Atlas cluster running, check out the MongoDB documentation for getting started.

Understanding the Shape of the Project

Because the backend and the frontend are two independent npm projects with no shared package, we'll set each one up on its own. The backend is an Express application with a Socket.io server attached to the same HTTP server. The frontend is a Phaser game bootstrapped with Vite. Neither project depends on the other at build time. They only talk to each other over HTTP and WebSockets at runtime.

By the end of this tutorial, the backend will have the following shape:

backend/
├── .env
├── config.ts
├── main.ts
├── routes.ts
├── libs/
│   └── mongodb.ts
├── game/
│   ├── state.ts
│   ├── coins.ts
│   └── socket.ts
└── types/
    └── index.ts

And the frontend will look like this:

frontend/
├── .env
├── index.html
└── src/
    ├── main.ts
    ├── style.css
    ├── api.ts
    ├── socket.ts
    ├── types.ts
    ├── screens/
    │   ├── lobby.ts
    │   └── results.ts
    └── game/
        └── game-scene.ts

Go ahead and create the two project directories, backend and frontend, side by side, and we'll fill each one in as we go.

Setting Up the Backend Project

Inside backend, initialize a Node.js project and install the runtime dependencies:

npm init -y
npm install express cors socket.io mongodb zod dotenv

Express is the HTTP layer, cors lets the frontend's origin talk to it, socket.io handles the real-time game traffic, the official mongodb package is our database driver, zod validates the two request bodies our REST layer accepts, and dotenv loads configuration from a .env file.

For TypeScript itself, install the following as dev dependencies:

npm install --save-dev typescript ts-node nodemon @types/express @types/node @types/cors

ts-node runs TypeScript directly without a separate compile step, and nodemon restarts the process whenever a file changes, which keeps the feedback loop tight while you build out the socket handlers.

Add a tsconfig.json at the root of backend:

{
    "compilerOptions": {
        "target": "ES2020",
        "module": "commonjs",
        "lib": ["ES2020"],
        "outDir": "./dist",
        "rootDir": "./",
        "strict": true,
        "esModuleInterop": true,
        "skipLibCheck": true,
        "forceConsistentCasingInFileNames": true,
        "resolveJsonModule": true
    },
    "include": ["./**/*.ts"],
    "exclude": ["node_modules", "dist"]
}

Then update the scripts block of backend/package.json so there's a development command and a production command:

"scripts": {
    "dev": "nodemon --exec ts-node main.ts",
    "build": "tsc",
    "start": "node dist/main.js"
}

Finally, create a backend/.env file. This is where every tunable in the game lives, from canvas dimensions to how often coins spawn:

PORT=3000
MONGO_URI=ADD_YOUR_CONNECTION_STRING_HERE
MONGO_DB_NAME=coin_race
FRONTEND_URL=http://localhost:5173

CANVAS_WIDTH=1920
CANVAS_HEIGHT=1080

PLAYER_SIZE=40
PLAYER_SPEED=10

COIN_SIZE=20
COIN_MAX_ON_SCREEN=10
COIN_SPAWN_INTERVAL_MS=2000

MATCH_DURATION_MS=30000
LEADERBOARD_LIMIT=10

Replace MONGO_URI with your own connection string. The coin_race database does not need to exist ahead of time. MongoDB creates it the first time we write to it. If you're using version control, make sure .env is in your .gitignore.

Loading Configuration Once at Startup

Nothing in this game should read process.env directly while a match is running. Player movement happens dozens of times per second, and re-parsing environment variables on every one of those calls would be wasteful for no benefit, since the values never change after the process boots. Instead, we'll read everything into a typed config object once.

Create backend/config.ts:

import "dotenv/config";

function intFromEnv(name: string, fallback: number): number {
    const raw = process.env[name];
    if (raw === undefined) return fallback;

    const parsed = parseInt(raw, 10);
    if (Number.isNaN(parsed)) {
        throw new Error(`Environment variable ${name} must be a number, got "${raw}".`);
    }

    return parsed;
}

The intFromEnv helper does two things worth calling out. If the variable is not set, it falls back to a sane default rather than blowing up, which keeps the .env file optional for anything with a reasonable default. But if the variable is set to something that is not a number, it throws immediately at startup instead of quietly producing NaN the first time a coin tries to spawn at an NaN coordinate. A typo in .env should fail loudly on day one, not three requests into a match.

With that helper in place, add the actual config object to the bottom of the same backend/config.ts file:

export const config = {
    PORT: intFromEnv("PORT", 3000),
    FRONTEND_URL: process.env.FRONTEND_URL ?? "http://localhost:5173",

    CANVAS_WIDTH: intFromEnv("CANVAS_WIDTH", 1920),
    CANVAS_HEIGHT: intFromEnv("CANVAS_HEIGHT", 1080),

    PLAYER_SIZE: intFromEnv("PLAYER_SIZE", 40),
    PLAYER_SPEED: intFromEnv("PLAYER_SPEED", 10),

    COIN_SIZE: intFromEnv("COIN_SIZE", 20),
    COIN_MAX_ON_SCREEN: intFromEnv("COIN_MAX_ON_SCREEN", 10),
    COIN_SPAWN_INTERVAL_MS: intFromEnv("COIN_SPAWN_INTERVAL_MS", 2000),

    MATCH_DURATION_MS: intFromEnv("MATCH_DURATION_MS", 30000),
    LEADERBOARD_LIMIT: intFromEnv("LEADERBOARD_LIMIT", 10),
} as const;

Every module in the backend that needs one of these values imports config from this file rather than touching process.env itself. MONGO_URI and MONGO_DB_NAME are the one exception. They stay in the MongoDB connection module we're about to write, since they're only ever read once, at connection time.

Connecting to MongoDB with a Cached Client

Opening a new MongoClient for every request is a common mistake. It adds connection latency to every single call and makes a clean shutdown much harder than it needs to be. A long-running server like this one should connect once when the process starts and reuse that same client for the lifetime of the process.

Create backend/libs/mongodb.ts:

import { MongoClient, Db } from "mongodb";

let client: MongoClient | null = null;
let db: Db | null = null;

export async function connectToMongoDB(): Promise<void> {
    if (client) return;

    const uri = process.env.MONGO_URI;
    const dbName = process.env.MONGO_DB_NAME;

    if (!uri || !dbName) {
        throw new Error("MONGO_URI and MONGO_DB_NAME must be set in environment");
    }

    client = new MongoClient(uri, { appName: "devrel-tutorial-typescript-multiplayergame" });

    await client.connect();
    db = client.db(dbName);

    console.log(`Connected to MongoDB — database: "${dbName}"`);
}

The module-level client and db variables start out null. connectToMongoDB returns immediately if a client already exists, so calling it more than once is harmless. Otherwise, it reads the two required environment variables, throws if either is missing, and connects.

The rest of the codebase never calls connectToMongoDB more than once, but it does need a way to get at the cached Db handle and a way to close the connection on shutdown:

export function getDb(): Db {
    if (!db) {
        throw new Error("MongoDB not connected. Call connectToMongoDB() first.");
    }
    return db;
}

export async function closeMongoDB(): Promise<void> {
    if (client) {
        await client.close();
        client = null;
        db = null;
    }
}

getDb is the function that every route and every socket handler calls to access the database. It throws if it's called before the connection is established, which should never happen in practice since we connect before the server starts listening. closeMongoDB exists for graceful shutdown, though this demo doesn't wire up signal handlers for it. Feel free to add SIGTERM/SIGINT handlers that call it if you want the server to shut down cleanly.

Defining the Player and Match Data Models

Before writing any routes, it's worth deciding what a player document and a match document actually look like, since TypeScript won't infer that for us. Both the database models and the socket payload shapes live in one file because the backend and frontend are separate npm projects with no shared package. The frontend keeps its own copy of these same types in frontend/src/types.ts, and the two have to be kept in sync by hand whenever a payload shape changes.

Create backend/types/index.ts, starting with the two collections that persist to MongoDB:

import { ObjectId } from "mongodb";

export interface Player {
    _id?: ObjectId;
    username: string;
    high_score: number;
    total_matches_played: number;
    total_wins: number;
    created_at: Date;
}

export type MatchStatus = "waiting" | "in_progress" | "finished";

export interface MatchPlayer {
    username: string;
    score: number;
}

export interface Match {
    _id?: ObjectId;
    status: MatchStatus;
    player_1: MatchPlayer;
    player_2: MatchPlayer | null;
    winner: string | null;
    created_at: Date;
    ended_at: Date | null;
}

A Player document tracks a running high_score alongside lifetime stats. A Match document embeds both players directly, rather than referencing them by ID, since a match only ever needs a username and a score for each side, and there is no reason to $lookup back into the players collection to render a scoreboard. player_2 starts out null because a match is created by whoever becomes player_1, and the second player joins later over a socket.

The rest of the file describes objects that never touch MongoDB. They only exist in memory while a match is being played, and as payloads on the wire between client and server:

export interface Coin {
    id: string;
    x: number;
    y: number;
}

export interface PlayerGameState {
    username: string;
    socketId: string;
    x: number;
    y: number;
    score: number;
}

export type GameStatus = "waiting" | "in_progress" | "finished";

export interface GameState {
    roomId: string;
    player1: PlayerGameState | null;
    player2: PlayerGameState | null;
    coins: Map<string, Coin>;
    status: GameStatus;
    matchTimer: ReturnType<typeof setTimeout> | null;
    coinSpawnInterval: ReturnType<typeof setInterval> | null;
}

GameState is the live, in-memory representation of a single match while it's being played. It's keyed by room ID and lives entirely in the Node.js process's memory, not in MongoDB. We'll build the module that manages these next.

Finally, the socket event payload types. Client-to-server payloads describe what a player's browser can ask the server to do:

export interface RoomJoinPayload {
    roomId: string;
    username: string;
}

export type MoveDirection = "up" | "down" | "left" | "right";

export interface PlayerMovePayload {
    roomId: string;
    direction: MoveDirection;
}

export interface CoinCollectPayload {
    roomId: string;
    coinId: string;
    playerX: number;
    playerY: number;
}

And server-to-client payloads describe what the server broadcasts back:

export interface GameStartPayload {
    players: {
        player1: string;
        player2: string;
    };
    coins: Coin[];
    durationMs: number;
    canvasWidth: number;
    canvasHeight: number;
    playerSize: number;
    coinSize: number;
}

export interface Scores {
    player1: { username: string; score: number };
    player2: { username: string; score: number };
}

export interface GameEndPayload {
    scores: Scores;
    winner: string | null;
}

Notice that GameStartPayload hands the client the canvas dimensions, player size, and coin size straight from the server's config. The frontend never hard-codes these values. It sizes its Phaser canvas and its square textures based entirely on what the server tells it, which means changing PLAYER_SIZE in the backend's .env file is the only change needed to resize every player square in the game.

Building the REST Layer

The REST surface for this game is deliberately small. Everything about playing a match, from joining a room to moving to collecting coins, happens over sockets. REST only handles the things that need to survive outside of any one match: registering a player, creating a match, and reading the leaderboard.

Create backend/routes.ts, starting with the username validation schema shared by two of the three endpoints:

import { Router, Request, Response } from "express";
import { z } from "zod";
import { config } from "./config";
import { getDb } from "./libs/mongodb";
import { Match, Player } from "./types/index";

const router = Router();

const UsernameSchema = z.object({
    username: z.string().min(1).max(20),
});

The schema caps a username at 20 characters, which matches the maxlength attribute on the frontend's username input. Keeping those two limits in sync means a player never sees a client-side character limit that's more permissive than what the server will actually accept.

The first endpoint registers a new player:

router.post("/players", async (req: Request, res: Response) => {
    const parsed = UsernameSchema.safeParse(req.body);
    if (!parsed.success) {
        res.status(400).json({ error: parsed.error.flatten() });
        return;
    }

    const { username } = parsed.data;

    try {
        const db = getDb();

        const existing = await db.collection<Player>("players").findOne({ username });
        if (existing) {
            res.status(409).json({ error: "Username already exists." });
            return;
        }

        const player: Player = {
            username,
            high_score: 0,
            total_matches_played: 0,
            total_wins: 0,
            created_at: new Date(),
        };

        const result = await db.collection<Player>("players").insertOne(player);
        res.status(201).json({ _id: result.insertedId, ...player });
    } catch (err) {
        console.error("POST /players", err);
        res.status(500).json({ error: "Internal server error" });
    }
});

There is no authentication in this demo. A username is the entire identity model, so the endpoint runs a findOne before inserting and responds with a 409 if that username is already taken. This is a small enough project that the race condition between the check and the insert is not worth guarding against with a unique index, though you'd want one in anything beyond a tutorial.

Next is the leaderboard, a read-only endpoint that sorts by high_score:

router.get("/players/leaderboard", async (_req: Request, res: Response) => {
    try {
        const db = getDb();

        const players = await db
            .collection<Player>("players")
            .find({})
            .sort({ high_score: -1 })
            .limit(config.LEADERBOARD_LIMIT)
            .project({ username: 1, high_score: 1, total_wins: 1, total_matches_played: 1 })
            .toArray();

        res.json(players);
    } catch (err) {
        console.error("GET /players/leaderboard", err);
        res.status(500).json({ error: "Internal server error" });
    }
});

Sorting descending by high_score and limiting to config.LEADERBOARD_LIMIT gives us the top N players. The .project() call trims the response down to only the fields the lobby actually renders, which keeps the payload small and avoids exposing anything we don't need to.

The last REST endpoint creates a match:

router.post("/rooms", async (req: Request, res: Response) => {
    const parsed = UsernameSchema.safeParse(req.body);
    if (!parsed.success) {
        res.status(400).json({ error: parsed.error.flatten() });
        return;
    }

    const { username } = parsed.data;

    try {
        const db = getDb();

        const player = await db.collection<Player>("players").findOne({ username });
        if (!player) {
            res.status(404).json({ error: "Player not found. Create a player first." });
            return;
        }

        const match: Match = {
            status: "waiting",
            player_1: { username, score: 0 },
            player_2: null,
            winner: null,
            created_at: new Date(),
            ended_at: null,
        };

        const result = await db.collection<Match>("matches").insertOne(match);

        res.status(201).json({ _id: result.insertedId, ...match });
    } catch (err) {
        console.error("POST /rooms", err);
        res.status(500).json({ error: "Internal server error" });
    }
});

export default router;

Whoever calls this endpoint becomes player_1, and the match starts in the waiting state with player_2 set to null. The response's _id is the room ID. That value, and nothing else, is what a second player needs to join the match. There is no lobby list and no way to browse open matches. You either have the ID, or you don't.

Wiring Express and Socket.io Together in main.ts

With the routes and the database module ready, we can assemble the actual server. The important detail here is that Socket.io does not run alongside Express. It attaches to the same underlying HTTP server, so both HTTP requests and WebSocket upgrades are served from one listening port.

Create backend/main.ts:

import express from "express";
import http from "http";
import cors from "cors";
import { Server } from "socket.io";
import { config } from "./config";
import { connectToMongoDB, getDb } from "./libs/mongodb";
import { registerGameSockets } from "./game/socket";
import apiRouter from "./routes";

const { PORT, FRONTEND_URL } = config;

async function bootstrap(): Promise<void> {
    await connectToMongoDB();

    const app = express();

    app.use(cors({ origin: FRONTEND_URL }));
    app.use(express.json());

    app.use(apiRouter);

    const httpServer = http.createServer(app);

    const io = new Server(httpServer, {
        cors: {
            origin: FRONTEND_URL,
            methods: ["GET", "POST"],
        },
    });

    registerGameSockets(io, getDb());

    httpServer.listen(PORT, () => {
        console.log(`Server listening on http://localhost:${PORT}`);
    });
}

bootstrap().catch((err) => {
    console.error("Failed to start server:", err);
    process.exit(1);
});

bootstrap connects to MongoDB before anything else, which means if the database is unreachable, the process fails fast instead of accepting HTTP requests it can't actually serve. From there, it builds the Express app, wraps it in a plain http.createServer, and hands that same server instance to new Server(httpServer, ...) from Socket.io. Both CORS configurations, one for Express and one for Socket.io, point at FRONTEND_URL from the config, since the frontend runs on a different origin during development.

The one function we haven't written yet is registerGameSockets. That's where all the actual game logic lives, and it's big enough to deserve its own module, along with a couple of helpers.

Managing In-Memory Game State

A match's live state, who's playing, where they are, what coins exist, changes dozens of times a second while a match is in progress. You could batch those writes to MongoDB every few hundred milliseconds, or only persist them at meaningful checkpoints, and either approach is worth reaching for once a room's state needs to survive a server restart. For this tutorial, we'll take the simpler route: keep one GameState object per active room in a plain in-memory Map, and commit to MongoDB at exactly the two moments that matter, when a match starts and when it ends.

Create backend/game/state.ts:

import { config } from "../config";
import { GameState, MoveDirection, PlayerGameState } from "../types/index";

const gameRooms = new Map<string, GameState>();

export function createGameState(roomId: string): GameState {
    const state: GameState = {
        roomId,
        player1: null,
        player2: null,
        coins: new Map(),
        status: "waiting",
        matchTimer: null,
        coinSpawnInterval: null,
    };
    gameRooms.set(roomId, state);
    return state;
}

export function getGameState(roomId: string): GameState | undefined {
    return gameRooms.get(roomId);
}

export function deleteGameState(roomId: string): void {
    gameRooms.delete(roomId);
}

The room ID, which is the string form of the match's MongoDB _id, is the key into gameRooms. createGameState seeds a fresh room with no players and no coins. deleteGameState is called once a match finishes, so a process that runs for a long time doesn't accumulate state for matches that ended hours ago.

Next, a couple of functions for placing and identifying players:

export function getStartingPosition(
    slot: 1 | 2
): { x: number; y: number } {
    const { CANVAS_WIDTH, CANVAS_HEIGHT, PLAYER_SIZE } = config;

    if (slot === 1) {
        return { x: PLAYER_SIZE * 2, y: Math.floor(CANVAS_HEIGHT / 2) };
    }
    return { x: CANVAS_WIDTH - PLAYER_SIZE * 3, y: Math.floor(CANVAS_HEIGHT / 2) };
}

export function addPlayerToState(
    state: GameState,
    slot: 1 | 2,
    username: string,
    socketId: string
): PlayerGameState {
    const { x, y } = getStartingPosition(slot);
    const playerState: PlayerGameState = { username, socketId, x, y, score: 0 };

    if (slot === 1) {
        state.player1 = playerState;
    } else {
        state.player2 = playerState;
    }

    return playerState;
}

export function getPlayerBySocketId(
    state: GameState,
    socketId: string
): PlayerGameState | null {
    if (state.player1?.socketId === socketId) return state.player1;
    if (state.player2?.socketId === socketId) return state.player2;
    return null;
}

Player 1 always spawns near the left edge and player 2 near the right edge, both vertically centered, based on the canvas dimensions and player size from config. The frontend runs the same calculation independently, so a player's sprite appears in the right spot before the very first movement broadcast arrives.

getPlayerBySocketId is worth pausing on, because it is the mechanism that keeps a client from acting as a player it doesn't own. When a movement or coin-collect event comes in, the server never trusts a username sent in the payload. It looks up which player is tied to the socket that sent the event, and only that player's state can change as a result. A client has no way to move the other player or claim the other player's coins, because there's no code path that lets a payload name which player it's acting as.

The last two functions are where the server enforces two invariants the game depends on: players can't leave the canvas, and a coin can't be claimed from across the screen.

export function applyMovement(
    player: PlayerGameState,
    direction: MoveDirection
): { x: number; y: number } {
    const { CANVAS_WIDTH, CANVAS_HEIGHT, PLAYER_SIZE, PLAYER_SPEED } = config;

    let { x, y } = player;

    switch (direction) {
        case "up":    y -= PLAYER_SPEED; break;
        case "down":  y += PLAYER_SPEED; break;
        case "left":  x -= PLAYER_SPEED; break;
        case "right": x += PLAYER_SPEED; break;
    }

    x = Math.max(0, Math.min(CANVAS_WIDTH - PLAYER_SIZE, x));
    y = Math.max(0, Math.min(CANVAS_HEIGHT - PLAYER_SIZE, y));

    player.x = x;
    player.y = y;

    return { x, y };
}

applyMovement takes one step of PLAYER_SPEED pixels in the requested direction, then clamps the result with Math.max/Math.min so it can never fall outside [0, CANVAS_WIDTH - PLAYER_SIZE] or [0, CANVAS_HEIGHT - PLAYER_SIZE]. This is the only function in the entire codebase that changes a player's position, and it always writes the clamped result back onto the player state before returning it. There is no way for a player to end up outside the canvas, no matter how fast someone mashes the arrow keys.

export function validateCoinCollect(
    player: PlayerGameState,
    coinX: number,
    coinY: number
): boolean {
    const { PLAYER_SIZE, COIN_SIZE, PLAYER_SPEED } = config;

    const tolerance = PLAYER_SPEED;

    const playerRight  = player.x + PLAYER_SIZE + tolerance;
    const playerBottom = player.y + PLAYER_SIZE + tolerance;
    const playerLeft   = player.x - tolerance;
    const playerTop    = player.y - tolerance;

    const coinRight  = coinX + COIN_SIZE;
    const coinBottom = coinY + COIN_SIZE;

    return (
        playerLeft   < coinRight  &&
        playerRight  > coinX      &&
        playerTop    < coinBottom &&
        playerBottom > coinY
    );
}

This is a standard axis-aligned bounding box overlap test between the player's square and the coin's square, padded on every side by one PLAYER_SPEED unit of tolerance. The client is the one that first notices an overlap, using Phaser's own physics, but the server never takes that claim at face value. It re-runs its own overlap check using the player's server-tracked position before letting a coin count. The tolerance exists solely to absorb the small amount of latency between when a client detects the overlap and when the server validates it. Without it, a coin sitting right at the edge of a fast-moving player's hitbox could get rejected by a matter of a pixel or two.

Spawning Coins on a Timer

Coins need to appear at random positions, capped at some maximum on screen at once, on a fixed interval, and only while a match is actually in progress. This is small enough to live in its own module.

Create backend/game/coins.ts:

import { randomUUID } from "crypto";
import { Server } from "socket.io";
import { config } from "../config";
import { Coin, GameState } from "../types/index";

function randomInt(min: number, max: number): number {
    return Math.floor(Math.random() * (max - min + 1)) + min;
}

export function generateCoin(): Coin {
    const { CANVAS_WIDTH, CANVAS_HEIGHT, COIN_SIZE } = config;

    return {
        id: randomUUID(),
        x: randomInt(0, CANVAS_WIDTH - COIN_SIZE),
        y: randomInt(0, CANVAS_HEIGHT - COIN_SIZE),
    };
}

generateCoin gives every coin a fresh UUID as its ID, which is how the client and server refer to a specific coin across every event related to it. The random x and y are bounded by COIN_SIZE, so the square is always drawn fully inside the canvas and never hangs off an edge.

The spawner itself runs on a setInterval scoped to a single room's GameState:

export function startCoinSpawner(
    io: Server,
    state: GameState
): void {
    state.coinSpawnInterval = setInterval(() => {
        if (state.status !== "in_progress") {
            stopCoinSpawner(state);
            return;
        }

        if (state.coins.size >= config.COIN_MAX_ON_SCREEN) return;

        const coin = generateCoin();
        state.coins.set(coin.id, coin);
        io.to(state.roomId).emit("coin:spawned", coin);
    }, config.COIN_SPAWN_INTERVAL_MS);
}

export function stopCoinSpawner(state: GameState): void {
    if (state.coinSpawnInterval !== null) {
        clearInterval(state.coinSpawnInterval);
        state.coinSpawnInterval = null;
    }
}

Every tick, it checks that the match is still in_progress before doing anything else, since an interval that outlives its match would keep spawning coins into a room nobody is playing in anymore. If the room is already at COIN_MAX_ON_SCREEN, the tick is a no-op. Otherwise, it generates a coin, stores it in the room's coins map, and emits coin:spawned to everyone in that Socket.io room. Notice that state.coins (the server's map) is updated before the broadcast goes out, which matters because that map is the server's own record of what's currently on screen, and it's what coin:collect checks against later.

Handling the Real-Time Game Loop

This is the heart of the game. Every socket event that a player's browser can send is registered here, and this is also where a match's ending gets decided and persisted back to MongoDB.

Create backend/game/socket.ts, starting with the imports and the function that ends a match:

import { Server, Socket } from "socket.io";
import { Db, ObjectId } from "mongodb";
import { config } from "../config";
import {
    RoomJoinPayload,
    PlayerMovePayload,
    CoinCollectPayload,
    GameStartPayload,
    GameEndPayload,
    Match,
} from "../types/index";
import {
    createGameState,
    getGameState,
    deleteGameState,
    addPlayerToState,
    getPlayerBySocketId,
    applyMovement,
    validateCoinCollect,
} from "./state";
import { startCoinSpawner, stopCoinSpawner } from "./coins";

async function endMatch(
    io: Server,
    db: Db,
    roomId: string
): Promise<void> {
    const state = getGameState(roomId);
    if (!state || state.status === "finished") return;

    state.status = "finished";
    stopCoinSpawner(state);

    if (state.matchTimer !== null) {
        clearTimeout(state.matchTimer);
        state.matchTimer = null;
    }

The early if (!state || state.status === "finished") return; guard is what makes endMatch safe to call exactly once, no matter what triggers it. In this game, it's only ever called by the match duration timer, but guarding against a second call costs nothing and rules out a whole class of bugs if you ever add another way to end a match, like a forfeit button.

With the room marked finished and its timers cleared, endMatch figures out the winner and writes the result to MongoDB:

    const p1 = state.player1!;
    const p2 = state.player2!;

    const winner =
        p1.score > p2.score
            ? p1.username
            : p2.score > p1.score
            ? p2.username
            : null; // draw

    const endedAt = new Date();

    await db.collection<Match>("matches").updateOne(
        { _id: new ObjectId(roomId) },
        {
            $set: {
                status: "finished",
                "player_1.score": p1.score,
                "player_2.score": p2.score,
                winner,
                ended_at: endedAt,
            },
        }
    );

The non-null assertions on state.player1! and state.player2! are safe here because endMatch is only ever scheduled by a timer that gets armed after both player slots are already filled, which we'll see shortly. The updateOne call uses dot notation to update player_1.score and player_2.score inside the embedded documents without having to reconstruct the whole Match document, and a null winner represents a draw.

The last thing endMatch does is update player stats and notify both clients:

    for (const { username, score } of [
        { username: p1.username, score: p1.score },
        { username: p2.username, score: p2.score },
    ]) {
        const isWinner = username === winner;
        await db.collection("players").updateOne(
            { username },
            {
                $inc: {
                    total_matches_played: 1,
                    total_wins: isWinner ? 1 : 0,
                },
                $max: { high_score: score },
            }
        );
    }

    const payload: GameEndPayload = {
        scores: {
            player1: { username: p1.username, score: p1.score },
            player2: { username: p2.username, score: p2.score },
        },
        winner,
    };

    io.to(roomId).emit("game:end", payload);
    deleteGameState(roomId);
}

$inc bumps total_matches_played for both players and total_wins for whichever one won. The $max operator is doing quiet but important work here: it only writes high_score if the new value is actually greater than what's already stored. That's the entire "update high score if this match beats it" rule from a single atomic operator, with no read-then-write round trip and no race condition if a player somehow finished two matches back to back. Once the database is updated, the server emits game:end to the room and calls deleteGameState to drop the room's in-memory state for good.

Now for the part of backend/game/socket.ts that registers every socket event. Start with the connection handler and room:join:

export function registerGameSockets(io: Server, db: Db): void {
    io.on("connection", (socket: Socket) => {
        console.log(`Socket connected: ${socket.id}`);

        socket.on("room:join", async (payload: RoomJoinPayload) => {
            const { roomId, username } = payload;

            try {
                let matchDoc = await db
                    .collection<Match>("matches")
                    .findOne({ _id: new ObjectId(roomId) });

                if (!matchDoc) {
                    socket.emit("error", { message: "Room not found." });
                    return;
                }

                if (matchDoc.status === "finished") {
                    socket.emit("error", { message: "This match has already ended." });
                    return;
                }

room:join is where a socket connection turns into an actual seat in a match. It starts by loading the match document from MongoDB using the room ID, which is really just the match's _id as a string. A missing document or a match that has already finished both get rejected immediately with an error event back to that one socket.

Next, the handler works out which slot, player 1 or player 2, the joining username belongs in:

                let slot: 1 | 2;

                if (matchDoc.player_1.username === username) {
                    slot = 1;
                } else if (!matchDoc.player_2) {
                    await db.collection<Match>("matches").updateOne(
                        { _id: new ObjectId(roomId) },
                        { $set: { player_2: { username, score: 0 } } }
                    );
                    slot = 2;
                } else if (matchDoc.player_2.username === username) {
                    slot = 2;
                } else {
                    socket.emit("error", { message: "Room is full." });
                    return;
                }

If the username matches player_1, that's slot 1, most likely the creator of the match, reconnecting. If there's no player_2 yet, this socket becomes player 2, and that assignment is persisted to MongoDB right away, so a second join attempt (say, from a page refresh) recognizes the same username as already seated in slot 2 rather than trying to add a third player. Anyone else gets turned away with "Room is full."

With a slot decided, the handler joins the actual Socket.io room and updates the in-memory game state:

                socket.join(roomId);

                let state = getGameState(roomId);
                if (!state) {
                    state = createGameState(roomId);
                }

                addPlayerToState(state, slot, username, socket.id);

                socket.emit("room:joined", { slot, roomId });

socket.join(roomId) is Socket.io's own room feature, and it's what makes io.to(roomId).emit(...) reach exactly the two sockets seated in this match and nobody else. getGameState/createGameState either finds the existing in-memory state for this room or creates it on the first join. addPlayerToState then places the player's starting position and registers their socket.id, which is the identity check every later movement and coin-collect event relies on.

The last piece of room:join is starting the actual match once both seats are filled:

                if (state.player1 && state.player2 && state.status === "waiting") {
                    state.status = "in_progress";

                    await db.collection<Match>("matches").updateOne(
                        { _id: new ObjectId(roomId) },
                        { $set: { status: "in_progress" } }
                    );

                    const startPayload: GameStartPayload = {
                        players: {
                            player1: state.player1.username,
                            player2: state.player2.username,
                        },
                        coins: [],
                        durationMs: config.MATCH_DURATION_MS,
                        canvasWidth: config.CANVAS_WIDTH,
                        canvasHeight: config.CANVAS_HEIGHT,
                        playerSize: config.PLAYER_SIZE,
                        coinSize: config.COIN_SIZE,
                    };

                    io.to(roomId).emit("game:start", startPayload);

                    startCoinSpawner(io, state);

                    state.matchTimer = setTimeout(
                        () => endMatch(io, db, roomId),
                        config.MATCH_DURATION_MS
                    );
                }
            } catch (err) {
                console.error("room:join error", err);
                socket.emit("error", { message: "Failed to join room." });
            }
        });

The state.status === "waiting" check ensures this block only ever runs once per match, even though room:join can fire multiple times as players reconnect. Once both slots are filled, the match flips to in_progress in both the in-memory state and MongoDB, game:start goes out to the room with every piece of sizing configuration the frontend needs, the coin spawner starts, and the match duration timer that eventually calls endMatch gets armed. This is also the only place that a timer is ever created, which is why endMatch can safely assume both players exist.

With joining handled, the movement and coin collection handlers are much shorter:

        socket.on("player:move", (payload: PlayerMovePayload) => {
            const { roomId, direction } = payload;
            const state = getGameState(roomId);

            if (!state || state.status !== "in_progress") return;

            const player = getPlayerBySocketId(state, socket.id);
            if (!player) return;

            const { x, y } = applyMovement(player, direction);

            io.to(roomId).emit("player:moved", { username: player.username, x, y });
        });

player:move never trusts the payload beyond a roomId and a direction. It resolves which player is actually pressing the key by looking up the calling socket's ID, applies the movement (which, as we saw earlier, always clamps to the canvas), and broadcasts the new authoritative position to the whole room. Neither client moves its local sprite in response to a key press. Both clients wait for this exact player:moved broadcast, including the client that pressed the key.

        socket.on("coin:collect", async (payload: CoinCollectPayload) => {
            const { roomId, coinId, playerX, playerY } = payload;
            const state = getGameState(roomId);

            if (!state || state.status !== "in_progress") return;

            const coin = state.coins.get(coinId);
            if (!coin) return;

            const player = getPlayerBySocketId(state, socket.id);
            if (!player) return;

            player.x = playerX;
            player.y = playerY;

            if (!validateCoinCollect(player, coin.x, coin.y)) {
                return;
            }

            state.coins.delete(coinId);
            player.score += 1;

            io.to(roomId).emit("coin:collected", {
                coinId,
                username: player.username,
                scores: {
                    player1: { username: state.player1!.username, score: state.player1!.score },
                    player2: { username: state.player2!.username, score: state.player2!.score },
                },
            });
        });

coin:collect is a claim, not a command, and the handler treats it that way. It first checks that the coin is still on the server's coins map, since a coin that's already been claimed by the other player (or never existed) is silently ignored. It resolves the calling player the same way player:move does, syncs the player's server-tracked position to whatever the client reported, and only then runs validateCoinCollect. If that check fails, meaning the claim doesn't line up with where the server thinks the player actually is, the event is dropped without any response at all. A bogus claim gets no feedback, not even an error, because there's nothing useful to tell a client that's either buggy or trying to cheat. Only a validated claim deletes the coin from the map, increments the score, and broadcasts coin:collected with the updated scores for both players.

The last handler just logs a disconnect and does nothing further:

        socket.on("disconnect", () => {
            console.log(`Socket disconnected: ${socket.id}`);
        });
    });
}

Per the design of this demo, a disconnect doesn't end the match or award an automatic win. If the match timer is still running, it will fire on schedule and call endMatch with whatever scores existed at that point. That keeps the disconnect logic dead simple, at the cost of a player being able to leave a match hanging until the clock runs out. For a tutorial, that trade-off is fine.

That's the entire backend. It's worth running npm run dev at this point just to confirm the server boots and connects to MongoDB before moving on to the frontend.

Setting Up the Frontend Project

Inside frontend, scaffold a Vite project configured for vanilla TypeScript:

npm create vite@latest . -- --template vanilla-ts
npm install phaser socket.io-client axios

Phaser is the game engine that renders the canvas and handles sprites, input, and overlap detection. socket.io-client is the browser counterpart to the backend's Socket.io server, and axios handles the three REST calls to the backend.

Create a frontend/.env file pointing at the backend:

VITE_API_URL=http://localhost:3000
VITE_SOCKET_URL=http://localhost:3000

Vite only exposes environment variables prefixed with VITE_ to client-side code, which is why both variables use that prefix. Both point at the same backend here, since the REST API and the Socket.io server are on the same Express process, but they're kept as separate variables in case you ever split them apart.

Defining the Page Markup and Styles

The Vite scaffold generated a frontend/index.html and a frontend/src/style.css filled with starter content, along with a few demo files that we won't need. You can delete the demo files and replace the contents of both of these files. All three screens of the game, the lobby, the match, and the results, live in this single HTML page. Our TypeScript code will show and hide them as the player moves through the game, so there's no need for a routing library.

In the <head> of frontend/index.html, the only change worth making is setting the <title> to "Coin Race". Then replace the <body> with the following, starting with the lobby:

<body>
    <div id="app">
        <section id="lobby" class="screen">
            <h1>Coin Race</h1>

            <div id="lobby-register">
                <label for="username">Username</label>
                <input id="username" type="text" maxlength="20" placeholder="Enter a username" />
                <button id="register-btn" type="button">Register</button>
                <p id="lobby-error" class="error hidden"></p>
            </div>

            <div id="lobby-actions" class="hidden">
                <p id="welcome-message"></p>

                <div class="action-row">
                    <button id="create-match-btn" type="button">Create Match</button>
                </div>

                <div class="action-row">
                    <input id="room-id-input" type="text" placeholder="Room ID" />
                    <button id="join-match-btn" type="button">Join Match</button>
                </div>
            </div>

            <div id="lobby-waiting" class="hidden">
                <p>Waiting for opponent...</p>
                <p>Share this Room ID: <strong id="waiting-room-id"></strong></p>
            </div>

            <div id="leaderboard">
                <h2>Leaderboard</h2>
                <ol id="leaderboard-list"></ol>
            </div>
        </section>

The lobby has three states, and only one of them is visible at a time. The lobby-register block is where a player enters a username, with lobby-error reserved for problems such as a username that's already taken. Once registered, the player sees lobby-actions, where they can either create a match or paste in a room ID to join one. The player who creates a match is moved to lobby-waiting, which displays the room ID they need to share with their opponent. Everything but the registration form starts with the hidden class. The leaderboard-list element is an empty list that gets filled in from the leaderboard endpoint.

Below the lobby, add the game and results screens:

        <section id="game-screen" class="screen hidden">
            <div id="hud">
                <span id="hud-player1"></span>
                <span id="hud-timer"></span>
                <span id="hud-player2"></span>
            </div>
            <div id="game-container"></div>
        </section>

        <section id="results" class="screen hidden">
            <h1>Match Over</h1>
            <p id="results-winner"></p>
            <p id="results-scores"></p>
            <button id="play-again-btn" type="button">Play Again</button>
        </section>
    </div>
    <script type="module" src="/src/main.ts"></script>
</body>

The hud element holds the score for each player on either side of the match timer, all of which the game scene updates as events come in from the server. The game-container element is empty on purpose. It's where Phaser injects its canvas when a match starts. The results screen has a spot for the winner, a spot for the final scores, and a button to go back to the lobby. Both of these screens start hidden. The <script> tag at the bottom is how Vite loads frontend/src/main.ts as the entry point for the application.

Next, replace the contents of frontend/src/style.css, starting with the base page styles:

:root {
    color-scheme: dark;
    font: 16px/1.4 system-ui, sans-serif;
    color: #e8e8ec;
    background: #14151c;
}

* {
    box-sizing: border-box;
}

body {
    margin: 0;
    min-height: 100vh;
    display: flex;
    align-items: center;
    justify-content: center;
}

#app {
    width: 100%;
    display: flex;
    align-items: center;
    justify-content: center;
}

.screen {
    width: 100%;
    max-width: 480px;
    padding: 32px;
    text-align: center;
}

.hidden {
    display: none;
}

The page uses a dark color scheme, and both body and app use flexbox to center their content, which is what keeps the 1920x1080 game canvas centered in the browser tab. Each screen is a narrow, centered column by default. The most important rule here is .hidden, because the entire screen switching approach depends on it. It's used for the three screens, as well as the smaller pieces inside the lobby, like the registration form, the match actions, the waiting message, and the error message. Adding or removing the hidden class is how we'll move between screens and between the states of the lobby.

Add the styles for the form controls and the lobby:

h1 {
    margin: 0 0 24px;
}

label {
    display: block;
    margin-bottom: 8px;
    color: #a8a8b3;
}

input[type="text"] {
    width: 100%;
    padding: 10px 12px;
    margin-bottom: 12px;
    border-radius: 6px;
    border: 1px solid #3a3b46;
    background: #1e1f29;
    color: #e8e8ec;
    font-size: 15px;
}

button {
    padding: 10px 18px;
    border-radius: 6px;
    border: none;
    background: #4c6ef5;
    color: #fff;
    font-size: 15px;
    cursor: pointer;
}

button:hover {
    background: #3b5bdb;
}

.action-row {
    display: flex;
    gap: 8px;
    margin-bottom: 12px;
}

.action-row input {
    margin-bottom: 0;
}

.error {
    color: #ff6b6b;
}

#leaderboard {
    margin-top: 32px;
    text-align: left;
}

#leaderboard-list {
    padding-left: 24px;
}

#lobby-waiting strong {
    color: #4c6ef5;
}

Most of this is cosmetic styling for inputs and buttons. The two classes worth pointing out are action-row, which places the room ID input and the join button side by side, and error, which colors the lobby's error message red. The waiting room ID is also highlighted, so it's easy to spot and share.

Finally, add the styles for the game and results screens:

/* --- Game screen --- */

#game-screen {
    max-width: none;
    padding: 16px;
}

#hud {
    width: 1920px;
    max-width: 100%;
    margin: 0 auto 8px;
    display: flex;
    justify-content: space-between;
    font-size: 20px;
    font-weight: 600;
}

#game-container {
    display: flex;
    justify-content: center;
}

/* --- Results screen --- */

#results-scores {
    font-size: 18px;
    margin-bottom: 24px;
}

The game screen removes the 480px limit that every other screen has, since the canvas is much wider. The hud is set to the same 1920px width as the canvas, so the scores line up with its left and right edges, with the timer sitting in the middle. With the page in place, we can move on to the TypeScript that drives it.

Mirroring the Backend's Types on the Frontend

Since there's no shared package between the two projects, the frontend needs its own copy of every type that describes a REST response or a socket payload. Create frontend/src/types.ts starting with the REST models:

export interface Player {
    _id: string;
    username: string;
    high_score: number;
    total_matches_played: number;
    total_wins: number;
    created_at: string;
}

export interface Match {
    _id: string;
    status: "waiting" | "in_progress" | "finished";
    player_1: { username: string; score: number };
    player_2: { username: string; score: number } | null;
    winner: string | null;
    created_at: string;
    ended_at: string | null;
}

export interface LeaderboardEntry {
    username: string;
    high_score: number;
    total_wins: number;
    total_matches_played: number;
}

These mirror the backend's Player and Match interfaces field for field, with one difference: dates come across JSON as strings, so created_at and ended_at are typed as string here instead of Date.

Next, the game objects and socket payloads, which need to match the backend's versions exactly, since they describe the wire format both sides agree on:

export interface Coin {
    id: string;
    x: number;
    y: number;
}

export type MoveDirection = "up" | "down" | "left" | "right";

export interface Scores {
    player1: { username: string; score: number };
    player2: { username: string; score: number };
}

export interface RoomJoinedPayload {
    slot: 1 | 2;
    roomId: string;
}

export interface GameStartPayload {
    players: {
        player1: string;
        player2: string;
    };
    coins: Coin[];
    durationMs: number;
    canvasWidth: number;
    canvasHeight: number;
    playerSize: number;
    coinSize: number;
}

export interface PlayerMovedPayload {
    username: string;
    x: number;
    y: number;
}

export interface CoinCollectedPayload {
    coinId: string;
    username: string;
    scores: Scores;
}

export interface GameEndPayload {
    scores: Scores;
    winner: string | null;
}

Finally, two event map interfaces that give socket.on and socket.emit full compile-time checking against these payload types, instead of writing a wrapper function around every single event:

export interface ServerToClientEvents {
    "room:joined": (payload: RoomJoinedPayload) => void;
    "game:start": (payload: GameStartPayload) => void;
    "coin:spawned": (coin: Coin) => void;
    "player:moved": (payload: PlayerMovedPayload) => void;
    "coin:collected": (payload: CoinCollectedPayload) => void;
    "game:end": (payload: GameEndPayload) => void;
    error: (payload: { message: string }) => void;
}

export interface ClientToServerEvents {
    "room:join": (payload: { roomId: string; username: string }) => void;
    "player:move": (payload: { roomId: string; direction: MoveDirection }) => void;
    "coin:collect": (payload: {
        roomId: string;
        coinId: string;
        playerX: number;
        playerY: number;
    }) => void;
}

Anywhere these two generics get passed to Socket<ServerToClientEvents, ClientToServerEvents>, TypeScript will flag a typo in an event name or a wrong payload shape at compile time rather than at runtime in the browser console.

Talking to the REST API

Create frontend/src/api.ts:

import axios from "axios";
import type { LeaderboardEntry, Match, Player } from "./types";

const api = axios.create({
    baseURL: import.meta.env.VITE_API_URL,
});

export async function createPlayer(username: string): Promise<Player | null> {
    try {
        const res = await api.post<Player>("/players", { username });
        return res.data;
    } catch (err) {
        if (axios.isAxiosError(err) && err.response?.status === 409) {
            return null;
        }
        throw err;
    }
}

createPlayer treats a 409 specially. Since there's no authentication in this game, a username that's already taken isn't really an error from the player's point of view, it just means they're a returning player signing back in with the same name. The function swallows that one status code and returns null instead of throwing an exception, and the lobby code we'll write shortly treats null the same way it treats a successful registration.

The other two functions are simpler, since neither has a special case to handle:

export async function createRoom(username: string): Promise<Match> {
    const res = await api.post<Match>("/rooms", { username });
    return res.data;
}

export async function getLeaderboard(): Promise<LeaderboardEntry[]> {
    const res = await api.get<LeaderboardEntry[]>("/players/leaderboard");
    return res.data;
}

createRoom returns the full Match document, and the lobby only cares about its _id, which becomes the room ID shared with the second player. getLeaderboard is a plain GET with no parameters, since LEADERBOARD_LIMIT is decided entirely on the backend.

Setting Up the Socket Connection

Create frontend/src/socket.ts:

import { io, type Socket } from "socket.io-client";
import type { ClientToServerEvents, ServerToClientEvents } from "./types";

export const socket: Socket<ServerToClientEvents, ClientToServerEvents> = io(
    import.meta.env.VITE_SOCKET_URL,
    { autoConnect: false }
);

export function connectSocket(): void {
    if (!socket.connected) socket.connect();
}

The socket is created once, at module load, but with autoConnect: false, so nothing actually opens a connection until a player is about to create or join a room. connectSocket is idempotent, calling it on an already-connected socket does nothing, which means the lobby code doesn't have to track connection state itself.

One more function belongs in this file:

export function removeAllGameListeners(): void {
    socket.off("coin:spawned");
    socket.off("player:moved");
    socket.off("coin:collected");
    socket.off("game:end");
}

The Phaser scene we're about to write subscribes to four in-match events every time a match starts. Since matches are never reused (the design intentionally forces a new match to be created after each one), those listeners need to be torn down when a match ends, or a second match on the same socket connection would stack duplicate handlers, and every event would fire twice. Notice this function deliberately leaves room:joined, game:start, and error alone, since those belong to the lobby and need to stay registered for the next match.

Building the Lobby Screen

The lobby is plain DOM manipulation, no Phaser involved, since the game canvas doesn't exist yet at this point. It's built against the elements we defined in frontend/index.html earlier: a registration form, a create/join section, a waiting state, and a leaderboard list.

Create frontend/src/screens/lobby.ts, starting with the type for the callback the lobby hands control off to once a match actually starts:

import { createPlayer, createRoom, getLeaderboard } from "../api";
import { connectSocket, socket } from "../socket";
import type { GameStartPayload } from "../types";

const USERNAME_STORAGE_KEY = "coin-race-username";

export type MatchReadyHandler = (
    roomId: string,
    mySlot: 1 | 2,
    startPayload: GameStartPayload
) => void;

function joinRoom(roomId: string, username: string): void {
    connectSocket();
    socket.emit("room:join", { roomId, username });
}

joinRoom is shared by both the "create match" and "join match" flows, since creating a match still means joining the room you just created over the socket. It's the only place in the lobby that opens the socket connection.

The bulk of the file is initLobby, which wires up every DOM element and event listener. Start with the setup:

export function initLobby(onMatchReady: MatchReadyHandler): () => void {
    const usernameInput = document.getElementById("username") as HTMLInputElement;
    const registerBtn = document.getElementById("register-btn") as HTMLButtonElement;
    const lobbyError = document.getElementById("lobby-error") as HTMLParagraphElement;
    const lobbyRegister = document.getElementById("lobby-register") as HTMLDivElement;
    const lobbyActions = document.getElementById("lobby-actions") as HTMLDivElement;
    const lobbyWaiting = document.getElementById("lobby-waiting") as HTMLDivElement;
    const welcomeMessage = document.getElementById("welcome-message") as HTMLParagraphElement;
    const createMatchBtn = document.getElementById("create-match-btn") as HTMLButtonElement;
    const roomIdInput = document.getElementById("room-id-input") as HTMLInputElement;
    const joinMatchBtn = document.getElementById("join-match-btn") as HTMLButtonElement;
    const waitingRoomId = document.getElementById("waiting-room-id") as HTMLElement;
    const leaderboardList = document.getElementById("leaderboard-list") as HTMLOListElement;

    let username = "";
    let pendingRoomId: string | null = null;
    let pendingSlot: 1 | 2 | null = null;

    const savedUsername = localStorage.getItem(USERNAME_STORAGE_KEY);
    if (savedUsername) usernameInput.value = savedUsername;

The three let variables track state across the lobby's lifetime, since this is plain DOM code with no framework managing state for us. username is set once registration succeeds. pendingRoomId and pendingSlot are set as soon as the socket confirms this client has a seat in a room, and both are read the moment game:start arrives. Restoring username from localStorage on load means a returning player doesn't have to retype their name every session.

A few small helper functions keep the event listeners themselves readable:

    function showError(message: string): void {
        lobbyError.textContent = message;
        lobbyError.classList.remove("hidden");
    }

    function clearError(): void {
        lobbyError.classList.add("hidden");
    }

    function resetToActions(): void {
        lobbyWaiting.classList.add("hidden");
        lobbyActions.classList.remove("hidden");
        pendingRoomId = null;
        pendingSlot = null;
    }

    async function loadLeaderboard(): Promise<void> {
        try {
            const entries = await getLeaderboard();
            leaderboardList.innerHTML = entries
                .map((e) => `<li>${e.username} — ${e.high_score}</li>`)
                .join("");
        } catch {
            leaderboardList.innerHTML = "<li>Unable to load leaderboard.</li>";
        }
    }

resetToActions is the escape hatch back to the create/join buttons from the "waiting for opponent" screen, used when a join attempt gets rejected. loadLeaderboard fetches and renders the top players, falling back to a plain message if the request fails rather than leaving the list blank with no explanation.

The registration button ties the username input to createPlayer:

    registerBtn.addEventListener("click", async () => {
        const value = usernameInput.value.trim();
        if (!value) {
            showError("Enter a username.");
            return;
        }

        clearError();
        try {
            await createPlayer(value);
            username = value;
            localStorage.setItem(USERNAME_STORAGE_KEY, username);
            welcomeMessage.textContent = `Welcome, ${username}!`;
            lobbyRegister.classList.add("hidden");
            lobbyActions.classList.remove("hidden");
        } catch {
            showError("Failed to register. Try again.");
        }
    });

Remember that createPlayer returns null rather than throwing on a 409, so a duplicate username still lands in the success branch here and the player moves straight to the create/join screen, exactly as if registration had succeeded for the first time.

Creating and joining a match are two short handlers:

    createMatchBtn.addEventListener("click", async () => {
        clearError();
        try {
            const match = await createRoom(username);
            pendingRoomId = match._id;
            joinRoom(match._id, username);
        } catch {
            showError("Failed to create match.");
        }
    });

    joinMatchBtn.addEventListener("click", () => {
        const roomId = roomIdInput.value.trim();
        if (!roomId) {
            showError("Enter a room ID.");
            return;
        }
        clearError();
        pendingRoomId = roomId;
        joinRoom(roomId, username);
    });

Creating a match calls the REST endpoint first to get a room ID, then immediately joins that same room over the socket. Joining an existing match skips the REST call entirely, since the room already exists, and goes straight to joinRoom with whatever ID was typed in.

The last three listeners handle what comes back from the server:

    socket.on("room:joined", ({ slot, roomId }) => {
        pendingSlot = slot;
        pendingRoomId = roomId;
        waitingRoomId.textContent = roomId;
        lobbyActions.classList.add("hidden");
        lobbyWaiting.classList.remove("hidden");
    });

    socket.on("error", ({ message }) => {
        resetToActions();
        showError(message);
    });

    socket.on("game:start", (payload) => {
        if (!pendingRoomId || !pendingSlot) return;
        onMatchReady(pendingRoomId, pendingSlot, payload);
    });

    void loadLeaderboard();

    return () => {
        clearError();
        resetToActions();
        void loadLeaderboard();
    };
}

room:joined confirms this socket now has a seat, records the slot, and switches to the "waiting for opponent" view, displaying the room ID so it can be shared with a second player. If anything goes wrong on the server's side, the generic error event resets the UI and shows whatever message the server sent. And game:start is the payoff: once both slots are filled and the server broadcasts the start payload, the lobby hands off to onMatchReady with the room ID, this client's slot, and the payload the Phaser scene needs to build itself. Finally, initLobby returns a small function that puts the lobby back into its create/join state and reloads the leaderboard. We'll call it from frontend/src/main.ts when a player comes back from the results screen, so the leaderboard reflects the match that just finished.

Building the Game Scene

This is the biggest single file in the frontend, and the one that most directly reflects the "server decides, client renders" philosophy this game is built around. Nothing in this file changes a score or moves a sprite in response to local input alone. It only forwards input to the server and reacts to what comes back.

Create frontend/src/game/game-scene.ts, starting with the class shape and its init lifecycle hook:

import Phaser from "phaser";
import type { GameEndPayload, GameStartPayload, MoveDirection, Scores } from "../types";
import { removeAllGameListeners, socket } from "../socket";

const MOVE_EMIT_INTERVAL_MS = 60;
const LOCAL_OUTLINE_COLOR = 0x4c6ef5;

export interface GameSceneData {
    roomId: string;
    mySlot: 1 | 2;
    startPayload: GameStartPayload;
    onEnd: (result: GameEndPayload) => void;
}

type PlayerSprite = Phaser.Types.Physics.Arcade.SpriteWithDynamicBody;
type CoinSprite = Phaser.Types.Physics.Arcade.SpriteWithDynamicBody;

export class GameScene extends Phaser.Scene {
    private sceneData!: GameSceneData;
    private player1!: PlayerSprite;
    private player2!: PlayerSprite;
    private localPlayer!: PlayerSprite;
    private coinSprites = new Map<string, CoinSprite>();
    private coinsGroup!: Phaser.Physics.Arcade.Group;
    private cursorKeys!: Phaser.Types.Input.Keyboard.CursorKeys;
    private wasdKeys!: { W: Phaser.Input.Keyboard.Key; A: Phaser.Input.Keyboard.Key; S: Phaser.Input.Keyboard.Key; D: Phaser.Input.Keyboard.Key };
    private moveAccumulator = 0;
    private matchEndAt = 0;

    constructor() {
        super("GameScene");
    }

    init(data: GameSceneData): void {
        this.sceneData = data;
        this.coinSprites.clear();
        this.moveAccumulator = 0;
    }
}

GameSceneData is what main.ts hands the scene when it starts. mySlot is how the scene knows which of the two sprites is the one the local player controls. init runs before create on every scene start and resets per-match state, which matters because Phaser scenes can be restarted, and stale data from a previous match should never leak into a new one.

The create method builds everything the match needs. It starts with the two-player sprites:

    create(): void {
        const { startPayload, mySlot } = this.sceneData;
        const { playerSize, coinSize, canvasWidth, canvasHeight, durationMs } = startPayload;

        this.matchEndAt = Date.now() + durationMs;

        this.generateTextures(playerSize, coinSize);

        const p1Start = this.getStartingPosition(1, canvasWidth, canvasHeight, playerSize);
        const p2Start = this.getStartingPosition(2, canvasWidth, canvasHeight, playerSize);

        this.player1 = this.physics.add.sprite(
            p1Start.x,
            p1Start.y,
            mySlot === 1 ? "player-outline" : "player"
        );
        this.player1.setOrigin(0, 0);
        this.player1.body.setAllowGravity(false);
        this.player1.body.setImmovable(true);

        this.player2 = this.physics.add.sprite(
            p2Start.x,
            p2Start.y,
            mySlot === 2 ? "player-outline" : "player"
        );
        this.player2.setOrigin(0, 0);
        this.player2.body.setAllowGravity(false);
        this.player2.body.setImmovable(true);

        this.localPlayer = mySlot === 1 ? this.player1 : this.player2;

Every sizing value used here, playerSize, coinSize, canvasWidth, canvasHeight, and the match's durationMs, comes straight from the server's startPayload. The local player's sprite gets the "player-outline" texture instead of the plain "player" texture, which is the only visual difference between the two squares, and it's purely cosmetic, so a player can tell which square is theirs at a glance. setAllowGravity(false) and setImmovable(true) matter here because Phaser's arcade physics defaults assume gravity and movable bodies, neither of which applies to a top-down game where the server dictates every position directly.

Next, the coin group and the local overlap check:

        this.coinsGroup = this.physics.add.group();

        this.physics.add.overlap(
            this.localPlayer,
            this.coinsGroup,
            (_player, coin) => this.handleCoinOverlap(coin as CoinSprite),
            undefined,
            this
        );

This is the one piece of client-side game logic the design deliberately calls for: collision detection happens locally using Phaser's built-in overlap check, but only against this.localPlayer, not the opponent's sprite. The callback doesn't touch a score or delete anything from the scene by itself. It only raises a claim, which we'll see in handleCoinOverlap shortly.

The rest of create binds input and subscribes to the server's events:

        const keyboard = this.input.keyboard;
        if (!keyboard) throw new Error("Keyboard input not available");
        this.cursorKeys = keyboard.createCursorKeys();
        this.wasdKeys = keyboard.addKeys("W,A,S,D") as typeof this.wasdKeys;

        socket.on("player:moved", ({ username, x, y }) => {
            const sprite = this.spriteForUsername(username);
            sprite?.setPosition(x, y);
        });

        socket.on("coin:spawned", (coin) => {
            if (this.coinSprites.has(coin.id)) return;
            const sprite = this.physics.add.sprite(coin.x, coin.y, "coin");
            sprite.setOrigin(0, 0);
            sprite.body.setAllowGravity(false);
            sprite.setData("id", coin.id);
            this.coinsGroup.add(sprite);
            this.coinSprites.set(coin.id, sprite);
        });

        socket.on("coin:collected", ({ coinId, scores }) => {
            this.removeCoinSprite(coinId);
            this.updateHud(scores);
        });

        socket.on("game:end", (result) => {
            removeAllGameListeners();
            this.sceneData.onEnd(result);
        });

        this.updateHud({
            player1: { username: startPayload.players.player1, score: 0 },
            player2: { username: startPayload.players.player2, score: 0 },
        });
    }

Look closely at player:moved: it's the only place either player's sprite position ever gets set. coin:spawned creates a new sprite and stores it by coin ID so it can be found again later, and coin:collected removes that sprite and refreshes the score display, again strictly in response to a server event rather than any local prediction. When the match ends, removeAllGameListeners tears down these four handlers before handing control back to main.ts through onEnd.

The update method runs every frame and is where local key presses actually turn into socket emissions:

    update(_time: number, delta: number): void {
        this.updateTimerHud();

        this.moveAccumulator += delta;
        if (this.moveAccumulator < MOVE_EMIT_INTERVAL_MS) return;
        this.moveAccumulator = 0;

        const up = this.cursorKeys.up.isDown || this.wasdKeys.W.isDown;
        const down = this.cursorKeys.down.isDown || this.wasdKeys.S.isDown;
        const left = this.cursorKeys.left.isDown || this.wasdKeys.A.isDown;
        const right = this.cursorKeys.right.isDown || this.wasdKeys.D.isDown;

        const directions: MoveDirection[] = [];
        if (up && !down) directions.push("up");
        if (down && !up) directions.push("down");
        if (left && !right) directions.push("left");
        if (right && !left) directions.push("right");

        for (const direction of directions) {
            socket.emit("player:move", { roomId: this.sceneData.roomId, direction });
        }
    }

The moveAccumulator throttle caps how often player:move gets emitted to once every MOVE_EMIT_INTERVAL_MS (60 milliseconds), regardless of how fast Phaser's own frame rate runs. Without it, a 144Hz monitor would flood the server with movement events at a rate the game doesn't actually need. Opposing keys on the same axis, like holding both up and down, cancel out rather than emitting either direction, which avoids sending a movement event that would just get undone by the next one.

The rest of frontend/src/game/game-scene.ts is smaller, supporting methods. First, the coin overlap handler that raises a claim:

    private handleCoinOverlap(coin: CoinSprite): void {
        const coinId = coin.getData("id") as string | undefined;
        if (!coinId || !this.coinSprites.has(coinId)) return;

        socket.emit("coin:collect", {
            roomId: this.sceneData.roomId,
            coinId,
            playerX: this.localPlayer.x,
            playerY: this.localPlayer.y,
        });
        this.removeCoinSprite(coinId);
    }

    private removeCoinSprite(coinId: string): void {
        const sprite = this.coinSprites.get(coinId);
        if (!sprite) return;
        sprite.destroy();
        this.coinSprites.delete(coinId);
    }

Notice that handleCoinOverlap calls removeCoinSprite immediately, before the server has confirmed anything. That's a small bit of optimistic UI: it stops the same overlap from firing the claim over and over again on every frame the two squares keep touching. If the server's validateCoinCollect check ends up rejecting the claim, this client's coin simply stays gone locally while the opponent's copy of that coin is still sitting on screen, since the server's own map never dropped it. In practice, that's rare enough with the tolerance built into the validation that it's not worth solving for in a demo.

The last few helpers handle rendering details rather than game logic:

    private spriteForUsername(username: string): PlayerSprite | null {
        const { players } = this.sceneData.startPayload;
        if (username === players.player1) return this.player1;
        if (username === players.player2) return this.player2;
        return null;
    }

    private getStartingPosition(
        slot: 1 | 2,
        canvasWidth: number,
        canvasHeight: number,
        playerSize: number
    ): { x: number; y: number } {
        if (slot === 1) {
            return { x: playerSize * 2, y: Math.floor(canvasHeight / 2) };
        }
        return { x: canvasWidth - playerSize * 3, y: Math.floor(canvasHeight / 2) };
    }

This getStartingPosition is a deliberate duplicate of the same-named function in the backend's game/state.ts. Both sides need to agree on where a sprite appears before the very first player:moved broadcast arrives, and since there's no shared package between the two projects, the calculation is written out twice rather than imported from anywhere.

The texture generator draws every square the game needs without loading a single image asset:

    private generateTextures(playerSize: number, coinSize: number): void {
        if (!this.textures.exists("player")) {
            const g = this.make.graphics({ x: 0, y: 0 });
            g.fillStyle(0xffffff, 1);
            g.fillRect(0, 0, playerSize, playerSize);
            g.generateTexture("player", playerSize, playerSize);
            g.destroy();
        }

        if (!this.textures.exists("player-outline")) {
            const g = this.make.graphics({ x: 0, y: 0 });
            g.fillStyle(0xffffff, 1);
            g.fillRect(0, 0, playerSize, playerSize);
            g.lineStyle(4, LOCAL_OUTLINE_COLOR, 1);
            g.strokeRect(2, 2, playerSize - 4, playerSize - 4);
            g.generateTexture("player-outline", playerSize, playerSize);
            g.destroy();
        }

        if (!this.textures.exists("coin")) {
            const g = this.make.graphics({ x: 0, y: 0 });
            g.fillStyle(0xffd43b, 1);
            g.fillRect(0, 0, coinSize, coinSize);
            g.generateTexture("coin", coinSize, coinSize);
            g.destroy();
        }
    }

Each block draws a rectangle with Phaser's Graphics object and bakes it into a reusable texture with generateTexture, then destroys the temporary graphics object since only the baked texture is needed after that. The this.textures.exists(...) guards mean these textures are only ever generated once, even if a player somehow ends up creating a second GameScene instance in the same browser tab.

The last two methods write to the HUD elements that sit outside the Phaser canvas in the surrounding HTML:

    private updateHud(scores: Scores): void {
        const p1El = document.getElementById("hud-player1");
        const p2El = document.getElementById("hud-player2");
        if (p1El) p1El.textContent = `${scores.player1.username}: ${scores.player1.score}`;
        if (p2El) p2El.textContent = `${scores.player2.username}: ${scores.player2.score}`;
    }

    private updateTimerHud(): void {
        const timerEl = document.getElementById("hud-timer");
        if (!timerEl) return;
        const remainingMs = Math.max(0, this.matchEndAt - Date.now());
        timerEl.textContent = `${Math.ceil(remainingMs / 1000)}s`;
    }
}

The countdown in updateTimerHud is purely cosmetic. It's computed locally from matchEndAt, which was set once in create from the server's durationMs, so it can tick smoothly every frame without asking the server for the time remaining. The actual end of the match is still decided entirely by the server's own setTimeout in backend/game/socket.ts. If this client's clock drifts, the worst that happens is the displayed countdown is a little off. The match still ends exactly when the server says it does.

Rendering the Results Screen and Wiring main.ts

The results screen is the simplest file in the project. Create frontend/src/screens/results.ts:

import type { GameEndPayload } from "../types";

export function renderResults(result: GameEndPayload, onPlayAgain: () => void): void {
    const winnerEl = document.getElementById("results-winner")!;
    const scoresEl = document.getElementById("results-scores")!;
    const playAgainBtn = document.getElementById("play-again-btn") as HTMLButtonElement;

    const { player1, player2 } = result.scores;

    winnerEl.textContent = result.winner ? `${result.winner} wins!` : "It's a draw!";
    scoresEl.textContent = `${player1.username}: ${player1.score}  —  ${player2.username}: ${player2.score}`;

    playAgainBtn.onclick = () => onPlayAgain();
}

It takes the same GameEndPayload that the socket delivered and a callback for the "play again" button, and does nothing more than fill in some text content. There's deliberately no way from this screen to rejoin the same match. Per the game's design, playing again always means creating a brand new match from the lobby.

Finally, frontend/src/main.ts ties the three screens together:

import "./style.css";
import Phaser from "phaser";
import { initLobby } from "./screens/lobby";
import { renderResults } from "./screens/results";
import { GameScene, type GameSceneData } from "./game/game-scene";
import type { GameEndPayload, GameStartPayload } from "./types";

const lobbyScreen = document.getElementById("lobby")!;
const gameScreen = document.getElementById("game-screen")!;
const resultsScreen = document.getElementById("results")!;
const gameContainer = document.getElementById("game-container")!;

function showScreen(screen: HTMLElement): void {
    for (const el of [lobbyScreen, gameScreen, resultsScreen]) {
        el.classList.toggle("hidden", el !== screen);
    }
}

let activeGame: Phaser.Game | null = null;

showScreen is the entire routing mechanism for this app. There's no client-side router, just three <section> elements in frontend/index.html that get shown or hidden with a hidden CSS class. activeGame holds the live Phaser instance so it can be destroyed cleanly when a match ends.

The two functions that drive the transitions between screens:

function handleMatchReady(
    roomId: string,
    mySlot: 1 | 2,
    startPayload: GameStartPayload
): void {
    showScreen(gameScreen);

    activeGame = new Phaser.Game({
        type: Phaser.AUTO,
        parent: gameContainer,
        width: startPayload.canvasWidth,
        height: startPayload.canvasHeight,
        backgroundColor: "#14151c",
        scale: {
            mode: Phaser.Scale.NONE,
        },
        physics: {
            default: "arcade",
            arcade: {
                gravity: { x: 0, y: 0 },
            },
        },
        scene: [GameScene],
    });

    const sceneData: GameSceneData = {
        roomId,
        mySlot,
        startPayload,
        onEnd: handleGameEnd,
    };
    activeGame.scene.start("GameScene", sceneData);
}

function handleGameEnd(result: GameEndPayload): void {
    if (activeGame) {
        activeGame.destroy(true);
        activeGame = null;
    }
    showScreen(resultsScreen);
    renderResults(result, () => {
        returnToLobby();
        showScreen(lobbyScreen);
    });
}

const returnToLobby = initLobby(handleMatchReady);

handleMatchReady is the callback the lobby invokes once game:start arrives. The Phaser game's width and height are set from startPayload.canvasWidth/canvasHeight, the exact same values the backend reads from its own .env file, so resizing the canvas is a one-line config change on the backend with nothing to update on the frontend. handleGameEnd destroys the Phaser instance entirely rather than just hiding it, since a new match always means a fresh GameScene, then shows the results screen with a "play again" callback that resets the lobby, refreshes the leaderboard, and routes back to it. The last line, initLobby(handleMatchReady), is what actually starts the whole application, and it holds on to the reset function the lobby returns as returnToLobby.

Running the Game End to End

With both projects built out, start the backend first:

cd backend
npm run dev

You should see Connected to MongoDB followed by Server listening on http://localhost:3000. In a second terminal, start the frontend:

cd frontend
npm run dev

Vite will print a local URL, typically http://localhost:5173. Open that URL in two separate browser tabs (or two different browsers entirely, since two tabs of the same browser work just as well for this).

In the first tab, register a username and click "Create Match." You'll land on a waiting screen with a room ID displayed. Copy that ID.

In the second tab, register a different username, paste the room ID into the "Room ID" field, and click "Join Match." As soon as the server sees both players seated, both tabs should flip to the game screen at the same moment, each showing two white squares, one of them outlined to mark which one you control. Move around with the arrow keys or WASD and watch coins spawn and disappear as either player touches them. After thirty seconds, both tabs should show the results screen with the final scores and a winner (or a draw). Head back to the lobby and check the leaderboard. If either player's score beats their previous high score, you should see it reflected there.

Conclusion

We built a real-time two-player game where MongoDB and Socket.io each do the job they're actually good at. MongoDB stores what needs to survive between matches: player accounts, match history, and a leaderboard sorted by high_score. Socket.io handles everything that happens inside the thirty-second window of a single match, with the server holding the only authoritative copy of every position, every coin, and every score. The client never gets final say over anything that affects the outcome of the game. It only forwards input and renders whatever the server broadcasts back.

A few things would need to change before this could handle more than a tutorial's worth of traffic. The in-memory gameRooms map in backend/game/state.ts lives on a single Node.js process, so running more than one server instance would need those rooms to be moved somewhere shared. MongoDB change streams on the matches collection are worth a look here, since every server instance could watch for the same match updates and stay in sync without owning the room's state itself. There's also no reconnection handling beyond the same username rejoining the same slot, and no automatic winner if a player disconnects mid-match.

As a next step, try adding a rematch button that creates a new match against the same opponent without a trip back through the lobby, or add a MongoDB index on high_score so the leaderboard query stays fast as the players collection grows.

Want to try this project for yourself? Clone the repository on GitHub.

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.

DEV Community: MongoDB

Leveraging Skills in a MongoDB MCP Agent

Skills in a ReAct (Reason - Act) Agent for MongoDB

Evolving from a one-pass MongoDB Server Agent

Reuse the scripting from the earlier article

Prerequisite

Getting Started

Add Skills to MCP

Skills add complexity to the Agent

The enhanced ReAct loop agent

The payoff: more sophisticated queries

The secret sauce in the agent-server.js code

Agentic code assistance

Summary

What agent memory actually asks of a database

Requirement 1: The write has to be atomic across three shapes of data

Requirement 2: You have to be able to rebuild state and prove it's the same state

Requirement 3: Retrieval has to be hybrid, and it should happen server-side

Requirement 4: Deletion has to be provable, and it reaches further than you think

The security layer, briefly

Reproducing it

Why you'd want this

What MongoDB is doing for this

MongoDB as a Vector Database for AI Agents

Introduction

What you will need

Understanding the Architecture

Step 1 — Setting Up Your Environment

Step 2 — Setting Up MongoDB Atlas

Step 3 — Creating the Vector Search Indexes

Step 4 — Ingesting Recipes from TheMealDB

Step 5 — Building the Agent

Step 6 — Running the Agent

Step 7 — Inspecting What MongoDB is Storing

Going Further

Conclusion

MongoDB as a Vector Database for AI Agents

Introduction

What you will need

Understanding the Architecture

Step 1 — Setting Up Your Environment

Step 2 — Setting Up MongoDB Atlas

Step 3 — Creating the Vector Search Indexes

Step 4 — Ingesting Recipes from TheMealDB

Step 5 — Building the Agent

Step 6 — Running the Agent

Step 7 — Inspecting What MongoDB is Storing

Going Further

Conclusion

Building an Intelligent Document Processing Pipeline with MongoDB Atlas

Why Keep the Evidence?

Where is The Risk

Keeping the Evidence with the Document

Keeping a Review Queue

Sometimes a Parser is Enough

Migration Path

What Survives is the Pipeline

Multiplayer Game Development with TypeScript, Phaser, Socket.io, and MongoDB

The Prerequisites

Understanding the Shape of the Project

Setting Up the Backend Project

Loading Configuration Once at Startup

Connecting to MongoDB with a Cached Client

Defining the Player and Match Data Models

Building the REST Layer

Wiring Express and Socket.io Together in main.ts

Managing In-Memory Game State

Spawning Coins on a Timer

Handling the Real-Time Game Loop

Setting Up the Frontend Project

Defining the Page Markup and Styles

Mirroring the Backend's Types on the Frontend

Talking to the REST API

Setting Up the Socket Connection

Building the Lobby Screen

Building the Game Scene

Rendering the Results Screen and Wiring main.ts

Running the Game End to End

Conclusion

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

The secret sauce in the `agent-server.js` code