diff --git a/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/solution.py b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/solution.py new file mode 100644 index 0000000..c2633c7 --- /dev/null +++ b/solutions/69f8e929da860fb4533faa2a_MCP-сервер_для_управления_памятью_агента/solution.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +# MCP-сервер для управления памятью агента +# Используем FastMCP, pydantic и python-dotenv + +import json +import os +from pathlib import Path +from typing import Any, Dict, List, Optional + +from dotenv import load_dotenv +from fastmcp import MCPServer, MCPRequest, MCPResponse +from pydantic import BaseModel, Field + +# Загружаем переменные окружения из .env +load_dotenv() + +# Путь к файлу хранения данных +DATA_FILE = Path(os.getenv("MEMORY_DATA_FILE", "memory_data.json")) + +# --------------------- +# Модели данных +# --------------------- +class MemoryItem(BaseModel): + key: str = Field(..., description="Ключ памяти") + value: Any = Field(..., description="Значение памяти") + namespace: Optional[str] = Field(None, description="Опциональное пространство имён") + +# --------------------- +# Хранилище памяти +# --------------------- +class MemoryStore: + def __init__(self, file_path: Path): + self.file_path = file_path + self._data: Dict[str, Any] = {} + self._load() + + def _load(self) -> None: + if self.file_path.exists(): + try: + with self.file_path.open("r", encoding="utf-8") as f: + self._data = json.load(f) + except Exception: + self._data = {} + else: + self._data = {} + + def _save(self) -> None: + with self.file_path.open("w", encoding="utf-8") as f: + json.dump(self._data, f, ensure_ascii=False, indent=2) + + def save(self, key: str, value: Any) -> None: + self._data[key] = value + self._save() + + def get(self, key: str) -> Any: + return self._data.get(key) + + def delete(self, key: str) -> bool: + if key in self._data: + del self._data[key] + self._save() + return True + return False + + def list_keys(self) -> List[str]: + return list(self._data.keys()) + + def save_with_namespace(self, key: str, value: Any, namespace: str) -> None: + ns_key = f"{namespace}:{key}" + self.save(ns_key, value) + + def get_by_namespace(self, key: str, namespace: str) -> Any: + ns_key = f"{namespace}:{key}" + return self.get(ns_key) + +# --------------------- +# MCP-обработчики +# --------------------- +class MemoryMCPHandler: + def __init__(self, store: MemoryStore): + self.store = store + + async def handle(self, request: MCPRequest) -> MCPResponse: + # Ожидаем, что payload содержит "action" и параметры + payload = request.payload + action = payload.get("action") + if not action: + return MCPResponse(error="Missing action") + + try: + if action == "save": + key = payload["key"] + value = payload["value"] + self.store.save(key, value) + return MCPResponse(result="ok") + elif action == "get": + key = payload["key"] + value = self.store.get(key) + return MCPResponse(result=value) + elif action == "delete": + key = payload["key"] + success = self.store.delete(key) + return MCPResponse(result=success) + elif action == "list_keys": + keys = self.store.list_keys() + return MCPResponse(result=keys) + elif action == "save_with_namespace": + key = payload["key"] + value = payload["value"] + namespace = payload["namespace"] + self.store.save_with_namespace(key, value, namespace) + return MCPResponse(result="ok") + elif action == "get_by_namespace": + key = payload["key"] + namespace = payload["namespace"] + value = self.store.get_by_namespace(key, namespace) + return MCPResponse(result=value) + else: + return MCPResponse(error="Unknown action") + except KeyError as e: + return MCPResponse(error=f"Missing parameter: {e}") + except Exception as e: + return MCPResponse(error=str(e)) + +# --------------------- +# Запуск сервера +# --------------------- +async def main(): + store = MemoryStore(DATA_FILE) + handler = MemoryMCPHandler(store) + server = MCPServer(handler=handler, host=os.getenv("HOST", "0.0.0.0"), port=int(os.getenv("PORT", 8000))) + await server.serve() + +if __name__ == "__main__": + import asyncio + asyncio.run(main())