MCP, RAG & CAG
Augmenting LLMs with external knowledge and tools.
Beyond the Training Data
LLMs are incredibly powerful, but they have a fundamental limitation: they only know what was in their training data, which has a cutoff date and doesn't include your private data. Three techniques close that gap, and they are not interchangeable. RAG (Retrieval-Augmented Generation) searches a knowledge base at query time and injects the best matches into the prompt. CAG (Cache-Augmented Generation) skips the search entirely by preloading the whole corpus once and letting the provider's cache keep it warm. MCP (Model Context Protocol) is a different axis altogether: a standard wire protocol that lets a model read live data and take actions through tools.
The first decision is usually not "which is best" but "does my knowledge fit in the context window?" If it does, CAG is simpler and faster. If it does not, you need RAG. If the model has to do something rather than know something, you need MCP.
RAG: Retrieval-Augmented Generation
RAG is the most common technique for grounding LLM responses in external knowledge. Instead of relying solely on the model's training data, RAG retrieves relevant information from external sources and includes it in the prompt.
How RAG Works
RAG Pipeline Flow
Figure 1: A query is embedded, matched against the vector store, and the top documents are injected into the prompt before generation
Why a Vector Database?
Steps 2 and 3 of Figure 1 are where a normal data store stops being useful, so it is worth being precise about what the retrieval step actually asks for. It is not "find rows where topic = 'vacation'" and not "find rows containing the word vacation." It is find the handful of vectors closest to this one in 384-dimensional space. That is a different question, and it is the reason a dedicated store shows up here.
A B-tree index cannot help with it. B-trees work by keeping a total order over a scalar key, and there is no such ordering to exploit across 384 dimensions at once. A relational database without a vector extension has only one option: scan every row, compute the distance, sort. That is fine for the four documents here and untenable at a million.Chroma instead maintains an approximate nearest-neighbour index, HNSW, which trades exact answers for sublinear search: it accepts an occasional missed neighbour in exchange for not looking at most of the collection.
Chroma is an open-source embedding database built around that index. It keeps the vector, the original document text, and arbitrary metadata together, which is why a single query() call returns everything the prompt needs: the text to inject and the title to cite. Keeping vectors in one system and documents in another means a second round trip and an ID mapping to maintain.
pgvector: if you already run PostgreSQL, it adds vector search to the database holding your documents instead of adding a service to operate, back up, and keep in sync. Qdrant, Weaviate, and Milvus are dedicated stores with more tuning and filtering capability, Pinecone is the managed option, and FAISS is a library rather than a server for when the index can live inside your process. They all implement the same idea; they differ in what you have to run.Detailed Example: Vacation Policy Query
Figure 1 shows the shape of the pipeline. What follows is what actually moves through it for a single question. Every number below came out of a real run against a real vector store, so start that store first and the rest of this section reproduces exactly.
Step 0: start Chroma and load the documents. Chroma runs as a container. The pinned tag matters: the client library and the server negotiate a wire protocol, so matching the tag to your installed chromadb avoids a version-skew error that looks like a connection failure.
# Start a throwaway Chroma server on port 8000 docker run -d --name chroma-rag -p 8000:8000 chromadb/chroma:1.5.9 # Confirm it is accepting requests before loading anything curl -s http://localhost:8000/api/v2/heartbeat # Client libraries used below pip install chromadb==1.5.9 sentence-transformers anthropic
Expected Output:
{"nanosecond heartbeat":1786801599715002622}The server starts empty. This script creates the collection and writes the four handbook documents into it, embedding each one with the same model that will later embed the queries. Run it once:
# ingest.py - create the collection and load the handbook into Chroma
#
# Run this once against the container started above. It is idempotent: any
# existing collection is dropped first, so re-running leaves four documents
# rather than eight.
import chromadb
from sentence_transformers import SentenceTransformer
DOCUMENTS = [
{
"title": "Employee Handbook - Section 5.2",
"text": "Vacation Policy: Full-time employees receive 20 days of paid "
"vacation per year. Vacation accrues monthly at 1.67 days per "
"month. Unused vacation rolls over up to 10 days. Employees "
"must request vacation at least 2 weeks in advance except for "
"emergencies.",
},
{
"title": "Remote Work Guidelines",
"text": "Remote Work Policy: Employees may work remotely up to 3 days "
"per week. Remote work requires manager approval and must be "
"scheduled in advance. Employees must be available during core "
"hours (10 AM - 3 PM) regardless of location.",
},
{
"title": "Benefits Guide",
"text": "Health Benefits: Company provides comprehensive health "
"insurance through BlueCross. Coverage starts on the first day "
"of the month following hire date. Premium is $150/month for "
"individuals, $350/month for families. Annual deductible: "
"$1,000 individual, $2,000 family.",
},
{
"title": "Performance Management",
"text": "Performance Review Process: Reviews conducted annually in "
"December. Employees receive feedback on goals, competencies, "
"and overall performance. Reviews determine merit increases and "
"promotion eligibility. Self-assessment due by December 1st, "
"manager review by December 15th.",
},
]
# HttpClient talks to the container. chromadb.Client() would instead run an
# in-process store that disappears when the script exits.
client = chromadb.HttpClient(host="localhost", port=8000)
print("server reachable:", client.heartbeat() > 0)
if any(c.name == "documents" for c in client.list_collections()):
client.delete_collection("documents")
collection = client.create_collection(
name="documents",
metadata={"hnsw:space": "cosine"},
)
embedder = SentenceTransformer("all-MiniLM-L6-v2")
texts = [doc["text"] for doc in DOCUMENTS]
collection.add(
ids=[f"doc_{i}" for i in range(len(DOCUMENTS))],
embeddings=embedder.encode(texts).tolist(),
documents=texts,
metadatas=[{"title": doc["title"]} for doc in DOCUMENTS],
)
stored = collection.get()
print("collection:", collection.name)
print("documents stored:", collection.count())
for record_id, meta in zip(stored["ids"], stored["metadatas"]):
print(f" {record_id}: {meta['title']}")Expected Output:
server reachable: True collection: documents documents stored: 4 doc_0: Employee Handbook - Section 5.2 doc_1: Remote Work Guidelines doc_2: Benefits Guide doc_3: Performance Management
Now that the collection exists, the HNSW index described earlier is no longer an abstraction. Its settings are readable, and the space confirms the distance function the similarity numbers depend on:
import chromadb
collection = chromadb.HttpClient(host="localhost", port=8000).get_collection("documents")
print(collection.configuration_json["hnsw"])Expected Output:
{'space': 'cosine', 'ef_construction': 100, 'ef_search': 100, 'max_neighbors': 16, 'resize_factor': 1.2, 'sync_threshold': 1000}hnsw:space metadata sets the distance function to cosine at creation time, and it is genuinely fixed from then on: calling modify() to change it raises ValueError: Changing the distance function of a collection once it is created is not supported currently. The similarity numbers below therefore only mean what they appear to mean because cosine was chosen here. Second, the embeddings are computed in your code and passed in explicitly. Omit embeddings= and Chroma embeds the documents itself with its own bundled model, downloading roughly 80 MB on first use. That default happens to be an ONNX build of the same all-MiniLM-L6-v2, so it would not break this example, which is exactly what makes the habit dangerous: the day you switch to a different model on one side only, nothing raises and the results quietly become meaningless.Steps 1 and 2: embed the query, then search. The question becomes a vector in the same space as the documents, and the search is a nearest-neighbour lookup over those vectors. Nothing about the words is compared directly.
from sentence_transformers import SentenceTransformer
embedder = SentenceTransformer("all-MiniLM-L6-v2")
query = "What is our company's vacation policy?"
query_embedding = embedder.encode([query])[0]
# Individual components are small and carry no meaning on their own;
# only the direction of the whole vector matters.
print(len(query_embedding))
print([round(float(v), 4) for v in query_embedding[:5]])Expected Output:
384 [0.0471, 0.0331, 0.0649, -0.0026, 0.0241]
Step 3: retrieve the top K. Chroma returns a cosine distance per document, which this code converts to a similarity with 1 - distance. Ask for the top 3 of the 4 documents loaded above:
# query_step3.py - the retrieval half of the pipeline, against the container
import chromadb
from sentence_transformers import SentenceTransformer
embedder = SentenceTransformer("all-MiniLM-L6-v2")
client = chromadb.HttpClient(host="localhost", port=8000)
collection = client.get_collection("documents")
query = "What is our company's vacation policy?"
query_embedding = embedder.encode([query])[0]
results = collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=3,
)
for rank in range(len(results["ids"][0])):
title = results["metadatas"][0][rank]["title"]
similarity = 1 - results["distances"][0][rank]
preview = results["documents"][0][rank][:52]
print(f"{rank + 1}. {title:33s} similarity = {similarity:.3f}")
print(f" {preview}...")Expected Output:
1. Employee Handbook - Section 5.2 similarity = 0.698 Vacation Policy: Full-time employees receive 20 days... 2. Remote Work Guidelines similarity = 0.324 Remote Work Policy: Employees may work remotely up t... 3. Benefits Guide similarity = 0.183 Health Benefits: Company provides comprehensive heal...
| Rank | Document | Similarity | Actually about vacation? |
|---|---|---|---|
| 1 | Employee Handbook - Section 5.2 | 0.698 | Yes |
| 2 | Remote Work Guidelines | 0.324 | No |
| 3 | Benefits Guide | 0.183 | No |
Step 4: build the prompt. The retrieved text is concatenated into a context block ahead of the question, with an instruction that confines the answer to it. This is the entire "augmented" part of Retrieval-Augmented Generation.
Context: Source 1 (Employee Handbook - Section 5.2): Vacation Policy: Full-time employees receive 20 days of paid vacation per year. Vacation accrues monthly at 1.67 days per month. Unused vacation rolls over up to 10 days. Employees must request vacation at least 2 weeks in advance except for emergencies. Source 2 (Remote Work Guidelines): Remote Work Policy: Employees may work remotely up to 3 days per week. Remote work requires manager approval and must be scheduled in advance. Employees must be available during core hours (10 AM - 3 PM) regardless of location. Source 3 (Benefits Guide): Health Benefits: Company provides comprehensive health insurance through BlueCross. Coverage starts on the first day of the month following hire date. Premium is $150/month for individuals, $350/month for families. Annual deductible: $1,000 individual, $2,000 family. Question: What is our company's vacation policy? Answer the question based on the context above.
Notice what came along for the ride. Sources 2 and 3 scored 0.324 and 0.183, they have nothing to do with vacation, and they are in the prompt anyway, because the retriever was asked for three documents and dutifully returned three. The model now has to ignore them, and you are paying for their tokens on every request. This is the ordinary case, not a malfunction, and it is why the scores in the table above deserve a threshold rather than blind trust.
Steps 5 and 6: generate, then attribute. That prompt goes to claude-sonnet-5 with max_tokens=1000 and no temperature, since current models reject the parameter. The response is returned alongside the documents and scores that produced it, so the caller can judge the evidence rather than trusting the answer blind. The next section is the full implementation of exactly these six steps.
Why RAG is Powerful
- Access to private/proprietary data
- Always up-to-date information
- Reduces hallucinations
- Attributable sources
- No model retraining needed
Key Components
- Document corpus/knowledge base
- Embedding model
- Vector database
- Retrieval algorithm
- LLM for generation
Complete RAG Implementation
ingest.py already populated, so it does no ingestion of its own: documents are embedded once, queries run thousands of times, and keeping the two apart is what lets you re-index without redeploying the service.# rag_system.py - Complete RAG implementation
#
# Reads the collection that ingest.py populated. Ingestion and serving are kept
# apart on purpose: documents are embedded once, queries run thousands of times.
import chromadb
from sentence_transformers import SentenceTransformer
from anthropic import Anthropic
class RAGSystem:
"""
Production RAG system with embeddings, vector search, and LLM generation.
"""
def __init__(
self,
chroma_host: str = "localhost",
chroma_port: int = 8000,
collection_name: str = "documents",
embedding_model: str = "all-MiniLM-L6-v2",
llm_model: str = "claude-sonnet-5",
api_key: str | None = None
):
# The same embedding model used at ingest time. Querying a collection
# with a different model returns nonsense: the vectors are not
# comparable, and nothing raises to tell you.
self.embedder = SentenceTransformer(embedding_model)
# Connect to the Chroma server and open the existing collection.
self.chroma_client = chromadb.HttpClient(host=chroma_host, port=chroma_port)
self.collection = self.chroma_client.get_collection(collection_name)
# LLM for generation
self.llm = Anthropic(api_key=api_key)
self.llm_model = llm_model
def retrieve(self, query: str, top_k: int = 3) -> list[dict]:
"""
Retrieve most relevant documents for a query.
Args:
query: User question
top_k: Number of documents to retrieve
Returns:
List of retrieved documents with metadata and scores
"""
# Embed query
query_embedding = self.embedder.encode([query])[0]
# Search vector database
results = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=top_k
)
# Format results
retrieved_docs = []
for i in range(len(results['ids'][0])):
retrieved_docs.append({
'id': results['ids'][0][i],
'text': results['documents'][0][i],
'metadata': results['metadatas'][0][i],
'similarity': 1 - results['distances'][0][i] # Convert distance to similarity
})
return retrieved_docs
def generate_answer(
self,
query: str,
retrieved_docs: list[dict],
include_sources: bool = True
) -> dict:
"""
Generate answer using retrieved context.
Args:
query: User question
retrieved_docs: Documents from retrieve()
include_sources: Include source citations
Returns:
Answer with metadata
"""
# Build context from retrieved documents
context = "\n\n".join([
f"Source {i+1} ({doc['metadata'].get('title', 'Unknown')}):\n{doc['text']}"
for i, doc in enumerate(retrieved_docs)
])
# Create prompt with context
system_prompt = """You are a helpful assistant that answers questions based on the provided context.
Guidelines:
1. Answer ONLY using information from the provided context
2. If the context doesn't contain enough information, say so
3. Cite sources when possible (Source 1, Source 2, etc.)
4. Be concise but complete"""
user_prompt = f"""Context:
{context}
Question: {query}
Answer the question based on the context above."""
# Call LLM
response = self.llm.messages.create(
model=self.llm_model,
max_tokens=1000,
# No temperature: current Claude models reject the parameter with a
# 400. Grounding comes from the retrieved context and the system
# prompt, not from a sampling setting. See lesson 3.
system=system_prompt,
messages=[{"role": "user", "content": user_prompt}]
)
answer = next(b.text for b in response.content if b.type == "text")
# Build result
result = {
'answer': answer,
'query': query,
'sources': [
{
'title': doc['metadata'].get('title', 'Unknown'),
'similarity': doc['similarity'],
'excerpt': doc['text'][:200] + '...'
}
for doc in retrieved_docs
] if include_sources else []
}
return result
def query(self, question: str, top_k: int = 3) -> dict:
"""
Complete RAG pipeline: retrieve + generate.
Args:
question: User question
top_k: Number of documents to retrieve
Returns:
Answer with sources
"""
# Step 1: Retrieve relevant documents
retrieved_docs = self.retrieve(question, top_k=top_k)
# Step 2: Generate answer
result = self.generate_answer(question, retrieved_docs)
return result
# Example Usage
if __name__ == "__main__":
import os
rag = RAGSystem(api_key=os.getenv("ANTHROPIC_API_KEY"))
questions = [
"How many vacation days do I get?",
"What is the remote work policy?",
"When are performance reviews?",
"How much does health insurance cost?"
]
for question in questions:
print(f"\n{'='*60}")
print(f"Q: {question}")
print('='*60)
result = rag.query(question, top_k=2)
print(f"\nA: {result['answer']}")
print(f"\nSources:")
for i, source in enumerate(result['sources'], 1):
print(f" {i}. {source['title']} (similarity: {source['similarity']:.2f})")Expected Output:
============================================================ Q: How many vacation days do I get? ============================================================ A: Full-time employees receive **20 days of paid vacation per year**, which accrues monthly at a rate of 1.67 days per month. Unused vacation days can roll over, up to a maximum of 10 days. Keep in mind that vacation requests must generally be made at least 2 weeks in advance, except in emergencies (Source 1). Sources: 1. Employee Handbook - Section 5.2 (similarity: 0.72) 2. Remote Work Guidelines (similarity: 0.30) ============================================================ Q: What is the remote work policy? ============================================================ A: According to Source 1 (Remote Work Guidelines), the remote work policy includes the following key points: - Employees may work remotely **up to 3 days per week**. - Remote work **requires manager approval** and must be **scheduled in advance**. - Regardless of work location, employees must be **available during core hours (10 AM - 3 PM)**. Sources: 1. Remote Work Guidelines (similarity: 0.69) 2. Employee Handbook - Section 5.2 (similarity: 0.29) ============================================================ Q: When are performance reviews? ============================================================ A: Performance reviews are conducted annually in December (Source 1). The process includes specific deadlines: employees must submit their self-assessment by December 1st, and managers complete their review by December 15th (Source 1). Sources: 1. Performance Management (similarity: 0.69) 2. Employee Handbook - Section 5.2 (similarity: 0.08) ============================================================ Q: How much does health insurance cost? ============================================================ A: Based on Source 1 (Benefits Guide), health insurance costs are: - **$150/month** for individual coverage - **$350/month** for family coverage Additionally, there are annual deductibles of **$1,000 for individuals** and **$2,000 for families**. Sources: 1. Benefits Guide (similarity: 0.58) 2. Employee Handbook - Section 5.2 (similarity: 0.17)
temperature. Current Claude models reject it outright, so grounding has to come from the retrieved context and the system prompt. The answer wording will vary slightly between runs; the similarity scores will not. When you are done, remove the container with docker rm -f chroma-rag: it stores its data inside the container, so deleting it discards the collection and re-running ingest.py rebuilds it from scratch.Common RAG Challenges
1. Retrieval Quality
Problem: Retrieved docs aren't relevant
Solution: Better embeddings, hybrid search, query rewriting
2. Context Length
Problem: Too many docs exceed LLM context window
Solution: Document chunking, re-ranking, summarization
3. Stale Information
Problem: Documents outdated
Solution: Regular re-indexing, metadata timestamps
4. Multi-Hop Reasoning
Problem: Answer requires multiple docs
Solution: Iterative retrieval, knowledge graphs
CAG: Cache-Augmented Generation
Everything above exists to answer one question: which slice of the corpus belongs in this prompt? CAG asks whether you need to choose at all. If the whole corpus fits in the context window, you can preload it once and let the provider's cache keep it warm, deleting the embedding model, the vector store, and every retrieval bug along with them.
What CAG Actually Is
CAG (Cache-Augmented Generation) preloads an entire knowledge base into the model's context and reuses the cached key/value state across requests, so queries are answered with no retrieval step. The term comes from Chan et al., "Don't Do RAG: When Cache-Augmented Generation is All You Need for Knowledge Tasks" (arXiv:2412.15605, December 2024). You will also see it written "Context-Augmented Generation"; both names refer to this same preload-and-cache technique.
RAG vs CAG: Where the Knowledge Lives
Figure 2: RAG (top) does work on every request; CAG (bottom) does the work once and reads it back from cache
A Working CAG System
cache_control breakpoint. The first call pays to write the cache, and every later call reads it back at a fraction of the price. Watch the token counters in the output: they are the only proof that caching is actually happening.# cag_system.py - Cache-Augmented Generation
#
# CAG skips retrieval entirely: the whole knowledge base is preloaded into the
# prompt once, and the provider's cache keeps it warm so later questions pay
# almost nothing to re-read it.
import os
import time
from anthropic import Anthropic
MODEL = "claude-sonnet-5"
class CAGSystem:
"""
Answer questions from a corpus small enough to live in the context window.
The corpus goes in the system prompt behind a cache_control breakpoint, so
the first call writes the cache and every later call reads it. There is no
embedding model, no vector store, and no retrieval step to get wrong.
"""
def __init__(self, documents: list[dict], api_key: str | None = None):
self.client = Anthropic(api_key=api_key)
corpus = "\n\n".join(
f"### {doc['title']}\n{doc['text']}" for doc in documents
)
# The cached prefix must be byte-for-byte identical on every call, so
# everything volatile (the question, timestamps) stays out of it.
self.system = [
{
"type": "text",
"text": (
"You answer questions using only the company handbook below.\n"
"Cite the section heading you used. If the handbook does not "
"cover it, say so.\n\n"
f"{corpus}"
),
"cache_control": {"type": "ephemeral"},
}
]
def ask(self, question: str) -> dict:
started = time.perf_counter()
response = self.client.messages.create(
model=MODEL,
max_tokens=300,
system=self.system,
messages=[{"role": "user", "content": question}],
)
elapsed = time.perf_counter() - started
usage = response.usage
return {
"answer": next(b.text for b in response.content if b.type == "text"),
"latency_s": elapsed,
"cache_write": usage.cache_creation_input_tokens,
"cache_read": usage.cache_read_input_tokens,
"uncached_input": usage.input_tokens,
"output": usage.output_tokens,
}
if __name__ == "__main__":
documents = [
{
"title": "Vacation Policy",
"text": (
"Full-time employees receive 20 days of paid vacation per year. "
"Vacation accrues monthly at 1.67 days per month. Unused vacation "
"rolls over up to 10 days. Employees must request vacation at "
"least 2 weeks in advance except for emergencies. Vacation "
"requests are approved by the direct manager and recorded in the "
"HR system. Employees who leave the company are paid out for "
"accrued but unused vacation, capped at 10 days."
),
},
{
"title": "Remote Work Guidelines",
"text": (
"Employees may work remotely up to 3 days per week. Remote work "
"requires manager approval and must be scheduled in advance. "
"Employees must be available during core hours (10 AM to 3 PM) "
"regardless of location. Fully remote arrangements require "
"director-level approval and are reviewed annually. The company "
"provides a one-time 500 dollar home office stipend."
),
},
{
"title": "Benefits Guide",
"text": (
"The company provides comprehensive health insurance through "
"BlueCross. Coverage starts on the first day of the month "
"following the hire date. Premium is 150 dollars per month for "
"individuals and 350 dollars per month for families. The annual "
"deductible is 1,000 dollars for individuals and 2,000 dollars "
"for families. Dental and vision are included at no extra cost. "
"The company matches 401(k) contributions up to 4 percent."
),
},
{
"title": "Performance Management",
"text": (
"Reviews are conducted annually in December. Employees receive "
"feedback on goals, competencies, and overall performance. "
"Reviews determine merit increases and promotion eligibility. "
"Self-assessment is due by December 1st and the manager review "
"by December 15th. Mid-year check-ins happen in June and are "
"not scored. Promotion decisions are ratified by a calibration "
"committee in January."
),
},
{
"title": "Expense Policy",
"text": (
"Expenses must be submitted within 30 days of being incurred. "
"Receipts are required for any single expense over 25 dollars. "
"Airfare must be booked in economy class for flights under six "
"hours. Meals during travel are reimbursed up to 75 dollars per "
"day. Client entertainment requires pre-approval from a "
"director. Personal vehicle mileage is reimbursed at the "
"current federal rate. Expenses submitted after 60 days are not "
"reimbursed except in documented emergencies."
),
},
{
"title": "Parental Leave",
"text": (
"Birthing parents receive 16 weeks of fully paid leave. "
"Non-birthing parents receive 8 weeks of fully paid leave. "
"Leave must be taken within 12 months of the birth or adoption. "
"Leave can be taken continuously or split into two blocks with "
"manager approval. Parental leave does not count against the "
"vacation balance and continues to accrue vacation normally. "
"Employees returning from parental leave may request a phased "
"return at 60 percent hours for the first four weeks."
),
},
{
"title": "Equipment and Security",
"text": (
"Every employee receives a company laptop refreshed every three "
"years. Full-disk encryption and the company MDM agent are "
"mandatory on any device that touches company data. Personal "
"devices may access email and chat only through the managed "
"container. Lost or stolen equipment must be reported to IT "
"within 24 hours. Software purchases over 100 dollars require "
"IT review for license and security compliance. Production "
"database access requires a documented business justification "
"and is granted for a maximum of 90 days before re-approval."
),
},
{
"title": "Learning and Development",
"text": (
"Each employee has an annual learning budget of 2,000 dollars "
"for courses, books, and conferences. Conference attendance "
"requires manager approval and a written summary shared with "
"the team afterwards. The company covers certification exam "
"fees for role-relevant certifications, including one retake. "
"Employees may use up to 5 working days per year for learning "
"activities. Tuition reimbursement for degree programs is "
"available after two years of service, capped at 10,000 dollars "
"per year, and requires a two-year commitment after completion."
),
},
]
cag = CAGSystem(documents, api_key=os.getenv("ANTHROPIC_API_KEY"))
questions = [
"How many vacation days do I get, and how many can I roll over?",
"What is the home office stipend?",
"What happens to my unused vacation if I quit?",
"How long do I have to submit an expense?",
]
for i, question in enumerate(questions, 1):
result = cag.ask(question)
print(f"\n{'=' * 62}")
print(f"Q{i}: {question}")
print("=" * 62)
print(result["answer"])
print(
f"\n cache write: {result['cache_write']:>5} "
f"cache read: {result['cache_read']:>5} "
f"uncached in: {result['uncached_input']:>3} "
f"out: {result['output']:>3} "
f"latency: {result['latency_s']:.2f}s"
)Expected Output:
============================================================== Q1: How many vacation days do I get, and how many can I roll over? ============================================================== Based on the **Vacation Policy** section: - Full-time employees receive **20 days of paid vacation per year**. - Unused vacation **rolls over up to 10 days**. cache write: 1223 cache read: 0 uncached in: 23 out: 62 latency: 1.65s ============================================================== Q2: What is the home office stipend? ============================================================== The home office stipend is **$500**, provided as a one-time payment (Remote Work Guidelines). cache write: 0 cache read: 1223 uncached in: 13 out: 35 latency: 1.41s ============================================================== Q3: What happens to my unused vacation if I quit? ============================================================== According to the **Vacation Policy** section: if you leave the company, you are paid out for accrued but unused vacation, capped at 10 days. cache write: 0 cache read: 1223 uncached in: 19 out: 86 latency: 2.05s ============================================================== Q4: How long do I have to submit an expense? ============================================================== Expenses must be submitted within 30 days of being incurred. Note that expenses submitted after 60 days are not reimbursed, except in documented emergencies. **Source: Expense Policy** cache write: 0 cache read: 1223 uncached in: 16 out: 65 latency: 1.33s
The counters tell the whole story. The first question wrote 1,223 tokens into the cache and read none. Every question after it read all 1,223 back and sent only 13 to 23 fresh tokens, because the only thing that changed was the question itself. There is no retrieval step anywhere, so there is no chance of retrieving the wrong thing: the model can see the entire handbook every time.
count_tokens reports 598, so every call reports cache write: 0, cache read: 0: no error, no warning, just silently paying full price forever. The minimum is 1,024 tokens on Sonnet 5, 512 on Opus 5, and 4,096 on Haiku 4.5. Always confirm with the usage counters rather than assuming. Lesson 17 covers the cache mechanics, including the silent invalidators that reset it.Choosing Between RAG and CAG
The deciding question is whether your knowledge fits, and it is worth measuring rather than guessing: a 1M-token window is large enough that many "we need a vector database" corpora turn out to fit comfortably.
- A product manual, a policy handbook, one codebase
- Low latency matters: no embedding or search hop
- You want the model to see everything, not a top-K guess
- You would rather run no extra infrastructure
- Millions of documents, or per-tenant data
- Content updates faster than you can re-warm a cache
- Per-user permissions decide what may be seen
- Attribution to a specific source document is required
MCP: Model Context Protocol
RAG and CAG both answer "what should the model know?" MCP answers a different question: "what should the model be able to do?" It is an open protocol for connecting AI assistants to external data sources and tools, a standardized way to give models "plugins", whether that means databases, APIs, file systems, or custom business logic.
What is MCP?
MCP (Model Context Protocol) is an open, standardized protocol for connecting AI assistants to external data sources and tools. It defines a universal interface for exposing resources (read-only data) and tools (actions) to any MCP-compatible AI system.
MCP Architecture
Figure 3: One AI application talks to many MCP servers over a single standardized protocol; each server fronts a different backend
Key Concepts
- Resources: read-only data sources (files, DB records, API responses)
- Tools: actions the AI can take (create, update, delete, execute)
- Prompts: pre-built prompt templates with variables
- Sampling: request the AI to generate content (for complex workflows)
Benefits
- Standardized protocol, not vendor-specific
- Composable: multiple MCP servers work together
- Secure: fine-grained access control
- Extensible: easy to add new integrations
Building an MCP Server
The server is read-only by design, which means it cannot create the database it serves. Run this once first, in the directory the server will run from:
# seed_db.py - create the SQLite database the MCP server exposes
#
# The server is read-only by design, so it cannot create its own data. Run this
# once, in the same directory the server runs from.
import sqlite3
conn = sqlite3.connect("company.db")
cursor = conn.cursor()
cursor.executescript("""
DROP TABLE IF EXISTS employees;
DROP TABLE IF EXISTS departments;
CREATE TABLE departments (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL
);
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
department TEXT NOT NULL,
salary INTEGER
);
""")
cursor.executemany("INSERT INTO departments VALUES (?, ?)", [
(1, "Engineering"),
(2, "Sales"),
(3, "Marketing"),
])
cursor.executemany("INSERT INTO employees VALUES (?, ?, ?, ?)", [
(1, "Alice", "Engineering", 145000),
(2, "Bob", "Engineering", 132000),
(3, "Carol", "Sales", 98000),
(4, "Dan", "Marketing", 91000),
(5, "Erin", "Engineering", 151000),
])
conn.commit()
for table in ("departments", "employees"):
count = cursor.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
print(f"{table}: {count} rows")
conn.close()Expected Output:
departments: 3 rows employees: 5 rows
# mcp_server.py - MCP server for database access
# Uses the official MCP Python SDK (pip install mcp)
from mcp.server import MCPServer
import sqlite3
import json
# Initialize the MCP server. In SDK v1 this class was called FastMCP and lived
# in mcp.server.fastmcp; v2 renamed it to MCPServer with no compatibility alias,
# so the old import raises ModuleNotFoundError.
mcp = MCPServer("company-db-server")
# Database connection
DB_PATH = "company.db"
def get_connection():
return sqlite3.connect(DB_PATH)
@mcp.resource("db://tables")
def list_tables() -> str:
"""List all available database tables."""
conn = get_connection()
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
conn.close()
return json.dumps({"tables": tables}, indent=2)
@mcp.resource("db://tables/{table_name}")
def read_table(table_name: str) -> str:
"""Read data from a specific table (limited to 100 rows)."""
conn = get_connection()
cursor = conn.cursor()
# Look the name up in sqlite_master instead of interpolating it blindly:
# a table name cannot be a bound parameter, so this is what keeps the
# f-string below from becoming an injection point.
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
if not cursor.fetchone():
conn.close()
return json.dumps({"error": f"Table '{table_name}' not found"})
cursor.execute(f"SELECT * FROM [{table_name}] LIMIT 100")
columns = [description[0] for description in cursor.description]
rows = cursor.fetchall()
conn.close()
data = [dict(zip(columns, row)) for row in rows]
return json.dumps(data, indent=2)
@mcp.tool()
def query_database(query: str) -> str:
"""
Execute a read-only SQL query on the database.
Args:
query: SQL SELECT query to execute
"""
# Safety: only allow SELECT queries
if not query.strip().upper().startswith("SELECT"):
return json.dumps({"error": "Only SELECT queries are allowed"})
conn = get_connection()
try:
cursor = conn.cursor()
cursor.execute(query)
columns = [description[0] for description in cursor.description]
rows = cursor.fetchall()
data = [dict(zip(columns, row)) for row in rows]
return json.dumps({
"query": query,
"row_count": len(data),
"results": data
}, indent=2)
except Exception as e:
return json.dumps({"error": f"Query failed: {str(e)}"})
finally:
conn.close()
@mcp.tool()
def get_schema(table_name: str) -> str:
"""
Get the schema (columns, types) of a database table.
Args:
table_name: Name of the table to inspect
"""
conn = get_connection()
cursor = conn.cursor()
# Validate table exists first
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
if not cursor.fetchone():
conn.close()
return json.dumps({"error": f"Table '{table_name}' not found"})
cursor.execute(f"PRAGMA table_info([{table_name}])")
schema = cursor.fetchall()
conn.close()
columns = []
for col in schema:
columns.append({
"name": col[1],
"type": col[2],
"nullable": not col[3],
"primary_key": bool(col[5])
})
return json.dumps({"table": table_name, "columns": columns}, indent=2)
# Run the server (uses stdio transport by default)
if __name__ == "__main__":
mcp.run()mcp.run() speaks JSON-RPC over stdin and stdout, so the process simply waits for a client to say something. There is no banner and no log line: stdout belongs to the protocol, which is also why a stray print() anywhere in an stdio MCP server corrupts the stream and breaks the connection. Use logging to stderr instead.Testing It Without an Assistant
Since the server says nothing on its own, you need a client to see anything at all. Wiring it into Claude Desktop works but makes for a slow debugging loop, and it blurs the two failure modes you most need to tell apart: a server that registered nothing, and a server whose tools registered fine but return the wrong data. This script launches the server as a subprocess and drives the protocol directly.
# test_client.py - exercise the server without wiring it into an assistant
#
# mcp_server.py speaks JSON-RPC over stdin/stdout, so running it directly prints
# nothing and simply waits. This client launches it as a subprocess and drives
# the protocol, which is the fastest way to tell "my tool logic is wrong" apart
# from "my server registered nothing".
import asyncio
import sys
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
def text_of(result):
"""Tool results arrive as a list of content blocks."""
return "".join(block.text for block in result.content if block.type == "text")
async def main():
params = StdioServerParameters(command=sys.executable, args=["mcp_server.py"])
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# What did the server actually advertise?
tools = await session.list_tools()
print("Tools:", [t.name for t in tools.tools])
resources = await session.list_resources()
print("Resources:", [str(r.uri) for r in resources.resources])
# Templated resources are listed separately from fixed ones.
templates = await session.list_resource_templates()
print("Templates:", [t.uri_template for t in templates.resource_templates])
print("\n--- read db://tables ---")
result = await session.read_resource("db://tables")
print(result.contents[0].text)
print("\n--- call get_schema('employees') ---")
result = await session.call_tool("get_schema", {"table_name": "employees"})
print(text_of(result))
print("\n--- call query_database (GROUP BY) ---")
result = await session.call_tool("query_database", {
"query": "SELECT department, COUNT(*) AS headcount FROM employees "
"GROUP BY department ORDER BY department"
})
print(text_of(result))
print("\n--- a write, which the server should refuse ---")
result = await session.call_tool("query_database", {
"query": "DELETE FROM employees"
})
print(text_of(result))
print("\n--- a table that does not exist ---")
result = await session.call_tool("get_schema", {"table_name": "nope"})
print(text_of(result))
asyncio.run(main())Expected Output:
Tools: ['query_database', 'get_schema']
Resources: ['db://tables']
Templates: ['db://tables/{table_name}']
--- read db://tables ---
{
"tables": [
"departments",
"employees"
]
}
--- call get_schema('employees') ---
{
"table": "employees",
"columns": [
{
"name": "id",
"type": "INTEGER",
"nullable": true,
"primary_key": true
},
{
"name": "name",
"type": "TEXT",
"nullable": false,
"primary_key": false
},
{
"name": "department",
"type": "TEXT",
"nullable": false,
"primary_key": false
},
{
"name": "salary",
"type": "INTEGER",
"nullable": true,
"primary_key": false
}
]
}
--- call query_database (GROUP BY) ---
{
"query": "SELECT department, COUNT(*) AS headcount FROM employees GROUP BY department ORDER BY department",
"row_count": 3,
"results": [
{
"department": "Engineering",
"headcount": 3
},
{
"department": "Marketing",
"headcount": 1
},
{
"department": "Sales",
"headcount": 1
}
]
}
--- a write, which the server should refuse ---
{"error": "Only SELECT queries are allowed"}
--- a table that does not exist ---
{"error": "Table 'nope' not found"}Two things in that transcript are worth a second look. The templated resource is listed separately from the fixed one, under Templates, so a client that only calls list_resources() never learns that db://tables/{table_name} exists. And get_schema reports the primary key as nullable, which is not a bug in the code but a genuine SQLite quirk: an INTEGER PRIMARY KEY column is an alias for the rowid and carries no NOT NULL flag, so PRAGMA table_info honestly reports notnull = 0.
get_schema to learn the shape of the table and then query_database to run the aggregate. You write the server once and every MCP client can use it.When to Use MCP
- Database access: Read and write with tools, schema inspection
- File system access: Read files, search, create documents
- API integrations: Slack, GitHub, Jira, external services
- Custom business logic: Pricing calculations, workflow automation
- Real-time data: Stock prices, weather, news feeds
MCP vs RAG: When to Use Which
✓ Use RAG when:
- You have static documents/knowledge base
- Need semantic search over large corpus
- Documents don't change frequently
- Read-only access is sufficient
- Example: Company handbook Q&A
✓ Use MCP when:
- Need real-time data access
- AI needs to perform actions (not just read)
- Multiple data sources/tools
- Dynamic data (databases, APIs)
- Example: AI assistant that updates CRM
Everything Else in the Prompt
RAG, CAG, and MCP decide what knowledge and capabilities reach the model, but they are not the only things competing for room in the context window. A production prompt also carries the system instructions, the conversation so far, the tool definitions, and whatever per-user state the application tracks. Deciding how to spend that budget, what to summarise, what to prune, and what to keep byte-stable so the cache survives, is its own discipline.
RAG vs MCP vs CAG: Comparison
| Aspect | RAG | MCP | CAG |
|---|---|---|---|
| Purpose | Retrieve the relevant slice at query time | Connect the model to tools and live data | Preload the whole corpus once and cache it |
| Primary Use | Q&A over a corpus too big to fit | AI agents that take actions | Q&A over a corpus that does fit |
| Data Access | Read-only (documents) | Read & Write (tools) | Read-only (documents) |
| Retrieval step | Yes, on every request | Not applicable | None: that is the point |
| Freshness | Re-index changed documents | Live on every call | Re-warm the cache when the corpus changes |
| Corpus size limit | Effectively unbounded | Not applicable | Must fit the context window |
| Moving parts | Embedding model + vector DB + retriever | A server per integration | None beyond the API call |
| Standardization | No standard (many implementations) | Yes: MCP, governed by the Linux Foundation | No standard (provider caching APIs differ) |
Best Practices
1. Start Simple, Add Complexity as Needed
Begin with a prompt that simply contains what the model needs. Add CAG when the corpus outgrows the prompt but still fits the window, RAG when it outgrows the window, and MCP when the model needs to act rather than read. Let requirements pull you up that ladder instead of starting at the top.
2. Monitor Context Token Usage
Context is not free. Track token usage and costs. Prioritize most relevant context. Use summarization for lengthy documents. Stay within context window limits to avoid truncation or errors.
3. Implement Context Caching
Cache frequently used context (user profiles, company info, common docs) to reduce costs and latency. Use prompt caching features when available. Invalidate caches appropriately when data changes.
4. Validate Retrieved Context
Check relevance of retrieved documents. Low similarity scores indicate poor retrieval. Re-rank results. Filter out irrelevant context before sending to LLM to improve response quality and reduce token waste.
5. Keep Context Fresh
Re-index documents regularly. Add timestamps to context. Invalidate caches when data changes. Stale context is worse than no context, it leads to incorrect, outdated answers that erode user trust.
6. Provide Source Attribution
Always cite where information came from. Include document titles, timestamps, URLs. Enables users to verify information. Builds trust in the system and allows users to dive deeper into sources.
Key Takeaways
- RAG retrieves at query time - Semantic search picks a slice of the corpus, for when the corpus is too large to fit
- CAG preloads and caches - Cache-Augmented Generation loads the whole corpus once and reuses the cached prefix, so there is no retrieval to get wrong
- "Does it fit?" is the deciding question - Measure your corpus before assuming you need a vector database
- MCP is a different axis - RAG and CAG supply knowledge, MCP supplies capability, so MCP stacks on top of either
- MCP is vendor-neutral now - Created by Anthropic, donated to the Linux Foundation's Agentic AI Foundation in December 2025
- Real similarity scores are lower than you expect - 0.68 was the top match here, not 0.94, so calibrate thresholds against your own corpus
- Top-K always returns K - It returns the least-bad documents even when nothing is relevant, so check the scores
- Verify caching with the usage counters - Below the minimum prefix, caching silently does nothing and reports zeros
- Current Claude models reject temperature - Ground answers with retrieved context and the system prompt instead
- Cite sources - Return the documents and scores you used, so users can verify the answer