Add server.py

server.py

@@ -0,0 +1,85 @@
+"""MCP server providing memory tools with namespace support.
+
+This server uses FastMCP and stores data in a JSON file (data.json).
+The JSON structure is a dictionary mapping keys to objects containing value and timestamp.
+"""
+
+import json
+import time
+from pathlib import Path
+
+from fastmcp import mcp
+from pydantic import BaseModel
+from dotenv import load_dotenv
+
+# Load environment variables if .env exists
+load_dotenv()
+
+DATA_FILE = Path("data.json")
+
+# Ensure data file exists
+if not DATA_FILE.exists():
+    DATA_FILE.write_text("{}")
+
+class DataEntry(BaseModel):
+    value: str
+    timestamp: float
+
+# Helper functions for file operations
+
+def _load_data() -> dict:
+    try:
+        text = DATA_FILE.read_text()
+        return json.loads(text) if text else {}
+    except json.JSONDecodeError:
+        return {}
+
+def _save_data(data: dict) -> None:
+    DATA_FILE.write_text(json.dumps(data, indent=2))
+
+# Tool implementations
+
+@mcp.tool()
+def save(key: str, value: str) -> str:
+    data = _load_data()
+    data[key] = DataEntry(value=value, timestamp=time.time()).dict()
+    _save_data(data)
+    return f"Saved key '{key}'."
+
+@mcp.tool()
+def get(key: str) -> str:
+    data = _load_data()
+    entry = data.get(key)
+    if not entry:
+        return f"Key '{key}' not found."
+    return entry["value"]
+
+@mcp.tool()
+def delete(key: str) -> str:
+    data = _load_data()
+    if key in data:
+        del data[key]
+        _save_data(data)
+        return f"Deleted key '{key}'."
+    return f"Key '{key}' not found."
+
+@mcp.tool()
+def list_keys() -> str:
+    data = _load_data()
+    return json.dumps(list(data.keys()))
+
+# Namespace tools
+
+@mcp.tool()
+def save_with_namespace(namespace: str, key: str, value: str) -> str:
+    namespaced_key = f"{namespace}:{key}"
+    return save(namespaced_key, value)
+
+@mcp.tool()
+def get_by_namespace(namespace: str, key: str) -> str:
+    namespaced_key = f"{namespace}:{key}"
+    return get(namespaced_key)
+
+# Main entrypoint
+if __name__ == "__main__":
+    mcp.run()