feat: solution for 'MCP-сервер для управления памятью агента'

.gitignore

@@ -0,0 +1,5 @@
+node_modules/
+.env
+dist/
+build/
+*.log

README.md

@@ -0,0 +1,34 @@
+# MCP-сервер для управления памятью агента
+
+Главная
+Мои задания
+MCP-сервер для управления памятью агента
+5Д
+EN
+MCP-сервер для управления памятью агента
+Зачёт
+Версия 1
+Дедлайн сдачи: 31.08.2026
+
+В работе
+
+Редактирование ответа
+
+Заполните ответ и отправьте работу на проверку преподавателю.
+
+Тип ответа
+Текст
+Ссылка
+Файлы
+Текст ответа
+Прикреплённые файлы
+Загрузить файл
+Отправить на проверку
+Отменить
+
+Задание
+
+Практическое задание: MCP-сервер для управления памятью агента
+Цель
+
+Создать MCP-сервер (Model Context Protocol) — самостоятельный серв

memory_server.py

@@ -0,0 +1,115 @@
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional, Dict, List
+
+from fastmcp import FastMCP
+from pydantic import BaseModel, ValidationError
+
+
+class MemoryEntry(BaseModel):
+    value: Any
+    timestamp: str
+
+
+class MemoryServer:
+    def __init__(self):
+        self.mcp = FastMCP("Memory-Server")
+        self.storage_path = Path("./memory_data.json")
+        # Register tools
+        self.mcp.tool()(self.save)
+        self.mcp.tool()(self.get)
+        self.mcp.tool()(self.delete)
+        self.mcp.tool()(self.list_keys)
+        self.mcp.tool()(self.save_with_namespace)
+        self.mcp.tool()(self.get_with_namespace)
+        self.mcp.tool()(self.delete_with_namespace)
+        self.mcp.tool()(self.list_keys_with_namespace)
+
+    def _load_memory(self) -> Dict[str, Dict[str, Any]]:
+        """Загружает память из JSON-файла."""
+        if not self.storage_path.exists():
+            return {}
+        try:
+            with open(self.storage_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+                # Validate structure
+                validated = {}
+                for k, v in data.items():
+                    try:
+                        entry = MemoryEntry(**v)
+                        validated[k] = entry.dict()
+                    except ValidationError:
+                        # Skip invalid entries
+                        continue
+                return validated
+        except (json.JSONDecodeError, OSError):
+            return {}
+
+    def _save_memory(self, data: Dict[str, Dict[str, Any]]) -> None:
+        """Сохраняет память в JSON-файл."""
+        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            with open(self.storage_path, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2, ensure_ascii=False)
+        except OSError as e:
+            # Log error if needed
+            print(f"Error saving memory: {e}")
+
+    # Core CRUD operations
+
+    def save(self, key: str, value: Any) -> bool:
+        """Сохраняет значение по ключу в память сервера."""
+        data = self._load_memory()
+        entry = {"value": value, "timestamp": datetime.utcnow().isoformat()}
+        data[key] = entry
+        self._save_memory(data)
+        return True
+
+    def get(self, key: str) -> Optional[Dict[str, Any]]:
+        """Возвращает значение по ключу с метаданными."""
+        data = self._load_memory()
+        entry = data.get(key)
+        return entry if entry else None
+
+    def delete(self, key: str) -> bool:
+        """Удаляет ключ из памяти."""
+        data = self._load_memory()
+        if key in data:
+            del data[key]
+            self._save_memory(data)
+            return True
+        return False
+
+    def list_keys(self) -> List[str]:
+        """Возвращает список всех ключей."""
+        data = self._load_memory()
+        return list(data.keys())
+
+    # Namespace-aware operations
+
+    def _compose_key(self, namespace: str, key: str) -> str:
+        return f"{namespace}:{key}"
+
+    def save_with_namespace(self, namespace: str, key: str, value: Any) -> bool:
+        composite = self._compose_key(namespace, key)
+        return self.save(composite, value)
+
+    def get_with_namespace(self, namespace: str, key: str) -> Optional[Dict[str, Any]]:
+        composite = self._compose_key(namespace, key)
+        return self.get(composite)
+
+    def delete_with_namespace(self, namespace: str, key: str) -> bool:
+        composite = self._compose_key(namespace, key)
+        return self.delete(composite)
+
+    def list_keys_with_namespace(self, namespace: str) -> List[str]:
+        data = self._load_memory()
+        prefix = f"{namespace}:"
+        return [k[len(prefix) :] for k in data.keys() if k.startswith(prefix)]
+
+
+if __name__ == "__main__":
+    server = MemoryServer()
+    print("Starting Memory MCP Server...")
+    server.mcp.run()

requirements.txt

@@ -0,0 +1,3 @@
+fastmcp
+pydantic
+python-dotenv