MCPcopy Create free account
hub / github.com/HKUDS/OpenOPC / _MemoryStore

Class _MemoryStore

tests/test_parallel_runtime_isolation.py:19–41  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

17
18
19class _MemoryStore:
20 def __init__(self, tasks: list[Task]) -> None:
21 self.tasks = {task.id: task for task in tasks}
22 self.saved: list[Task] = []
23 self.renewed: list[str] = []
24
25 @property
26 def is_ready(self) -> bool:
27 return True
28
29 async def get_task(self, task_id: str) -> Task | None:
30 return self.tasks.get(task_id)
31
32 async def save_task(self, task: Task) -> None:
33 self.tasks[task.id] = task
34 self.saved.append(task)
35
36 async def get_session_transcript(self, session_id: str) -> list[dict]:
37 return []
38
39 async def renew_task_lock(self, task_id: str) -> bool:
40 self.renewed.append(task_id)
41 return True
42
43
44def _ui_chat_store() -> SimpleNamespace:

Calls

no outgoing calls