From deda284278b0dff31a431aaafd379803ef8e75eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=B4=D0=B5=D0=BB=D0=B8=D0=BD=D0=B0=20=D0=A1=D0=B0?= =?UTF-8?q?=D1=82=D1=82=D0=B0=D1=80=D0=BE=D0=B2=D0=B0?= Date: Thu, 28 May 2026 06:39:47 +0000 Subject: [PATCH] memory_server.py created --- memory_server.py | 99 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 memory_server.py diff --git a/memory_server.py b/memory_server.py new file mode 100644 index 0000000..e5d9f83 --- /dev/null +++ b/memory_server.py @@ -0,0 +1,99 @@ +import json +import os +from datetime import datetime +from fastmcp import MCP, Tool + +# In-memory store backed by JSON file +DATA_FILE = "memory_store.json" + +class MemoryStore: + def __init__(self): + if not os.path.exists(DATA_FILE): + with open(DATA_FILE, "w", encoding="utf-8") as f: + json.dump({}, f) + with open(DATA_FILE, "r", encoding="utf-8") as f: + self.store = json.load(f) + + def _save(self): + with open(DATA_FILE, "w", encoding="utf-8") as f: + json.dump(self.store, f, ensure_ascii=False, indent=2) + + def _key(self, key: str, namespace: str = None) -> str: + return f"{namespace}:{key}" if namespace else key + + # CRUD operations + def save(self, key: str, value: str): + self.store[self._key(key)] = {"value": value, "timestamp": datetime.utcnow().isoformat()} + self._save() + + def get(self, key: str) -> str: + return self.store.get(self._key(key), {}).get("value") + + def delete(self, key: str): + self.store.pop(self._key(key), None) + self._save() + + def list_keys(self): + return [k for k in self.store.keys()] + + # Namespace helpers + def save_with_namespace(self, namespace: str, key: str, value: str): + self.store[self._key(key, namespace)] = {"value": value, "timestamp": datetime.utcnow().isoformat()} + self._save() + + def get_by_namespace(self, namespace: str, key: str) -> str: + return self.store.get(self._key(key, namespace), {}).get("value") + +# Instantiate store +store = MemoryStore() + +# Define tools for MCP +@Tool(name="save", description="Save a value under a key") +def save_tool(args): + key = args["key"] + value = args["value"] + store.save(key, value) + return f"Saved {key}" + +@Tool(name="get", description="Get value by key") +def get_tool(args): + key = args["key"] + val = store.get(key) + return val if val is not None else "Key not found" + +@Tool(name="delete", description="Delete a key") +def delete_tool(args): + key = args["key"] + store.delete(key) + return f"Deleted {key}" + +@Tool(name="list_keys", description="List all keys") +def list_keys_tool(_): + return ", ".join(store.list_keys()) or "No keys" + +@Tool(name="save_with_namespace", description="Save value under key in namespace") +def save_ns_tool(args): + ns = args["namespace"] + key = args["key"] + val = args["value"] + store.save_with_namespace(ns, key, val) + return f"Saved {ns}:{key}" + +@Tool(name="get_by_namespace", description="Get value by key in namespace") +def get_ns_tool(args): + ns = args["namespace"] + key = args["key"] + val = store.get_by_namespace(ns, key) + return val if val is not None else "Key not found" + +# Create MCP instance and register tools +mcp = MCP() +mcp.register_tool(save_tool) +mcp.register_tool(get_tool) +mcp.register_tool(delete_tool) +mcp.register_tool(list_keys_tool) +mcp.register_tool(save_ns_tool) +mcp.register_tool(get_ns_tool) + +if __name__ == "__main__": + mcp.run(transport="stdio")