Добавлен memory_server.py

memory_server.py

@@ -1,44 +1,45 @@
+# memory_server.py
+
+"""
+MCP-сервер для управления памятью с поддержкой namespaces.
+"""
+
 from fastmcp import FastMCP
 import json
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Optional, List
-import fnmatch
+from typing import Any, Optional, List, Dict
 
 
 class MemoryServer:
     """
-    MCP‑сервер для управления памятью с поддержкой namespaces.
+    Сервер памяти. Хранит данные в JSON‑файле и предоставляет инструменты
+    для сохранения, чтения, удаления и поиска ключей.
     """
 
-    def __init__(self):
-        # Инициализируем FastMCP
+    def __init__(self) -> None:
+        # Инициализируем FastMCP с именем сервера
         self.mcp = FastMCP("Memory-Server")
-        # Путь к файлу хранения данных
-        self.storage_path = Path("./memory_data.json")
 
-    # ------------------------------------------------------------------
-    # Вспомогательные методы для работы с JSON‑файлом
-    # ------------------------------------------------------------------
-    def _load_memory(self) -> dict:
+        # Путь к файлу хранения данных
+        self.storage_path: Path = Path("./memory_data.json")
+        self.memory: Dict[str, Dict[str, Any]] = self._load_memory()
+
+    def _load_memory(self) -> Dict[str, Dict[str, Any]]:
         """Загружает память из JSON‑файла."""
         if not self.storage_path.exists():
             return {}
         with open(self.storage_path, "r", encoding="utf-8") as f:
             return json.load(f)
 
-    def _save_memory(self, data: dict):
+    def _save_memory(self) -> None:
         """Сохраняет память в JSON‑файл."""
         self.storage_path.parent.mkdir(parents=True, exist_ok=True)
         with open(self.storage_path, "w", encoding="utf-8") as f:
-            json.dump(data, f, indent=2, ensure_ascii=False)
-
-    # ------------------------------------------------------------------
-    # Инструменты FastMCP
-    # ------------------------------------------------------------------
+            json.dump(self.memory, f, indent=2, ensure_ascii=False)
 
     @self.mcp.tool()
-    def save(key: str, value: Any) -> bool:
+    def save(self, key: str, value: Any) -> bool:
         """
         Сохраняет значение по ключу в память сервера.
 
@@ -47,21 +48,17 @@ class MemoryServer:
             value: Любое сериализуемое значение.
 
         Returns:
-            True при успешном сохранении, False иначе.
+            True при успешном сохранении.
         """
-        data = self._load_memory()
-        # Проверяем наличие опасных символов в ключе
-        if "/" in key or "\\" in key:
-            return False
-        data[key] = {
+        self.memory[key] = {
             "value": value,
             "timestamp": datetime.utcnow().isoformat(),
         }
-        self._save_memory(data)
+        self._save_memory()
         return True
 
     @self.mcp.tool()
-    def get(key: str) -> Optional[dict]:
+    def get(self, key: str) -> Optional[Dict[str, Any]]:
         """
         Возвращает значение по ключу с метаданными.
 
@@ -70,12 +67,11 @@ class MemoryServer:
 
         Returns:
             Словарь с полями {"key": ..., "value": ..., "timestamp": ...}
-            или None если ключ не найден.
+            или None, если ключ не найден.
         """
-        data = self._load_memory()
-        if key not in data:
+        entry = self.memory.get(key)
+        if entry is None:
             return None
-        entry = data[key]
         return {
             "key": key,
             "value": entry["value"],
@@ -83,7 +79,7 @@ class MemoryServer:
         }
 
     @self.mcp.tool()
-    def delete(key: str) -> bool:
+    def delete(self, key: str) -> bool:
         """
         Удаляет ключ из памяти сервера.
 
@@ -93,15 +89,14 @@ class MemoryServer:
         Returns:
             True при успешном удалении, False если ключ не найден.
         """
-        data = self._load_memory()
-        if key not in data:
-            return False
-        del data[key]
-        self._save_memory(data)
-        return True
+        if key in self.memory:
+            del self.memory[key]
+            self._save_memory()
+            return True
+        return False
 
     @self.mcp.tool()
-    def list_keys(pattern: str = "*") -> List[str]:
+    def list_keys(self, pattern: str = "*") -> List[str]:
         """
         Возвращает список всех ключей с поддержкой wildcard‑паттерна.
 
@@ -111,15 +106,17 @@ class MemoryServer:
         Returns:
             Список совпадающих ключей.
         """
-        data = self._load_memory()
-        return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)]
+        import fnmatch
 
-    # ------------------------------------------------------------------
-    # Инструменты с поддержкой namespaces
-    # ------------------------------------------------------------------
+        return [k for k in self.memory.keys() if fnmatch.fnmatch(k, pattern)]
 
     @self.mcp.tool()
-    def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool:
+    def save_with_namespace(
+        self,
+        key: str,
+        value: Any,
+        namespace: str = "default",
+    ) -> bool:
         """
         Сохраняет значение с указанием пространства имён.
 
@@ -131,11 +128,11 @@ class MemoryServer:
         Returns:
             True при успехе, False иначе.
         """
-        full_key = f"{namespace}:{key}"
-        return self.save(full_key, value)
+        ns_key = f"{namespace}:{key}"
+        return self.save(ns_key, value)
 
     @self.mcp.tool()
-    def get_by_namespace(namespace: str = "default") -> List[dict]:
+    def get_by_namespace(self, namespace: str = "default") -> List[Dict[str, Any]]:
         """
         Возвращает все ключи из указанного namespace.
 
@@ -145,24 +142,21 @@ class MemoryServer:
         Returns:
             Список словарей с метаданными всех ключей namespace.
         """
-        data = self._load_memory()
+        result: List[Dict[str, Any]] = []
         prefix = f"{namespace}:"
-        result = []
-        for k, v in data.items():
+        for k, entry in self.memory.items():
             if k.startswith(prefix):
                 result.append(
                     {
                         "key": k,
-                        "value": v["value"],
-                        "timestamp": v["timestamp"],
+                        "value": entry["value"],
+                        "timestamp": entry["timestamp"],
                     }
                 )
         return result
 
 
-# ----------------------------------------------------------------------
-# Запуск сервера
-# ----------------------------------------------------------------------
 if __name__ == "__main__":
     server = MemoryServer()
+    # Запускаем сервер с транспортом stdio
     server.mcp.run(transport="stdio", show_banner=False, log_level="ERROR")