How to Build an MCP Server for MongoDB

uv init --python 3.11 mongo-tools && cd mongo-tools
uv add "mcp[cli]==2.0.0" "pymongo==4.11" mongomock

import os
from datetime import datetime
from typing import Any

from bson import ObjectId
from pymongo import MongoClient

from mcp.server.mcpserver import MCPServer

# One connection, reused across every tool call. Point MONGO_URI at a
# read-only user so a bug in a tool can never write or drop a collection.
MONGO_URI = os.environ["MONGO_URI"]
DB_NAME = os.environ.get("MONGO_DB", "app")
MAX_LIMIT = 50

client: MongoClient = MongoClient(MONGO_URI)
db = client[DB_NAME]

mcp = MCPServer(name="mongo-tools", version="1.0.0")


def _safe_filter(filter: dict[str, Any]) -> dict[str, Any]:
    """Allow only flat field == value equality. Reject anything that could
    inject a MongoDB operator: a $-prefixed key ($where runs JavaScript on the
    server, $gt/$ne turn into match-all), or a dotted key that reaches into a
    subdocument the caller was never meant to touch."""
    if not isinstance(filter, dict):
        raise ValueError("filter must be an object of field: value pairs")
    clean: dict[str, Any] = {}
    for key, value in filter.items():
        if key.startswith("$") or "." in key:
            raise ValueError(f"illegal filter key: {key!r}")
        if isinstance(value, (dict, list)):
            raise ValueError(f"value for {key!r} must be a scalar, not an operator object")
        clean[key] = value
    return clean


def _jsonable(doc: dict[str, Any]) -> dict[str, Any]:
    """BSON types are not JSON-serializable. Coerce ObjectId and datetime to
    strings so the document survives the JSON-RPC boundary intact."""
    out: dict[str, Any] = {}
    for key, value in doc.items():
        if isinstance(value, ObjectId):
            out[key] = str(value)
        elif isinstance(value, datetime):
            out[key] = value.isoformat()
        else:
            out[key] = value
    return out

@mcp.tool()
def list_collections() -> list[str]:
    """List the collection names in the database."""
    return sorted(db.list_collection_names())


@mcp.tool()
def find_documents(
    collection: str, filter: dict[str, Any] | None = None, limit: int = 20
) -> list[dict[str, Any]]:
    """Find documents in a collection by exact field match. `filter` is a flat
    map of field: value equality pairs; returns at most `limit` documents."""
    query = _safe_filter(filter or {})
    capped = max(1, min(limit, MAX_LIMIT))
    cursor = db[collection].find(query).limit(capped)
    return [_jsonable(doc) for doc in cursor]


@mcp.tool()
def count_documents(collection: str, filter: dict[str, Any] | None = None) -> int:
    """Count the documents in a collection matching a flat equality filter."""
    return db[collection].count_documents(_safe_filter(filter or {}))

if __name__ == "__main__":
    # stdio is the transport a local client (Claude Desktop, Cursor) launches.
    mcp.run("stdio")

"""End-to-end test: seed an in-memory MongoDB, drive the tools, and assert
that operator injection is refused. Uses mongomock, so no real database or
network is required. Run it with: uv run python test_server.py"""
import mongomock

import server

# Swap the live connection for an in-memory one and seed three users.
server.db = mongomock.MongoClient()["app"]
server.db.users.insert_many(
    [
        {"name": "ada", "status": "active"},
        {"name": "grace", "status": "active"},
        {"name": "alan", "status": "disabled"},
    ]
)

# 1. list_collections sees the seeded collection.
assert "users" in server.list_collections()

# 2. An equality filter returns the matching docs, with _id coerced to a string.
active = server.find_documents("users", {"status": "active"})
assert len(active) == 2
assert all(isinstance(doc["_id"], str) for doc in active)

# 3. count_documents agrees with find.
assert server.count_documents("users", {"status": "active"}) == 2

# 4. Every injection attempt is refused as a tool error, not executed.
for attack in ({"$where": "true"}, {"name": {"$ne": ""}}, {"a.b": 1}):
    try:
        server.find_documents("users", attack)
        raise SystemExit(f"injection not blocked: {attack}")
    except ValueError:
        pass

print("ALL CHECKS PASSED")

uv run python test_server.py
# ALL CHECKS PASSED

{
  "mcpServers": {
    "mongo-tools": {
      "command": "uv",
      "args": ["--directory", "/absolute/path/to/mongo-tools", "run", "python", "server.py"],
      "env": {
        "MONGO_URI": "mongodb://readonly:secret@localhost:27017/",
        "MONGO_DB": "app"
      }
    }
  }
}