From ad5b50084016540562dc231eea76ba1b3c79940d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9A=D0=B8=D1=80=D0=B8=D0=BB=D0=BB=20=D0=9A=D1=83=D1=82?= =?UTF-8?q?=D0=BB=D0=B0=D1=85=D0=BC=D0=B5=D1=82=D0=BE=D0=B2?= Date: Mon, 25 May 2026 22:57:29 +0000 Subject: [PATCH] Add solution for 69f8e929da860fb4533faa2a --- memory_server.py | 138 +++++++++++++---------------------------------- 1 file changed, 38 insertions(+), 100 deletions(-) diff --git a/memory_server.py b/memory_server.py index 50dabfb..ef2d00d 100644 --- a/memory_server.py +++ b/memory_server.py @@ -1,137 +1,75 @@ -from __future__ import annotations - -import fnmatch +from fastmcp import FastMCP import json from datetime import datetime from pathlib import Path from typing import Any, Optional - -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 - +import fnmatch 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._register_tools() + 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: + 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]: - return { - "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) + @self.mcp.tool() + def save(self, key: str, value: Any) -> bool: 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]: - key = validate_token(key, "key", allow_colon=True) + @self.mcp.tool() + def get(self, key: str) -> Optional[dict]: data = self._load_memory() - record = data.get(key) - if record is None: - return None - return {"key": key, **record} + if key in data: + entry = data[key] + return {"key": key, "value": entry["value"], "timestamp": entry["timestamp"]} + return None - def _delete_entry(self, key: str) -> bool: - key = validate_token(key, "key", allow_colon=True) + @self.mcp.tool() + def delete(self, key: str) -> bool: data = self._load_memory() - if key not in data: - return False - del data[key] - self._save_memory(data) - return True + if key in data: + del data[key] + self._save_memory(data) + 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: - namespace = validate_token(namespace, "namespace") - key = validate_token(key, "key") - return self._save_entry(f"{namespace}:{key}", value) + @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) - def _get_namespace_entries(self, namespace: str = "default") -> list[dict]: - namespace = validate_token(namespace, "namespace") + @self.mcp.tool() + 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): - if key.startswith(prefix): - result.append({"key": key, **data[key]}) + for k, v in data.items(): + if k.startswith(prefix): + 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' + ) \ No newline at end of file