diff --git a/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_client.py b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_client.py index 8af601a..1438a1d 100644 --- a/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_client.py +++ b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/memory_client.py @@ -1,39 +1,169 @@ -import asyncio -from fastmcp import Client +# memory_server.py -async def main() -> None: - # Подключаемся к серверу через stdio, указывая путь к исполняемому скрипту сервера - client = Client("python memory_server.py") - - await client.connect() - - try: - # Сохраняем данные в namespace "default" - result = await client.call_tool( - "save_with_namespace", - {"key": "username", "value": "Алексей", "namespace": "default"}, - ) - print(f"Сохранено: {result}") - - # Читаем все ключи из 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']}") - - # Поиск ключей по шаблону - keys = await client.call_tool( - "list_keys", - {"pattern": "*name"}, - ) - print(f"Ключи с 'name': {keys}") - - finally: - await client.close() +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") + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + def _load_memory(self) -> dict: + """Загружает память из 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): + """Сохраняет память в 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) + + def _current_timestamp(self) -> str: + return datetime.now(timezone.utc).isoformat() + + # ------------------------------------------------------------------ + # Basic tools + # ------------------------------------------------------------------ + @self.mcp.tool() + def save(key: str, value: Any) -> bool: + """ + Сохраняет значение по ключу в память сервера. + + Args: + key: Идентификатор для сохранения (уникальный ключ). + value: Любое сериализуемое значение. + + Returns: + True при успешном сохранении, False иначе. + """ + memory = self._load_memory() + memory[key] = {"value": value, "timestamp": self._current_timestamp()} + try: + self._save_memory(memory) + return True + except Exception: + return False + + @self.mcp.tool() + def get(key: str) -> Optional[Dict[str, Any]]: + """ + Возвращает значение по ключу с метаданными. + + Args: + key: Идентификатор для поиска. + + Returns: + Словарь с полями {"key": ..., "value": ..., "timestamp": ...} + или None если ключ не найден. + """ + memory = self._load_memory() + entry = memory.get(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 если ключ не найден. + """ + memory = self._load_memory() + if key not in memory: + return False + del memory[key] + try: + self._save_memory(memory) + return True + except Exception: + return False + + @self.mcp.tool() + def list_keys(pattern: str = "*") -> List[str]: + """ + Возвращает список всех ключей с поддержкой wildcard-паттерна. + + Args: + pattern: Паттерн для фильтрации (поддерживает * и ?). + + Returns: + Список совпадающих ключей. + """ + memory = self._load_memory() + return [k for k in memory.keys() if fnmatch.fnmatch(k, pattern)] + + # ------------------------------------------------------------------ + # Namespace-aware tools + # ------------------------------------------------------------------ + @self.mcp.tool() + def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool: + """ + Сохраняет значение с указанием пространства имён. + + Args: + key: Идентификатор. + value: Значение для сохранения. + namespace: Пространство имён (по умолчанию 'default'). + + Returns: + True при успехе, False иначе. + """ + namespaced_key = f"{namespace}:{key}" + return self.save(namespaced_key, value) + + @self.mcp.tool() + def get_by_namespace(namespace: str = "default") -> List[Dict[str, Any]]: + """ + Возвращает все ключи из указанного namespace. + + Args: + namespace: Пространство имён для чтения. + + Returns: + Список словарей с метаданными всех ключей namespace. + """ + memory = self._load_memory() + prefix = f"{namespace}:" + result = [] + for k, v in memory.items(): + if k.startswith(prefix): + result.append( + { + "key": k, + "value": v["value"], + "timestamp": v["timestamp"], + } + ) + return result + + +# ---------------------------------------------------------------------- +# Server entry point +# ---------------------------------------------------------------------- if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file + server = MemoryServer() + server.mcp.run( + transport="stdio", + show_banner=False, + log_level="ERROR", + ) \ No newline at end of file