Локальный семантический поиск по заметкам как MCP-сервер: нарезка под лимит модели и кэш

У меня есть папка с конспектами созвонов и расшифровками встреч, и искать в ней по слову неудобно: я помню, что обсуждали найм, а в файле написано «собеседование» и «кандидат», и обычный поиск по тексту такой файл не находит. Отправлять заметки во внешний сервис ради эмбеддингов мне не хотелось, поэтому я собрал маленький MCP-сервер, который ищет по смыслу прямо на моей машине, и подключил его к Claude Code. Всё работает без платных API, а заметки не покидают компьютер.

В статье разберу, как устроен движок и сервер, и три места, на которых легко споткнуться: молчаливое обрезание текста моделью, лишний вывод в стандартный поток, из-за которого ломается протокол, и проверка пути в инструменте чтения. Отдельно покажу кэш, который сам понимает, что устарел. Код ниже взят из рабочих файлов, лишнее я убрал.

Что получилось

Два файла на Python и папка с заметками:

search_index.py   # движок: нарезка, эмбеддинги, кэш, поиск
server.py         # MCP-сервер: три инструмента поверх движка
notes/            # сами заметки, .md и .txt
.index/           # кэш, создаётся сам, в git не попадает

Сервер отдаёт Claude Code три инструмента:

• search_notes(query, limit=5) ищет по смыслу и возвращает фрагменты с именем файла и оценкой близости;

• list_notes() показывает список заметок с размером и датой изменения;

• read_note(filename) отдаёт заметку целиком.

Все три только читают, и я так и пометил их в коде, чтобы клиент мог вызывать их без лишних вопросов.

Для тех, кто с MCP не работал: это протокол, по которому программа вроде Claude Code запускает вашу маленькую программу-сервер и спрашивает у неё, какие инструменты она умеет. Дальше модель сама решает, когда какой вызвать. Общаются они через стандартный ввод и вывод, никакого сетевого порта открывать не надо.

Стек и версии

[project]
requires-python = ">=3.11"
dependencies = [
    "mcp==1.27.2",
    "sentence-transformers==5.5.1",
    "numpy==2.4.6",
]

[tool.uv]
package = false

Версии я зафиксировал жёстко. package = false говорит uv, что это приложение для прямого запуска и собирать его как пакет не надо. Минимальный Python 3.11 появился из-за numpy 2.4.6: я сначала поставил >=3.10, а менеджер пакетов напомнил, что эта версия numpy 3.10 уже не поддерживает.

Эмбеддинги (это набор чисел, по которому сравнивают смысл текстов) считает модель paraphrase-multilingual-MiniLM-L12-v2. Она весит около 120 МБ, знает больше пятидесяти языков, включая русский и английский, и выдаёт векторы длиной 384. Скачивается один раз при первом запуске, потом работает без интернета. Вместе с sentence-transformers ставится PyTorch, это ещё один-два гигабайта на диске, один раз.

В плане у меня была модель покрупнее, paraphrase-multilingual-mpnet-base-v2. Она точнее, но около гигабайта и медленнее. Для нескольких десятков заметок такая точность мне не нужна, поэтому она осталась в списке «если не хватит».

Грабля первая: модель молча режет длинный вход

У MiniLM жёсткий лимит в 128 токенов на вход (токеном называют кусок слова, которым модель «читает» текст). Всё, что длиннее, она отбрасывает без предупреждения: ошибки нет, поиск работает, просто хвост каждого длинного фрагмента в индексе не участвует.

Сначала я собирался резать заметки по 400–600 символов, как обычно советуют. Но у русского текста токенов на символ выходит больше, чем у английского, и 600 символов могут не поместиться в 128 токенов, а с ними потерялся бы и хвост фрагмента.

Поэтому главным ограничителем я сделал токены, а символы оставил ориентиром. Считает токены токенайзер самой модели, и лимит я тоже читаю из модели:

CHAR_TARGET = 500      # ориентир размера фрагмента (символы)
CHAR_MAX = 600         # верхняя граница фрагмента (символы)
CHAR_OVERLAP = 80      # перекрытие между соседними фрагментами (символы)
TOKEN_MARGIN = 4       # запас под спец-токены модели ([CLS]/[SEP] и т.п.)
FALLBACK_MAX_SEQ = 128 # если у модели вдруг не считался лимит

...
self.model = SentenceTransformer(MODEL_NAME)
self.tokenizer = self.model.tokenizer
max_seq = int(getattr(self.model, "max_seq_length", FALLBACK_MAX_SEQ) or FALLBACK_MAX_SEQ)
self.token_cap = max(8, max_seq - TOKEN_MARGIN)

Запас в четыре токена нужен под служебные метки, которые модель сама добавляет к началу и концу текста. Итого потолок на фрагмент получается 124 токена.

Сама нарезка идёт в два прохода. Сначала текст разбивается на «кирпичики» по строкам; слишком длинную строку я режу по концам предложений, а слишком длинное предложение по символам:

def _split_units(text: str) -> list[str]:
    units: list[str] = []
    for line in text.split("n"):
        line = line.strip()
        if not line:
            continue
        if len(line) <= CHAR_MAX:
            units.append(line)
            continue
        for part in re.split(r"(?<=[.!?…])s+", line):
            part = part.strip()
            if not part:
                continue
            if len(part) <= CHAR_MAX:
                units.append(part)
            else:
                for i in range(0, len(part), CHAR_MAX):
                    units.append(part[i:i + CHAR_MAX])
    return units

Потом каждый кирпичик проверяется на токены. Если он сам не влезает, его режут уже точно по токенам, через encode и decode:

def _split_by_tokens(tokenizer, text: str, token_cap: int) -> list[str]:
    ids = tokenizer.encode(text, add_special_tokens=False)
    if len(ids) <= token_cap:
        return [text]
    pieces = []
    for i in range(0, len(ids), token_cap):
        piece = tokenizer.decode(ids[i:i + token_cap], skip_special_tokens=True).strip()
        if piece:
            pieces.append(piece)
    return pieces

Второй проход жадно складывает кирпичики во фрагменты и проверяет оба ограничения сразу, и символьное, и токенное. Когда фрагмент заполнился, в начало следующего переносится хвост предыдущего на 80 символов, чтобы мысль, разрезанная посередине, не терялась:

chunks: list[str] = []
cur = ""
for unit in safe_units:
    candidate = f"{cur} {unit}".strip() if cur else unit
    too_long = len(candidate) > CHAR_MAX or _count_tokens(tokenizer, candidate) > token_cap
    if cur and too_long:
        chunks.append(cur)
        tail = cur[-CHAR_OVERLAP:]
        start = f"{tail} {unit}".strip()
        if len(start) <= CHAR_MAX and _count_tokens(tokenizer, start) <= token_cap:
            cur = start
        else:
            cur = unit
    else:
        cur = candidate
if cur:
    chunks.append(cur)

Число токенов считается заново на каждом шаге, это не самое быстрое решение, зато код остаётся простым. Скорость нарезки я не замерял.

На трёх первых тестовых заметках получилось 12 фрагментов. Размер по символам вышел от 323 до 492, в среднем 414. По токенам от 88 до 123 при потолке 124. Выходит, на русском тексте токенный потолок срабатывает раньше символьного, и фрагменты получаются короче «целевых» 500 символов, но ничего не обрезается. Эти цифры я брал из статистики самотеста.

Кэш, который сам понимает, что устарел

Считать эмбеддинги для всех заметок при каждом запросе незачем. Я кладу их в файл .index/cache.pkl, а ключом служит относительный путь заметки плюс время её изменения. Одного имени файла не хватило бы, ведь в разных подпапках могут лежать файлы с одинаковым названием.

Но у кэша есть вторая причина стать недействительным: если я поменяю модель или параметры нарезки, старые векторы будут от другой нарезки и сравнивать с ними нельзя. Для этого в кэш записан словарь параметров, и при любом расхождении кэш собирается с нуля:

def _current_params(token_cap: int) -> dict:
    return {
        "version": CACHE_VERSION,
        "model": MODEL_NAME,
        "char_target": CHAR_TARGET,
        "char_max": CHAR_MAX,
        "char_overlap": CHAR_OVERLAP,
        "token_cap": token_cap,
    }

Поле version я завёл на случай, когда поменяю саму структуру кэша: увеличиваю число, и все старые файлы считаются чужими.

Запись кэша идёт через временный файл с последующей заменой, чтобы сбой посреди записи не оставил полуфайл:

def _save_cache(cache: dict) -> None:
    INDEX_DIR.mkdir(exist_ok=True)
    tmp = CACHE_FILE.with_suffix(".tmp")
    with open(tmp, "wb") as fh:
        pickle.dump(cache, fh)
    tmp.replace(CACHE_FILE)  # атомарная замена

Битый кэш при чтении не роняет сервер: ловится исключение, пишется предупреждение, индекс строится заново.

При сборке индекса для каждого файла сравнивается время изменения, и пересчитываются только изменившиеся:

for rel, abs_path, mtime in notes:
    current_rel.add(rel)
    entry = files_cache.get(rel)
    if entry and entry["mtime"] == mtime:
        file_chunks = entry["chunks"]
        file_emb = entry["embeddings"]
    else:
        text = _read_text(abs_path, rel)
        if text is None or not text.strip():
            files_cache.pop(rel, None)
            continue
        texts = _chunk_text(self.tokenizer, text, self.token_cap)
        if not texts:
            files_cache.pop(rel, None)
            continue
        file_emb = self._embed(texts)
        file_chunks = [
            {"rel_path": rel, "index": i + 1, "text": t,
             "n_tokens": _count_tokens(self.tokenizer, t)}
            for i, t in enumerate(texts)
        ]
        files_cache[rel] = {"mtime": mtime, "chunks": file_chunks, "embeddings": file_emb}
        n_recomputed += 1

    chunk_meta.extend(file_chunks)
    emb_blocks.append(file_emb)

for rel in list(files_cache.keys()):
    if rel not in current_rel:
        del files_cache[rel]

Последний цикл убирает из кэша файлы, которых на диске уже нет, поэтому удалённая заметка пропадает из поиска сама.

Дальше матрица склеивается в один массив, np.vstack(emb_blocks). Если заметок нет совсем, создаётся пустая матрица нужной формы, (0, 384), чтобы код поиска не ломался на пустой папке.

Нагляднее всего это видно на проверке. Я добавил четвёртую заметку про маркетинговую кампанию и повторил поиск, не вводя никаких команд переиндексации. В логе появилось «файлов 4, пересчитано файлов: 1»: движок досчитал только новый файл, и заметка сразу нашлась по запросу про продвижение и рекламу нового тарифа в соцсетях. Повторный прогон без изменений дал «пересчитано файлов: 0».

Правда, время изменения файла признак слабый: если содержимое поменялось, а mtime по какой-то причине совпал (например, файл восстановили из архива с сохранением даты), кэш этого не заметит. На своих заметках я с этим не сталкивался, поэтому хэш содержимого добавлять не стал.

Поиск: косинусная близость и фоллбэк по словам

Эмбеддинги я считаю с normalize_embeddings=True. Тогда векторы имеют единичную длину, и косинусная близость сводится к обычному скалярному произведению: умножаю матрицу фрагментов на вектор запроса и получаю близость ко всем фрагментам разом.

q_emb = self.model.encode([query], normalize_embeddings=True, convert_to_numpy=True)[0]
sims = self.embeddings @ q_emb

Чистый семантический поиск плохо работает с точными названиями, числами и редкими терминами. Запрос из одного слова, например имя компании, у эмбеддинг-модели часто даёт расплывчатый результат: она ловит общую тему и теряет конкретное слово. Поэтому я добавил запасной поиск по словам, который включается в двух случаях: запрос короткий (не больше двух слов) или лучшая близость ниже порога.

KEYWORD_SIM_THRESHOLD = 0.35   # если лучшая близость ниже — включаем фоллбэк
KEYWORD_SHORT_QUERY = 2        # запрос из ≤2 слов — тоже включаем фоллбэк
KEYWORD_WEIGHT = 0.15          # вклад одного совпавшего слова в итоговый рейтинг

q_words = _query_words(query)
use_kw = bool(q_words) and (
    len(q_words) <= KEYWORD_SHORT_QUERY or float(sims.max()) < KEYWORD_SIM_THRESHOLD
)
kw_scores = np.zeros(len(sims), dtype=np.float32)
if use_kw:
    for i, meta in enumerate(self.chunk_meta):
        kw_scores[i] = _keyword_score(meta["text"], q_words)

combined = sims + KEYWORD_WEIGHT * kw_scores
order = np.argsort(-combined)[:limit]

Оценка по словам считает, сколько слов запроса нашлось во фрагменте: точное слово даёт 1,0, а вхождение подстрокой (чтобы цеплялись формы слова) только 0,5:

def _keyword_score(text: str, q_words: list[str]) -> float:
    text_low = text.lower()
    text_words = set(_WORD_RE.findall(text_low))
    score = 0.0
    for w in q_words:
        if w in text_words:
            score += 1.0
        elif len(w) >= 4 and w in text_low:
            score += 0.5
    return score

Подстрочное совпадение разрешено только для слов от четырёх букв, иначе короткое слово находилось бы внутри десятков чужих. Числа из запроса сохраняются: в _query_words отбрасываются слова короче двух символов, кроме цифр. Итоговый рейтинг складывается из чистой близости и взвешенной оценки по словам, но в выдачу я отдаю только чистую близость, score: по ней в ответе видно, насколько фрагмент совпал по смыслу и стоит ли ему верить.

Пороги 0,35 и вес 0,15 я проверял только на нескольких заметках. На большой коллекции их, скорее всего, придётся пересматривать.

На тестовых заметках всё вело себя ожидаемо: запрос «продажи» поднимал на первое место заметку про продажи, «роадмап» заметку про продукт, а имя компании из заметки и число «200» выходили на первое место точно благодаря фоллбэку, без которого эти два запроса были бы лотереей.

Сервер: три инструмента на FastMCP

Сервер собран на FastMCP из официального SDK. Индекс один на весь процесс, модель в память грузится лениво при первом обращении и дальше переиспользуется:

from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations
import search_index as engine

mcp = FastMCP("notes")
_index = engine.SearchIndex()

_READONLY = ToolAnnotations(
    read_only_hint=True,
    destructive_hint=False,
    idempotent_hint=True,
    open_world_hint=False,
)

Аннотации сообщают клиенту, что инструмент ничего не меняет, повторный вызов безопасен и в «большой мир» он не ходит. Я сначала пробовал задать их словарём с ключами в camelCase, как показано в одном из руководств, но в актуальной версии SDK корректнее объект ToolAnnotations с полями в snake_case, который к тому же проверяется типами.

Сам поисковый инструмент получился коротким. Описание параметров лежит в Field, и именно оно попадает в схему инструмента, которую читает модель:

@mcp.tool(title="Поиск по заметкам", annotations=_READONLY)
def search_notes(
    query: Annotated[str, Field(description="Что ищем — тема или вопрос (рус./англ.).")],
    limit: Annotated[int, Field(ge=1, le=50, description="Сколько фрагментов вернуть (1–50).")] = engine.DEFAULT_LIMIT,
) -> str:
    """Семантический поиск по заметкам. Возвращает самые подходящие фрагменты
    с указанием файла-источника и оценкой близости. При коротком запросе или слабой
    близости дополняется поиском по словам. Новые и изменённые заметки
    подхватываются автоматически."""
    query = (query or "").strip()
    if not query:
        return 'Пустой запрос. Пример: search_notes("условия по оплате").'
    _index.build()  # переиндексирует только изменённые файлы
    results = _index.search(query, limit=limit)
    if not results:
        return f"По запросу «{query}» ничего не найдено. Список заметок — через list_notes()."
    return _format_search_results(query, results)

Автообновление держится на вызове _index.build() перед каждым поиском: он проходит по файлам, сверяет даты и, если ничего не менялось, почти ничего не делает, потому что считать заново нечего.

Отвечает инструмент обычным текстом в markdown, и для каждого фрагмента я выдаю имя файла, номер фрагмента, близость и метод:

out.append(
    f"{rank}. **{r['rel_path']}** · фрагмент {r['chunk_index']} · "
    f"близость {r['score']} ({method})"
)
out.append(f"   {r['text']}")

Все сообщения об ошибках написаны так, чтобы модель могла сама исправиться: пустой запрос подсказывает пример вызова, пустая выдача отсылает к list_notes().

Грабля вторая: любой лишний вывод в stdout ломает протокол

Сервер говорит с клиентом через стандартный вывод, и каждая лишняя строка, которую что-то напечатает в stdout, портит обмен сообщениями. Библиотеки машинного обучения любят печатать предупреждения и прогресс-бары. Поэтому в движке отключён прогресс-бар (show_progress_bar=False), логирование идёт в stderr, а у токенайзера приглушены предупреждения:

os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

try:
    from transformers.utils import logging as hf_logging
    hf_logging.set_verbosity_error()
except Exception:
    pass

Последний блок убирает предупреждение «длина больше 128». Оно мешало в самотесте, а пользы от него нет: я сам проверяю кандидатов на превышение и дроблю до того, как они попадут в модель. Когда я проверял сервер реальным клиентом по stdio, то отдельно убедился, что stdout остаётся чистым.

Грабля третья: инструмент, который читает файлы по имени от модели

read_note(filename) принимает строку, которую подставила модель, а модель в свою очередь могла получить её из текста заметки или из чьей-то просьбы. Если слепо склеить путь, ../../что-то-важное выведет сервер за пределы папки. Поэтому путь нормализуется и проверяется на принадлежность корню:

def _is_within(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
        return True
    except ValueError:
        return False

...
notes_root = engine.NOTES_DIR.resolve()
try:
    target = (engine.NOTES_DIR / filename).resolve()
except Exception:
    return f"Некорректное имя файла: {filename!r}."

if not _is_within(target, notes_root):
    return "Доступ только к файлам внутри папки заметок — попытка выйти за её пределы отклонена."
if not target.exists() or not target.is_file():
    return _not_found_hint(filename)
if target.suffix.lower() not in engine.NOTE_EXTENSIONS:
    return f"Это не заметка (.md/.txt): {filename}."

resolve() раскрывает .. и символические ссылки до настоящего пути, и только после этого проверяется, лежит ли он внутри корня. Проверка по расширению нужна отдельно: даже внутри папки читаются только .md и .txt.

Подсказка при ошибке тоже сделана с расчётом на модель. Если имя неоднозначно (несколько файлов с одинаковым именем в разных подпапках), в ответе перечисляются все совпадения и просьба указать относительный путь. Если файла нет совсем, показывается список доступных заметок, и модель с первого раза подбирает правильное имя:

def _not_found_hint(filename: str) -> str:
    names = [rel for rel, _abs, _mtime in engine._scan_notes()]
    base = Path(filename).name.lower()
    same_name = [n for n in names if Path(n).name.lower() == base]
    if len(same_name) > 1:
        listing = "n".join(f"- {n}" for n in same_name)
        return f"Имя «{filename}» неоднозначно — несколько файлов с таким именем:n{listing}n..."
    listing = "n".join(f"- {n}" for n in names) or "(заметок нет)"
    return f"Заметка «{filename}» не найдена. Доступные заметки:n{listing}"

В самотесте сервера (python server.py --selftest) я проверяю оба случая: попытку прочитать файл за пределами папки заметок через ../ и обращение к несуществующему файлу. Первая отклоняется, вторая возвращает список.

Пути: почему сервер не зависит от текущей папки

Claude Code запускает сервер из какой-то своей рабочей папки, и если искать notes/ относительно неё, то в другом проекте сервер просто не найдёт заметок. Поэтому корень вычисляется от самого файла:

def _resolve_base_dir() -> Path:
    env = os.environ.get("CLAUDE_PROJECT_DIR")
    if env:
        candidate = Path(env).expanduser().resolve()
        if (candidate / "notes").exists():
            return candidate
    return Path(__file__).resolve().parent

BASE_DIR = _resolve_base_dir()
NOTES_DIR = BASE_DIR / "notes"
INDEX_DIR = BASE_DIR / ".index"
CACHE_FILE = INDEX_DIR / "cache.pkl"

Переменную окружения я учитываю только если в указанной папке действительно есть notes/, иначе откатываюсь к расположению файла, и случайная переменная от другого проекта сервер не собьёт.

Мелочи, которые не должны ронять индекс

Отдельный набор случаев у любой пользовательской папки: пустой файл, файл в другой кодировке, бинарник с расширением .txt. Движок такие файлы пропускает с предупреждением в лог:

def _read_text(abs_path: Path, rel: str) -> str | None:
    try:
        return abs_path.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        logger.warning("Файл не в UTF-8 (или бинарный) — пропускаю: %s", rel)
        return None
    except Exception as exc:
        logger.warning("Не смог прочитать %s (%s) — пропускаю", rel, exc)
        return None

На пустой папке поиск возвращает понятное сообщение, что заметок нет. Вложенные подпапки внутри notes/ поддерживаются, список файлов собирается через rglob.

Подключение к Claude Code

Регистрация одной командой. Путь к проекту подставляете свой:

claude mcp add notes -- uv run --directory <папка проекта> python server.py
claude mcp list
# notes: ... - ✓ Connected

✓ Connected значит, что Claude Code сам запустил сервер этой командой и прошёл рукопожатие по протоколу, то есть команда рабочая. Есть нюанс, который меня один раз сбил: новый MCP-сервер виден только в новой сессии. Если добавить его посреди работы, инструментов в текущем чате не будет, нужно открыть новый.

Вживую я проверял так: в новом чате попросил найти материал про найм. Claude Code вызвал search_notes, получил пять фрагментов, и на первом месте оказалась заметка про собеседование с близостью 0,5735. После этого он собрал по фрагментам краткую сводку про кандидата.

Для проверки без клиента у меня есть два самотеста: python search_index.py "запрос" печатает топ-5 фрагментов вместе со статистикой нарезки, а python server.py --selftest вызывает все три инструмента напрямую.

Что осталось неидеальным

• Заметок у меня пока всего четыре, и тестовых. Пороги фоллбэка (0,35 и вес 0,15) проверены только на них, на настоящей большой коллекции поведение может отличаться. Сказать, как поиск поведёт себя на тысяче заметок, я не могу, проверять не пробовал.

• Кэш хранится в pickle. Для файла, который создаёт и читает одна и та же программа на своей машине, это допустимо, но чужой файл из такого кэша открывать нельзя. Если кто-то подложит вам свой cache.pkl, при загрузке выполнится чужой код.

• Устаревание файла я определяю по времени изменения. Хэш содержимого надёжнее, но дороже, и я его не добавлял.

• Ключевой поиск проходит по всем фрагментам циклом на Python. На сотнях фрагментов незаметно, на сотнях тысяч стоило бы переделать.

• MiniLM я взял как компромисс по размеру: она хуже улавливает тонкие различия в смысле, чем большие модели, и длинные мысли, растянутые на несколько абзацев, теряют связность из-за нарезки на 100-токенные куски.

• Читаются только .md и .txt в UTF-8. PDF и документы Word движок не открывает, их нужно предварительно переводить в текст.

• Сервер работает только локально через stdio. Удалённого доступа и веб-интерфейса нет, я их сознательно оставил за рамками.

Итог в цифрах

• Лимит модели 128 токенов, потолок на фрагмент 124. Реальные русские фрагменты получились от 323 до 492 символов и от 88 до 123 токенов, то есть ничего не обрезается.

• На трёх первых заметках индекс дал 12 фрагментов. При добавлении четвёртой заметки пересчитался один файл из четырёх, при повторном запуске без изменений ни одного.

• Три зависимости с жёстко зафиксированными версиями, два файла кода, модель около 120 МБ, без интернета после первой загрузки.

• Три грабли: модель молча обрезает вход сверх 128 токенов, любой лишний вывод в stdout ломает MCP-обмен, а путь, пришедший от модели, нужно проверять на принадлежность папке заметок.

Если вам нужен просто поиск, ядро сводится к нарезке с проверкой по токенам, кэшу по времени изменения и умножению матрицы на вектор, а обёртка на FastMCP занимает меньше двухсот строк.