From 6010476f4c6332810dc9075500baa4d99c37a1f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9C=D0=B0=D1=80=D0=B0=D1=82=20=D0=A4=D0=B0=D0=B7=D1=8B?= =?UTF-8?q?=D0=BB=D0=BE=D0=B2?= Date: Fri, 8 May 2026 10:32:55 +0000 Subject: [PATCH] Add MCP Memory Server with 6 tools: save, get, delete, list_keys, save_with_namespace, get_by_namespace --- memory_server.py | 160 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 160 insertions(+) create mode 100644 memory_server.py diff --git a/memory_server.py b/memory_server.py new file mode 100644 index 0000000..b64cf89 --- /dev/null +++ b/memory_server.py @@ -0,0 +1,160 @@ +from fastmcp import FastMCP + +import json +from datetime import datetime +from pathlib import Path +from typing import Any, Optional +import fnmatch + + +class MemoryServer: + def __init__(self): + self.mcp = FastMCP("Memory-Server") + self.storage_path = Path("./memory_data.json") + self.register_tools() + + def _load_memory(self) -> dict: + """Load memory from JSON file.""" + 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): + """Save memory to JSON file.""" + 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) + + def register_tools(self): + @self.mcp.tool() + def save(key: str, value: Any) -> bool: + """Save a value by key to server memory. + + Args: + key: Unique identifier for storage. + value: Any serializable value + + Returns: + True if successfully saved, False otherwise. + """ + if key and "/../" in key: + return False + try: + data = self._load_memory() + data[key] = { + "value": value, + "timestamp": datetime.now().isoformat() + } + self._save_memory(data) + return True + except Exception: + return False + + @self.mcp.tool() + def get(key: str) -> Optional[dict]: + """Get a value by key with metadata. + + Args: + key: Unique identifier to look up. + + Returns: + Dict with {"key": ..., "value": ..., "timestamp": ...} + or None if key not found. + """ + try: + data = self._load_memory() + if key in data: + return { + "key": key, + "value": data[key]["value"], + "timestamp": data[key]["timestamp"] + } + return None + except Exception: + return None + + @self.mcp.tool() + def delete(key: str) -> bool: + """Delete a key from server memory. + + Args: + key: Unique identifier to delete. + + Returns: + True if successfully deleted, False if key not found. + """ + try: + data = self._load_memory() + if key in data: + del data[key] + self._save_memory(data) + return True + return False + except Exception: + return False + + @self.mcp.tool() + def list_keys(pattern: str = "*") -> list[str]: + """List all keys matching a wildcard pattern. + + Args: + pattern: Filter pattern (supports * and ?) + + Returns: + List of matching keys. + """ + try: + data = self._load_memory() + return [k for k in data.keys() if fnmatch.fnmatch(k, pattern)] + except Exception: + return [] + + @self.mcp.tool() + def save_with_namespace(key: str, value: Any, namespace: str = "default") -> bool: + """Save a value with a namespace. + + Args: + key: Unique identifier. + value: Value to store. + namespace: Namespace (default 'default'). + + Returns: + True if successfully saved, False otherwise. + """ + combined_key = f"{namespace}:{key}" + return save(combined_key, value) + + @self.mcp.tool() + def get_by_namespace(namespace: str = "default") -> list[dict]: + """Get all keys from a specified namespace. + + Args: + namespace: Namespace to read from. + + Returns: + List of dicts with metadata for all keys in namespace. + """ + prefix = f"{namespace}:" + try: + data = self._load_memory() + result = [] + for key, val_data in data.items(): + if key.startswith(prefix): + result.append({ + "key": key.replace(prefix, "", 1), + "value": val_data["value"], + "timestamp": val_data["timestamp"] + }) + return result + except Exception: + return [] + + +if __name__ == "__main__": + server = MemoryServer() + server.mcp.run( + transport="stdio", + show_banner=False, + log_level='ERROR' + )