feat: add text processing with media

README.md

@@ -59,6 +59,9 @@ with cortext.Cortext(":memory:") as memory:
     print(len(emb))
 ```
 
+For binary attachments, pass `Media(data, mimetype)` or raw bytes with
+`media_mimetype` to `process_text_with_media`.
+
 Python 3.10+. Binding **1.2.4** tracks engine **v1.2.4**.
 
 ## Why not pypi.org?

cortext/__init__.py

@@ -411,6 +411,30 @@ def _configure_library(lib: ctypes.CDLL) -> ctypes.CDLL:
     ]
     lib.cortext_process_audio_json_with_options.restype = ctypes.c_void_p
 
+    text_media_json = getattr(lib, "cortext_process_text_with_media_json", None)
+    text_media_json_with_options = getattr(
+        lib, "cortext_process_text_with_media_json_with_options", None
+    )
+    lib._cortext_text_media_json_available = text_media_json_with_options is not None
+    if text_media_json is not None:
+        text_media_json.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_char_p,
+            ctypes.c_char_p,
+            ctypes.POINTER(_NativeMedia),
+        ]
+        text_media_json.restype = ctypes.c_void_p
+
+    if text_media_json_with_options is not None:
+        text_media_json_with_options.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_char_p,
+            ctypes.c_char_p,
+            ctypes.POINTER(_NativeMedia),
+            ctypes.POINTER(_NativeProcessJsonOptions),
+        ]
+        text_media_json_with_options.restype = ctypes.c_void_p
+
     lib.cortext_process_audio_with_media_json.argtypes = [
         ctypes.c_void_p,
         ctypes.POINTER(ctypes.c_float),
@@ -1134,6 +1158,52 @@ def process_text(
             )
         )
 
+    def process_text_with_media_json(
+        self,
+        text: str,
+        source_id: str,
+        media: Media | bytes | bytearray | memoryview | None = None,
+        media_mimetype: str | None = None,
+        include_embedding: bool = True,
+        retention: Retention | int | None = None,
+    ) -> str:
+        if not getattr(self._lib, "_cortext_text_media_json_available", False):
+            raise RuntimeError(
+                "loaded native library does not support text media processing"
+            )
+        native_media, media_buffer, mimetype_buffer = _native_media(
+            media, media_mimetype
+        )
+        options = _native_process_json_options(include_embedding, retention)
+        _ = (media_buffer, mimetype_buffer)
+        return self._call_json(
+            self._lib.cortext_process_text_with_media_json_with_options,
+            text.encode("utf-8"),
+            source_id.encode("utf-8"),
+            ctypes.byref(native_media),
+            ctypes.byref(options),
+        )
+
+    def process_text_with_media(
+        self,
+        text: str,
+        source_id: str,
+        media: Media | bytes | bytearray | memoryview | None = None,
+        media_mimetype: str | None = None,
+        include_embedding: bool = True,
+        retention: Retention | int | None = None,
+    ) -> dict[str, Any]:
+        return json.loads(
+            self.process_text_with_media_json(
+                text,
+                source_id,
+                media,
+                media_mimetype,
+                include_embedding,
+                retention,
+            )
+        )
+
     def embed_text_json(self, text: str) -> str:
         return self._call_json(
             self._lib.cortext_embed_text_json,

tests/test_cortext.py

@@ -128,6 +128,69 @@ def test_process_text_and_embed_text(tmp_path):
         clear_materialize_cache()
 
 
+def test_text_media_capability_only_requires_options_symbol():
+    import cortext
+
+    class FakeFunction:
+        pass
+
+    class FakeLibrary:
+        def __init__(self):
+            self.functions = {}
+
+        def __getattr__(self, name):
+            if name == "cortext_process_text_with_media_json":
+                raise AttributeError(name)
+            function = self.functions.setdefault(name, FakeFunction())
+            return function
+
+    lib = cortext._configure_library(FakeLibrary())
+
+    assert lib._cortext_text_media_json_available is True
+    assert "cortext_process_text_with_media_json_with_options" in lib.functions
+
+
+def test_process_text_with_media_forwards_binary_bytes_and_mimetype():
+    import ctypes
+
+    import cortext
+
+    native_fn = object()
+    fake_lib = type("FakeLib", (), {
+        "cortext_process_text_with_media_json_with_options": native_fn,
+        "_cortext_text_media_json_available": True,
+    })()
+    engine = object.__new__(cortext.Cortext)
+    engine._lib = fake_lib
+    engine._handle = None
+    captured = {}
+
+    def call_json(fn, *args):
+        captured["fn"] = fn
+        captured["text"] = args[0]
+        captured["source_id"] = args[1]
+        media = ctypes.cast(
+            args[2], ctypes.POINTER(cortext._NativeMedia)
+        ).contents
+        captured["media"] = ctypes.string_at(media.data, media.size)
+        captured["mimetype"] = ctypes.string_at(media.mimetype)
+        return '{"ok": true}'
+
+    engine._call_json = call_json
+    payload = b"\x00\xff\x01"
+
+    assert engine.process_text_with_media(
+        "text", "source", payload, "application/octet-stream"
+    ) == {"ok": True}
+    assert captured == {
+        "fn": native_fn,
+        "text": b"text",
+        "source_id": b"source",
+        "media": payload,
+        "mimetype": b"application/octet-stream",
+    }
+
+
 def test_embedding_length_stable(tmp_path):
     import cortext
     from cortext.model import clear_materialize_cache