From 1ba0420539e9c6e9c70228e5719a876ed7df21d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9C=D0=B0=D1=80=D0=B8=D1=8F=20=D0=91=D0=B5=D1=80=D0=B4?= =?UTF-8?q?=D0=BD=D0=B8=D0=BA=D0=BE=D0=B2=D0=B0?= Date: Wed, 27 May 2026 07:45:36 +0000 Subject: [PATCH] =?UTF-8?q?MCP-=D1=81=D0=B5=D1=80=D0=B2=D0=B5=D1=80=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20=D1=83=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=20=D0=BF=D0=B0=D0=BC=D1=8F=D1=82=D1=8C=D1=8E?= =?UTF-8?q?=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82=D0=B0:=20memory=5Fserver.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../memory_server.py | 115 ++++++++++++++++++ 1 file changed, 115 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..0b77e4f --- /dev/null +++ b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py @@ -0,0 +1,115 @@ +<|channel|>final code<|message|>import json +from datetime import datetime +from pathlib import Path +from typing import Any, Optional + +from fastmcp import FastMCP + + +class MemoryServer: + def __init__(self): + 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) + if isinstance(data, dict): + return data + except json.JSONDecodeError: + pass + return {} + + 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: + """Сохраняет значение по ключу в память сервера.""" + if not isinstance(key, str) or not key.strip(): + 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]: + """Возвращает значение по ключу с метаданными.""" + if not isinstance(key, str): + return 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: + """Удаляет ключ из памяти сервера.""" + if not isinstance(key, str): + return 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-паттерна.""" + if not isinstance(pattern, str): + pattern = "*" + data = self._load_memory() + import fnmatch + + 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: + """Сохраняет значение с указанием пространства имён.""" + if not isinstance(namespace, str) or not namespace.strip(): + return False + full_key = f"{namespace}:{key}" + return self.save(full_key, value) + + @self.mcp.tool() + def get_by_namespace(namespace: str = "default") -> list[dict]: + """Возвращает все ключи из указанного namespace.""" + if not isinstance(namespace, str) or not namespace.strip(): + return [] + data = self._load_memory() + prefix = f"{namespace}:" + result = [] + for k, v in data.items(): + if k.startswith(prefix): + key_part = k[len(prefix) :] + result.append( + {"key": key_part, "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