brojs-task-69f8e929da860fb4533faa2a/answer.md

{ "task": { "_id": "69f8e929da860fb4533faa2a", "title": "MCP-сервер для управления памятью агента", "description": "# Практическое задание: MCP-сервер для управления памятью агента\n\n## Цель\n\nСоздать MCP-сервер (Model Context Protocol) — самостоятельный сервис, который предоставляет другим агентам инструменты для работы с памятью через стандартизированный протокол. Это первый шаг к построению распределённых многоагентных систем, где агенты могут обмениваться данными и координировать работу через единое хранилище.\n\nВ отличие от локальной памяти внутри одного скрипта, MCP-сервер — это независимый процесс с API, доступный для любого клиента (другого агента, CLI-утилиты или другого сервиса).\n\n---\n\n## Стек и установка\n\n- Python 3.10+\n- Пакеты: fastmcp (фреймворк MCP), pydantic (валидация данных)\n\nУстановка:\n\nbash\npip install fastmcp pydantic python-dotenv\n\n\n> Важно: fastmcp — это библиотека для создания серверов и клиентов MCP. Она позволяет регистрировать инструменты, которые становятся доступны агентам через протокол stdio или SSE.\n\n---\n\n## Что нужно сделать\n\n### 1. Создать MCP-сервер с базовыми инструментами памяти\n\nСоздай файл memory_server.py, в котором определишь класс сервера на основе FastMCP:\n\npython\nfrom fastmcp import FastMCP\nimport json\nfrom datetime import datetime\nfrom pathlib import Path\nfrom typing import Any, Optional\n\nclass MemoryServer:\n def __init__(self):\n self.mcp = FastMCP(\"Memory-Server\")\n self.storage_path = Path(\"./memory_data.json\")\n \n def _load_memory(self) -> dict:\n \"\"\"Загружает память из JSON-файла.\"\"\"\n if not self.storage_path.exists():\n return {}\n with open(self.storage_path, 'r', encoding='utf-8') as f:\n return json.load(f)\n \n def _save_memory(self, data: dict):\n \"\"\"Сохраняет память в JSON-файл.\"\"\"\n self.storage_path.parent.mkdir(parents=True, exist_ok=True)\n with open(self.storage_path, 'w', encoding='utf-8') as f:\n json.dump(data, f, indent=2, ensure_ascii=False)\n \n @self.mcp.tool()\n def save(key: str, value: Any) -> bool:\n \"\"\"Сохраняет значение по ключу в память сервера.\n \n Args:\n key: Идентификатор для сохранения (уникальный ключ).\n value: Любое сериализуемое значение.\n \n Returns:\n True при успешном сохранении, False иначе.\n \"\"\"\n # Реализуй здесь...\n \n @self.mcp.tool()\n def get(key: str) -> Optional[dict]:\n \"\"\"Возвращает значение по ключу с метаданными.\n \n Args:\n key: Идентификатор для поиска.\n \n Returns:\n Словарь с полями {\"key\": ..., \"value\": ..., \"timestamp\": ...} \n или None если ключ не найден.\n \"\"\"\n # Реализуй здесь...\n \n @self.mcp.tool()\n def delete(key: str) -> bool:\n \"\"\"Удаляет ключ из памяти сервера.\n \n Args:\n key: Идентификатор для удаления.\n \n Returns:\n True при успешном удалении, False если ключ не найден.\n \"\"\"\n # Реализуй здесь...\n \n @self.mcp.tool()\n def list_keys(pattern: str = \"*\") -> list[str]:\n \"\"\"Возвращает список всех ключей с поддержкой wildcard-паттерна.\n \n Args:\n pattern: Паттерн для фильтрации (поддерживает * и ?).\n \n Returns:\n Список совпадающих ключей.\n \"\"\"\n # Реализуй здесь...\n\n\n### 2. Добавить инструменты с поддержкой namespaces\n\nNamespace — это пространство имён для изоляции данных. Например, один агент может работать в namespace agent_1, другой — в agent_2. Данные не пересекаются.\n\nДобавь следующие инструменты:\n\npython\n @self.mcp.tool()\n def save_with_namespace(key: str, value: Any, namespace: str = \"default\") -> bool:\n \"\"\"Сохраняет значение с указанием пространства имён.\n \n Args:\n key: Идентификатор.\n value: Значение для сохранения.\n namespace: Пространство имён (по умолчанию 'default').\n \n Returns:\n True при успехе, False иначе.\n \"\"\"\n # Ключ в хранилище должен формироваться как: f\"{namespace}:{key}\"\n \n @self.mcp.tool()\n def get_by_namespace(namespace: str = \"default\") -> list[dict]:\n \"\"\"Возвращает все ключи из указанного namespace.\n \n Args:\n namespace: Пространство имён для чтения.\n \n Returns:\n Список словарей с метаданными всех ключей namespace.\n \"\"\"\n # Ищи ключи, начинающиеся на f\"{namespace}:\"\n\n\n### 3. Реализовать сохранение данных\n\nСоздай файловое хранилище в формате JSON. Структура файла:\n\njson\n{\n \"default:user_name\": {\n \"value\": \"Алексей\",\n \"timestamp\": \"2026-05-15T10:30:00\"\n },\n \"agent_1:session_token\": {\n \"value\": \"abc123xyz\", \n \"timestamp\": \"2026-05-15T10:31:00\"\n }\n}\n\n\nКаждое значение должно храниться с метаданными (timestamp). Ключи формируются как namespace:key для namespace-based инструментов.\n\n### 4. Написать клиент для тестирования\n\nСоздай файл memory_client.py, который демонстрирует работу с сервером через MCP-клиент:\n\npython\nfrom fastmcp import Client\nimport asyncio\n\nasync def main():\n # Подключение к MCP-серверу через stdio\n client = Client(\"python memory_server.py\")\n \n await client.connect()\n \n try:\n # Сохранение данных в namespace \"default\"\n result = await client.call_tool(\n \"save_with_namespace\", \n {\"key\": \"username\", \"value\": \"Алексей\", \"namespace\": \"default\"}\n )\n print(f\"Сохранено: {result}\")\n \n # Чтение данных\n result = await client.call_tool(\n \"get_by_namespace\", \n {\"namespace\": \"default\"}\n )\n print(f\"Данные namespace 'default':\")\n for item in result:\n print(f\" {item['key']}: {item['value']}\")\n \n # Поиск по паттерну\n keys = await client.call_tool(\"list_keys\", {\"pattern\": \"*name\"})\n print(f\"Ключи с 'name': {keys}\")\n \n finally:\n await client.close()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n\n\n### 5. Добавить запуск сервера\n\nВ конце файла memory_server.py добавь блок запуска:\n\npython\nif __name__ == \"__main__\":\n server = MemoryServer()\n server.mcp.run(\n transport=\"stdio\",\n show_banner=False,\n log_level='ERROR'\n )\n\n\n---\n\n## Подсказки\n\n- Импорты: понадобятся FastMCP, Client из fastmcp; json, datetime стандартные библиотеки; Path из pathlib.\n- Wildcard паттерн: используй fnmatch.fnmatch(key, pattern) для фильтрации ключей.\n- Структура данных: каждый ключ в JSON должен хранить значение + timestamp. Namespace добавляется к ключу как префикс: namespace:key.\n- Асинхронность: клиент работает асинхронно (async def, await). Сервер запускается синхронно через mcp.run().\n- Безопасность: валидируй ввод — ключи не должны содержать опасных символов (например, пути /../).\n\n---\n\n## Ожидаемый результат\n\nПри запуске клиента в консоли должен появиться вывод:\n\ntext\nСохранено: True\nДанные namespace 'default':\n username: Алексей\nКлючи с 'name': ['default:user_name']\n\n\nСервер работает как отдельный процесс, клиент подключается к нему через stdio и вызывает инструменты удалённо.\n\n---\n\n## Критерии оценки\n\n| Критерий | Баллы |\n|----------|-------|\n| Сервер корректно зарегистрирован с FastMCP и минимум 6 инструментов (save, get, delete, list_keys, save_with_namespace, get_by_namespace) | 2 |\n| Инструменты правильно оформлены через декоратор @mcp.tool() с docstring и типизацией аргументов | 2 |\n| Используется транспорт stdio в mcp.run(), сервер запускается корректно | 1 |\n| Данные сохраняются в JSON-файл с метаданными (timestamp) | 2 |\n| Поддержка namespace для изоляции данных (ключи формируются как namespace:key) | 2 |\n| Реализован клиент для тестирования с примером вызова инструментов | 1 |\n| Итого | 10 |\n\n---\n\n## Ссылки\n\n- FastMCP documentation — создание MCP-серверов и клиентов\n- Model Context Protocol (MCP) — спецификация протокола\n- LangChain AI - MCP Integration — интеграция MCP с LangChain", "course": "698b49da77cb6d4d2e43ce78", "lesson": { "_id": "69b035ab2a32e307dd6e1fee", "name": "Implementing Model Context Protocol in LangChain", "date": "2026-03-10T15:30:00.000Z", "id": "69b035ab2a32e307dd6e1fee" }, "gradeType": "passfail", "maxScore": 100, "assignmentType": "individual", "allowSubmissions": true, "allowRevisions": false, "creator": { "sub": "8ae348ab-f32e-46c7-9960-8b7fdf0e4c0e", "name": "Александр Примаков" }, "teamGrades": [], "created": "2026-05-04T18:44:57.314Z", "id": "69f8e929da860fb4533faa2a" } }