✨ Multi-modal agent config

Author: Zhi-a

sdk/nexent/multi_modal/load_save_object.py

@@ -6,6 +6,7 @@
 import requests
 
 from .utils import (
+    UrlType,
     is_url,
     generate_object_name,
     detect_content_type_from_bytes,
@@ -36,20 +37,28 @@ def _get_client(self) -> Any:
             raise ValueError("Storage client is not initialized.")
         return self._storage_client
 
-    def download_file_from_url(self, url: str, timeout: int = 30) -> Optional[bytes]:
+    def download_file_from_url(
+            self,
+            url: str,
+            url_type: UrlType,
+            timeout: int = 30
+    ) -> Optional[bytes]:
         """
         Download file content from S3 URL or HTTP/HTTPS URL as bytes.
         """
         if not url:
             return None
 
+        if not url_type:
+            raise ValueError("url_type must be provided for download_file_from_url")
+
         try:
-            if url.startswith(('http://', 'https://')):
+            if url_type in ("http", "https"):
                 response = requests.get(url, timeout=timeout)
                 response.raise_for_status()
                 return response.content
 
-            if url.startswith('s3://') or url.startswith('/'):
+            if url_type == "s3":
                 client = self._get_client()
                 bucket, object_name = parse_s3_url(url)
 
@@ -68,7 +77,7 @@ def download_file_from_url(self, url: str, timeout: int = 30) -> Optional[bytes]
                 except Exception as exc:
                     raise ValueError(f"Failed to read stream content: {exc}") from exc
 
-            raise ValueError(f"Unsupported URL format: {url[:50]}...")
+            raise ValueError(f"Unsupported URL type: {url_type}")
 
         except Exception as exc:
             logger.error(f"Failed to download file from URL: {exc}")
@@ -115,22 +124,24 @@ def decorator(func: Callable):
             def wrapper(*args, **kwargs):
                 def _transform_single_value(param_name: str, value: Any,
                                             transformer: Optional[Callable[[bytes], Any]]) -> Any:
-                    if isinstance(value, str) and is_url(value):
-                        bytes_data = self.download_file_from_url(value)
-
-                        if bytes_data is None:
-                            raise ValueError(f"Failed to download file from URL: {value}")
-
-                        if transformer:
-                            transformed_data = transformer(bytes_data)
-                            logger.info(
-                                f"Downloaded {param_name} from URL and transformed "
-                                f"using {transformer.__name__}"
-                            )
-                            return transformed_data
-
-                        logger.info(f"Downloaded {param_name} from URL as bytes (binary stream)")
-                        return bytes_data
+                    if isinstance(value, str):
+                        url_type = is_url(value)
+                        if url_type:
+                            bytes_data = self.download_file_from_url(value, url_type=url_type)
+
+                            if bytes_data is None:
+                                raise ValueError(f"Failed to download file from URL: {value}")
+
+                            if transformer:
+                                transformed_data = transformer(bytes_data)
+                                logger.info(
+                                    f"Downloaded {param_name} from URL and transformed "
+                                    f"using {transformer.__name__}"
+                                )
+                                return transformed_data
+
+                            logger.info(f"Downloaded {param_name} from URL as bytes (binary stream)")
+                            return bytes_data
 
                     raise ValueError(
                         f"Parameter '{param_name}' is not a URL string. "

sdk/nexent/multi_modal/utils.py

@@ -2,40 +2,53 @@
 import logging
 from datetime import datetime
 import uuid
-from typing import Tuple
+from typing import Literal, Optional, Tuple
 import mimetypes
 from pathlib import PurePosixPath
 
 
 logger = logging.getLogger("multi_modal")
 
+UrlType = Literal["http", "https", "s3"]
 
-def is_url(url: str) -> bool:
+
+def is_url(url: str) -> Optional[UrlType]:
     """
-    Check if a string is a URL (S3 or HTTP/HTTPS)
+    Classify a string URL as HTTP(S) or S3.
 
     Args:
-        url: String to check
+        url: URL candidate
 
     Returns:
-        True if it is a URL, False otherwise
+        'http', 'https', or 's3' when the input matches the respective
+        scheme. Returns None when the input is not a supported URL.
     """
     if not url or not isinstance(url, str):
-        return False
+        return None
 
-    # Check for HTTP/HTTPS URLs
-    if url.startswith(("http://", "https://")):
-        return True
+    url = url.strip()
 
-    # Check for S3 URLs
-    if url.startswith("s3://"):
-        return True
+    if url.startswith("http://"):
+        return "http"
 
-    # Check for MinIO path format: /bucket/key
-    if url.startswith("/") and len(url.split("/")) >= 3:
-        return True
+    if url.startswith("https://"):
+        return "https"
 
-    return False
+    if url.startswith("s3://"):
+        bucket_path = url.replace("s3://", "", 1)
+        bucket_object = bucket_path.split("/", 1)
+        if len(bucket_object) == 2 and all(bucket_object):
+            return "s3"
+        return None
+
+    if url.startswith("/"):
+        stripped = url.lstrip("/")
+        parts = stripped.split("/", 1)
+        if len(parts) == 2 and all(parts):
+            return "s3"
+        return None
+
+    return None
 
 
 def bytes_to_base64(bytes_data: bytes, content_type: str = "application/octet-stream") -> str:

test/sdk/multi_modal/test_load_save_object.py

@@ -37,7 +37,10 @@ def raise_for_status(self):
             return None
 
     monkeypatch.setattr(lso.requests, "get", lambda url, timeout: _Response())
-    data = manager.download_file_from_url("https://example.com/file.png")
+    data = manager.download_file_from_url(
+        "https://example.com/file.png",
+        url_type="https",
+    )
     assert data == b"binary"
 
 
@@ -49,7 +52,7 @@ def get_file_stream(self, object_name: str, bucket: str) -> Tuple[bool, Any]:
             return True, io.BytesIO(b"payload")
 
     manager = make_manager(_FakeClient())
-    data = manager.download_file_from_url("s3://bucket/path/to/object")
+    data = manager.download_file_from_url("s3://bucket/path/to/object", url_type="s3")
     assert data == b"payload"
 
 
@@ -59,25 +62,26 @@ def get_file_stream(self, object_name: str, bucket: str):
             return False, "boom"
 
     manager = make_manager(_FailingClient())
-    assert manager.download_file_from_url("s3://bucket/object") is None
+    assert manager.download_file_from_url("s3://bucket/object", url_type="s3") is None
 
 
 def test_download_file_from_s3_missing_method_returns_none():
     class _InvalidClient:
         pass
 
     manager = make_manager(_InvalidClient())
-    assert manager.download_file_from_url("s3://bucket/object") is None
+    assert manager.download_file_from_url("s3://bucket/object", url_type="s3") is None
 
 
-def test_download_file_with_unsupported_scheme_returns_none():
+def test_download_file_requires_url_type():
     manager = make_manager()
-    assert manager.download_file_from_url("ftp://unsupported/path") is None
+    with pytest.raises(ValueError):
+        manager.download_file_from_url("https://example.com/file.png", url_type=None)  # type: ignore[arg-type]
 
 
 def test_download_file_empty_url_returns_none():
     manager = make_manager()
-    assert manager.download_file_from_url("") is None
+    assert manager.download_file_from_url("", url_type="https") is None
 
 
 def test_download_file_stream_read_failure(monkeypatch):
@@ -93,7 +97,7 @@ def get_file_stream(self, object_name: str, bucket: str):
             return True, _FailingStream()
 
     manager = make_manager(_Client())
-    assert manager.download_file_from_url("s3://bucket/object") is None
+    assert manager.download_file_from_url("s3://bucket/object", url_type="s3") is None
 
 
 def test_upload_bytes_to_minio_generates_object_name(monkeypatch):
@@ -195,7 +199,7 @@ def handler(image):
 
     result = handler("https://example.com/img.png")
 
-    download_mock.assert_called_once_with("https://example.com/img.png")
+    download_mock.assert_called_once_with("https://example.com/img.png", url_type="https")
     assert result == b"file-bytes"
 
 
@@ -242,7 +246,7 @@ def handler(image, other=None):
         return image, other
 
     result = handler("https://example.com/a.png")
-    download_mock.assert_called_once_with("https://example.com/a.png")
+    download_mock.assert_called_once_with("https://example.com/a.png", url_type="https")
     assert result == (b"bytes", None)
 
 

test/sdk/multi_modal/test_utils.py

@@ -6,16 +6,18 @@
 
 
 def test_is_url_variants():
-    assert utils.is_url("https://example.com/image.png")
-    assert utils.is_url("s3://bucket/key")
-    assert utils.is_url("/bucket/key")
-    assert not utils.is_url("not-a-url")
-    assert not utils.is_url(123)  # type: ignore[arg-type]
+    assert utils.is_url("https://example.com/image.png") == "https"
+    assert utils.is_url("http://example.com/image.png") == "http"
+    assert utils.is_url("s3://bucket/key") == "s3"
+    assert utils.is_url("/bucket/key") == "s3"
+    assert utils.is_url("not-a-url") is None
+    assert utils.is_url(123) is None  # type: ignore[arg-type]
 
 
 def test_is_url_requires_bucket_and_key():
-    assert not utils.is_url("/bucket")
-    assert not utils.is_url("")
+    assert utils.is_url("/bucket") is None
+    assert utils.is_url("s3://bucket/") is None
+    assert utils.is_url("") is None
 
 
 def test_bytes_to_base64_and_back():
@@ -167,3 +169,37 @@ def test_parse_s3_url_requires_object_name():
         utils.parse_s3_url("/bucket")
 
 
+def test_base64_to_bytes_header_without_base64_flag():
+    payload = base64.b64encode(b"json-bytes").decode("utf-8")
+    decoded, content_type = utils.base64_to_bytes(
+        f"data:application/json,{payload}"
+    )
+    assert decoded == b"json-bytes"
+    assert content_type == "application/json"
+
+
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        (b"\x00\x00\x00 qt  " + b"\x00" * 6, "video/quicktime"),
+        (b"OggS" + b"\x00" * 8, "audio/ogg"),
+        (b"fLaC" + b"\x00" * 8, "audio/flac"),
+        (b"\x1a\x45\xdf\xa3" + b"\x00" * 8, "video/webm"),
+        (b"RIFF" + b"\x00" * 4 + b"AVI ", "video/x-msvideo"),
+    ],
+)
+def test_detect_content_type_expanded_signatures(payload: bytes, expected: str):
+    assert utils.detect_content_type_from_bytes(payload) == expected
+
+
+def test_detect_content_type_mp3_frame_sync():
+    payload = b"\xff\xfb" + b"\x00" * 4
+    assert utils.detect_content_type_from_bytes(payload) == "audio/mpeg"
+
+
+@pytest.mark.parametrize("value", ["", None])
+def test_parse_s3_url_rejects_empty(value):
+    with pytest.raises(ValueError):
+        utils.parse_s3_url(value)  # type: ignore[arg-type]
+
+