From 4f93cda649f961665d3109105d8048d3aa226e07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=B4=D0=B5=D0=BB=D0=B8=D0=BD=D0=B0=20=D0=A1=D0=B0?= =?UTF-8?q?=D1=82=D1=82=D0=B0=D1=80=D0=BE=D0=B2=D0=B0?= Date: Sun, 31 May 2026 15:37:16 +0000 Subject: [PATCH] add file --- memory_server.py | 131 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 memory_server.py diff --git a/memory_server.py b/memory_server.py new file mode 100644 index 0000000..356df48 --- /dev/null +++ b/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' + ) \ No newline at end of file