From 2f0aaddb8a962100e94e25d2c6e29433cd3eee52 Mon Sep 17 00:00:00 2001 From: lonpatovaadelina Date: Thu, 28 May 2026 14:05:50 +0000 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20memory=5Fclient.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- memory_client.py | 269 +---------------------------------------------- 1 file changed, 1 insertion(+), 268 deletions(-) diff --git a/memory_client.py b/memory_client.py index dab8527..48de9e1 100644 --- a/memory_client.py +++ b/memory_client.py @@ -1,270 +1,3 @@ -# 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 # выполняет тесты \ No newline at end of file +python-dotenv==1.0.1 # загрузка переменных окружения \ No newline at end of file