86 lines
2.0 KiB
Python
86 lines
2.0 KiB
Python
"""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()
|