Every coding agent I use starts each session knowing nothing about the project. It has not read the PRD for the feature it is building. It does not know we decided on JSONB for filters six months ago, or why. It writes #c62828 into a component because it has never seen the design system. So I paste things in, again, every session.
We were spending too much time and tokens waiting and repeating the same things over and over. Explaining architectural and design decisions agreed upon at the beginning of the project. Wondering why we were burning through tokens as we built more features.
The fix I settled on is a small MCP server that runs on my laptop. It indexes the project’s PRDs, architecture decision records, design system docs, agent profiles, Jira tickets, and source code into LanceDB, a vector database that lives in a folder on disk. Any MCP client (Claude Code, Claude Desktop, Cursor) can then search it. Nothing leaves the machine: the embedding model runs locally too.
It is about 250 lines of Python across four files.
One caveat is that there can be drift when your MCP repo sits outside your codebase repo and you need to update 2 repos with changes. You can fix it by using hooks, so when a PR is created on the codebase repo open a PR on the MCP repo with corresponding changes.
Or have a live Context Graph service that is updated for the entire team to share. Look here: Obie Fernandez – Nexus
The pieces
MCP, the Model Context Protocol, is how a client discovers and calls tools on a server. A local server is just a process the client starts and talks to over stdin and stdout. No ports, no auth.
LanceDB is an embedded database, like SQLite for vectors. lancedb.connect("~/.local/share/kb") opens a directory; there is no server to run. It does vector search, full-text search, and both at once (hybrid), with SQL-style filters on regular columns.
The embedding model is BAAI/bge-small-en-v1.5 through sentence-transformers. It is small (384 dimensions), downloads once from Hugging Face, and runs fine on a laptop CPU. Later I swapped it for nomic-embed-text running in Ollama; that has its own section below.
The knowledge base is a plain folder of Markdown, one subfolder per kind of document:
~/kb/
prd/ saved-searches.md, billing-v2.md
adr/ 0007-use-postgres-jsonb-for-filters.md
design-system/ color.md, spacing.md, buttons.md
agents/ reviewer.md, planner.mdThe codebase stays where it is. The indexer reads it in place.
Setup
uv init --python 3.12 kb-mcp && cd kb-mcp
uv add "mcp[cli]" lancedb sentence-transformersThis pulls in PyTorch for the embedding model, so the first install is large: the virtualenv came to about 1.1 GB. The Ollama setup further down avoids that.
A note on versions, because it will save you an hour: this is written against mcp 2.3 and lancedb 0.40. In the 2.x Python SDK, FastMCP was renamed MCPServer and moved to mcp.server.mcpserver. Most MCP tutorials still show from mcp.server.fastmcp import FastMCP, which now fails on import with a message pointing at the migration guide.
One table, a kind column
I started with a table per document type and dropped it quickly. One table with a kind column is simpler to keep in sync, and the search tool can still filter to one kind when the agent knows what it wants. The shared module defines the schema and where things live:
import os
from pathlib import Path
import lancedb
from lancedb.embeddings import get_registry
from lancedb.pydantic import LanceModel, Vector
KB_ROOT = Path(os.environ.get("KB_ROOT", "~/kb")).expanduser()
CODE_ROOTS = [Path(p).expanduser() for p in os.environ.get("KB_CODE", "").split(":") if p]
DB_PATH = Path(os.environ.get("KB_DB", "~/.local/share/kb")).expanduser()
# folder under KB_ROOT -> the kind stored on each chunk
KINDS = {"prd": "prd", "adr": "adr", "design-system": "design", "agents": "agent"}
embedder = get_registry().get("sentence-transformers").create(name="BAAI/bge-small-en-v1.5")
class Chunk(LanceModel):
kind: str
path: str
title: str
file_hash: str
text: str = embedder.SourceField()
vector: Vector(embedder.ndims()) = embedder.VectorField()
def open_table():
db = lancedb.connect(DB_PATH)
return db.create_table("chunks", schema=Chunk, exist_ok=True)The embedding function is attached to the schema. LanceDB stores that in the table’s metadata, so table.add() embeds the text column on the way in, and table.search("some words") embeds the query with the same model. I never call the model myself, and the indexer and the server can never drift onto different models.
Local embeddings with nomic-embed-text
sentence-transformers works, but it drags PyTorch into the server’s virtualenv, and the model loads into every process that opens the table. I now run embeddings through Ollama with nomic-embed-text instead. Ollama is a separate local daemon, so the model loads once and stays warm, and the MCP server only makes HTTP calls to localhost:11434.
nomic-embed-text is also a better fit for this job. It produces 768-dimension vectors, takes 2,048 tokens of input in Ollama’s build, four times bge-small’s 512, so an 80-line code chunk usually fits without truncation, and it is about 270 MB to pull.
ollama pull nomic-embed-text
uv add ollama
uv remove sentence-transformersRemoving sentence-transformers took my virtualenv from 1.1 GB to 355 MB, with no PyTorch left in it.
LanceDB has an ollama embedding function built in, so swapping is nearly one line. Nearly, because of one detail that is easy to miss: nomic-embed-text was trained with task prefixes. Documents should be embedded as search_document: ... and queries as search_query: .... Leave them off and it still returns vectors, just worse ones, and nothing warns you. LanceDB’s built-in Ollama embedder sends text as-is, so I subclass it and add the prefixes. In kb.py, the embedder line becomes:
from lancedb.embeddings import get_registry, register
from lancedb.embeddings.ollama import OllamaEmbeddings
@register("nomic-ollama")
class NomicOllama(OllamaEmbeddings):
"""nomic-embed-text was trained with task prefixes; leaving them off costs accuracy."""
def compute_source_embeddings(self, texts, *args, **kwargs):
texts = self.sanitize_input(texts)
return self.generate_embeddings([f"search_document: {t}" for t in texts])
def compute_query_embeddings(self, query, *args, **kwargs):
return self.generate_embeddings([f"search_query: {query}"])
embedder = get_registry().get("nomic-ollama").create(name="nomic-embed-text")LanceDB already calls compute_source_embeddings when rows are added and compute_query_embeddings when you search, hybrid search included, so those two overrides put each prefix in the right place. @register matters as well. The table’s metadata stores the embedder by its registered name, nomic-ollama, so any process that opens the table has to import kb.py first. The indexer and the server both do.
Two things to know when you switch:
- Rebuild from scratch. The vector column is now 768 wide instead of 384, and vectors from different models cannot be compared anyway. Delete the
KB_DBfolder and runingest.pyagain. - Ollama has to be running before the server starts.
embedder.ndims()embeds a test string at import time to find the vector width, so if the daemon is down, the server dies on startup and the client shows it as failed to connect. Running Ollama as a login item orbrew services start ollamatakes care of it.
On the sample project, the same queries returned the same top results as with bge-small. [WHETHER NOMIC FOUND THINGS ON YOUR REAL PROJECT THAT BGE-SMALL MISSED, OR WAS NOTICEABLY SLOWER TO INDEX]
Chunking
How you split documents decides most of how good search feels. I use two rules.
Markdown is split at headings, one chunk per section, and every chunk starts with the document title. A section called “Consequences” means nothing alone; “ADR 0007: Store filter sets as JSONB / Consequences” is findable.
def split_markdown(text: str, doc_title: str):
"""One chunk per heading section, with the doc title and heading kept on each chunk."""
sections = re.split(r"(?m)^(?=#{1,3} )", text)
for section in sections:
section = section.strip()
if not section:
continue
heading = section.splitlines()[0].lstrip("# ").strip()
yield f"{doc_title} > {heading}", f"{doc_title}\n\n{section}"Code is split into overlapping windows of 80 lines, with the file path and line range at the top. It is crude. A parser-aware splitter (tree-sitter, one chunk per function) would do better, and it is the next thing I would add. In practice the agent mostly needs to land in the right file, and it can read the rest itself.
def split_code(text: str, rel: str, window: int = 80, step: int = 60):
"""Overlapping line windows. Crude, but code search mostly needs to land near the right lines."""
lines = text.splitlines()
for start in range(0, max(len(lines), 1), step):
part = lines[start : start + window]
if not part:
break
title = f"{rel}:{start + 1}-{start + len(part)}"
yield title, f"{title}\n\n" + "\n".join(part)
if start + window >= len(lines):
breakRe-indexing only what changed
Each chunk stores a SHA-256 of the file it came from. On each run the indexer loads the stored hash per path, skips files whose hash matches, deletes and re-adds the ones that changed, and removes chunks for files that no longer exist:
def main():
table = open_table()
indexed = {}
if table.count_rows():
for row in table.search().select(["path", "file_hash"]).limit(None).to_list():
indexed[row["path"]] = row["file_hash"]
seen, added = set(), 0
for kind, path in discover():
key = str(path)
seen.add(key)
file_hash = hashlib.sha256(path.read_bytes()).hexdigest()
if indexed.get(key) == file_hash:
continue
table.delete(where_path(key))
rows = list(chunks_for(kind, path, file_hash))
if rows:
table.add(rows)
added += len(rows)
for gone in set(indexed) - seen:
table.delete(where_path(gone))
table.create_index("text", config=FTS(), replace=True)
print(f"{added} chunks added, {len(set(indexed) - seen)} files removed, {table.count_rows()} chunks total")The last line rebuilds the full-text index, which the hybrid search needs. Running it on a sample project, then again with nothing changed:
$ uv run python ingest.py
14 chunks added, 0 files removed, 14 chunks total
$ uv run python ingest.py
0 chunks added, 0 files removed, 14 chunks total[NUMBERS FROM YOUR REAL PROJECT: FILES, CHUNKS, FIRST INDEX TIME, INCREMENTAL TIME, SIZE ON DISK]
The server
The server exposes two tools and one prompt:
from pathlib import Path
from typing import Literal
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from kb import CODE_ROOTS, KB_ROOT, open_table
mcp = MCPServer(
"kb",
instructions=(
"Project knowledge: PRDs, architecture decision records (ADRs), the design system, "
"agent profiles and the codebase. Search here before guessing at requirements, "
"past decisions or design tokens."
),
)
table = open_table()
Kind = Literal["prd", "adr", "design", "agent", "code"]
@mcp.tool()
def search(query: str, kind: Kind | None = None, limit: int = 8) -> list[dict]:
"""Hybrid (vector + keyword) search over the knowledge base.
Use kind to narrow it: prd for requirements, adr for past architecture decisions,
design for tokens and component rules, agent for agent profiles, code for source.
"""
q = table.search(query, query_type="hybrid").limit(limit)
if kind:
q = q.where(f"kind = '{kind}'", prefilter=True)
return [
{"kind": r["kind"], "path": r["path"], "title": r["title"], "text": r["text"]}
for r in q.to_list()
]
@mcp.tool()
def read_document(path: str) -> str:
"""Return a whole file from the knowledge base or codebase, by the path a search result gave."""
target = Path(path).resolve()
roots = [KB_ROOT.resolve(), *(r.resolve() for r in CODE_ROOTS)]
if not any(target.is_relative_to(root) for root in roots):
raise ToolError(f"{path} is outside the indexed folders")
return target.read_text()
@mcp.prompt()
def agent(name: str) -> str:
"""Load an agent profile from agents/<name>.md as the working instructions."""
profile = (KB_ROOT / "agents" / f"{name}.md").resolve()
if not profile.is_relative_to((KB_ROOT / "agents").resolve()):
raise ValueError("bad agent name")
return profile.read_text()
if __name__ == "__main__":
mcp.run()A few choices in there are deliberate.
Hybrid search, not pure vectors. Vector search is good at “why do we store filters this way” and bad at exact names. Ask it for MAX_SAVED and it returns things that are about limits in general. Keyword search finds the identifier. query_type="hybrid" runs both and merges the results with reciprocal rank fusion, which is the default reranker.
kind is a Literal. The SDK turns it into an enum in the tool’s JSON schema, so the client sees exactly which values are valid. It also means the value is checked before it reaches the where clause, so it cannot be used to inject a filter.
Search returns chunks; read_document returns whole files. Chunks keep the search response small. When a chunk is the right one, the agent asks for the whole file. The path check stops it from reading anything outside the indexed folders: asking for /etc/passwd gets “outside the indexed folders”. In the 2.x SDK, raise ToolError for errors like this. Any other exception is treated as a crash and logged with a full traceback.
Agent profiles are a prompt, not a tool. MCP prompts show up as slash commands in Claude Code, so /mcp__kb__agent reviewer loads the reviewer profile as the instructions for the session. The profiles are also indexed, so a search for “who reviews PRs” still finds them.
The instructions string matters. It is sent to the client when it connects, and it is the only hint the model gets about when to reach for this server rather than grepping around. Mine says, in effect, “search here before guessing”.
One trap with stdio servers: stdout is the protocol channel. A stray print() in the server corrupts the stream and the client drops the connection. Log to stderr. The indexer is a separate script, so it can print whatever it likes.
Wiring it into Claude Code
claude mcp add kb --scope user \
-e KB_ROOT=$HOME/kb \
-e KB_CODE=$HOME/Workspace/myapp \
-e KB_DB=$HOME/.local/share/kb \
-- uv run --directory $HOME/Workspace/kb-mcp python server.py--scope user makes it available in every project. uv run --directory means the client does not need to know about the virtualenv. Run /mcp inside Claude Code to check it connected.
For Claude Desktop or Cursor, the same command and env vars go in their JSON config under mcpServers.
I tested it with a small client script before pointing a real agent at it. Each query is followed by the top results:
how many saved searches can a user keep
code app/saved_searches.py:1-11
prd PRD: Saved searches > Requirements
why are filters stored as json (kind=adr)
adr ADR 0007: Store filter sets as JSONB
what color for the delete button (kind=design)
design Color tokens > Destructive actionsThe first one is the case I care about. One question returns both the requirement (“max 20”) and the code that enforces it, without the agent knowing either exists.
Measuring it
Once agents were calling the server on their own, I wanted to know two things: how fast search is, and how much each call costs in context.
The second one has a catch. The MCP server never sees the model’s token usage; the client counts that and does not report it back. What the server can see is the size of every response it sends, and that response lands in the agent’s context window. That is the part of the token bill this server is responsible for, so that is what it measures, at roughly four characters per token. It is an estimate, but it is fine for comparing one call with another.
Everything goes into a decorator that times each call and appends a line of JSON to a log next to the database:
import json
import statistics
import time
from datetime import datetime, timedelta, timezone
from functools import wraps
from kb import DB_PATH
LOG = DB_PATH / "metrics.jsonl"
def approx_tokens(value) -> int:
"""About 4 characters per token. Rough, but good enough to compare one call with another."""
text = value if isinstance(value, str) else json.dumps(value)
return len(text) // 4
def metered(fn):
"""Record wall time and response size for every call, including failed ones."""
@wraps(fn)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result, error = None, None
try:
result = fn(*args, **kwargs)
return result
except Exception as exc:
error = type(exc).__name__
raise
finally:
entry = {
"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"tool": fn.__name__,
"args": kwargs,
"ms": round((time.perf_counter() - start) * 1000, 1),
"tokens": approx_tokens(result) if result is not None else 0,
"results": len(result) if isinstance(result, list) else None,
"error": error,
}
LOG.parent.mkdir(parents=True, exist_ok=True)
with LOG.open("a") as f:
f.write(json.dumps(entry, default=str) + "\n")
return wrapper
def summarize(hours: int = 24) -> str:
if not LOG.exists():
return "no calls recorded yet"
since = datetime.now(timezone.utc) - timedelta(hours=hours)
rows = [json.loads(line) for line in LOG.read_text().splitlines()]
rows = [r for r in rows if datetime.fromisoformat(r["ts"]) >= since]
lines = [f"last {hours}h", f"{'tool':<15}{'calls':>6}{'errors':>7}{'p50 ms':>9}{'p95 ms':>9}{'avg tok':>9}{'total tok':>11}"]
for tool in sorted({r["tool"] for r in rows}):
calls = [r for r in rows if r["tool"] == tool]
ms = sorted(r["ms"] for r in calls)
tokens = [r["tokens"] for r in calls]
p95 = ms[min(len(ms) - 1, round(0.95 * (len(ms) - 1)))]
lines.append(
f"{tool:<15}{len(calls):>6}{sum(1 for r in calls if r['error']):>7}"
f"{statistics.median(ms):>9.0f}{p95:>9.0f}{statistics.mean(tokens):>9.0f}{sum(tokens):>11}"
)
return "\n".join(lines)
if __name__ == "__main__":
import sys
print(summarize(int(sys.argv[1]) if len(sys.argv) > 1 else 24))In the server, each tool gets @metered under @mcp.tool(), and the summary is exposed as a resource:
from metrics import metered, summarize
@mcp.tool()
@metered
def search(query: str, kind: Kind | None = None, limit: int = 8) -> list[dict]:
...
@mcp.resource("kb://stats")
def stats() -> str:
"""Call counts, latency and approximate response tokens per tool, last 24 hours."""
return summarize()The order matters. @wraps keeps the original signature visible, and the SDK builds the tool’s JSON schema from that signature, so the client still sees query, kind and limit. The log write sits in a finally block, so failed calls are recorded too, with the exception name.
The stats are a resource rather than a tool on purpose. Every tool’s description goes into the agent’s context on every session, and the agent has no use for latency numbers. In Claude Code you pull them in yourself with @kb:kb://stats, or run uv run python metrics.py in a terminal:
last 24h
tool calls errors p50 ms p95 ms avg tok total tok
read_document 2 1 0 0 48 95
search 6 0 18 124 734 4405That is from the sample project. Search takes 18 ms at the median. The 124 ms p95 is the first call, which pays to warm up the model and the index. The token column is the more useful one: with the default limit=8, every search puts about 700 tokens into context, even on tiny documents. On a real codebase with 80-line chunks it will be several times that, which makes limit the first thing worth tuning.
To get real numbers I pointed it at Plamotaku: its PRD and ADR documents plus src, supabase and scripts, about 270 files and 32,000 lines of TypeScript and SQL. (Not the repo root. That would have pulled in 1,600 build files from .next and a Python virtualenv, neither of which is in the skip list.) I built the index once with each embedding model, then ran the same 20 questions through the MCP server, the kind an agent asks: “who is allowed to edit a collection”, “why not a custom API server”, “kit_prices”. On an M1 Pro with 16 GB:
| bge-small (in process) | nomic-embed-text (Ollama) | |
|---|---|---|
| Full index, 651 chunks | 43 s | 79 s |
| Re-index, nothing changed | 7.9 s | 1.6 s |
| Index on disk | 14 MB | 15 MB |
| Server ready after start | 6.3 s | 1.3 s |
| Search p50 / p95 | 28 / 47 ms | 37 / 76 ms |
Tokens per search, limit=8 | 4,828 avg, 8,134 max | 4,517 avg, 7,113 max |
Tokens per search, limit=4 | 2,178 avg | 2,098 avg |
Search speed is a non-issue with either model. Tens of milliseconds disappears next to the seconds the model spends thinking. nomic is a little slower per query because every query is an HTTP call to Ollama, but it starts five times faster, and a re-index with nothing to do takes 1.6 seconds instead of 8, because the server and the indexer no longer load a model of their own. The slow full index only happens once.
Tokens are the real cost. At limit=8, a single search on a real codebase put almost 5,000 tokens into the agent’s context, and the worst one over 8,000, nearly all of it code chunks. An agent that searches five times while planning a change has spent 25,000 tokens before writing a line. Halving the limit halves the cost, and the result that matters is nearly always in the top four anyway. On this repo, 4 is the better default, and an agent that needs more can ask for it.
As for which model finds things better: the top result was the same for 13 of the 20 questions. Hybrid search does a lot of the work, since exact names like kit_prices or createClient are found by the keyword side either way. Where they differed, nomic was usually better. For “supabase client for server components” it ranked the ADR on the three Supabase client factories first, and for “user profile page” it found the actual profile page. It also put the “Non-goals” section in the top three for “what is out of scope”, which bge-small missed. Both missed on “who is allowed to edit a collection”: neither ranked the row level security ADR or the policy migration in the top three, which is the answer. Code that enforces a rule rarely uses the words a person would use to ask about it.
The log also keeps every query. Reading what an agent actually searched for shows which documents it goes looking for and does not find.
Things I would add next, in rough order:
- Function-level code chunks with tree-sitter, so a code result is a whole function instead of an arbitrary window.
- Metadata from front matter. An ADR’s status (proposed, accepted, superseded) belongs in its own column, so search can skip superseded decisions.
- One index per project. Point
KB_DBat a folder inside each repo and register the server with--scope project, so each repo carries its own knowledge.