From 8cde94ff2c502d38ec165fb6c6ed39ce6ce3979b Mon Sep 17 00:00:00 2001 From: RomanKurlygin Date: Wed, 27 May 2026 08:21:12 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20memory=5Fserver.py=20=E2=80=94=20build?= =?UTF-8?q?=5Fserver()=20+=20main.py=20=D0=B4=D0=BB=D1=8F=20=D0=B0=D0=B2?= =?UTF-8?q?=D1=82=D0=BE=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D0=BA=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- memory_server.py | 96 ++++++++++++++++++++++++++---------------------- 1 file changed, 52 insertions(+), 44 deletions(-) diff --git a/memory_server.py b/memory_server.py index e920706..654f9c3 100644 --- a/memory_server.py +++ b/memory_server.py @@ -4,13 +4,13 @@ from __future__ import annotations import fnmatch import json import re -from datetime import datetime, timezone +from datetime import datetime from pathlib import Path from typing import Any, Optional from fastmcp import FastMCP -SAFE_KEY_PATTERN = re.compile(r"^[A-Za-z0-9_\-:.]+$") +_UNSAFE_KEY = re.compile(r"\.\.|/|\\") class MemoryServer: @@ -20,30 +20,29 @@ class MemoryServer: self._register_tools() def _validate_key(self, key: str) -> bool: - return bool(key) and bool(SAFE_KEY_PATTERN.match(key)) + return bool(key) and _UNSAFE_KEY.search(key) is None - def _load_memory(self) -> dict[str, dict[str, Any]]: + def _load_memory(self) -> dict: if not self.storage_path.exists(): return {} - with open(self.storage_path, encoding="utf-8") as file: - return json.load(file) + with open(self.storage_path, encoding="utf-8") as f: + return json.load(f) - def _save_memory(self, data: dict[str, dict[str, Any]]) -> None: + def _save_memory(self, data: dict) -> None: self.storage_path.parent.mkdir(parents=True, exist_ok=True) - with open(self.storage_path, "w", encoding="utf-8") as file: - json.dump(data, file, indent=2, ensure_ascii=False) + with open(self.storage_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) - def _entry(self, key: str, value: Any) -> dict[str, Any]: - return { - "value": value, - "timestamp": datetime.now(timezone.utc).isoformat(), - } + def _timestamp(self) -> str: + return datetime.now().isoformat(timespec="seconds") + + def _make_entry(self, value: Any) -> dict: + return {"value": value, "timestamp": self._timestamp()} def _register_tools(self) -> None: mcp = self.mcp - server = self - @mcp.tool + @mcp.tool() def save(key: str, value: Any) -> bool: """Сохраняет значение по ключу в память сервера. @@ -54,17 +53,17 @@ class MemoryServer: Returns: True при успешном сохранении, False иначе. """ - if not server._validate_key(key): + if not self._validate_key(key): return False try: - data = server._load_memory() - data[key] = server._entry(key, value) - server._save_memory(data) + data = self._load_memory() + data[key] = self._make_entry(value) + self._save_memory(data) return True - except (OSError, TypeError, ValueError): + except (TypeError, ValueError, OSError): return False - @mcp.tool + @mcp.tool() def get(key: str) -> Optional[dict]: """Возвращает значение по ключу с метаданными. @@ -74,18 +73,18 @@ class MemoryServer: Returns: Словарь с полями key, value, timestamp или None если ключ не найден. """ - if not server._validate_key(key): + if not self._validate_key(key): return None - record = server._load_memory().get(key) - if record is None: + entry = self._load_memory().get(key) + if entry is None: return None return { "key": key, - "value": record.get("value"), - "timestamp": record.get("timestamp"), + "value": entry["value"], + "timestamp": entry["timestamp"], } - @mcp.tool + @mcp.tool() def delete(key: str) -> bool: """Удаляет ключ из памяти сервера. @@ -95,16 +94,16 @@ class MemoryServer: Returns: True при успешном удалении, False если ключ не найден. """ - if not server._validate_key(key): + if not self._validate_key(key): return False - data = server._load_memory() + data = self._load_memory() if key not in data: return False del data[key] - server._save_memory(data) + self._save_memory(data) return True - @mcp.tool + @mcp.tool() def list_keys(pattern: str = "*") -> list[str]: """Возвращает список всех ключей с поддержкой wildcard-паттерна. @@ -114,11 +113,13 @@ class MemoryServer: Returns: Список совпадающих ключей. """ - keys = list(server._load_memory().keys()) - return sorted(key for key in keys if fnmatch.fnmatch(key, pattern)) + keys = list(self._load_memory().keys()) + return sorted(k for k in keys if fnmatch.fnmatch(k, pattern)) - @mcp.tool - def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool: + @mcp.tool() + def save_with_namespace( + key: str, value: Any, namespace: str = "default" + ) -> bool: """Сохраняет значение с указанием пространства имён. Args: @@ -129,10 +130,10 @@ class MemoryServer: Returns: True при успехе, False иначе. """ - full_key = f"{namespace}:{key}" - return save(full_key, value) + storage_key = f"{namespace}:{key}" + return save(storage_key, value) - @mcp.tool + @mcp.tool() def get_by_namespace(namespace: str = "default") -> list[dict]: """Возвращает все ключи из указанного namespace. @@ -143,24 +144,31 @@ class MemoryServer: Список словарей с метаданными всех ключей namespace. """ prefix = f"{namespace}:" + data = self._load_memory() result: list[dict] = [] - for storage_key, record in server._load_memory().items(): + for storage_key, entry in data.items(): if not storage_key.startswith(prefix): continue short_key = storage_key[len(prefix) :] result.append( { "key": short_key, - "value": record.get("value"), - "timestamp": record.get("timestamp"), - "storage_key": storage_key, + "value": entry["value"], + "timestamp": entry["timestamp"], } ) return sorted(result, key=lambda item: item["key"]) + _ = (save, get, delete, list_keys, save_with_namespace, get_by_namespace) + + +def build_server(storage_path: str | Path = "./memory_data.json") -> MemoryServer: + """Сборка MCP-сервера — точка входа для автопроверки.""" + return MemoryServer(storage_path=storage_path) + if __name__ == "__main__": - server = MemoryServer() + server = build_server() server.mcp.run( transport="stdio", show_banner=False,