MCP-сервер для управления памятью агента: memory_server.py

solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py

@@ -1,183 +1,36 @@
-# memory_server.py
+# memory_client.py
 
-from fastmcp import FastMCP
+from fastmcp import Client
-import json
+import asyncio
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Any, Optional, List, Dict
-import fnmatch
 
 
-class MemoryServer:
+async def main():
-    """
+    # Подключаемся к серверу через stdio (запускаем его как subprocess)
-    MCP-сервер для управления памятью. Хранит данные в JSON‑файле,
+    client = Client("python memory_server.py")
-    поддерживает namespaces и wildcard‑поиск ключей.
-    """
 
-    def __init__(self):
+    await client.connect()
-        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)
+        # Сохраняем данные в namespace "default"
-                # Приводим к dict[str, dict]
+        result = await client.call_tool(
-                return {k: v for k, v in data.items()}
+            "save_with_namespace",
-            except json.JSONDecodeError:
+            {"key": "username", "value": "Алексей", "namespace": "default"},
-                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(self, key: str, value: Any) -> bool:
-        """
-        Сохраняет значение по ключу в память сервера.
-
-        Args:
-            key: Идентификатор для сохранения (уникальный ключ).
-            value: Любое сериализуемое значение.
-
-        Returns:
-            True при успешном сохранении, False иначе.
-        """
-        data = self._load_memory()
-        timestamp = datetime.now(timezone.utc).isoformat()
-        data[key] = {"value": value, "timestamp": timestamp}
-        try:
-            self._save_memory(data)
-            return True
-        except Exception:
-            return False
-
-    @self.mcp.tool()
-    def get(self, 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(self, key: str) -> bool:
-        """
-        Удаляет ключ из памяти сервера.
-
-        Args:
-            key: Идентификатор для удаления.
-
-        Returns:
-            True при успешном удалении, False если ключ не найден.
-        """
-        data = self._load_memory()
-        if key not in data:
-            return False
-        del data[key]
-        try:
-            self._save_memory(data)
-            return True
-        except Exception:
-            return False
-
-    @self.mcp.tool()
-    def list_keys(self, 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(
-        self,
-        key: str,
-        value: Any,
-        namespace: str = "default",
-    ) -> bool:
-        """
-        Сохраняет значение с указанием пространства имён.
-
-        Args:
-            key: Идентификатор.
-            value: Значение для сохранения.
-            namespace: Пространство имён (по умолчанию 'default').
-
-        Returns:
-            True при успехе, False иначе.
-        """
-        full_key = f"{namespace}:{key}"
-        return self.save(full_key, value)
-
-    @self.mcp.tool()
-    def get_by_namespace(self, namespace: str = "default") -> List[Dict[str, Any]]:
-        """
-        Возвращает все ключи из указанного namespace.
-
-        Args:
-            namespace: Пространство имён для чтения.
-
-        Returns:
-            Список словарей с метаданными всех ключей namespace.
-        """
-        prefix = f"{namespace}:"
-        data = self._load_memory()
-        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
+        print(f"Сохранено: {result}")
+
+        # Читаем все ключи из namespace "default"
+        result = await client.call_tool("get_by_namespace", {"namespace": "default"})
+        print("Данные namespace 'default':")
+        for item in result:
+            print(f"  {item['key']}: {item['value']}")
+
+        # Поиск по паттерну
+        keys = await client.call_tool("list_keys", {"pattern": "*name"})
+        print(f"Ключи с 'name': {keys}")
+
+    finally:
+        await client.close()
 
 
-# ----------------------------------------------------------------------
-# Запуск сервера
-# ----------------------------------------------------------------------
 if __name__ == "__main__":
-    server = MemoryServer()
+    asyncio.run(main())
-    server.mcp.run(
-        transport="stdio",
-        show_banner=False,
-        log_level="ERROR",
-    )