fix: memory_server.py — MCP-сервер для управления памятью агента

memory_server.py

@@ -12,89 +12,143 @@ class MemoryServer:
         self.mcp = FastMCP("Memory-Server")
         self.storage_path = Path("./memory_data.json")
         # Register tools
-        self._register_tools()
+        self.mcp.tool()(self.save)
+        self.mcp.tool()(self.get)
+        self.mcp.tool()(self.delete)
+        self.mcp.tool()(self.list_keys)
+        self.mcp.tool()(self.save_with_namespace)
+        self.mcp.tool()(self.get_by_namespace)
 
-    def _load_memory(self) -> Dict[str, dict]:
+    def _load_memory(self) -> Dict[str, Dict]:
+        """Load memory from JSON file."""
         if not self.storage_path.exists():
             return {}
         with self.storage_path.open("r", encoding="utf-8") as f:
             return json.load(f)
 
-    def _save_memory(self, data: Dict[str, dict]) -> None:
+    def _save_memory(self, data: Dict[str, Dict]):
+        """Save memory to JSON file."""
         self.storage_path.parent.mkdir(parents=True, exist_ok=True)
         with self.storage_path.open("w", encoding="utf-8") as f:
             json.dump(data, f, indent=2, ensure_ascii=False)
 
     def _validate_key(self, key: str) -> bool:
-        # Disallow path traversal characters
+        """Simple validation to avoid path traversal etc."""
-        return ".." not in key and "/" not in key and "\\" not in key
+        if ".." in key or "/" in key or "\\" in key:
+            return False
+        return True
 
-    def _register_tools(self):
+    def save(self, key: str, value: Any) -> bool:
-        @self.mcp.tool()
+        """
-        def save(key: str, value: Any) -> bool:
+        Save a value under a key.
-            """Save a value under a key."""
+
-            if not self._validate_key(key):
+        Args:
-                return False
+            key: Unique identifier.
-            data = self._load_memory()
+            value: Any JSON-serializable value.
-            data[key] = {"value": value, "timestamp": datetime.utcnow().isoformat()}
+
+        Returns:
+            True on success, False otherwise.
+        """
+        if not self._validate_key(key):
+            return False
+        data = self._load_memory()
+        data[key] = {
+            "value": value,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        self._save_memory(data)
+        return True
+
+    def get(self, key: str) -> Optional[Dict]:
+        """
+        Retrieve a value with metadata.
+
+        Args:
+            key: Identifier to look up.
+
+        Returns:
+            Dict with keys 'key', 'value', 'timestamp' or None.
+        """
+        if not self._validate_key(key):
+            return None
+        data = self._load_memory()
+        entry = data.get(key)
+        if entry is None:
+            return None
+        return {
+            "key": key,
+            "value": entry["value"],
+            "timestamp": entry["timestamp"]
+        }
+
+    def delete(self, key: str) -> bool:
+        """
+        Delete a key from memory.
+
+        Args:
+            key: Identifier to delete.
+
+        Returns:
+            True if the key existed and was removed, False otherwise.
+        """
+        if not self._validate_key(key):
+            return False
+        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]:
-        def get(key: str) -> Optional[dict]:
+        """
-            """Retrieve a value with metadata."""
+        List all keys matching a wildcard pattern.
-            if not self._validate_key(key):
-                return 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()
+        Args:
-        def delete(key: str) -> bool:
+            pattern: Wildcard pattern (* and ? supported).
-            """Delete a key."""
-            if not self._validate_key(key):
-                return False
-            data = self._load_memory()
-            if key in data:
-                del data[key]
-                self._save_memory(data)
-                return True
-            return False
 
-        @self.mcp.tool()
+        Returns:
-        def list_keys(pattern: str = "*") -> List[str]:
+            List of matching keys.
-            """List keys matching a wildcard pattern."""
+        """
-            data = self._load_memory()
+        data = self._load_memory()
-            return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)]
+        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:
-        def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool:
+        """
-            """Save a value with a namespace."""
+        Save a value with a namespace prefix.
-            if not self._validate_key(key) or not self._validate_key(namespace):
-                return False
-            full_key = f"{namespace}:{key}"
-            return save(full_key, value)
 
-        @self.mcp.tool()
+        Args:
-        def get_by_namespace(namespace: str = "default") -> List[dict]:
+            key: Identifier.
-            """Get all entries from a namespace."""
+            value: JSON-serializable value.
-            if not self._validate_key(namespace):
+            namespace: Namespace name.
-                return []
+
-            data = self._load_memory()
+        Returns:
-            prefix = f"{namespace}:"
+            True on success, False otherwise.
-            result = []
+        """
-            for k, v in data.items():
+        full_key = f"{namespace}:{key}"
-                if k.startswith(prefix):
+        return self.save(full_key, value)
-                    result.append(
+
-                        {
+    def get_by_namespace(self, namespace: str = "default") -> List[Dict]:
-                            "key": k[len(prefix) :],
+        """
-                            "value": v["value"],
+        Retrieve all entries belonging to a namespace.
-                            "timestamp": v["timestamp"],
+
-                        }
+        Args:
-                    )
+            namespace: Namespace to query.
-            return result
+
+        Returns:
+            List of dictionaries with metadata for each entry.
+        """
+        prefix = f"{namespace}:"
+        data = self._load_memory()
+        result = []
+        for k, v in data.items():
+            if k.startswith(prefix):
+                result.append({
+                    "key": k,
+                    "value": v["value"],
+                    "timestamp": v["timestamp"]
+                })
+        return result
 
 
 if __name__ == "__main__":
@@ -102,5 +156,5 @@ if __name__ == "__main__":
     server.mcp.run(
         transport="stdio",
         show_banner=False,
-        log_level="ERROR",
+        log_level="ERROR"
     )