Add solution for 69f8e929da860fb4533faa2a

memory_server.py

@@ -1,137 +1,75 @@
-from __future__ import annotations
+from fastmcp import FastMCP
-
-import fnmatch
 import json
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Optional
-
+import fnmatch
-from fastmcp import FastMCP
-
-
-def validate_token(value: str, field: str, allow_colon: bool = False) -> str:
-    value = value.strip()
-    if not value:
-        raise ValueError(f"{field} must not be empty")
-
-    forbidden = ["..", "/", "\\"]
-    if not allow_colon:
-        forbidden.append(":")
-
-    for marker in forbidden:
-        if marker in value:
-            raise ValueError(f"{field} contains forbidden sequence: {marker}")
-    return value
-
 
 class MemoryServer:
-    def __init__(self, storage_path: Path | None = None):
+    def __init__(self):
         self.mcp = FastMCP("Memory-Server")
-        self.storage_path = storage_path or Path("./memory_data.json")
+        self.storage_path = Path("./memory_data.json")
-        self._register_tools()
 
     def _load_memory(self) -> dict:
         if not self.storage_path.exists():
             return {}
-        with open(self.storage_path, "r", encoding="utf-8") as f:
+        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:
+        with open(self.storage_path, 'w', encoding='utf-8') as f:
             json.dump(data, f, indent=2, ensure_ascii=False)
 
-    def _make_record(self, value: Any) -> dict[str, Any]:
+    @self.mcp.tool()
-        return {
+    def save(self, key: str, value: Any) -> bool:
-            "value": value,
-            "timestamp": datetime.now().isoformat(timespec="seconds"),
-        }
-
-    def _save_entry(self, key: str, value: Any) -> bool:
-        key = validate_token(key, "key", allow_colon=True)
         data = self._load_memory()
-        data[key] = self._make_record(value)
+        data[key] = {"value": value, "timestamp": datetime.utcnow().isoformat()}
         self._save_memory(data)
         return True
 
-    def _get_entry(self, key: str) -> Optional[dict]:
+    @self.mcp.tool()
-        key = validate_token(key, "key", allow_colon=True)
+    def get(self, key: str) -> Optional[dict]:
         data = self._load_memory()
-        record = data.get(key)
+        if key in data:
-        if record is None:
+            entry = data[key]
-            return None
+            return {"key": key, "value": entry["value"], "timestamp": entry["timestamp"]}
-        return {"key": key, **record}
+        return None
 
-    def _delete_entry(self, key: str) -> bool:
+    @self.mcp.tool()
-        key = validate_token(key, "key", allow_colon=True)
+    def delete(self, key: str) -> bool:
         data = self._load_memory()
-        if key not in data:
+        if key in data:
-            return False
+            del data[key]
-        del data[key]
+            self._save_memory(data)
-        self._save_memory(data)
+            return True
-        return True
+        return False
 
-    def _list_entry_keys(self, pattern: str = "*") -> list[str]:
+    @self.mcp.tool()
+    def list_keys(self, pattern: str = "*") -> list[str]:
         data = self._load_memory()
-        return sorted(key for key in data if fnmatch.fnmatch(key, pattern))
+        return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)]
 
-    def _save_namespaced(self, key: str, value: Any, namespace: str = "default") -> bool:
+    @self.mcp.tool()
-        namespace = validate_token(namespace, "namespace")
+    def save_with_namespace(self, key: str, value: Any, namespace: str = "default") -> bool:
-        key = validate_token(key, "key")
+        full_key = f"{namespace}:{key}"
-        return self._save_entry(f"{namespace}:{key}", value)
+        return self.save(full_key, value)
 
-    def _get_namespace_entries(self, namespace: str = "default") -> list[dict]:
+    @self.mcp.tool()
-        namespace = validate_token(namespace, "namespace")
+    def get_by_namespace(self, namespace: str = "default") -> list[dict]:
+        data = self._load_memory()
         prefix = f"{namespace}:"
-        data = self._load_memory()
         result = []
-        for key in sorted(data):
+        for k, v in data.items():
-            if key.startswith(prefix):
+            if k.startswith(prefix):
-                result.append({"key": key, **data[key]})
+                short_key = k[len(prefix):]
+                result.append({"key": short_key, "value": v["value"], "timestamp": v["timestamp"]})
         return result
 
-    def _register_tools(self) -> None:
-        @self.mcp.tool()
-        def save(key: str, value: Any) -> bool:
-            """Save a serializable value under a key."""
-
-            return self._save_entry(key, value)
-
-        @self.mcp.tool()
-        def get(key: str) -> Optional[dict]:
-            """Return a saved value with its metadata or None."""
-
-            return self._get_entry(key)
-
-        @self.mcp.tool()
-        def delete(key: str) -> bool:
-            """Delete a key from memory if it exists."""
-
-            return self._delete_entry(key)
-
-        @self.mcp.tool()
-        def list_keys(pattern: str = "*") -> list[str]:
-            """List saved keys filtered by a wildcard pattern."""
-
-            return self._list_entry_keys(pattern)
-
-        @self.mcp.tool()
-        def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool:
-            """Save a value inside a namespace using namespace:key format."""
-
-            return self._save_namespaced(key, value, namespace)
-
-        @self.mcp.tool()
-        def get_by_namespace(namespace: str = "default") -> list[dict]:
-            """Return all records that belong to the namespace."""
-
-            return self._get_namespace_entries(namespace)
-
-
 if __name__ == "__main__":
     server = MemoryServer()
     server.mcp.run(
         transport="stdio",
         show_banner=False,
-        log_level="ERROR",
+        log_level='ERROR'
     )