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")