From df34678fa169e46c0888eabf109ef30f388d78c7 Mon Sep 17 00:00:00 2001 From: lonpatovaadelina Date: Tue, 26 May 2026 09:28:23 +0000 Subject: [PATCH] =?UTF-8?q?MCP=E2=80=91=D1=81=D0=B5=D1=80=D0=B2=D0=B5?= =?UTF-8?q?=D1=80=20=D0=B4=D0=BB=D1=8F=20=D1=83=D0=BF=D1=80=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=BF=D0=B0=D0=BC=D1=8F=D1=82?= =?UTF-8?q?=D1=8C=D1=8E=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82=D0=B0:=20memory?= =?UTF-8?q?=5Fserver.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../memory_server.py | 189 ++++++++++++++++++ 1 file changed, 189 insertions(+) create mode 100644 solutions/69f8e929da860fb4533faa2a_MCP_сервер_для_управления_памятью_агента/memory_server.py diff --git a/solutions/69f8e929da860fb4533faa2a_MCP_сервер_для_управления_памятью_агента/memory_server.py b/solutions/69f8e929da860fb4533faa2a_MCP_сервер_для_управления_памятью_агента/memory_server.py new file mode 100644 index 0000000..76af4fc --- /dev/null +++ b/solutions/69f8e929da860fb4533faa2a_MCP_сервер_для_управления_памятью_агента/memory_server.py @@ -0,0 +1,189 @@ +# memory_server.py + +""" +FastMCP сервер для управления памятью агента. +Поддерживает операции сохранения, чтения, удаления и списка ключей, +а также работу с пространствами имён (namespaces). +""" + +from fastmcp import FastMCP +import json +from datetime import datetime +from pathlib import Path +from typing import Any, Optional, Dict, List +import fnmatch + + +class MemoryServer: + """ + Сервер памяти. Хранит данные в JSON‑файле с метаданными. + """ + + def __init__(self) -> None: + self.mcp = FastMCP("Memory-Server") + self.storage_path = Path("./memory_data.json") + + # ------------------------------------------------------------------ + # Внутренние методы работы с файлом + # ------------------------------------------------------------------ + def _load_memory(self) -> Dict[str, Any]: + """Загружает память из JSON‑файла.""" + if not self.storage_path.exists(): + return {} + with open(self.storage_path, "r", encoding="utf-8") as f: + try: + data = json.load(f) + except json.JSONDecodeError: + # Если файл повреждён – начинаем с пустого словаря + data = {} + return data + + def _save_memory(self, data: Dict[str, Any]) -> 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) + + # ------------------------------------------------------------------ + # Инструменты без namespace + # ------------------------------------------------------------------ + @self.mcp.tool() + def save(key: str, value: Any) -> bool: + """ + Сохраняет значение по ключу в память сервера. + + Args: + key: Идентификатор для сохранения (уникальный ключ). + value: Любое сериализуемое значение. + + Returns: + True при успешном сохранении, False иначе. + """ + if not isinstance(key, str) or "/" in key: + return False + data = self._load_memory() + data[key] = { + "value": value, + "timestamp": datetime.utcnow().isoformat(), + } + self._save_memory(data) + return True + + @self.mcp.tool() + def get(key: str) -> Optional[Dict[str, Any]]: + """ + Возвращает значение по ключу с метаданными. + + Args: + key: Идентификатор для поиска. + + Returns: + Словарь с полями {"key": ..., "value": ..., "timestamp": ...} + или None если ключ не найден. + """ + data = self._load_memory() + entry = data.get(key) + if entry is None: + return None + return { + "key": key, + "value": entry["value"], + "timestamp": entry["timestamp"], + } + + @self.mcp.tool() + def delete(key: str) -> bool: + """ + Удаляет ключ из памяти сервера. + + Args: + key: Идентификатор для удаления. + + Returns: + True при успешном удалении, False если ключ не найден. + """ + data = self._load_memory() + if key in data: + del data[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)] + + # ------------------------------------------------------------------ + # Инструменты с namespace + # ------------------------------------------------------------------ + @self.mcp.tool() + def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool: + """ + Сохраняет значение с указанием пространства имён. + + Args: + key: Идентификатор. + value: Значение для сохранения. + namespace: Пространство имён (по умолчанию 'default'). + + Returns: + True при успехе, False иначе. + """ + if not isinstance(key, str) or "/" in key: + return False + full_key = f"{namespace}:{key}" + data = self._load_memory() + 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[str, Any]]: + """ + Возвращает все ключи из указанного namespace. + + Args: + namespace: Пространство имён для чтения. + + Returns: + Список словарей с метаданными всех ключей namespace. + """ + prefix = f"{namespace}:" + data = self._load_memory() + result = [] + for full_key, entry in data.items(): + if full_key.startswith(prefix): + key_part = full_key[len(prefix) :] + result.append( + { + "key": key_part, + "value": entry["value"], + "timestamp": entry["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