feat: Support LoadArtifacts parsing for binary text documents (e.g., DOCX)

Co-authored-by: Xuan Yang <xygoogle@google.com>
PiperOrigin-RevId: 948373781

Author: xuanyang15

src/google/adk/models/google_llm.py

@@ -501,6 +501,7 @@ async def wait_5_seconds(tool_context: Any = None) -> Any:
     )
 
   async def _preprocess_request(self, llm_request: LlmRequest) -> None:
+    from ..tools import load_artifacts_tool  # pylint: disable=import-outside-toplevel
 
     if self._api_backend == GoogleLLMVariant.GEMINI_API:
       # Using API key from Google AI Studio to call model doesn't support labels.
@@ -528,6 +529,24 @@ async def _preprocess_request(self, llm_request: LlmRequest) -> None:
           llm_request.config.system_instruction = None
           await self._adapt_computer_use_tool(llm_request)
 
+    # Sanitize inputs by ensuring unsupported inline types (e.g. DOCX from UI)
+    # are converted to plain text using load_artifacts_tool._as_safe_part_for_llm.
+    if llm_request.contents:
+      for content in llm_request.contents:
+        if not content.parts:
+          continue
+        new_parts = []
+        for part in content.parts:
+          if part.inline_data:
+            # GE inline_data does not preserve filenames, so we pass a dummy
+            # 'inline-file' name as a placeholder for
+            # _as_safe_part_for_llm's required artifact_name argument.
+            part = load_artifacts_tool._as_safe_part_for_llm(  # pylint: disable=protected-access
+                part, 'inline-file'
+            )
+          new_parts.append(part)
+        content.parts = new_parts
+
   def _merge_tracking_headers(self, headers: dict[str, str]) -> dict[str, str]:
     """Merge tracking headers to the given headers."""
     return merge_tracking_headers(headers)

src/google/adk/tools/load_artifacts_tool.py

@@ -16,10 +16,14 @@
 
 import base64
 import binascii
+import io
 import json
 import logging
+import re
+import struct
 from typing import Any
 from typing import TYPE_CHECKING
+import zipfile
 
 from google.genai import types
 from typing_extensions import override
@@ -96,6 +100,42 @@ def _maybe_base64_to_bytes(data: str) -> bytes | None:
       return None
 
 
+def _try_extract_docx_text(data: bytes) -> str | None:
+  """Extracts raw text from a DOCX binary."""
+  # We use regex instead of standard XML parser to avoid XML bomb vulnerabilities,
+  # and cap the zip extraction at 10 MB to prevent zip bombs.
+  try:
+    with zipfile.ZipFile(io.BytesIO(data)) as docx_zip:
+      if 'word/document.xml' not in docx_zip.namelist():
+        return None
+      with docx_zip.open('word/document.xml') as xml_file:
+        xml_content = xml_file.read(10 * 1024 * 1024).decode(
+            'utf-8', errors='ignore'
+        )
+
+      # Find the prefix for the WordprocessingML namespace
+      # xmlns:w="..." or xmlns:something="..."
+      ns_match = re.search(
+          r'xmlns:(\w+)="http://schemas.openxmlformats.org/wordprocessingml/2006/main"',
+          xml_content,
+      )
+      prefix = ns_match.group(1) if ns_match else 'w'
+
+      p_tag = f'{prefix}:p'
+      t_tag = f'{prefix}:t'
+
+      paragraphs = []
+      for p in re.split(rf'<{p_tag}(?:[^>]*)>', xml_content):
+        texts = re.findall(rf'<{t_tag}(?:[^>]*)>([^<]*)</{t_tag}>', p)
+        if texts:
+          paragraphs.append(''.join(texts))
+
+      return '\n'.join(paragraphs)
+  except (zipfile.BadZipFile, KeyError, struct.error) as e:
+    logger.debug('Failed to parse docx layout: %s', e)
+    return None
+
+
 def _as_safe_part_for_llm(
     artifact: types.Part, artifact_name: str
 ) -> types.Part:
@@ -125,7 +165,23 @@ def _as_safe_part_for_llm(
       return types.Part.from_text(text=data)
     data = decoded
 
-  if mime_type.startswith('text/') or mime_type in _TEXT_LIKE_MIME_TYPES:
+  # Attempt DOCX extraction if file seems to be a docx document.
+  is_docx = mime_type in (
+      'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
+      'application/octet-stream',
+  ) or artifact_name.lower().endswith('.docx')
+  if is_docx:
+    extracted_text = _try_extract_docx_text(data)
+    if extracted_text is not None:
+      return types.Part.from_text(text=extracted_text)
+
+  # Fallback to general text extraction
+  is_text_like = (
+      mime_type.startswith('text/')
+      or mime_type in _TEXT_LIKE_MIME_TYPES
+      or artifact_name.lower().endswith(('.csv', '.txt', '.json', '.xml'))
+  )
+  if is_text_like:
     try:
       return types.Part.from_text(text=data.decode('utf-8'))
     except UnicodeDecodeError:

tests/unittests/models/test_google_llm.py

@@ -29,6 +29,7 @@
 from google.adk.models.google_llm import Gemini
 from google.adk.models.llm_request import LlmRequest
 from google.adk.models.llm_response import LlmResponse
+from google.adk.tools import load_artifacts_tool
 from google.adk.utils._client_labels_utils import _AGENT_ENGINE_TELEMETRY_ENV_VARIABLE_NAME
 from google.adk.utils._client_labels_utils import _AGENT_ENGINE_TELEMETRY_TAG
 from google.adk.utils._google_client_headers import get_tracking_headers
@@ -1037,6 +1038,42 @@ async def test_preprocess_request_handles_backend_specific_fields(
     assert llm_request_with_files.config.labels == expected_labels
 
 
+@pytest.mark.asyncio
+async def test_preprocess_request_converts_inline_data_safely():
+  """Tests that _preprocess_request uses _as_safe_part_for_llm to sanitize inline data."""
+  with mock.patch.object(
+      load_artifacts_tool, "_as_safe_part_for_llm", autospec=True
+  ) as mock_safe_part:
+    # Arrange
+    mock_safe_part.return_value = Part.from_text(text="safe_text")
+    my_gemini_llm = Gemini(model="gemini-2.5-flash")
+
+    my_llm_request = LlmRequest(
+        model="gemini-2.5-flash",
+        contents=[
+            Content(
+                role="user",
+                parts=[
+                    Part(
+                        inline_data=types.Blob(
+                            data=b"some bytes",
+                            mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+                        )
+                    )
+                ],
+            )
+        ],
+    )
+
+    # Act
+    await my_gemini_llm._preprocess_request(my_llm_request)  # pylint: disable=protected-access
+
+    # Assert
+    mock_safe_part.assert_called_once()
+    assert mock_safe_part.call_args[0][1] == "inline-file"
+    assert my_llm_request.contents[0].parts[0].text == "safe_text"
+
+
 @pytest.mark.asyncio
 async def test_generate_content_async_stream_aggregated_content_regardless_of_finish_reason():
   """Test that aggregated content is generated regardless of finish_reason."""