add file

memory_server.py

@@ -0,0 +1,131 @@
+from fastmcp import FastMCP
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional
+import fnmatch
+
+class MemoryServer:
+    def __init__(self):
+        self.mcp = FastMCP("Memory-Server")
+        self.storage_path = Path("./memory_data.json")
+
+    def _load_memory(self) -> dict:
+        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):
+        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)
+
+    @self.mcp.tool()
+    def save(key: str, value: Any) -> bool:
+        """
+        Сохраняет значение по ключу в память сервера.
+        Args:
+            key: Идентификатор для сохранения (уникальный ключ).
+            value: Любое сериализуемое значение.
+        Returns:
+            True при успешном сохранении, False иначе.
+        """
+        data = self._load_memory()
+        full_key = f"default:{key}"
+        data[full_key] = {
+            "value": value,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        self._save_memory(data)
+        return True
+
+    @self.mcp.tool()
+    def get(key: str) -> Optional[dict]:
+        """
+        Возвращает значение по ключу с метаданными.
+        Args:
+            key: Идентификатор для поиска.
+        Returns:
+            Словарь с полями {"key": ..., "value": ..., "timestamp": ...} или None если ключ не найден.
+        """
+        data = self._load_memory()
+        full_key = f"default:{key}"
+        if full_key in data:
+            item = data[full_key]
+            return {"key": full_key, "value": item["value"], "timestamp": item["timestamp"]}
+        return None
+
+    @self.mcp.tool()
+    def delete(key: str) -> bool:
+        """
+        Удаляет ключ из памяти сервера.
+        Args:
+            key: Идентификатор для удаления.
+        Returns:
+            True при успешном удалении, False если ключ не найден.
+        """
+        data = self._load_memory()
+        full_key = f"default:{key}"
+        if full_key in data:
+            del data[full_key]
+            self._save_memory(data)
+            return True
+        return False
+
+    @self.mcp.tool()
+    def list_keys(pattern: str = "*") -> list[str]:
+        """
+        Возвращает список всех ключей с поддержкой wildcard-паттерна.
+        Args:
+            pattern: Паттерн для фильтрации (поддерживает * и ?).
+        Returns:
+            Список совпадающих ключей.
+        """
+        data = self._load_memory()
+        return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)]
+
+    @self.mcp.tool()
+    def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool:
+        """
+        Сохраняет значение с указанием пространства имён.
+        Args:
+            key: Идентификатор.
+            value: Значение для сохранения.
+            namespace: Пространство имён (по умолчанию 'default').
+        Returns:
+            True при успехе, False иначе.
+        """
+        data = self._load_memory()
+        full_key = f"{namespace}:{key}"
+        data[full_key] = {
+            "value": value,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        self._save_memory(data)
+        return True
+
+    @self.mcp.tool()
+    def get_by_namespace(namespace: str = "default") -> list[dict]:
+        """
+        Возвращает все ключи из указанного namespace.
+        Args:
+            namespace: Пространство имён для чтения.
+        Returns:
+            Список словарей с метаданными всех ключей namespace.
+        """
+        data = self._load_memory()
+        prefix = f"{namespace}:"
+        result = []
+        for k, v in data.items():
+            if k.startswith(prefix):
+                result.append({"key": k, "value": v["value"], "timestamp": v["timestamp"]})
+        return result
+
+if __name__ == "__main__":
+    server = MemoryServer()
+    server.mcp.run(
+        transport="stdio",
+        show_banner=False,
+        log_level='ERROR'
+    )