feat: solution for 69f8e929da860fb4533faa2a

solutions/69f8e929da860fb4533faa2a/solution.py

@@ -0,0 +1,116 @@
+# memory_server.py
+from fastmcp import FastMCP
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional, List
+import fnmatch
+
+class MemoryServer:
+    def __init__(self):
+        self.mcp = FastMCP("Memory-Server")
+        self.storage_path = Path("./memory_data.json")
+
+    def _load_memory(self) -> dict:
+        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):
+        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(self, key: str, value: Any) -> bool:
+        """Сохраняет значение по ключу в память сервера."""
+        data = self._load_memory()
+        data[key] = {"value": value, "timestamp": datetime.utcnow().isoformat()}
+        self._save_memory(data)
+        return True
+
+    @self.mcp.tool
+    def get(self, key: str) -> Optional[dict]:
+        """Возвращает значение по ключу с метаданными."""
+        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:
+        """Удаляет ключ из памяти сервера."""
+        data = self._load_memory()
+        if key in data:
+            del data[key]
+            self._save_memory(data)
+            return True
+        return False
+
+    @self.mcp.tool
+    def list_keys(self, pattern: str = "*") -> List[str]:
+        """Возвращает список всех ключей с поддержкой wildcard-паттерна."""
+        data = self._load_memory()
+        return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)]
+
+    @self.mcp.tool
+    def save_with_namespace(self, key: str, value: Any, namespace: str = "default") -> bool:
+        """Сохраняет значение с указанием пространства имён."""
+        full_key = f"{namespace}:{key}"
+        return self.save(full_key, value)
+
+    @self.mcp.tool
+    def get_by_namespace(self, namespace: str = "default") -> List[dict]:
+        """Возвращает все ключи из указанного namespace."""
+        data = self._load_memory()
+        result = []
+        prefix = f"{namespace}:"
+        for k, v in data.items():
+            if k.startswith(prefix):
+                result.append({"key": k[len(prefix):], "value": v["value"], "timestamp": v["timestamp"]})
+        return result
+
+if __name__ == "__main__":
+    server = MemoryServer()
+    server.mcp.run(
+        transport="stdio",
+        show_banner=False,
+        log_level='ERROR'
+    )
+
+# memory_client.py
+import asyncio
+from fastmcp import Client
+
+async def main():
+    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}")
+
+        # Чтение данных
+        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()
+
+if __name__ == "__main__":
+    asyncio.run(main())