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

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

@@ -1,39 +1,169 @@
-import asyncio
-from fastmcp import Client
+# memory_server.py
 
-async def main() -> None:
-    # Подключаемся к серверу через stdio, указывая путь к исполняемому скрипту сервера
-    client = Client("python memory_server.py")
-
-    await client.connect()
-
-    try:
-        # Сохраняем данные в namespace "default"
-        result = await client.call_tool(
-            "save_with_namespace",
-            {"key": "username", "value": "Алексей", "namespace": "default"},
-        )
-        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()
+from fastmcp import FastMCP
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional, List, Dict
+import fnmatch
 
 
+class MemoryServer:
+    def __init__(self):
+        self.mcp = FastMCP("Memory-Server")
+        self.storage_path = Path("./memory_data.json")
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+    def _load_memory(self) -> dict:
+        """Загружает память из 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):
+        """Сохраняет память в 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)
+
+    def _current_timestamp(self) -> str:
+        return datetime.now(timezone.utc).isoformat()
+
+    # ------------------------------------------------------------------
+    # Basic tools
+    # ------------------------------------------------------------------
+    @self.mcp.tool()
+    def save(key: str, value: Any) -> bool:
+        """
+        Сохраняет значение по ключу в память сервера.
+
+        Args:
+            key: Идентификатор для сохранения (уникальный ключ).
+            value: Любое сериализуемое значение.
+
+        Returns:
+            True при успешном сохранении, False иначе.
+        """
+        memory = self._load_memory()
+        memory[key] = {"value": value, "timestamp": self._current_timestamp()}
+        try:
+            self._save_memory(memory)
+            return True
+        except Exception:
+            return False
+
+    @self.mcp.tool()
+    def get(key: str) -> Optional[Dict[str, Any]]:
+        """
+        Возвращает значение по ключу с метаданными.
+
+        Args:
+            key: Идентификатор для поиска.
+
+        Returns:
+            Словарь с полями {"key": ..., "value": ..., "timestamp": ...}
+            или None если ключ не найден.
+        """
+        memory = self._load_memory()
+        entry = memory.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 если ключ не найден.
+        """
+        memory = self._load_memory()
+        if key not in memory:
+            return False
+        del memory[key]
+        try:
+            self._save_memory(memory)
+            return True
+        except Exception:
+            return False
+
+    @self.mcp.tool()
+    def list_keys(pattern: str = "*") -> List[str]:
+        """
+        Возвращает список всех ключей с поддержкой wildcard-паттерна.
+
+        Args:
+            pattern: Паттерн для фильтрации (поддерживает * и ?).
+
+        Returns:
+            Список совпадающих ключей.
+        """
+        memory = self._load_memory()
+        return [k for k in memory.keys() if fnmatch.fnmatch(k, pattern)]
+
+    # ------------------------------------------------------------------
+    # Namespace-aware tools
+    # ------------------------------------------------------------------
+    @self.mcp.tool()
+    def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool:
+        """
+        Сохраняет значение с указанием пространства имён.
+
+        Args:
+            key: Идентификатор.
+            value: Значение для сохранения.
+            namespace: Пространство имён (по умолчанию 'default').
+
+        Returns:
+            True при успехе, False иначе.
+        """
+        namespaced_key = f"{namespace}:{key}"
+        return self.save(namespaced_key, value)
+
+    @self.mcp.tool()
+    def get_by_namespace(namespace: str = "default") -> List[Dict[str, Any]]:
+        """
+        Возвращает все ключи из указанного namespace.
+
+        Args:
+            namespace: Пространство имён для чтения.
+
+        Returns:
+            Список словарей с метаданными всех ключей namespace.
+        """
+        memory = self._load_memory()
+        prefix = f"{namespace}:"
+        result = []
+        for k, v in memory.items():
+            if k.startswith(prefix):
+                result.append(
+                    {
+                        "key": k,
+                        "value": v["value"],
+                        "timestamp": v["timestamp"],
+                    }
+                )
+        return result
+
+
+# ----------------------------------------------------------------------
+# Server entry point
+# ----------------------------------------------------------------------
 if __name__ == "__main__":
-    asyncio.run(main())
+    server = MemoryServer()
+    server.mcp.run(
+        transport="stdio",
+        show_banner=False,
+        log_level="ERROR",
+    )