# Память и состояние в MCP-сервере Данные между вызовами без сессий и свой сервер памяти на SQLite > Страница сайта MCP Docs — руководство (https://mcpdocs.ru/svoy-server/pamyat/). Тип: Руководство. Код проверен: Python 3.13, пакет mcp 2.3.0 — по stdio, с перезапуском процесса. Ревизия спецификации 2026-07-28, сверено 13.10.2026. С ревизии 2026-07-28 в MCP нет сессий: любой запрос может прийти на любой экземпляр сервера. Значит, всё, что должно жить между вызовами, сервер хранит сам и выдаёт клиенту явный идентификатор. Здесь — два частых случая: черновик, который собирается за несколько вызовов, и долговременная память ассистента. _Без сессий состояние живёт в хранилище сервера под явным идентификатором, а клиент передаёт этот идентификатор в каждом вызове._ ## Почему не держать состояние в памяти процесса - Удалённый сервер обычно работает в нескольких экземплярах за балансировщиком; следующий вызов может попасть на другой. - Заголовка `Mcp-Session-Id` больше нет, и привязать клиента к экземпляру нечем. - После перезапуска процесса всё, что было в переменных, теряется. ## Явный идентификатор вместо сессии Первый вызов создаёт объект и возвращает его идентификатор; модель передаёт его в следующих вызовах. Владелец проверяется при каждом обращении: идентификатор — не пароль. _order_server.py_ ```python """Состояние без сессии: черновик заказа живёт под явным идентификатором.""" import secrets from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.mcpserver.exceptions import ToolError mcp = MCPServer("Заказы", version="1.0.0") class Order(BaseModel): order_id: str items: dict[str, int] # артикул -> количество # В примере — словарь в памяти. В работе — база или Redis: запросы # одного клиента могут приходить на разные экземпляры сервера. DRAFTS: dict[str, dict] = {} def caller() -> str: token = get_access_token() return (token.subject or token.client_id) if token else "local" def draft_of(order_id: str) -> dict: draft = DRAFTS.get(order_id) # Проверяем владельца при каждом вызове: идентификатор — не пароль if draft is None or draft["owner"] != caller(): raise ToolError(f"Черновика {order_id} нет или он не ваш") return draft @mcp.tool() def start_order() -> Order: """Создать черновик заказа. Верните order_id в следующих вызовах.""" order_id = "ord_" + secrets.token_urlsafe(9) DRAFTS[order_id] = {"owner": caller(), "items": {}} return Order(order_id=order_id, items={}) @mcp.tool() def add_item(order_id: str, sku: str, qty: int = 1) -> Order: """Добавить товар в черновик заказа.""" if qty < 1: raise ToolError("Количество должно быть больше нуля") items = draft_of(order_id)["items"] items[sku] = items.get(sku, 0) + qty return Order(order_id=order_id, items=items) @mcp.tool() def show_order(order_id: str) -> Order: """Показать черновик заказа.""" return Order(order_id=order_id, items=draft_of(order_id)["items"]) if __name__ == "__main__": mcp.run() ``` _Проверка_ ```text start_order() → {"order_id": "ord_c20o5oyms6qC", "items": {}} add_item(order_id, "A-100", qty=2) → {"items": {"A-100": 2}} add_item(order_id, "A-100") → {"items": {"A-100": 3}} show_order("ord_fake") → ошибка: Черновика ord_fake нет или он не ваш ``` В примере черновики хранятся в словаре — для показа. В работе берите базу или Redis с временем жизни записей. ## Свой сервер памяти «Память» для ассистента — это сервер с инструментами «запомнить», «вспомнить», «забыть», который хранит записи на диске. Заметки каждого пользователя отделены: по HTTP владелец берётся из токена, по stdio это один локальный пользователь. _memory_server.py_ ```python """MCP-сервер «Память»: заметки, которые переживают перезапуск клиента и сервера.""" import sqlite3 from collections.abc import AsyncIterator from contextlib import asynccontextmanager from mcp.server import MCPServer from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.mcpserver import Context from mcp.server.mcpserver.exceptions import ToolError DB_PATH = "memory.sqlite3" @asynccontextmanager async def lifespan(server: MCPServer) -> AsyncIterator[sqlite3.Connection]: db = sqlite3.connect(DB_PATH, check_same_thread=False) db.execute("""CREATE TABLE IF NOT EXISTS notes ( id INTEGER PRIMARY KEY, owner TEXT NOT NULL, text TEXT NOT NULL, tags TEXT NOT NULL DEFAULT '', created TEXT DEFAULT CURRENT_TIMESTAMP)""") try: yield db finally: db.close() mcp = MCPServer("Память", version="1.0.0", lifespan=lifespan) def owner() -> str: """Чьи это заметки: субъект токена по HTTP, «local» по stdio.""" token = get_access_token() return (token.subject or token.client_id) if token else "local" @mcp.tool() def remember(ctx: Context, text: str, tags: list[str] | None = None) -> dict[str, int]: """Запомнить факт. tags — метки для поиска, например ["клиент", "договор"].""" db = ctx.request_context.lifespan_context cur = db.execute("INSERT INTO notes(owner, text, tags) VALUES (?, ?, ?)", (owner(), text, ",".join(tags or []))) db.commit() return {"id": cur.lastrowid} @mcp.tool() def recall(ctx: Context, query: str = "", limit: int = 10) -> list[dict]: """Найти заметки по слову в тексте или метке. Пустой запрос — последние заметки.""" db = ctx.request_context.lifespan_context rows = db.execute( "SELECT id, text, tags, created FROM notes WHERE owner = ? " "AND (text LIKE ? OR tags LIKE ?) ORDER BY id DESC LIMIT ?", (owner(), f"%{query}%", f"%{query}%", min(limit, 50)), ).fetchall() return [{"id": r[0], "text": r[1], "tags": r[2].split(",") if r[2] else [], "created": r[3]} for r in rows] @mcp.tool() def forget(ctx: Context, note_id: int) -> str: """Удалить заметку по id.""" db = ctx.request_context.lifespan_context cur = db.execute("DELETE FROM notes WHERE id = ? AND owner = ?", (note_id, owner())) db.commit() if cur.rowcount == 0: raise ToolError(f"Заметки {note_id} нет") return "Удалено" if __name__ == "__main__": mcp.run() ``` - Соединение с базой открывается один раз в `lifespan` и доступно инструментам через `ctx.request_context.lifespan_context`. В SDK второй версии `lifespan` для HTTP выполняется один раз при запуске, а не на каждое соединение. - `ctx: Context` в параметрах SDK подставляет сам; в схему инструмента он не попадает. - Запросы к базе только с параметрами (`?`), лимит записей ограничен в коде. Проверка: две заметки записаны одним процессом сервера, найдены после его перезапуска. _Проверка_ ```text remember("Договор с ООО Ромашка до 31.12", tags=["договор"]) → {"id": 1} remember("Позвонить Иванову в пятницу") → {"id": 2} # новый процесс сервера recall("договор") → [{"id": 1, "text": "Договор с ООО Ромашка до 31.12", "tags": ["договор"], …}] forget(99) → ошибка: Заметки 99 нет ``` Готовый эталонный сервер памяти на графе знаний — [Memory](https://mcpprotocol.ru/katalog/memory/) в каталоге справочника. ## Многошаговые запросы и несколько экземпляров Если инструменту не хватает данных, сервер может вернуть результат `input_required` и попросить клиента повторить запрос с ответом пользователя. Промежуточное состояние такого запроса SDK запечатывает в поле `requestState` и отдаёт клиенту. Ключ для запечатывания по умолчанию создаётся в каждом процессе, поэтому сервер в нескольких экземплярах должен получить общий ключ: `MCPServer(..., request_state_security=RequestStateSecurity(keys=[...]))`, где класс импортируется из `mcp.server.request_state`. Механизм — в разделе [«Шаблоны обмена»](https://mcpdocs.ru/protocol/patterns/). ## Безопасность данных - Не храните в памяти ассистента пароли, номера карт и паспортные данные: всё записанное модель потом может прочитать и пересказать. - Дайте пользователю способ удалить свои записи — инструмент `forget` и удаление по запросу. - Для удалённого сервера отделяйте данные пользователей по субъекту токена, а не по значению, которое прислал клиент. > Готовые серверы — в [каталоге MCP-серверов](https://mcpprotocol.ru/katalog/) справочника. Свой удалённый сервер можно проверить на соответствие ревизии 2026-07-28 [бесплатной проверкой](https://mcpprotocol.ru/mcp-check/).