RAG, Vector Databases, and MCP: Wiring Them Together for Production

Why This Combination Matters

Most RAG tutorials stop at the same point: embed some documents, stuff them into a vector store, retrieve the top-k chunks, and paste them into a prompt. That gets you a demo. It does not get you a system another team can call, monitor, version, and trust.

Three pieces close that gap:

• RAG – the retrieval-augmented generation pattern itself: chunk, embed, retrieve, ground the model's answer in real data.
• A vector database – the durable, queryable index that makes retrieval fast and scalable instead of a linear scan through embeddings in memory.
• MCP (Model Context Protocol) – the standard that lets any MCP-compatible host (Claude Desktop, Claude Code, your own agent runtime) call that retrieval capability as a tool, instead of every team hand-rolling its own glue code between the model and the data.

Put together, the pattern looks like this:

Host (Claude / Claude Code / your agent)
        │  MCP protocol (JSON-RPC over stdio or HTTP+SSE)
        ▼
MCP Server ("docs-search")
        │  calls
        ▼
RAG Retrieval Layer  →  Vector DB (Chroma / pgvector / Qdrant)
                              │
                        Embedding Model

The host never talks to your vector database directly. It talks to a tool. That one architectural decision is what turns a notebook prototype into something you can put behind an SLA.

Part 1: RAG, Built for Production, Not for a Demo

The two places demo-quality RAG breaks in production are chunking and retrieval quality. Fix those first.

Chunking With Overlap and Metadata

from dataclasses import dataclass
from typing import List

@dataclass
class Chunk:
    text: str
    source: str
    chunk_id: str
    page: int | None = None

def chunk_document(text: str, source: str, chunk_size: int = 800,
                    overlap: int = 120) -> List[Chunk]:
    """Sliding-window chunking with overlap to avoid cutting
    context across boundaries — the single highest-leverage
    fix for weak retrieval."""
    chunks = []
    start = 0
    idx = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        piece = text[start:end]
        chunks.append(
            Chunk(text=piece, source=source, chunk_id=f"{source}-{idx}")
        )
        start += chunk_size - overlap
        idx += 1
    return chunks

Two things matter here that most tutorials skip: overlap (so an answer that straddles a chunk boundary doesn't get orphaned) and metadata on every chunk (source, page, chunk_id) so the model — and your logs — can cite where an answer came from.

Embedding With Batching and Retry

import time
from openai import OpenAI

client = OpenAI()

def embed_batch(texts: list[str], model: str = "text-embedding-3-large",
                 max_retries: int = 3) -> list[list[float]]:
    for attempt in range(max_retries):
        try:
            resp = client.embeddings.create(model=model, input=texts)
            return [d.embedding for d in resp.data]
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

Batch embedding calls (not one request per chunk) and add exponential backoff — at index-build time you're often pushing tens of thousands of chunks through the embedding API, and that's where rate limits bite.

Part 2: The Vector Database Layer

A vector database earns its place the moment your corpus is too large to hold in memory, or the moment you need filtered retrieval (by tenant, document type, date range) alongside similarity search. Here's a production-shaped setup using Chroma, with the pattern identical if you swap in pgvector or Qdrant.

import chromadb
from chromadb.config import Settings

client = chromadb.PersistentClient(path="./vector_store")

collection = client.get_or_create_collection(
    name="product_docs",
    metadata={"hnsw:space": "cosine"}  # cosine similarity, HNSW index
)

def index_chunks(chunks: list[Chunk]):
    embeddings = embed_batch([c.text for c in chunks])
    collection.upsert(
        ids=[c.chunk_id for c in chunks],
        embeddings=embeddings,
        documents=[c.text for c in chunks],
        metadatas=[{"source": c.source, "page": c.page or 0} for c in chunks],
    )

def retrieve(query: str, top_k: int = 5, source_filter: str | None = None):
    q_embedding = embed_batch([query])[0]
    where = {"source": source_filter} if source_filter else None
    results = collection.query(
        query_embeddings=[q_embedding],
        n_results=top_k,
        where=where,
    )
    return list(zip(results["documents"][0], results["metadatas"][0]))

Notice upsert, not insert — production indexes get re-crawled and re-embedded constantly, and re-indexing should be idempotent by chunk_id. Notice also the where filter — real retrieval almost always needs a metadata constraint alongside the similarity search, or you'll surface the right kind of chunk from the wrong tenant's documents.

A layer worth adding before this goes live: a semantic cache in front of the vector query. If the same or a near-duplicate question comes in repeatedly (which it will, in any real user base), you don't want to re-embed and re-query every time. A thin cache keyed on embedding similarity — check for a cached answer within a cosine-distance threshold before hitting the vector DB — cuts both latency and embedding-API cost substantially in high-traffic RAG deployments.

Part 3: Exposing Retrieval as an MCP Tool

This is the piece that makes the difference between "a RAG pipeline I run in a notebook" and "a capability any Claude-based host can use." Instead of embedding your retrieval logic into every application that needs it, you expose it once, as an MCP server, and any compliant host — Claude Desktop, Claude Code, a custom agent — can call it the same way.

# mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncio

app = Server("docs-search")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description=(
                "Search the product documentation vector index and "
                "return the most relevant passages with their sources."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "The search query"},
                    "top_k": {"type": "integer", "default": 5},
                    "source_filter": {"type": "string", "description": "Optional source doc to restrict to"},
                },
                "required": ["query"],
            },
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name != "search_docs":
        raise ValueError(f"Unknown tool: {name}")

    results = retrieve(
        query=arguments["query"],
        top_k=arguments.get("top_k", 5),
        source_filter=arguments.get("source_filter"),
    )

    formatted = "\n\n".join(
        f"[Source: {meta['source']}, page {meta['page']}]\n{doc}"
        for doc, meta in results
    )
    return [TextContent(type="text", text=formatted or "No matching passages found.")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

Register it with Claude Code or Claude Desktop with a short config entry:

{
  "mcpServers": {
    "docs-search": {
      "command": "python",
      "args": ["mcp_server.py"]
    }
  }
}

From that point on, when a developer working in Claude Code asks a question that needs grounding in your documentation, the host discovers the search_docs tool, calls it with the right arguments, gets back cited passages, and folds them into its answer — with no custom integration code per application. That is the actual point of MCP: one retrieval service, called the same way by every host that speaks the protocol, instead of a bespoke RAG wrapper duplicated inside every app, chatbot, and IDE plugin your organization builds.

Production Considerations Before You Ship This

• Observability – log every tool call: query text, top_k, latency, which chunks were returned, and — if you can capture it — whether the final answer used them. Without this, you're debugging RAG quality blind.
• Freshness – decide explicitly how re-indexing happens (scheduled crawl, webhook on document change, or both) and make upsert idempotent so partial re-index failures don't corrupt the collection.
• Access control at the MCP boundary – the MCP server, not the LLM, is the right place to enforce which documents a given caller is allowed to search. Filter by tenant/user in the call_tool handler before the query ever reaches the vector database.
• Timeouts and fallbacks – a vector DB query that hangs should not hang the whole conversation. Set a hard timeout on retrieve() and have the tool return a clear "search unavailable" message rather than blocking.
• Evaluation – keep a small, versioned set of query/expected-passage pairs and re-run it whenever you change the chunking strategy, the embedding model, or the index. Chunking changes are the single most common silent cause of retrieval regressions.

Closing

RAG gives you the pattern, the vector database gives you the scale, and MCP gives you the interface that lets any host reuse the pipeline without re-implementing it. None of the three pieces is complicated on its own — the production value comes from wiring them together deliberately: idempotent indexing, filtered retrieval, a caching layer in front of the vector store, and access control enforced at the tool boundary rather than left to the model's judgment.