File size: 1,786 Bytes
88d42c6
 
f30f44c
 
 
 
 
 
 
 
 
 
 
 
 
88d42c6
f30f44c
 
 
 
88d42c6
 
f30f44c
 
 
 
 
 
 
 
 
88d42c6
f30f44c
 
 
 
88d42c6
f30f44c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
# mcp/workspace.py

#!/usr/bin/env python3
"""MedGenesis – lightweight per‑user workspace helper

Persists the user’s most recent queries/results in **Streamlit `session_state`**.
Keeps only the last *N* (default = 10) items to avoid bloat.
Provides three public helpers:
    • `get_workspace()`   → returns the list (for sidebar rendering)
    • `save_query(q,res)` → prepends new item & trims to max_len
    • `clear_workspace()` → wipe slate (user button use‑case)
"""

from __future__ import annotations

import streamlit as st
from typing import List, Dict

_KEY = "saved_queries"
_MAX = 10  # adjust if you want deeper history


def get_workspace() -> List[Dict]:
    """Return the current saved‑query list (initialises if absent)."""
    if _KEY not in st.session_state:
        st.session_state[_KEY] = []
    return st.session_state[_KEY]


def save_query(query: str, result: Dict, *, max_len: int = _MAX) -> None:
    """Prepend a (query, result) record, trim list to *max_len*."""
    ws = get_workspace()

    # Deduplicate identical query strings (keep newest on top)
    ws[:] = [item for item in ws if item["query"] != query]

    ws.insert(0, {"query": query, "result": result})
    del ws[max_len:]  # in‑place trim


def clear_workspace() -> None:
    """Remove all saved queries for current user session."""
    st.session_state[_KEY] = []


# ------------------------------------------------------------------
# 👟 Quick manual test (run:  python -m mcp.workspace)
# ------------------------------------------------------------------
if __name__ == "__main__":
    print("Before:", get_workspace())
    save_query("demo", {"answer": 42})
    print("After :", get_workspace())
    clear_workspace()
    print("Cleared:", get_workspace())