diff --git a/memory_server.py b/memory_server.py index 1e5eee6..e54e093 100644 --- a/memory_server.py +++ b/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") \ No newline at end of file