Merge pull request #41506 from BerriAI/litellm_vertex_gcs_file_content_streaming

feat(vertex_ai): stream GCS batch output files from /v1/files/{id}/content

litellm/files/main.py

@@ -58,6 +58,7 @@ from litellm.types.llms.openai import (
 )
 from litellm.types.router import *
 from litellm.types.utils import (
+    FILE_CONTENT_STREAMING_PROVIDERS,
     OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS,
     LlmProviders,
 )
@@ -79,7 +80,22 @@ def _should_sdk_support_streaming(
     """
     Return whether file content streaming is supported for the provider.
     """
-    return custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS
+    return custom_llm_provider in FILE_CONTENT_STREAMING_PROVIDERS
+
+
+def _file_content_logging_obj(kwargs: dict[str, object], _is_async: bool) -> LiteLLMLoggingObj:
+    logging_obj: Final = kwargs.get("litellm_logging_obj")
+    if isinstance(logging_obj, LiteLLMLoggingObj):
+        return logging_obj
+    return LiteLLMLoggingObj(
+        model="",
+        messages=[],
+        stream=False,
+        call_type="afile_content" if _is_async else "file_content",
+        start_time=time.time(),
+        litellm_call_id=str(kwargs.get("litellm_call_id") or uuid_module.uuid4()),
+        function_id=str(kwargs.get("id") or ""),
+    )
 
 
 openai_files_instance: Final = OpenAIFilesAPI()
@@ -868,18 +884,21 @@ def file_content(
         )
 
         _is_async: Final = kwargs.pop("afile_content", False) is True
+        litellm_params_dict["api_key"] = optional_params.api_key
+        litellm_params_dict["api_base"] = optional_params.api_base
 
         if stream and _should_sdk_support_streaming(custom_llm_provider):
             return file_content_streaming(
                 file_id=file_id,
                 model=model,
                 custom_llm_provider=custom_llm_provider,
+                file_content_request=_file_content_request,
                 extra_headers=extra_headers,
-                extra_body=extra_body,
                 chunk_size=chunk_size,
                 optional_params=optional_params,
+                litellm_params=litellm_params_dict,
                 timeout=timeout,
-                logging_obj=cast(LiteLLMLoggingObj | None, kwargs.get("litellm_logging_obj")),
+                logging_obj=_file_content_logging_obj(kwargs, _is_async),
                 _is_async=_is_async,
                 client=client,
             )
@@ -890,27 +909,12 @@ def file_content(
             provider=LlmProviders(custom_llm_provider),
         )
         if provider_config is not None:
-            litellm_params_dict["api_key"] = optional_params.api_key
-            litellm_params_dict["api_base"] = optional_params.api_base
-
-            logging_obj = kwargs.get("litellm_logging_obj")
-            if logging_obj is None:
-                logging_obj = LiteLLMLoggingObj(
-                    model="",
-                    messages=[],
-                    stream=False,
-                    call_type="afile_content" if _is_async else "file_content",
-                    start_time=time.time(),
-                    litellm_call_id=kwargs.get("litellm_call_id", str(uuid_module.uuid4())),
-                    function_id=str(kwargs.get("id") or ""),
-                )
-
             response = base_llm_http_handler.retrieve_file_content(
                 file_content_request=_file_content_request,
                 provider_config=provider_config,
                 litellm_params=litellm_params_dict,
                 headers=extra_headers or {},
-                logging_obj=logging_obj,
+                logging_obj=_file_content_logging_obj(kwargs, _is_async),
                 _is_async=_is_async,
                 client=(client if client is not None and isinstance(client, (HTTPHandler, AsyncHTTPHandler)) else None),
                 timeout=timeout,
@@ -1000,24 +1004,24 @@ def file_content_streaming(
     file_id: str,
     model: str | None,
     custom_llm_provider: FileContentProvider | str | None,
+    file_content_request: FileContentRequest,
     extra_headers: dict[str, str] | None,
-    extra_body: dict[str, str] | None,
     chunk_size: int,
     optional_params: GenericLiteLLMParams,
+    litellm_params: dict,
     timeout: float | httpx.Timeout,
-    logging_obj: LiteLLMLoggingObj | None,
+    logging_obj: LiteLLMLoggingObj,
     _is_async: bool,
-    client: OpenAI | AsyncOpenAI | None,
+    client: OpenAI | AsyncOpenAI | HTTPHandler | AsyncHTTPHandler | None,
 ) -> FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult]:
-    if logging_obj is not None:
-        logging_obj.model = model or ""
-        logging_obj.model_call_details["model"] = model or ""
-        logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
+    logging_obj.model = model or ""
+    logging_obj.model_call_details["model"] = model or ""
+    logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
 
-        litellm_params: Final = logging_obj.model_call_details.get("litellm_params", {}) or {}
-        if optional_params.api_base is not None:
-            litellm_params["api_base"] = optional_params.api_base
-        logging_obj.model_call_details["litellm_params"] = litellm_params
+    logged_litellm_params: Final = logging_obj.model_call_details.get("litellm_params", {}) or {}
+    if optional_params.api_base is not None:
+        logged_litellm_params["api_base"] = optional_params.api_base
+    logging_obj.model_call_details["litellm_params"] = logged_litellm_params
 
     def _wrap_streaming_result(
         response: FileContentStreamingResult,
@@ -1044,22 +1048,45 @@ def file_content_streaming(
         )
         response = openai_files_instance.file_content_streaming(
             _is_async=_is_async,
-            file_content_request=FileContentRequest(
-                file_id=file_id,
-                extra_headers=extra_headers,
-                extra_body=extra_body,
-            ),
+            file_content_request=file_content_request,
             api_base=openai_creds.api_base,
             api_key=openai_creds.api_key,
             timeout=timeout,
             max_retries=optional_params.max_retries,
             organization=openai_creds.organization,
             chunk_size=chunk_size,
-            client=client,
+            client=client if isinstance(client, (OpenAI, AsyncOpenAI)) else None,
+        )
+    elif custom_llm_provider == LlmProviders.VERTEX_AI.value:
+        if not _is_async:
+            raise litellm.exceptions.BadRequestError(
+                message="Streaming 'file_content' for vertex_ai is only supported through 'afile_content'.",
+                model="n/a",
+                llm_provider=custom_llm_provider,
+                response=httpx.Response(
+                    status_code=400,
+                    content="Unsupported provider",
+                    request=httpx.Request(method="file_content", url="https://github.com/BerriAI/litellm"),
+                ),
+            )
+        vertex_files_config: Final = ProviderConfigManager.get_provider_files_config(
+            model="",
+            provider=LlmProviders.VERTEX_AI,
+        )
+        assert vertex_files_config is not None
+        response = base_llm_http_handler.async_retrieve_file_content_streaming(
+            file_content_request=file_content_request,
+            provider_config=vertex_files_config,
+            litellm_params=litellm_params,
+            headers=extra_headers or {},
+            logging_obj=logging_obj,
+            chunk_size=chunk_size,
+            client=client if isinstance(client, AsyncHTTPHandler) else None,
+            timeout=timeout,
         )
     else:
         raise litellm.exceptions.BadRequestError(
-            message=f"LiteLLM doesn't support {custom_llm_provider} for streaming 'file_content'. Supported providers are {sorted(OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS)}.",
+            message=f"LiteLLM doesn't support {custom_llm_provider} for streaming 'file_content'. Supported providers are {sorted(FILE_CONTENT_STREAMING_PROVIDERS)}.",
             model="n/a",
             llm_provider=custom_llm_provider,
             response=httpx.Response(

litellm/files/types.py

@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Iterator, Mapping
 from typing import Literal, NamedTuple
 
 FileContentProvider = Literal[
@@ -8,4 +8,4 @@ FileContentProvider = Literal[
 
 class FileContentStreamingResult(NamedTuple):
     stream_iterator: Iterator[bytes] | AsyncIterator[bytes]
-    headers: dict[str, str]
+    headers: Mapping[str, str]

litellm/llms/base_llm/files/transformation.py

@@ -1,10 +1,11 @@
 from abc import ABC, abstractmethod
-from collections.abc import Iterator, Mapping
+from collections.abc import AsyncGenerator, Iterator, Mapping
 from typing import TYPE_CHECKING, Any, Union
 
 import httpx
 from openai.types.file_deleted import FileDeleted
 
+from litellm.files.types import FileContentStreamingResult
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.types.files import TwoStepFileUploadConfig
 from litellm.types.llms.openai import (
@@ -196,6 +197,18 @@ class BaseFilesConfig(BaseConfig):
     ) -> "HttpxBinaryResponseContent":
         """Transform file content response into OpenAI format."""
 
+    async def transform_file_content_stream(
+        self,
+        *,
+        stream_iterator: AsyncGenerator[bytes, None],
+        headers: Mapping[str, str],
+        request_url: str,
+        logging_obj: LiteLLMLoggingObj,
+        litellm_params: dict,
+    ) -> FileContentStreamingResult:
+        """Transform a streamed file content body. Passes the upstream bytes and headers through by default."""
+        return FileContentStreamingResult(stream_iterator=stream_iterator, headers=headers)
+
     def transform_request(
         self,
         model: str,

litellm/llms/custom_httpx/llm_http_handler.py

@@ -1,14 +1,27 @@
 import asyncio
 import json
 import ssl
-from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
+from collections.abc import AsyncGenerator, AsyncIterator, Coroutine, Iterator, Mapping, Sequence
 from contextlib import asynccontextmanager
 from functools import lru_cache
 from types import MappingProxyType, ModuleType
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict, TypeVar, Union, cast, get_type_hints
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Final,
+    Literal,
+    NamedTuple,
+    Optional,
+    TypedDict,
+    TypeVar,
+    Union,
+    cast,
+    get_type_hints,
+)
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
 import httpx
+from httpx import USE_CLIENT_DEFAULT
 from httpx._types import FileContent
 from openai.types.file_deleted import FileDeleted
 
@@ -19,6 +32,7 @@ import litellm.types.utils
 from litellm._logging import _redact_string, verbose_logger
 from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta
 from litellm.constants import MAX_FILE_LIST_LIMIT, REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
+from litellm.files.types import FileContentStreamingResult
 from litellm.litellm_core_utils.agentic_loop_settings import (
     DEFAULT_MAX_AGENTIC_LOOPS,
     validated_max_agentic_loops,
@@ -288,6 +302,39 @@ def _aws_signing_overrides(optional_params: Mapping[str, Any], litellm_params: M
     )
 
 
+class _PreparedFileContentRequest(NamedTuple):
+    url: str
+    params: dict
+    headers: dict
+
+
+async def _aiter_bytes_then_close(response: httpx.Response, *, chunk_size: int) -> AsyncGenerator[bytes, None]:
+    try:
+        async for chunk in response.aiter_bytes(chunk_size=chunk_size):
+            yield chunk
+    finally:
+        await response.aclose()
+
+
+_DECODED_BODY_STALE_HEADERS: Final[frozenset[str]] = frozenset({"content-encoding", "content-length"})
+
+
+def _decoded_body_headers(response: httpx.Response) -> httpx.Headers:
+    """
+    `aiter_bytes` yields the decoded body, so the upstream transfer headers only
+    describe the bytes on the wire when no content-encoding was applied.
+    """
+    if response.headers.get("content-encoding", "identity").lower() == "identity":
+        return response.headers
+    return httpx.Headers(
+        [
+            (name, value)
+            for name, value in response.headers.multi_items()
+            if name.lower() not in _DECODED_BODY_STALE_HEADERS
+        ]
+    )
+
+
 def _collect_ws_project_quota_callbacks() -> tuple[ProjectQuotaCallback, ...]:
     """Duck-type discover proxy hooks exposing per-frame project ITPM/OTPM
     enforcement, so the Responses WebSocket loop can charge every
@@ -5080,35 +5127,16 @@ class BaseLLMHTTPHandler:
         else:
             sync_httpx_client = client
 
-        # Get URL and params from provider config
-        url, params = provider_config.transform_file_content_request(
+        prepared: Final = self._prepare_file_content_request(
             file_content_request=file_content_request,
-            optional_params={},
+            provider_config=provider_config,
             litellm_params=litellm_params,
-        )
-
-        # Validate environment and get headers
-        headers = provider_config.validate_environment(
-            api_key=litellm_params.get("api_key"),
             headers=headers,
-            model="",
-            messages=[],
-            optional_params={},
-            litellm_params=litellm_params,
-        )
-
-        logging_obj.pre_call(
-            input="",
-            api_key="",
-            additional_args={
-                "api_base": url,
-                "headers": headers,
-                "file_id": file_content_request.get("file_id"),
-            },
+            logging_obj=logging_obj,
         )
 
         try:
-            response: Final = sync_httpx_client.get(url=url, headers=headers, params=params)
+            response: Final = sync_httpx_client.get(url=prepared.url, headers=prepared.headers, params=prepared.params)
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -5143,35 +5171,18 @@ class BaseLLMHTTPHandler:
         else:
             async_httpx_client = client
 
-        # Get URL and params from provider config
-        url, params = provider_config.transform_file_content_request(
+        prepared: Final = self._prepare_file_content_request(
             file_content_request=file_content_request,
-            optional_params={},
+            provider_config=provider_config,
             litellm_params=litellm_params,
-        )
-
-        # Validate environment and get headers
-        headers = provider_config.validate_environment(
-            api_key=litellm_params.get("api_key"),
             headers=headers,
-            model="",
-            messages=[],
-            optional_params={},
-            litellm_params=litellm_params,
-        )
-
-        logging_obj.pre_call(
-            input="",
-            api_key="",
-            additional_args={
-                "api_base": url,
-                "headers": headers,
-                "file_id": file_content_request.get("file_id"),
-            },
+            logging_obj=logging_obj,
         )
 
         try:
-            response: Final = await async_httpx_client.get(url=url, headers=headers, params=params)
+            response: Final = await async_httpx_client.get(
+                url=prepared.url, headers=prepared.headers, params=prepared.params
+            )
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -5188,6 +5199,93 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
         )
 
+    async def async_retrieve_file_content_streaming(
+        self,
+        file_content_request: "FileContentRequest",
+        provider_config: BaseFilesConfig,
+        litellm_params: dict,
+        headers: dict,
+        logging_obj: LiteLLMLoggingObj,
+        chunk_size: int,
+        client: AsyncHTTPHandler | None = None,
+        timeout: float | httpx.Timeout | None = None,
+    ) -> FileContentStreamingResult:
+        """
+        Async retrieve file content by ID as a byte stream, without buffering the body.
+        """
+        async_httpx_client: Final = (
+            client if client is not None else get_async_httpx_client(llm_provider=provider_config.custom_llm_provider)
+        )
+
+        prepared: Final = self._prepare_file_content_request(
+            file_content_request=file_content_request,
+            provider_config=provider_config,
+            litellm_params=litellm_params,
+            headers=headers,
+            logging_obj=logging_obj,
+        )
+
+        request: Final = async_httpx_client.client.build_request(
+            "GET",
+            prepared.url,
+            headers=prepared.headers,
+            params=httpx.QueryParams(HTTPHandler.extract_query_params(prepared.url)).merge(prepared.params),
+            timeout=USE_CLIENT_DEFAULT if timeout is None else httpx.Timeout(timeout),
+        )
+        try:
+            response: Final = await async_httpx_client.client.send(request, stream=True)
+        except Exception as e:  # noqa: BLE001  # _handle_error maps every failure kind, like the buffered fetch
+            raise self._handle_error(e=e, provider_config=provider_config)
+
+        if response.status_code >= 400:
+            error_body: Final = await response.aread()
+            await response.aclose()
+            raise provider_config.get_error_class(
+                error_message=error_body.decode("utf-8", errors="replace"),
+                status_code=response.status_code,
+                headers=response.headers,
+            )
+
+        return await provider_config.transform_file_content_stream(
+            stream_iterator=_aiter_bytes_then_close(response, chunk_size=chunk_size),
+            headers=_decoded_body_headers(response),
+            request_url=str(response.request.url),
+            logging_obj=logging_obj,
+            litellm_params=litellm_params,
+        )
+
+    @staticmethod
+    def _prepare_file_content_request(
+        file_content_request: "FileContentRequest",
+        provider_config: BaseFilesConfig,
+        litellm_params: dict,
+        headers: dict,
+        logging_obj: LiteLLMLoggingObj,
+    ) -> "_PreparedFileContentRequest":
+        url, params = provider_config.transform_file_content_request(
+            file_content_request=file_content_request,
+            optional_params={},
+            litellm_params=litellm_params,
+        )
+        request_headers: Final = provider_config.validate_environment(
+            api_key=litellm_params.get("api_key"),
+            headers=headers,
+            model="",
+            messages=[],
+            optional_params={},
+            litellm_params=litellm_params,
+        )
+        logging_obj.pre_call(
+            input="",
+            api_key="",
+            additional_args={
+                "api_base": url,
+                "headers": request_headers,
+                "file_id": file_content_request.get("file_id"),
+            },
+        )
+        return _PreparedFileContentRequest(url=url, params=params, headers=request_headers)
+
     def _prepare_fake_stream_request(
         self,
         stream: bool,

litellm/llms/vertex_ai/files/transformation.py

@@ -5,7 +5,10 @@ import json
 import os
 import re
 import time
-from collections.abc import Callable, Iterable, Iterator, Mapping
+from collections.abc import AsyncGenerator, Callable, Iterable, Iterator, Mapping
+from contextlib import aclosing
+from dataclasses import dataclass
+from types import MappingProxyType
 from typing import Any, Final, TypedDict
 from urllib.parse import quote, unquote
 
@@ -16,6 +19,7 @@ from typing_extensions import ReadOnly, Required
 
 import litellm
 from litellm._uuid import uuid
+from litellm.files.types import FileContentStreamingResult
 from litellm.files.utils import FilesAPIUtils
 from litellm.litellm_core_utils.cloud_storage_security import (
     VERTEX_AI_MANAGED_GCS_PREFIX,
@@ -81,6 +85,8 @@ _EMBED_REQUEST_FIELD_BY_GEMINI_PARAM: Final = (
     ("title", "title"),
 )
 _VERTEX_BATCH_FANNED_OUT_KEY_PATTERN: Final = re.compile(r"(?P<custom_id>[^#]*)#(?P<index>\d+)/(?P<total>\d+)")
+_JSONL_NEWLINE: Final = b"\n"
+_BATCH_OUTPUT_FIRST_ROW_PEEK_LIMIT_BYTES: Final = 32 * 1024 * 1024
 
 
 class _GcsObjectMetadataJson(TypedDict, total=False):
@@ -257,6 +263,118 @@ def _is_vertex_embeddings_batch_output_row(vertex_output_row: Mapping[str, objec
     return bool(vertex_output_row.get("status")) and isinstance(request_data, dict) and "content" in request_data
 
 
+def _is_vertex_generate_content_batch_output_row(vertex_output_row: Mapping[str, object]) -> bool:
+    """
+    Whether a Vertex batch output row came from a `GenerateContentRequest`. Anything
+    else (a plain JSON line, an OpenAI batch row) is not a Vertex batch output.
+    """
+    if not (
+        "request" in vertex_output_row and "response" in vertex_output_row and "processed_time" in vertex_output_row
+    ):
+        return False
+    response: Final = vertex_output_row.get("response")
+    return (isinstance(response, dict) and ("candidates" in response or "promptFeedback" in response)) or bool(
+        vertex_output_row.get("status")
+    )
+
+
+def _try_parse_vertex_batch_output_row(line: bytes) -> _VertexBatchRow | None:
+    try:
+        row: Final = _parse_vertex_batch_output_row(line.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError):
+        return None
+    return row if isinstance(row, dict) else None
+
+
+def _first_non_empty_jsonl_line(lines: Iterable[bytes]) -> bytes | None:
+    return next((stripped for line in lines if (stripped := line.strip())), None)
+
+
+async def _peek_first_jsonl_line(
+    chunks: AsyncGenerator[bytes, None],
+    *,
+    peek_limit_bytes: int,
+) -> tuple[bytes | None, bytes]:
+    """
+    Reads from `chunks` until the first non-empty line is complete, returning it with
+    everything read so far so the caller can replay the bytes. Stops peeking once the
+    buffered prefix exceeds `peek_limit_bytes` without a newline, so a large file that
+    is not JSONL is never buffered in full.
+    """
+    buffered: bytes = b""  # rebind-ok: accumulates the prefix read while looking for the first newline
+    async for chunk in chunks:
+        buffered = buffered + chunk
+        first_line = _first_non_empty_jsonl_line(buffered.split(_JSONL_NEWLINE)[:-1])
+        if first_line is not None:
+            return first_line, buffered
+        if len(buffered) > peek_limit_bytes:
+            return None, buffered
+    return _first_non_empty_jsonl_line(buffered.split(_JSONL_NEWLINE)), buffered
+
+
+async def _prepend_bytes(prefix: bytes, chunks: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]:
+    async with aclosing(chunks):
+        if prefix:
+            yield prefix
+        async for chunk in chunks:
+            yield chunk
+
+
+async def _aiter_jsonl_lines(chunks: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]:
+    """Yields stripped, non-empty JSONL lines from a byte stream, holding at most one partial line."""
+    pending: bytes = b""  # rebind-ok: carries the partial trailing line over to the next chunk
+    async with aclosing(chunks):
+        async for chunk in chunks:
+            *complete_lines, pending = (pending + chunk).split(_JSONL_NEWLINE)
+            for line in complete_lines:
+                if stripped := line.strip():
+                    yield stripped
+    if tail := pending.strip():
+        yield tail
+
+
+async def _aiter_single_chunk(content: bytes) -> AsyncGenerator[bytes, None]:
+    yield content
+
+
+async def _aread_all(chunks: AsyncGenerator[bytes, None]) -> bytes:
+    async with aclosing(chunks):
+        return b"".join(tuple([chunk async for chunk in chunks]))
+
+
+def _headers_without_content_length(headers: Mapping[str, str]) -> Mapping[str, str]:
+    return MappingProxyType({key: value for key, value in headers.items() if key.lower() != "content-length"})
+
+
+@dataclass(frozen=True, slots=True)
+class _VertexBatchOutputRowTransformContext:
+    vertex_gemini_config: VertexGeminiConfig
+    logging_obj: Logging
+    mock_httpx_response: httpx.Response
+
+
+def _new_vertex_batch_output_row_transform_context() -> _VertexBatchOutputRowTransformContext:
+    batch_transform_logging_obj: Final = Logging(
+        model="",
+        messages=[],
+        stream=False,
+        call_type="batch_transform",
+        start_time=time.time(),
+        litellm_call_id="",
+        function_id="",
+    )
+    batch_transform_logging_obj.optional_params = {}
+    return _VertexBatchOutputRowTransformContext(
+        vertex_gemini_config=VertexGeminiConfig(),
+        logging_obj=batch_transform_logging_obj,
+        mock_httpx_response=httpx.Response(
+            status_code=200,
+            headers={"content-type": "application/json"},
+            request=httpx.Request(method="POST", url="https://example.com"),
+        ),
+    )
+
+
 def _openai_batch_output_row(
     custom_id: str,
     body: Mapping[str, object] | None = None,
@@ -1074,6 +1192,84 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
 
         return HttpxBinaryResponseContent(response=raw_response)
 
+    async def transform_file_content_stream(
+        self,
+        *,
+        stream_iterator: AsyncGenerator[bytes, None],
+        headers: Mapping[str, str],
+        request_url: str,
+        logging_obj: LiteLLMLoggingObj,
+        litellm_params: dict,
+    ) -> FileContentStreamingResult:
+        """
+        Streams file content, converting a Vertex AI batch output to OpenAI format row by
+        row when the first row identifies one, so peak memory stays at about one row.
+
+        Embeddings batch outputs are grouped by entry and so are transformed in full.
+        Everything else is passed through unchanged, including a row that fails to
+        transform mid-stream.
+        """
+        if litellm.disable_vertex_batch_output_transformation:
+            return FileContentStreamingResult(stream_iterator=stream_iterator, headers=headers)
+
+        first_line, buffered = await _peek_first_jsonl_line(
+            stream_iterator,
+            peek_limit_bytes=_BATCH_OUTPUT_FIRST_ROW_PEEK_LIMIT_BYTES,
+        )
+        replayed_stream: Final = _prepend_bytes(buffered, stream_iterator)
+        first_row: Final = None if first_line is None else _try_parse_vertex_batch_output_row(first_line)
+        if first_row is None:
+            return FileContentStreamingResult(stream_iterator=replayed_stream, headers=headers)
+
+        if _is_vertex_embeddings_batch_output_row(first_row):
+            transformed_content: Final = self._try_transform_vertex_batch_output_to_openai(
+                content=await _aread_all(replayed_stream),
+                logging_obj=logging_obj,
+                model=_model_from_managed_gcs_url(request_url),
+            )
+            return FileContentStreamingResult(
+                stream_iterator=_aiter_single_chunk(transformed_content),
+                headers=MappingProxyType({**headers, "content-length": str(len(transformed_content))}),
+            )
+
+        if not _is_vertex_generate_content_batch_output_row(first_row):
+            return FileContentStreamingResult(stream_iterator=replayed_stream, headers=headers)
+
+        return FileContentStreamingResult(
+            stream_iterator=self._aiter_openai_batch_output_rows(_aiter_jsonl_lines(replayed_stream)),
+            headers=_headers_without_content_length(headers),
+        )
+
+    async def _aiter_openai_batch_output_rows(self, lines: AsyncGenerator[bytes, None]) -> AsyncGenerator[bytes, None]:
+        context: Final = _new_vertex_batch_output_row_transform_context()
+        async with aclosing(lines):
+            first_line: Final = await anext(lines, None)
+            if first_line is None:
+                return
+            yield self._transform_vertex_batch_output_line(first_line, context=context)
+            async for line in lines:
+                yield _JSONL_NEWLINE + self._transform_vertex_batch_output_line(line, context=context)
+
+    def _transform_vertex_batch_output_line(
+        self,
+        line: bytes,
+        *,
+        context: _VertexBatchOutputRowTransformContext,
+    ) -> bytes:
+        vertex_output: Final = _try_parse_vertex_batch_output_row(line)
+        if vertex_output is None:
+            return line
+        try:
+            openai_output: Final = self._transform_single_vertex_batch_output_to_openai(
+                vertex_output=vertex_output,
+                vertex_gemini_config=context.vertex_gemini_config,
+                logging_obj=context.logging_obj,
+                mock_httpx_response=context.mock_httpx_response,
+            )
+        except Exception:  # noqa: BLE001  # a row that fails to transform is passed through raw, like the buffered path
+            return line
+        return json.dumps(openai_output).encode("utf-8")
+
     def _try_transform_vertex_batch_output_to_openai(
         self,
         content: bytes,
@@ -1120,38 +1316,13 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
             # first line is not valid UTF-8/JSON) raises and falls through to the
             # passthrough below, leaving the content untouched.
             first_row: Final = _parse_vertex_batch_output_row(first_line)
-            is_vertex_batch_output: Final = _is_vertex_embeddings_batch_output_row(first_row) or (
-                "request" in first_row
-                and "response" in first_row
-                and "processed_time" in first_row
-                and (
-                    "candidates" in first_row.get("response", {})
-                    or "promptFeedback" in first_row.get("response", {})
-                    or bool(first_row.get("status"))
-                )
-            )
-            if not is_vertex_batch_output:
+            if not (
+                _is_vertex_embeddings_batch_output_row(first_row)
+                or _is_vertex_generate_content_batch_output_row(first_row)
+            ):
                 return content
 
-            vertex_gemini_config: Final = VertexGeminiConfig()
-            # Use a fresh Logging object for the per-row transform so we never
-            # mutate the caller's (which already ran pre_call with its own
-            # model/start_time/optional_params).
-            batch_transform_logging_obj: Final = Logging(
-                model="",
-                messages=[],
-                stream=False,
-                call_type="batch_transform",
-                start_time=time.time(),
-                litellm_call_id="",
-                function_id="",
-            )
-            batch_transform_logging_obj.optional_params = {}
-            mock_httpx_response: Final = httpx.Response(
-                status_code=200,
-                headers={"content-type": "application/json"},
-                request=httpx.Request(method="POST", url="https://example.com"),
-            )
+            context: Final = _new_vertex_batch_output_row_transform_context()
 
             all_lines = itertools.chain((first_line,), lines)
 
@@ -1173,9 +1344,9 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
                 try:
                     openai_output = self._transform_single_vertex_batch_output_to_openai(
                         vertex_output=_parse_vertex_batch_output_row(line),
-                        vertex_gemini_config=vertex_gemini_config,
-                        logging_obj=batch_transform_logging_obj,
-                        mock_httpx_response=mock_httpx_response,
+                        vertex_gemini_config=context.vertex_gemini_config,
+                        logging_obj=context.logging_obj,
+                        mock_httpx_response=context.mock_httpx_response,
                     )
                 except Exception:
                     return content

litellm/proxy/openai_files_endpoints/file_content_streaming_handler.py

@@ -5,7 +5,7 @@ from fastapi.responses import StreamingResponse
 
 import litellm
 from litellm.files.types import FileContentProvider, FileContentStreamingResult
-from litellm.types.utils import OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS
+from litellm.types.utils import FILE_CONTENT_STREAMING_PROVIDERS
 
 if TYPE_CHECKING:
     from litellm.proxy._types import UserAPIKeyAuth
@@ -43,6 +43,7 @@ class FileContentStreamingHandler:
                 data=resolved_streaming_data,
                 credentials=credentials,
                 file_id=original_file_id,
+                include_internal_credentials=True,
             )
             resolved_streaming_data.pop("model", None)
             resolved_streaming_provider: Final = cast(str, credentials["custom_llm_provider"])
@@ -64,7 +65,7 @@ class FileContentStreamingHandler:
         *,
         custom_llm_provider: str,
     ) -> bool:
-        return custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS
+        return custom_llm_provider in FILE_CONTENT_STREAMING_PROVIDERS
 
     @staticmethod
     async def stream_file_content_with_logging(

litellm/types/utils.py

@@ -4137,6 +4137,10 @@ OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: set[str] = {
     LlmProviders.LITELLM_PROXY.value,
 }
 
+FILE_CONTENT_STREAMING_PROVIDERS: Final[frozenset[str]] = frozenset(
+    {*OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS, LlmProviders.VERTEX_AI.value}
+)
+
 ListBatchesSupportedProvider = Literal["openai", "azure", "hosted_vllm", "litellm_proxy", "vertex_ai"]
 
 LIST_BATCHES_SUPPORTED_PROVIDERS: Final[frozenset[str]] = frozenset(get_args(ListBatchesSupportedProvider))

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_streaming.py

@@ -15,9 +15,15 @@ replaced by a list-based pipeline:
   4. A tuple-wrapped file handle uploaded through the real create_file ordering
      keeps every row, including entry 0 (no partial upload from a consumed
      cursor).
+  5. Downloading a GCS object through ``async_retrieve_file_content_streaming``
+     yields the body as it arrives instead of buffering it, keeps the upstream
+     ``content-type`` / ``content-length``, transforms a Vertex batch output
+     row by row, and closes the response when the consumer is done.
 """
 
+import asyncio
 import gc
+import gzip
 import io
 import json
 import tempfile
@@ -27,20 +33,22 @@ import tracemalloc
 import httpx
 import pytest
 
+import litellm
+from litellm.files.types import FileContentStreamingResult
 from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.llms.base_llm.files.transformation import BaseFileUploadStream
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
+from litellm.llms.vertex_ai.common_utils import VertexAIError
 from litellm.llms.vertex_ai.files.transformation import (
     VertexAIFilesConfig,
-    _OpenAIToVertexBatchUploadStream,
     _get_litellm_batch_custom_id_from_labels,
     _iter_openai_jsonl_entries,
     _iter_openai_jsonl_lines,
     _openai_batch_jsonl_entry_to_vertex_rows,
+    _OpenAIToVertexBatchUploadStream,
 )
-from litellm.types.llms.openai import CreateFileRequest
-from litellm.llms.vertex_ai.common_utils import VertexAIError
+from litellm.types.llms.openai import CreateFileRequest, FileContentRequest
 
 
 def _upload_stream(transformed) -> BaseFileUploadStream:
@@ -586,3 +594,321 @@ class TestStreamingMediaUpload:
         monkeypatch.setattr(tempfile, "TemporaryFile", lambda *a, **k: (created.append(1), real_tempfile(*a, **k))[1])
         await self._run(_make_openai_jsonl_bytes(50))
         assert created == []
+
+
+_MANAGED_OUTPUT_FILE_ID = (
+    "gs://test-bucket/litellm-vertex-files/publishers/google/models/gemini-2.5-flash/abc/predictions.jsonl"
+)
+
+
+def _vertex_batch_output_row(custom_id: str, text: str) -> bytes:
+    return json.dumps(
+        {
+            "status": "",
+            "processed_time": "2024-11-01T18:13:16.826+00:00",
+            "request": {"labels": {"litellm_custom_id": custom_id}, "contents": [{"parts": [{"text": "hi"}]}]},
+            "response": {
+                "candidates": [{"content": {"parts": [{"text": text}], "role": "model"}, "finishReason": "STOP"}],
+                "usageMetadata": {"promptTokenCount": 1, "candidatesTokenCount": 2, "totalTokenCount": 3},
+                "modelVersion": "gemini-2.5-flash@default",
+            },
+        }
+    ).encode("utf-8")
+
+
+def _vertex_embeddings_output_row(key: str, values: list[float]) -> bytes:
+    return json.dumps(
+        {
+            "key": key,
+            "request": {"content": {"parts": [{"text": "hello world"}]}},
+            "response": {"embedding": {"values": values}, "usageMetadata": {"promptTokenCount": 2}},
+        }
+    ).encode("utf-8")
+
+
+def _gcs_download_mock(raw_chunks: list[bytes], headers: dict[str, str]):
+    """A fake GCS `alt=media` endpoint that serves the object one raw chunk at a
+    time, recording the request and how many chunks the consumer has pulled so
+    far, so a test can tell streaming apart from buffering."""
+    state = {"urls": [], "headers": [], "served": 0, "closed": False}
+
+    async def body():
+        for chunk in raw_chunks:
+            state["served"] += 1
+            yield chunk
+            await asyncio.sleep(0)
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        state["urls"].append(str(request.url))
+        state["headers"].append(dict(request.headers))
+        response = httpx.Response(200, content=body(), headers=headers)
+        original_aclose = response.aclose
+
+        async def aclose():
+            state["closed"] = True
+            await original_aclose()
+
+        response.aclose = aclose
+        return response
+
+    return handler, state
+
+
+class _StaticTokenFilesConfig(VertexAIFilesConfig):
+    """Vertex files config with a fixed access token, so no ADC lookup runs in tests."""
+
+    def get_access_token(self, credentials, project_id, _retry_reauth=False):
+        return "test-token", "test-project"
+
+
+def _stable_row_fields(jsonl: bytes) -> list[tuple]:
+    """Project OpenAI batch output rows onto the fields the transform derives from
+    the Vertex row, leaving out the ids and timestamps it generates per call."""
+    rows = [json.loads(line) for line in jsonl.split(b"\n") if line]
+    return [
+        (
+            row["custom_id"],
+            row["error"],
+            row["response"]["status_code"],
+            row["response"]["body"]["model"],
+            row["response"]["body"]["choices"][0]["message"]["content"],
+            row["response"]["body"]["usage"]["total_tokens"],
+        )
+        for row in rows
+    ]
+
+
+class TestFileContentStreaming:
+    """End-to-end against a faked GCS media endpoint. These fail if the retrieval
+    buffers the object before yielding, drops or duplicates bytes across chunk
+    boundaries, loses the upstream headers, or leaks the httpx response."""
+
+    async def _open(self, raw_chunks: list[bytes], headers: dict[str, str], chunk_size: int = 16):
+        mock, state = _gcs_download_mock(raw_chunks, headers)
+        result = await BaseLLMHTTPHandler().async_retrieve_file_content_streaming(
+            file_content_request=FileContentRequest(file_id=_MANAGED_OUTPUT_FILE_ID),
+            provider_config=_StaticTokenFilesConfig(),
+            litellm_params={"gcs_bucket_name": "test-bucket"},
+            headers={},
+            logging_obj=_logging_obj(),
+            chunk_size=chunk_size,
+            client=_async_handler_with(mock),
+        )
+        return result, state
+
+    async def test_plain_object_streams_through_with_upstream_headers(self):
+        raw = b'{"line": 1}\n{"line": 2}\n' * 40
+        raw_chunks = [raw[i : i + 100] for i in range(0, len(raw), 100)]
+        upstream = {"content-type": "application/octet-stream", "content-length": str(len(raw))}
+
+        result, state = await self._open(raw_chunks, upstream, chunk_size=7)
+
+        assert state["urls"] == [
+            "https://storage.googleapis.com/storage/v1/b/test-bucket/o/"
+            "litellm-vertex-files%2Fpublishers%2Fgoogle%2Fmodels%2Fgemini-2.5-flash%2Fabc%2Fpredictions.jsonl?alt=media"
+        ]
+        assert state["headers"][0]["authorization"] == "Bearer test-token"
+        assert result.headers["content-type"] == "application/octet-stream"
+        assert result.headers["content-length"] == str(len(raw))
+
+        received = [chunk async for chunk in result.stream_iterator]
+        assert b"".join(received) == raw
+        assert len(received) > 1
+        assert state["closed"] is True
+
+    async def test_body_is_yielded_before_the_object_is_fully_served(self):
+        raw_chunks = [b'{"line": %d}\n' % i for i in range(50)]
+        result, state = await self._open(raw_chunks, {"content-type": "application/octet-stream"}, chunk_size=8)
+
+        first = await anext(result.stream_iterator)
+
+        assert first
+        assert state["served"] < len(raw_chunks)
+        assert state["closed"] is False
+
+    async def test_gzip_encoded_object_is_decoded_without_stale_transfer_headers(self):
+        raw = b'{"line": 1}\n{"line": 2}\n' * 200
+        encoded = gzip.compress(raw)
+        upstream = {
+            "content-type": "application/octet-stream",
+            "content-encoding": "gzip",
+            "content-length": str(len(encoded)),
+        }
+
+        result, state = await self._open([encoded[i : i + 64] for i in range(0, len(encoded), 64)], upstream)
+        streamed = b"".join([chunk async for chunk in result.stream_iterator])
+
+        assert streamed == raw
+        assert result.headers["content-type"] == "application/octet-stream"
+        assert "content-encoding" not in result.headers
+        assert "content-length" not in result.headers
+        assert state["closed"] is True
+
+    async def test_vertex_batch_output_is_transformed_row_by_row(self):
+        rows = [_vertex_batch_output_row(f"request-{i}", f"answer {i}") for i in range(30)]
+        raw = b"\n".join(rows) + b"\n"
+        raw_chunks = [raw[i : i + 333] for i in range(0, len(raw), 333)]
+        expected = VertexAIFilesConfig()._try_transform_vertex_batch_output_to_openai(
+            content=raw, logging_obj=_logging_obj(), model="gemini-2.5-flash"
+        )
+        assert expected != raw
+
+        result, state = await self._open(
+            raw_chunks,
+            {"content-type": "application/octet-stream", "content-length": str(len(raw))},
+            chunk_size=97,
+        )
+        first = await anext(result.stream_iterator)
+        assert json.loads(first)["custom_id"] == "request-0"
+        assert state["served"] < len(raw_chunks)
+
+        rest = [chunk async for chunk in result.stream_iterator]
+        streamed = b"".join([first, *rest])
+        assert _stable_row_fields(streamed) == _stable_row_fields(expected)
+        assert len(_stable_row_fields(streamed)) == len(rows)
+        assert streamed.count(b"\n") == expected.count(b"\n")
+        assert len(rest) == len(rows) - 1
+        assert result.headers["content-type"] == "application/octet-stream"
+        assert "content-length" not in result.headers
+        assert state["closed"] is True
+
+    async def test_last_row_without_trailing_newline_and_unparseable_row_are_kept(self):
+        broken = b'{"custom_id": "request-1", "response": {"candidates": [}'
+        rows = [_vertex_batch_output_row("request-0", "first"), broken, _vertex_batch_output_row("request-2", "last")]
+        raw = b"\n".join(rows)
+        raw_chunks = [raw[i : i + 41] for i in range(0, len(raw), 41)]
+
+        result, state = await self._open(raw_chunks, {}, chunk_size=29)
+        streamed_lines = b"".join([chunk async for chunk in result.stream_iterator]).split(b"\n")
+
+        assert len(streamed_lines) == len(rows)
+        assert json.loads(streamed_lines[0])["custom_id"] == "request-0"
+        assert json.loads(streamed_lines[0])["response"]["body"]["choices"][0]["message"]["content"] == "first"
+        assert streamed_lines[1] == broken
+        assert json.loads(streamed_lines[2])["custom_id"] == "request-2"
+        assert json.loads(streamed_lines[2])["response"]["body"]["choices"][0]["message"]["content"] == "last"
+        assert state["closed"] is True
+
+    async def test_transform_opt_out_streams_raw_batch_output(self, monkeypatch):
+        monkeypatch.setattr("litellm.disable_vertex_batch_output_transformation", True)
+        raw = b"\n".join(_vertex_batch_output_row(f"request-{i}", "x") for i in range(3)) + b"\n"
+
+        result, _ = await self._open([raw], {"content-length": str(len(raw))})
+
+        assert b"".join([chunk async for chunk in result.stream_iterator]) == raw
+        assert result.headers["content-length"] == str(len(raw))
+
+    async def test_embeddings_batch_output_is_transformed_with_updated_content_length(self):
+        rows = [_vertex_embeddings_output_row(f"request-{i}", [0.1 * i, 0.2]) for i in range(3)]
+        raw = b"\n".join(rows) + b"\n"
+        raw_chunks = [raw[i : i + 50] for i in range(0, len(raw), 50)]
+
+        result, _ = await self._open(raw_chunks, {"content-length": str(len(raw))}, chunk_size=64)
+        streamed = b"".join([chunk async for chunk in result.stream_iterator])
+
+        transformed = [json.loads(line) for line in streamed.split(b"\n") if line]
+        assert [row["custom_id"] for row in transformed] == ["request-0", "request-1", "request-2"]
+        assert transformed[1]["response"]["body"]["data"][0]["embedding"] == [0.1, 0.2]
+        assert transformed[1]["response"]["body"]["model"] == "gemini-2.5-flash"
+        assert result.headers["content-length"] == str(len(streamed))
+
+    async def test_object_without_newlines_streams_after_the_peek_limit(self):
+        piece = b"\xff" * (1024 * 1024)
+        raw_chunks = [piece] * 40
+
+        result, state = await self._open(raw_chunks, {"content-type": "image/png"}, chunk_size=len(piece))
+        first = await anext(result.stream_iterator)
+
+        assert state["served"] < len(raw_chunks)
+        rest = [chunk async for chunk in result.stream_iterator]
+        assert len(first) + sum(len(chunk) for chunk in rest) == len(piece) * len(raw_chunks)
+        assert set(first) == {0xFF} and all(set(chunk) == {0xFF} for chunk in rest)
+        assert result.headers["content-type"] == "image/png"
+
+    async def test_consumer_stopping_early_closes_the_response(self):
+        raw_chunks = [b'{"line": %d}\n' % i for i in range(50)]
+        result, state = await self._open(raw_chunks, {})
+
+        await anext(result.stream_iterator)
+        await result.stream_iterator.aclose()
+
+        assert state["closed"] is True
+
+    async def test_gcs_error_raises_and_closes_the_response(self):
+        state = {"closed": False}
+
+        async def handler(request: httpx.Request) -> httpx.Response:
+            response = httpx.Response(403, json={"error": {"message": "forbidden"}})
+            original_aclose = response.aclose
+
+            async def aclose():
+                state["closed"] = True
+                await original_aclose()
+
+            response.aclose = aclose
+            return response
+
+        with pytest.raises(VertexAIError) as exc_info:
+            await BaseLLMHTTPHandler().async_retrieve_file_content_streaming(
+                file_content_request=FileContentRequest(file_id=_MANAGED_OUTPUT_FILE_ID),
+                provider_config=_StaticTokenFilesConfig(),
+                litellm_params={"gcs_bucket_name": "test-bucket"},
+                headers={},
+                logging_obj=_logging_obj(),
+                chunk_size=16,
+                client=_async_handler_with(handler),
+            )
+
+        assert exc_info.value.status_code == 403
+        assert "forbidden" in str(exc_info.value)
+        assert state["closed"] is True
+
+    async def test_afile_content_stream_routes_vertex_ai_to_the_gcs_stream(self):
+        raw = b'{"line": 1}\n{"line": 2}\n' * 20
+        mock, state = _gcs_download_mock(
+            [raw[i : i + 64] for i in range(0, len(raw), 64)], {"content-length": str(len(raw))}
+        )
+
+        result = await litellm.afile_content(
+            file_id=_MANAGED_OUTPUT_FILE_ID,
+            custom_llm_provider="vertex_ai",
+            stream=True,
+            api_key="test-token",
+            gcs_bucket_name="test-bucket",
+            client=_async_handler_with(mock),
+        )
+
+        assert isinstance(result, FileContentStreamingResult)
+        assert result.headers["content-length"] == str(len(raw))
+        assert state["urls"][0].endswith("predictions.jsonl?alt=media")
+        assert b"".join([chunk async for chunk in result.stream_iterator]) == raw
+        assert state["closed"] is True
+
+    async def test_afile_content_without_stream_keeps_buffered_vertex_response(self):
+        raw = b'{"line": 1}\n{"line": 2}\n'
+        mock, _ = _gcs_download_mock([raw], {"content-length": str(len(raw))})
+
+        result = await litellm.afile_content(
+            file_id=_MANAGED_OUTPUT_FILE_ID,
+            custom_llm_provider="vertex_ai",
+            api_key="test-token",
+            gcs_bucket_name="test-bucket",
+            client=_async_handler_with(mock),
+        )
+
+        assert result.response.content == raw
+
+    def test_sync_file_content_stream_is_rejected_for_vertex_ai(self):
+        mock, state = _gcs_download_mock([b"x"], {})
+
+        with pytest.raises(litellm.BadRequestError, match="afile_content"):
+            litellm.file_content(
+                file_id=_MANAGED_OUTPUT_FILE_ID,
+                custom_llm_provider="vertex_ai",
+                stream=True,
+                api_key="test-token",
+                gcs_bucket_name="test-bucket",
+                client=_async_handler_with(mock),
+            )
+
+        assert state["urls"] == []

tests/test_litellm/proxy/openai_files_endpoint/test_files_endpoint.py

@@ -3384,12 +3384,14 @@ def test_get_file_content_provider_only_resolves_named_vertex_credentials(
 
     async def _mock_afile_content(**kwargs):
         captured_kwargs.update(kwargs)
-        return HttpxBinaryResponseContent(
-            response=httpx.Response(
-                status_code=200,
-                content=b"vertex-bytes",
-                headers={"content-type": "application/octet-stream"},
-            )
+
+        async def _stream():
+            yield b"vertex-"
+            yield b"bytes"
+
+        return FileContentStreamingResult(
+            stream_iterator=_stream(),
+            headers={"content-type": "application/octet-stream"},
         )
 
     monkeypatch.setattr(litellm, "afile_content", _mock_afile_content)
@@ -3414,6 +3416,7 @@ def test_get_file_content_provider_only_resolves_named_vertex_credentials(
     assert response.status_code == 200, response.text
     assert response.content == b"vertex-bytes"
     assert captured_kwargs.get("file_id") == "file-abc123"
+    assert captured_kwargs.get("stream") is True
     _assert_vertex_named_credentials_attached(captured_kwargs)
     proxy_logging_obj.post_call_failure_hook.assert_not_called()