From c6bdc65bba496573c753d4cf745f2d6782999788 Mon Sep 17 00:00:00 2001 From: balabanovan530 <175+balabanovan530@noreply.localhost> Date: Wed, 3 Jun 2026 12:00:42 +0000 Subject: [PATCH] Add server.py --- server.py | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) create mode 100644 server.py diff --git a/server.py b/server.py new file mode 100644 index 0000000..4f61d3f --- /dev/null +++ b/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()