У меня есть папка с конспектами созвонов и расшифровками встреч, и искать в ней по слову неудобно: я помню, что обсуждали найм, а в файле написано «собеседование» и «кандидат», и обычный поиск по тексту такой файл не находит. Отправлять заметки во внешний сервис ради эмбеддингов мне не хотелось, поэтому я собрал маленький 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 занимает меньше двухсот строк.