commit 1fbb6def58e23715afcbd28c441aee06fa56e3ed Author: kuzakhmetovartur Date: Wed Jun 24 14:37:42 2026 +0300 feat: solution for 'MCP-сервер для управления памятью агента' diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b16538b --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +node_modules/ +.env +dist/ +build/ +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..b19b497 --- /dev/null +++ b/README.md @@ -0,0 +1,34 @@ +# MCP-сервер для управления памятью агента + +Главная +Мои задания +MCP-сервер для управления памятью агента +5Д +EN +MCP-сервер для управления памятью агента +Зачёт +Версия 1 +Дедлайн сдачи: 31.08.2026 + +В работе + +Редактирование ответа + +Заполните ответ и отправьте работу на проверку преподавателю. + +Тип ответа +Текст +Ссылка +Файлы +Текст ответа +Прикреплённые файлы +Загрузить файл +Отправить на проверку +Отменить + +Задание + +Практическое задание: MCP-сервер для управления памятью агента +Цель + +Создать MCP-сервер (Model Context Protocol) — самостоятельный серв \ No newline at end of file diff --git a/memory_server.py b/memory_server.py new file mode 100644 index 0000000..47758ed --- /dev/null +++ b/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() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4ba1984 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +fastmcp +pydantic +python-dotenv \ No newline at end of file