persistence_tools.py•2.28 kB
"""
Persistence tools for the MCP server.
"""
import logging
from typing import Any, Optional
from persistence import load_persistent_info, save_persistent_info
from . import mcp
@mcp.tool()
async def update_persistent_info(key: str, value: Any) -> str:
"""
Update or add a key-value pair to the persistent info JSON file.
Args:
key: The key to update.
value: The value to set.
Returns:
Success or error message.
"""
logging.info("update_persistent_info called with key=%r, value=%r", key, value)
try:
info = load_persistent_info()
info[key] = value
save_persistent_info(info)
return f"Persistent info updated: {key} = {value!r}"
except Exception as e:
logging.error("Error updating persistent info: %s", e)
return f"Error updating persistent info: {e}"
@mcp.tool()
async def get_persistent_info(key: Optional[str] = None) -> Any:
"""
Retrieve persistent info from disk.
Args:
key: If provided, return only this key's value. Otherwise, return all info.
Returns:
The value or dict, or error message.
"""
logging.info("get_persistent_info called with key=%r", key)
try:
info = load_persistent_info()
if key is not None:
return info.get(key, f"Key '{key}' not found in persistent info.")
return info
except Exception as e:
logging.error("Error retrieving persistent info: %s", e)
return f"Error retrieving persistent info: {e}"
@mcp.tool()
async def delete_persistent_info_key(key: str) -> str:
"""
Delete a key from the persistent info JSON file.
Args:
key: The key to delete.
Returns:
Success or error message.
"""
logging.info("delete_persistent_info_key called with key=%r", key)
try:
info = load_persistent_info()
if key in info:
del info[key]
save_persistent_info(info)
return f"Key '{key}' deleted from persistent info."
else:
return f"Key '{key}' not found in persistent info."
except Exception as e:
logging.error("Error deleting key from persistent info: %s", e)
return f"Error deleting key from persistent info: {e}"