# memory_server.py 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") # ------------------------------------------------------------------ # Вспомогательные методы для работы с файлом # ------------------------------------------------------------------ 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: return json.load(f) 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) # ------------------------------------------------------------------ # Базовые инструменты # ------------------------------------------------------------------ @self.mcp.tool() def save(key: str, value: Any) -> bool: """ Сохраняет значение по ключу в память сервера. Args: key: Идентификатор для сохранения (уникальный ключ). value: Любое сериализуемое значение. Returns: True при успешном сохранении, False иначе. """ data = self._load_memory() full_key = f"default:{key}" data[full_key] = { "value": value, "timestamp": datetime.now(timezone.utc).isoformat(), } try: self._save_memory(data) return True except Exception: return False @self.mcp.tool() def get(key: str) -> Optional[Dict[str, Any]]: """ Возвращает значение по ключу с метаданными. Args: key: Идентификатор для поиска. Returns: Словарь с полями {"key": ..., "value": ..., "timestamp": ...} или None если ключ не найден. """ data = self._load_memory() full_key = f"default:{key}" entry = data.get(full_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 если ключ не найден. """ data = self._load_memory() full_key = f"default:{key}" if full_key in data: del data[full_key] self._save_memory(data) return True return False @self.mcp.tool() def list_keys(pattern: str = "*") -> List[str]: """ Возвращает список всех ключей с поддержкой wildcard‑паттерна. Args: pattern: Паттерн для фильтрации (поддерживает * и ?). Returns: Список совпадающих ключей. """ data = self._load_memory() return [ key.split(":", 1)[1] for key in data.keys() if fnmatch.fnmatch(key, f"*:{pattern}") ] # ------------------------------------------------------------------ # Инструменты с поддержкой namespaces # ------------------------------------------------------------------ @self.mcp.tool() def save_with_namespace( key: str, value: Any, namespace: str = "default", ) -> bool: """ Сохраняет значение с указанием пространства имён. Args: key: Идентификатор. value: Значение для сохранения. namespace: Пространство имён (по умолчанию 'default'). Returns: True при успехе, False иначе. """ data = self._load_memory() full_key = f"{namespace}:{key}" data[full_key] = { "value": value, "timestamp": datetime.now(timezone.utc).isoformat(), } try: self._save_memory(data) return True except Exception: return False @self.mcp.tool() def get_by_namespace(namespace: str = "default") -> List[Dict[str, Any]]: """ Возвращает все ключи из указанного namespace. Args: namespace: Пространство имён для чтения. Returns: Список словарей с метаданными всех ключей namespace. """ data = self._load_memory() result: List[Dict[str, Any]] = [] prefix = f"{namespace}:" for full_key, entry in data.items(): if full_key.startswith(prefix): key_part = full_key[len(prefix) :] result.append( { "key": key_part, "value": entry["value"], "timestamp": entry["timestamp"], } ) return result # ---------------------------------------------------------------------- # Запуск сервера # ---------------------------------------------------------------------- if __name__ == "__main__": server = MemoryServer() server.mcp.run( transport="stdio", show_banner=False, log_level="ERROR", ) ``` ```python # memory_client.py """ Клиент для тестирования работы MCP‑сервера памяти. Использует fastmcp.Client для подключения к серверу, запускаемому в файле memory_server.py через stdio. Демонстрирует сохранение, чтение и поиск данных с поддержкой namespaces. """ from fastmcp import Client import asyncio async def main() -> None: """ Основная асинхронная функция: подключается к серверу, вызывает инструменты и выводит результаты. """ # Создаём клиент, указывая путь к исполняемому файлу сервера client = Client("python memory_server.py") # Открываем соединение await client.connect() try: # 1. Сохраняем данные в namespace "default" result = await client.call_tool( "save_with_namespace", {"key": "username", "value": "Алексей", "namespace": "default"}, ) print(f"Сохранено: {result}") # 2. Читаем все ключи из 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']}") # 3. Ищем ключи, содержащие подстроку 'name' keys = await client.call_tool( "list_keys", {"pattern": "*name"}, ) print("Ключи с 'name':", keys) finally: # Закрываем соединение await client.close() if __name__ == "__main__": asyncio.run(main()) ``` ```text # requirements.txt fastmcp==0.1.5 # библиотека для MCP‑серверов и клиентов pydantic==2.7.4 # валидация данных (не используется напрямую, но указана) python-dotenv==1.0.1 # загрузка переменных окружения langchain==0.3.0 # требуется по условию задания ``` **Пояснения** * В `memory_server.py` реализованы все требуемые инструменты с поддержкой namespaces и сохранением метаданных (`timestamp`). * Файл `memory_client.py` демонстрирует работу клиента, подключающегося к серверу через stdio. * Добавлен файл `requirements.txt`, в котором указана зависимость `langchain`, как того требует преподаватель. Теперь проект полностью соответствует заданию и может быть запущен командой: ```bash pip install -r requirements.txt python memory_server.py # запускает сервер # В другом терминале: python memory_client.py # выполняет тесты