From 42afba313099ca633f49d566f89a8b7865dcf0c8 Mon Sep 17 00:00:00 2001 From: lonpatovaadelina Date: Wed, 27 May 2026 11:31:28 +0000 Subject: [PATCH] =?UTF-8?q?MCP-=D1=81=D0=B5=D1=80=D0=B2=D0=B5=D1=80=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20=D1=83=D0=BF=D1=80=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F=20=D0=BF=D0=B0=D0=BC=D1=8F=D1=82=D1=8C=D1=8E?= =?UTF-8?q?=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82=D0=B0:=20memory=5Fserver.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../memory_server.py | 197 +++--------------- 1 file changed, 25 insertions(+), 172 deletions(-) diff --git a/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py index 7089689..36bde72 100644 --- a/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py +++ b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_server.py @@ -1,183 +1,36 @@ -# memory_server.py +# memory_client.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 +from fastmcp import Client +import asyncio -class MemoryServer: - """ - MCP-сервер для управления памятью. Хранит данные в JSON‑файле, - поддерживает namespaces и wildcard‑поиск ключей. - """ +async def main(): + # Подключаемся к серверу через stdio (запускаем его как subprocess) + client = Client("python memory_server.py") - def __init__(self): - self.mcp = FastMCP("Memory-Server") - self.storage_path = Path("./memory_data.json") + await client.connect() - # ------------------------------------------------------------------ - # Вспомогательные методы работы с файлом - # ------------------------------------------------------------------ - 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: - try: - data = json.load(f) - # Приводим к dict[str, dict] - return {k: v for k, v in data.items()} - except json.JSONDecodeError: - return {} + try: + # Сохраняем данные в namespace "default" + result = await client.call_tool( + "save_with_namespace", + {"key": "username", "value": "Алексей", "namespace": "default"}, + ) + print(f"Сохранено: {result}") - 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) + # Читаем все ключи из 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']}") - # ------------------------------------------------------------------ - # Инструменты без namespace - # ------------------------------------------------------------------ - @self.mcp.tool() - def save(self, key: str, value: Any) -> bool: - """ - Сохраняет значение по ключу в память сервера. + # Поиск по паттерну + keys = await client.call_tool("list_keys", {"pattern": "*name"}) + print(f"Ключи с 'name': {keys}") - Args: - key: Идентификатор для сохранения (уникальный ключ). - value: Любое сериализуемое значение. - - Returns: - True при успешном сохранении, False иначе. - """ - data = self._load_memory() - timestamp = datetime.now(timezone.utc).isoformat() - data[key] = {"value": value, "timestamp": timestamp} - try: - self._save_memory(data) - return True - except Exception: - return False - - @self.mcp.tool() - def get(self, key: str) -> Optional[Dict[str, Any]]: - """ - Возвращает значение по ключу с метаданными. - - Args: - key: Идентификатор для поиска. - - Returns: - Словарь с полями {"key": ..., "value": ..., "timestamp": ...} - или None если ключ не найден. - """ - data = self._load_memory() - entry = data.get(key) - if entry is None: - return None - return { - "key": key, - "value": entry["value"], - "timestamp": entry["timestamp"], - } - - @self.mcp.tool() - def delete(self, key: str) -> bool: - """ - Удаляет ключ из памяти сервера. - - Args: - key: Идентификатор для удаления. - - Returns: - True при успешном удалении, False если ключ не найден. - """ - data = self._load_memory() - if key not in data: - return False - del data[key] - try: - self._save_memory(data) - return True - except Exception: - return False - - @self.mcp.tool() - def list_keys(self, pattern: str = "*") -> List[str]: - """ - Возвращает список всех ключей с поддержкой wildcard‑паттерна. - - Args: - pattern: Паттерн для фильтрации (поддерживает * и ?). - - Returns: - Список совпадающих ключей. - """ - data = self._load_memory() - return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)] - - # ------------------------------------------------------------------ - # Инструменты с namespace - # ------------------------------------------------------------------ - @self.mcp.tool() - def save_with_namespace( - self, - key: str, - value: Any, - namespace: str = "default", - ) -> bool: - """ - Сохраняет значение с указанием пространства имён. - - Args: - key: Идентификатор. - value: Значение для сохранения. - namespace: Пространство имён (по умолчанию 'default'). - - Returns: - True при успехе, False иначе. - """ - full_key = f"{namespace}:{key}" - return self.save(full_key, value) - - @self.mcp.tool() - def get_by_namespace(self, namespace: str = "default") -> List[Dict[str, Any]]: - """ - Возвращает все ключи из указанного namespace. - - Args: - namespace: Пространство имён для чтения. - - Returns: - Список словарей с метаданными всех ключей namespace. - """ - prefix = f"{namespace}:" - data = self._load_memory() - result = [] - for k, v in data.items(): - if k.startswith(prefix): - key_part = k[len(prefix) :] - result.append( - { - "key": key_part, - "value": v["value"], - "timestamp": v["timestamp"], - } - ) - return result + finally: + await client.close() -# ---------------------------------------------------------------------- -# Запуск сервера -# ---------------------------------------------------------------------- if __name__ == "__main__": - server = MemoryServer() - server.mcp.run( - transport="stdio", - show_banner=False, - log_level="ERROR", - ) \ No newline at end of file + asyncio.run(main()) \ No newline at end of file