Add resource upload APIs

memory_system_api/service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import asyncio
 from typing import Any, Awaitable, Callable
+from urllib.parse import urlparse
 
 from .clients import EverOSMemorySystemClient, OpenVikingMemorySystemClient
 from .schemas import (
@@ -13,6 +14,8 @@ from .schemas import (
     MessageIngestRequest,
     MessageIngestResponse,
     ProfileResponse,
+    ResourceMutationResponse,
+    ResourceUploadRequest,
     SearchRequest,
     SearchResponse,
     SessionContextRequest,
@@ -30,6 +33,49 @@ class MemorySystemService:
         account = backends["openviking"].result if backends["openviking"].status == "success" else None
         return AccountResponse(status=self._aggregate_status(backends), account=account, backends=backends)
 
+    async def upload_resource(self, request: ResourceUploadRequest) -> ResourceMutationResponse:
+        credential = self.openviking.credential_for_user(request.user_id, request.user_key)
+
+        async def upload_openviking() -> dict[str, Any]:
+            if self._is_remote_url(request.path):
+                return await self.openviking.add_resource(
+                    credential,
+                    path=request.path,
+                    to=request.to,
+                    reason=request.reason,
+                    wait=request.wait,
+                    directly_upload_media=request.directly_upload_media,
+                )
+
+            temp_upload = await self.openviking.upload_temp_file(credential, request.path)
+            temp_file_id = self._temp_file_id_from_result(temp_upload)
+            return await self.openviking.add_resource(
+                credential,
+                temp_file_id=temp_file_id,
+                to=request.to,
+                reason=request.reason,
+                wait=request.wait,
+                directly_upload_media=request.directly_upload_media,
+            )
+
+        backends = {"openviking": await self._capture(upload_openviking)}
+        resource = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return ResourceMutationResponse(status=self._aggregate_status(backends), resource=resource, backends=backends)
+
+    async def delete_resource(
+        self,
+        user_id: str,
+        user_key: str,
+        uri: str,
+        recursive: bool = True,
+    ) -> ResourceMutationResponse:
+        credential = self.openviking.credential_for_user(user_id, user_key)
+        backends = {
+            "openviking": await self._capture(lambda: self.openviking.delete_resource(credential, uri, recursive)),
+        }
+        resource = backends["openviking"].result if backends["openviking"].status == "success" else None
+        return ResourceMutationResponse(status=self._aggregate_status(backends), resource=resource, backends=backends)
+
     async def ingest_messages(self, request: MessageIngestRequest) -> MessageIngestResponse:
         messages = self._messages_from_request(request)
         if not messages:
@@ -204,6 +250,16 @@ class MemorySystemService:
             messages.append({"role": "assistant", "content": request.assistant_message})
         return messages
 
+    def _is_remote_url(self, path: str) -> bool:
+        return urlparse(path).scheme in {"http", "https"}
+
+    def _temp_file_id_from_result(self, result: Any) -> str:
+        data = result.get("result") if isinstance(result, dict) and isinstance(result.get("result"), dict) else result
+        temp_file_id = data.get("temp_file_id") if isinstance(data, dict) else None
+        if not temp_file_id:
+            raise ValueError("OpenViking temp upload response missing temp_file_id")
+        return str(temp_file_id)
+
     async def _run_backends(self, **calls: Callable[[], Awaitable[Any]]) -> dict[str, BackendStatus]:
         names = list(calls)
         results = await asyncio.gather(*(self._capture(calls[name]) for name in names))