Add memory management APIs for OpenViking: list, read, write, and delete memories

memory_system_api/service.py

@@ -11,6 +11,8 @@ from .schemas import (
     BackendStatus,
     CommitResponse,
     ExtractResponse,
+    MemoryOperationResponse,
+    MemoryWriteRequest,
     MessageIngestRequest,
     MessageIngestResponse,
     ProfileResponse,
@@ -76,6 +78,63 @@ class MemorySystemService:
         resource = backends["openviking"].result if backends["openviking"].status == "success" else None
         return ResourceMutationResponse(status=self._aggregate_status(backends), resource=resource, backends=backends)
 
+    async def list_memories(
+        self,
+        user_id: str,
+        user_key: str,
+        uri: str = "viking://user/memories",
+        recursive: bool = True,
+    ) -> MemoryOperationResponse:
+        credential = self.openviking.credential_for_user(user_id, user_key)
+        backends = {
+            "openviking": await self._capture(lambda: self.openviking.list_memories(credential, uri, recursive)),
+        }
+        memory = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return MemoryOperationResponse(status=self._aggregate_status(backends), memory=memory, backends=backends)
+
+    async def read_memory(
+        self,
+        user_id: str,
+        user_key: str,
+        uri: str,
+    ) -> MemoryOperationResponse:
+        credential = self.openviking.credential_for_user(user_id, user_key)
+        backends = {
+            "openviking": await self._capture(lambda: self.openviking.read_memory(credential, uri)),
+        }
+        memory = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return MemoryOperationResponse(status=self._aggregate_status(backends), memory=memory, backends=backends)
+
+    async def write_memory(self, request: MemoryWriteRequest) -> MemoryOperationResponse:
+        credential = self.openviking.credential_for_user(request.user_id, request.user_key)
+        backends = {
+            "openviking": await self._capture(
+                lambda: self.openviking.write_memory(
+                    credential,
+                    uri=request.uri,
+                    content=request.content,
+                    mode=request.mode,
+                    wait=request.wait,
+                )
+            ),
+        }
+        memory = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return MemoryOperationResponse(status=self._aggregate_status(backends), memory=memory, backends=backends)
+
+    async def delete_memory(
+        self,
+        user_id: str,
+        user_key: str,
+        uri: str,
+        recursive: bool = False,
+    ) -> MemoryOperationResponse:
+        credential = self.openviking.credential_for_user(user_id, user_key)
+        backends = {
+            "openviking": await self._capture(lambda: self.openviking.delete_memory(credential, uri, recursive)),
+        }
+        memory = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return MemoryOperationResponse(status=self._aggregate_status(backends), memory=memory, backends=backends)
+
     async def ingest_messages(self, request: MessageIngestRequest) -> MessageIngestResponse:
         messages = self._messages_from_request(request)
         if not messages: