Память и состояние в MCP-сервере

"""Состояние без сессии: черновик заказа живёт под явным идентификатором."""
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()

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 нет или он не ваш

"""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()

remember("Договор с ООО Ромашка до 31.12", tags=["договор"])  → {"id": 1}
remember("Позвонить Иванову в пятницу")                       → {"id": 2}
# новый процесс сервера
recall("договор") → [{"id": 1, "text": "Договор с ООО Ромашка до 31.12", "tags": ["договор"], …}]
forget(99)        → ошибка: Заметки 99 нет