Update app.py

app.py

@@ -27,6 +27,9 @@ from PIL import Image
 from huggingface_hub import InferenceClient
 import time
 import tempfile
+import uuid
+import threading
+from datetime import datetime
 
 # Optional imports for Kokoro TTS (loaded lazily)
 import numpy as np
@@ -743,6 +746,225 @@ def Generate_Speech(  # <-- MCP tool #4 (Generate Speech)
         raise gr.Error(f"Error during speech generation: {str(e)}")
 
 
+# ==========================
+# JSON Memory System (MCP tools #7–#10 if enabled)
+# ==========================
+
+# Implementation goals (aligned with Gradio MCP docs):
+#  * Each function has a rich docstring (used for tool description)
+#  * Type hints + Annotated param docs become the schema
+#  * Zero external dependencies (pure stdlib JSON file persistence)
+#  * Safe concurrent access via a process‑local lock
+#  * Human‑readable & recoverable even if file becomes corrupted
+
+MEMORY_FILE = os.path.join(os.path.dirname(__file__), "memories.json")
+_MEMORY_LOCK = threading.RLock()
+_MAX_MEMORIES = 10_000  # soft cap to avoid unbounded growth
+
+
+def _now_iso() -> str:
+    return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+
+
+def _load_memories() -> List[Dict[str, str]]:
+    """Internal helper: load memory list from disk.
+
+    Returns an empty list if the file does not exist or is unreadable.
+    If the JSON is corrupted, a *.corrupt backup is written once and a
+    fresh empty list is returned (fail‑open philosophy for tool usage).
+    """
+    if not os.path.exists(MEMORY_FILE):
+        return []
+    try:
+        with open(MEMORY_FILE, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        if isinstance(data, list):
+            # Filter only dict items containing required keys if present
+            cleaned: List[Dict[str, str]] = []
+            for item in data:
+                if isinstance(item, dict) and "id" in item and "text" in item:
+                    cleaned.append(item)
+            return cleaned
+        return []
+    except Exception:
+        # Backup corrupted file once
+        try:
+            backup = MEMORY_FILE + ".corrupt"
+            if not os.path.exists(backup):
+                os.replace(MEMORY_FILE, backup)
+        except Exception:
+            pass
+        return []
+
+
+def _save_memories(memories: List[Dict[str, str]]) -> None:
+    """Persist memory list atomically to disk (write temp then replace)."""
+    tmp_path = MEMORY_FILE + ".tmp"
+    with open(tmp_path, "w", encoding="utf-8") as f:
+        json.dump(memories, f, ensure_ascii=False, indent=2)
+    os.replace(tmp_path, MEMORY_FILE)
+
+
+def Save_Memory(  # <-- MCP tool (Save memory)
+    text: Annotated[str, "Raw textual content to remember (will be stored verbatim)."],
+    tags: Annotated[str, "Optional comma-separated tags for lightweight categorization (e.g. 'user, preference')."] = "",
+) -> str:
+    """Store a new memory entry in a local JSON file (no external DB required).
+
+    Each memory is a JSON object with: id (UUID4), text, timestamp (UTC), tags.
+    The file `memories.json` lives beside this script; it is created on demand.
+
+    Behavior:
+      * Trims surrounding whitespace in `text`; rejects empty result.
+      * Deduplicates only if the newest existing memory has identical text
+        (cheap heuristic to avoid accidental double submissions).
+      * Soft cap (`_MAX_MEMORIES`); oldest entries are dropped if exceeded.
+
+    Returns:
+      Human‑readable confirmation containing the new memory UUID prefix.
+
+    Example:
+      Save_Memory(text="User prefers dark mode", tags="preference, ui")
+
+    Limitations:
+      * Not encrypted; do not store secrets.
+      * Concurrency protection is process‑local (thread lock). For multi‑process
+        usage you would need an OS file lock (out of scope here).
+    """
+    text_clean = (text or "").strip()
+    if not text_clean:
+        return "Error: memory text is empty."
+
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+        if memories and memories[-1].get("text") == text_clean:
+            return "Skipped: identical to last stored memory."
+
+        mem_id = str(uuid.uuid4())
+        entry = {
+            "id": mem_id,
+            "text": text_clean,
+            "timestamp": _now_iso(),
+            "tags": tags.strip(),
+        }
+        memories.append(entry)
+        if len(memories) > _MAX_MEMORIES:
+            # Drop oldest overflow
+            overflow = len(memories) - _MAX_MEMORIES
+            memories = memories[overflow:]
+        _save_memories(memories)
+    return f"Memory saved: {mem_id}"
+
+
+def List_Memories(  # <-- MCP tool (List memories)
+    limit: Annotated[int, "Maximum number of most recent memories to return (1–200)."] = 20,
+    include_tags: Annotated[bool, "If true, include tags column in output."] = True,
+) -> str:
+    """Return the N most recent memories in reverse chronological order.
+
+    Args:
+      limit: Upper bound of entries (clamped to 1–200).
+      include_tags: Whether to display tags in the formatted lines.
+
+    Format:
+      Each line: `<uuid_prefix> [YYYY-MM-DD HH:MM:SS] <text>` (+ ` | tags: ...` if present & enabled).
+
+    Notes:
+      * If no memories exist, returns a friendly message instead of empty string.
+      * UUID prefix is first 8 chars to keep responses compact (full id retained on disk).
+    """
+    limit = max(1, min(200, limit))
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+    if not memories:
+        return "No memories stored yet."
+    # Already chronological (append order); display newest first
+    chosen = memories[-limit:][::-1]
+    lines: List[str] = []
+    for m in chosen:
+        base = f"{m['id'][:8]} [{m.get('timestamp','?')}] {m.get('text','')}"
+        if include_tags and m.get("tags"):
+            base += f" | tags: {m['tags']}"
+        lines.append(base)
+    return "\n".join(lines)
+
+
+def Search_Memories(  # <-- MCP tool (Search memories)
+    query: Annotated[str, "Case-insensitive substring search; space-separated terms are ANDed."],
+    limit: Annotated[int, "Maximum number of matches (1–200)."] = 20,
+) -> str:
+    """Search memory text (and tags) for all provided terms.
+
+    Matching logic:
+      * Query is split on whitespace -> terms.
+      * A memory matches if every term appears (case-insensitive) in either
+        the text or the tag string.
+      * Results are sorted by timestamp descending (newest first).
+
+    Args:
+      query: One or more words. Empty query is rejected.
+      limit: Clamp 1–200.
+
+    Returns:
+      Formatted lines (same style as List_Memories) or 'No matches'.
+    """
+    q = (query or "").strip()
+    if not q:
+        return "Error: empty query."
+    terms = [t.lower() for t in q.split() if t.strip()]
+    if not terms:
+        return "Error: no valid search terms."
+    limit = max(1, min(200, limit))
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+    # Newest first iteration for early cutoff
+    matches: List[Dict[str, str]] = []
+    for m in reversed(memories):  # newest backward
+        hay = (m.get("text", "") + " " + m.get("tags", "")).lower()
+        if all(t in hay for t in terms):
+            matches.append(m)
+            if len(matches) >= limit:
+                break
+    if not matches:
+        return f"No matches for: {query}"
+    lines = [f"{m['id'][:8]} [{m.get('timestamp','?')}] {m.get('text','')}" + (f" | tags: {m['tags']}" if m.get('tags') else "") for m in matches]
+    return "\n".join(lines)
+
+
+def Delete_Memory(  # <-- MCP tool (Delete memory)
+    memory_id: Annotated[str, "Full UUID or a unique prefix (>=4 chars) of the memory id to delete."],
+) -> str:
+    """Delete a single memory by exact UUID or unique prefix.
+
+    Args:
+      memory_id: Full UUID (recommended) or prefix (minimum 4 chars). If the
+        prefix matches multiple entries, no deletion occurs (safety).
+
+    Returns:
+      * Success message with full UUID
+      * Disambiguation notice if prefix is not unique
+      * Not-found message
+    """
+    key = (memory_id or "").strip().lower()
+    if len(key) < 4:
+        return "Error: supply at least 4 characters of the id."
+    with _MEMORY_LOCK:
+        memories = _load_memories()
+        matched = [m for m in memories if m["id"].lower().startswith(key)]
+        if not matched:
+            return "Memory not found."
+        if len(matched) > 1 and key != matched[0]["id"].lower():
+            # ambiguous prefix
+            sample = ", ".join(m["id"][:8] for m in matched[:5])
+            more = "…" if len(matched) > 5 else ""
+            return f"Ambiguous prefix (matches {len(matched)} ids: {sample}{more}). Provide more characters."
+        # Unique match
+        target_id = matched[0]["id"]
+        memories = [m for m in memories if m["id"] != target_id]
+        _save_memories(memories)
+    return f"Deleted memory: {target_id}"
+
+
 # ======================
 # UI: four-tab interface
 # ======================
@@ -879,6 +1101,43 @@ kokoro_interface = gr.Interface(
     flagging_mode="never",
 )
 
+# --- Memory Manager tab (JSON persistence) ---
+with gr.Blocks(title="Memory Manager") as memory_interface:
+    gr.Markdown(
+        """
+        ### JSON Memory Manager
+        Lightweight, local persistence (no external DB). Four MCP tools are exposed:
+        - Save_Memory
+        - List_Memories
+        - Search_Memories
+        - Delete_Memory
+        Data stored in `memories.json` (UUID, text, timestamp, optional tags). Not encrypted—avoid secrets.
+        """
+    )
+    with gr.Tab("Save"):
+        in_text = gr.Textbox(label="Memory Text", lines=3)
+        in_tags = gr.Textbox(label="Tags (comma separated)")
+        out_save = gr.Textbox(label="Result", interactive=False)
+        save_btn = gr.Button("Save Memory")
+        save_btn.click(Save_Memory, inputs=[in_text, in_tags], outputs=out_save)
+    with gr.Tab("List"):
+        in_limit = gr.Slider(1, 200, value=20, step=1, label="Limit")
+        in_show_tags = gr.Checkbox(value=True, label="Include Tags")
+        out_list = gr.Textbox(label="Memories", lines=10)
+        list_btn = gr.Button("List Recent")
+        list_btn.click(List_Memories, inputs=[in_limit, in_show_tags], outputs=out_list)
+    with gr.Tab("Search"):
+        in_query = gr.Textbox(label="Query", placeholder="keywords…")
+        in_search_limit = gr.Slider(1, 200, value=20, step=1, label="Limit")
+        out_search = gr.Textbox(label="Matches", lines=10)
+        search_btn = gr.Button("Search")
+        search_btn.click(Search_Memories, inputs=[in_query, in_search_limit], outputs=out_search)
+    with gr.Tab("Delete"):
+        in_mem_id = gr.Textbox(label="Memory UUID or prefix")
+        out_delete = gr.Textbox(label="Result", interactive=False)
+        del_btn = gr.Button("Delete")
+        del_btn.click(Delete_Memory, inputs=[in_mem_id], outputs=out_delete)
+
 # ==========================
 # Image Generation (Serverless)
 # ==========================
@@ -1201,12 +1460,14 @@ _interfaces = [
     fetch_interface,
     concise_interface,
     code_interface,
+    memory_interface,
     kokoro_interface,
 ]
 _tab_names = [
     "Fetch Webpage",
     "DuckDuckGo Search",
     "Python Code Executor",
+    "Memory Manager",
     "Kokoro TTS",
 ]