scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 87

fix(async): move remote image fetches off the event loop for Snowflake, Bedrock invoke Claude, Mantle and Gemini

Browse source
This commit is contained in:
mateo-berri 2026-09-04 17:48:56 -07:00
parent b3c867c7b2
commit 199b44a475
15 changed files with 644 additions and 157 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
litellm/litellm_core_utils/prompt_templates/image_handling.py
View file

@ -2,7 +2,11 @@
Helper functions to handle images passed in messages
"""
import asyncio
import base64
from collections.abc import Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final
from httpx import Response
@ -12,6 +16,7 @@ from litellm import verbose_logger
from litellm.caching.caching import InMemoryCache
from litellm.constants import MAX_IMAGE_URL_DOWNLOAD_SIZE_MB
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.types.llms.openai import AllMessageValues
MAX_IMGS_IN_MEMORY: Final = 10
@ -124,3 +129,101 @@ def convert_url_to_base64(url: str) -> str:
raise litellm.ImageFetchError(
f"Error: Unable to fetch image from URL after 3 attempts. url={url}",
)
_REMOTE_URL_PREFIXES: Final = ("http://", "https://")
@dataclass(frozen=True, slots=True)
class _RemoteImage:
part: Mapping[str, object]
image_url: Mapping[str, object] | None
url: str
@dataclass(frozen=True, slots=True)
class _RemoteFile:
part: Mapping[str, object]
file: Mapping[str, object]
url: str
def _as_mapping(value: object) -> Mapping[str, object] | None:
return value if isinstance(value, Mapping) else None # pyright: ignore[reportUnknownVariableType] # fields are parsed one by one
def _remote_url(candidate: object) -> str | None:
return candidate if isinstance(candidate, str) and candidate.startswith(_REMOTE_URL_PREFIXES) else None
def _parse_remote_part(part: object) -> _RemoteImage | _RemoteFile | None:
fields: Final = _as_mapping(part)
if fields is None:
return None
if fields.get("type") == "image_url":
image_url: Final = fields.get("image_url")
image_url_fields: Final = _as_mapping(image_url)
url: Final = _remote_url(image_url_fields.get("url") if image_url_fields is not None else image_url)
return _RemoteImage(fields, image_url_fields, url) if url is not None else None
file: Final = _as_mapping(fields.get("file")) if fields.get("type") == "file" else None
file_url: Final = _remote_url(file.get("file_id")) if file is not None else None
return _RemoteFile(fields, file, file_url) if file is not None and file_url is not None else None
_PDF_FORMAT: Final = MappingProxyType({"format": "application/pdf"})
def _inferred_format(file: Mapping[str, object], url: str) -> Mapping[str, str]:
return _PDF_FORMAT if "format" not in file and url.lower().endswith(".pdf") else MappingProxyType({})
def _inlined_image_url(image_url: Mapping[str, object] | None, data_url: str) -> Mapping[str, object] | str:
return {**image_url, "url": data_url} if image_url is not None else data_url # mutable-ok: json-serialized part
def _inlined_file(file: Mapping[str, object], url: str, data_url: str) -> Mapping[str, object]:
kept: Final = {k: v for k, v in file.items() if k != "file_id"} # mutable-ok: json-serialized message part
return {**kept, **_inferred_format(file, url), "file_data": data_url} # mutable-ok: json-serialized part
def _inline(remote: _RemoteImage | _RemoteFile, data_url: str) -> Mapping[str, object]:
match remote:
case _RemoteImage(part, image_url, _):
return {**part, "image_url": _inlined_image_url(image_url, data_url)} # mutable-ok: json-serialized part
case _RemoteFile(part, file, url):
return {**part, "file": _inlined_file(file, url, data_url)} # mutable-ok: json-serialized message part
def _content_parts(message: Mapping[str, object]) -> tuple[object, ...]:
content: Final = message.get("content")
return tuple(content) if isinstance(content, list) else () # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType] # parts are parsed one by one
def _inline_message(message: AllMessageValues, data_urls: Mapping[str, str]) -> AllMessageValues:
parts: Final = _content_parts(message)
if not parts:
return message
inlined_parts: Final = [ # mutable-ok: content must stay a list for the transforms' isinstance checks
_inline(remote, data_urls[remote.url]) if (remote := _parse_remote_part(part)) is not None else part
for part in parts
]
inlined_message: Final = {**message, "content": inlined_parts} # mutable-ok: json-serialized message
return inlined_message # pyright: ignore[reportReturnType] # the same message with its remote parts inlined
async def async_inline_remote_media(
messages: list[AllMessageValues], # mutable-ok: every transform_request takes list[AllMessageValues]
) -> list[AllMessageValues]: # mutable-ok: every transform_request takes list[AllMessageValues]
remote_urls: Final = tuple(
dict.fromkeys(
remote.url
for message in messages
for part in _content_parts(message)
if (remote := _parse_remote_part(part)) is not None
)
)
if not remote_urls:
return messages
data_urls: Final = await asyncio.gather(*(async_convert_url_to_base64(url) for url in remote_urls))
inlined: Final = MappingProxyType(dict(zip(remote_urls, data_urls, strict=True)))
return [_inline_message(message, inlined) for message in messages] # mutable-ok: transform_request takes a list

4
litellm/llms/base_llm/chat/transformation.py
View file

@ -411,6 +411,10 @@ class BaseConfig(ABC):
def has_custom_stream_wrapper(self) -> bool:
return False
@property
def uses_async_transform_request(self) -> bool:
return False
@property
def supports_stream_param_in_request_body(self) -> bool:
"""

61
litellm/llms/bedrock/chat/invoke_transformations/anthropic_claude3_transformation.py
View file

@ -8,7 +8,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
convert_to_anthropic_image_obj,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_convert_url_to_base64,
async_inline_remote_media,
convert_url_to_base64,
)
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
@ -172,6 +172,10 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
return _anthropic_request
@property
def uses_async_transform_request(self) -> bool:
return True
async def async_transform_request(
self,
model: str,
@ -180,26 +184,14 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
litellm_params: dict,
headers: dict,
) -> dict:
_anthropic_request: Final = self._build_bedrock_anthropic_request_base(
return self.transform_request(
model=model,
messages=messages,
messages=await async_inline_remote_media(messages),
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
await self._async_convert_document_url_sources_to_base64(_anthropic_request)
beta_list: Final = self._compute_bedrock_invoke_beta_headers(
model=model,
messages=messages,
optional_params=optional_params,
headers=headers,
)
if beta_list:
_anthropic_request["anthropic_beta"] = beta_list
return _anthropic_request
def _build_bedrock_anthropic_request_base(
self,
model: str,
@ -321,45 +313,6 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
"data": image_chunk["data"],
}
async def _async_convert_document_url_sources_to_base64(self, anthropic_request: dict) -> None:
"""
Async version of document URL conversion for async completion paths.
"""
messages: Final = anthropic_request.get("messages")
if not isinstance(messages, list):
return
for message in messages:
if not isinstance(message, dict):
continue
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if not isinstance(block, dict) or block.get("type") != "document":
continue
source = block.get("source")
if not isinstance(source, dict) or source.get("type") != "url":
continue
source_url = source.get("url")
if not isinstance(source_url, str):
continue
inferred_format: str | None = None
if source_url.lower().endswith(".pdf"):
inferred_format = "application/pdf"
base64_url = await async_convert_url_to_base64(url=source_url)
image_chunk = convert_to_anthropic_image_obj(
openai_image_url=base64_url,
format=inferred_format,
)
block["source"] = {
"type": "base64",
"media_type": image_chunk["media_type"],
"data": image_chunk["data"],
}
def _normalize_bedrock_tool_search_tools(self, optional_params: dict) -> dict:
"""
Convert tool search entries to the format supported by the Bedrock Invoke API.

15
litellm/llms/bedrock/chat/mantle/transformation.py
View file

@ -10,6 +10,7 @@ at a different endpoint (bedrock-mantle.{region}.api.aws) with AWS SigV4 auth.
from collections.abc import AsyncIterator, Iterator
from typing import TYPE_CHECKING, Any, Final
from litellm.litellm_core_utils.prompt_templates.image_handling import async_inline_remote_media
from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
AmazonAnthropicClaudeConfig,
)
@ -110,21 +111,13 @@ class AmazonMantleConfig(AmazonAnthropicClaudeConfig):
litellm_params: dict,
headers: dict,
) -> dict:
model_id: Final = model.replace("mantle/", "", 1)
request: Final = self._build_bedrock_anthropic_request_base(
model=model_id,
messages=messages,
return self.transform_request(
model=model,
messages=await async_inline_remote_media(messages),
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
await self._async_convert_document_url_sources_to_base64(request)
return self._restore_mantle_body_fields(
request=request,
model_id=model_id,
optional_params=optional_params,
)
@staticmethod
def _restore_mantle_body_fields(request: dict, model_id: str, optional_params: dict) -> dict:

194
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -116,6 +116,7 @@ from litellm.types.llms.anthropic_skills import (
Skill,
)
from litellm.types.llms.openai import (
AllMessageValues,
CreateBatchRequest,
CreateFileRequest,
FileContentRequest,
@ -488,7 +489,7 @@ class BaseLLMHTTPHandler:
def completion(
self,
model: str,
messages: list,
messages: list[AllMessageValues],
api_base: str | None,
custom_llm_provider: str,
model_response: ModelResponse,
@ -507,7 +508,7 @@ class BaseLLMHTTPHandler:
shared_session: Optional["ClientSession"] = None,
):
json_mode: Final[bool] = optional_params.pop("json_mode", False)
extra_body: Final[dict | None] = optional_params.pop("extra_body", None)
extra_body: Final[Mapping[str, object] | None] = optional_params.pop("extra_body", None)
provider_config = provider_config or ProviderConfigManager.get_provider_chat_config(
model=model, provider=litellm.LlmProviders(custom_llm_provider)
@ -522,14 +523,17 @@ class BaseLLMHTTPHandler:
)
# get config from model, custom llm provider
headers = provider_config.validate_environment(
api_key=api_key,
headers=headers or {},
model=model,
messages=messages,
optional_params=optional_params,
api_base=api_base,
litellm_params=litellm_params,
request_headers: Final = cast( # cast-ok: validate_environment is declared as a bare dict
"dict[str, object]",
provider_config.validate_environment(
api_key=api_key,
headers=headers or {},
model=model,
messages=messages,
optional_params=optional_params,
api_base=api_base,
litellm_params=litellm_params,
),
)
api_base = provider_config.get_complete_url(
@ -541,93 +545,117 @@ class BaseLLMHTTPHandler:
litellm_params=litellm_params,
)
data: dict[str, object] = provider_config.transform_request(
model=model,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
if extra_body is not None:
data = {**data, **extra_body}
headers, signed_json_body = provider_config.sign_request(
headers=headers,
optional_params={
**optional_params,
**_aws_signing_overrides(optional_params, litellm_params),
},
request_data=data,
api_base=api_base,
api_key=api_key,
stream=stream,
fake_stream=fake_stream,
model=model,
)
## LOGGING
logging_obj.pre_call(
input=messages,
api_key=api_key,
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": headers,
},
)
# Check if stream was converted for WebSearch interception
# This is set by the async_pre_request_hook in WebSearchInterceptionLogger
if litellm_params.get("_websearch_interception_converted_stream", False):
logging_obj.model_call_details["websearch_interception_converted_stream"] = True
if acompletion is True:
if stream is True:
data = self._add_stream_param_to_request_body(
data=data,
provider_config=provider_config,
def sign_and_log(
transformed: dict[str, object], # mutable-ok: async_completion takes dict
) -> tuple[dict[str, object], dict[str, object], bytes | None]: # mutable-ok: async_completion takes dict
data: Final = {**transformed, **extra_body} if extra_body is not None else transformed
signed: Final = cast( # cast-ok: sign_request is declared as a bare dict
"tuple[dict[str, object], bytes | None]",
provider_config.sign_request(
headers=request_headers,
optional_params={
**optional_params,
**_aws_signing_overrides(optional_params, litellm_params),
},
request_data=data,
api_base=api_base,
api_key=api_key,
stream=stream,
fake_stream=fake_stream,
)
model=model,
),
)
logging_obj.pre_call(
input=messages,
api_key=api_key,
additional_args={
"complete_input_dict": data,
"api_base": api_base,
"headers": signed[0],
},
)
if litellm_params.get("_websearch_interception_converted_stream", False):
logging_obj.model_call_details["websearch_interception_converted_stream"] = True
return data, signed[0], signed[1]
def dispatch_async(
data: dict[str, object], # mutable-ok: async_completion takes dict
signed_headers: dict[str, object], # mutable-ok: async_completion takes dict
signed_json_body: bytes | None,
):
async_client: Final = client if isinstance(client, AsyncHTTPHandler) else None
if stream is True:
return self.acompletion_stream_function(
model=model,
messages=messages,
api_base=api_base,
headers=headers,
headers=signed_headers,
custom_llm_provider=custom_llm_provider,
provider_config=provider_config,
timeout=timeout,
logging_obj=logging_obj,
data=data,
data=self._add_stream_param_to_request_body(
data=data,
provider_config=provider_config,
fake_stream=fake_stream,
),
fake_stream=fake_stream,
client=(client if client is not None and isinstance(client, AsyncHTTPHandler) else None),
client=async_client,
litellm_params=litellm_params,
json_mode=json_mode,
optional_params=optional_params,
signed_json_body=signed_json_body,
)
return self.async_completion(
custom_llm_provider=custom_llm_provider,
provider_config=provider_config,
api_base=api_base,
headers=signed_headers,
data=data,
timeout=timeout,
model=model,
model_response=model_response,
logging_obj=logging_obj,
api_key=api_key,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
encoding=encoding,
client=async_client,
json_mode=json_mode,
signed_json_body=signed_json_body,
shared_session=shared_session,
)
else:
return self.async_completion(
custom_llm_provider=custom_llm_provider,
provider_config=provider_config,
api_base=api_base,
headers=headers,
data=data,
timeout=timeout,
model=model,
model_response=model_response,
logging_obj=logging_obj,
api_key=api_key,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
encoding=encoding,
client=(client if client is not None and isinstance(client, AsyncHTTPHandler) else None),
json_mode=json_mode,
signed_json_body=signed_json_body,
shared_session=shared_session,
if acompletion is True and provider_config.uses_async_transform_request:
async def transform_then_dispatch():
transformed: Final = cast( # cast-ok: async_transform_request is declared as a bare dict
"dict[str, object]",
await provider_config.async_transform_request(
model=model,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
headers=request_headers,
),
)
return await dispatch_async(*sign_and_log(transformed))
return transform_then_dispatch()
data, signed_headers, signed_json_body = sign_and_log(
provider_config.transform_request(
model=model,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
headers=request_headers,
)
)
if acompletion is True:
return dispatch_async(data, signed_headers, signed_json_body)
if stream is True:
data = self._add_stream_param_to_request_body(
@ -641,7 +669,7 @@ class BaseLLMHTTPHandler:
custom_llm_provider=custom_llm_provider,
logging_obj=logging_obj,
api_base=api_base,
headers=headers,
headers=signed_headers,
data=data,
signed_json_body=signed_json_body,
messages=messages,
@ -651,7 +679,7 @@ class BaseLLMHTTPHandler:
completion_stream, headers = self.make_sync_call(
provider_config=provider_config,
api_base=api_base,
headers=headers,
headers=signed_headers,
data=data,
signed_json_body=signed_json_body,
original_data=data,
@ -684,7 +712,7 @@ class BaseLLMHTTPHandler:
sync_httpx_client=sync_httpx_client,
provider_config=provider_config,
api_base=api_base,
headers=headers,
headers=signed_headers,
data=data,
signed_json_body=signed_json_body,
timeout=timeout,

16
litellm/llms/snowflake/chat/transformation.py
View file

@ -23,6 +23,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
create_anthropic_image_param,
select_anthropic_content_block_type_for_file,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import async_inline_remote_media
from litellm.llms.anthropic.chat.handler import ModelResponseIterator as AnthropicStreamParser
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.anthropic.common_utils import normalize_cache_control_in_anthropic_payload
@ -421,6 +422,21 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
return self._transform_request_anthropic(model, messages, optional_params, stream, extra_body)
return self._transform_request_openai(model, messages, optional_params, stream, extra_body)
@property
def uses_async_transform_request(self) -> bool:
return True
async def async_transform_request(
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict[str, object], # mutable-ok: BaseConfig signature
litellm_params: dict[str, object], # mutable-ok: BaseConfig signature
headers: dict[str, object], # mutable-ok: BaseConfig signature
) -> dict[str, object]: # mutable-ok: BaseConfig signature
inlined_messages: Final = await async_inline_remote_media(messages) if _is_claude_model(model) else messages
return self.transform_request(model, inlined_messages, optional_params, litellm_params, headers)
def _transform_request_openai(
self,
model: str,

9
litellm/llms/vertex_ai/gemini/transformation.py
View file

@ -27,6 +27,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
convert_to_gemini_tool_call_result,
response_schema_prompt,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import async_inline_remote_media
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.vertex_ai.common_utils import pop_vertex_request_labels
from litellm.types.files import (
@ -1348,13 +1349,15 @@ async def async_transform_request_body(
vertex_auth_header=vertex_auth_header,
)
if _openai_messages_may_need_sync_gcs_metadata_fetch(messages):
inlined_messages: Final = await async_inline_remote_media(messages) if custom_llm_provider == "gemini" else messages
if _openai_messages_may_need_sync_gcs_metadata_fetch(inlined_messages):
# _transform_request_body may issue a sync httpx.get (up to 5s timeout)
# via _get_gcs_object_content_type to fetch GCS object metadata. Run the
# whole sync transformation on a worker thread so it does not block the
# async event loop.
return await asyncify(_transform_request_body)(
messages=messages,
messages=inlined_messages,
model=model,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
@ -1363,7 +1366,7 @@ async def async_transform_request_body(
)
return _transform_request_body(
messages=messages,
messages=inlined_messages,
model=model,
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,

8
litellm/main.py
View file

@ -4474,7 +4474,7 @@ def _complete_snowflake(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client = _dispatch_client_http(ctx)
injected_client: Final = _dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
@ -4486,11 +4486,11 @@ def _complete_snowflake(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
client: Final = (
injected_client if injected_client is not None else (HTTPHandler(timeout=timeout) if stream is False else None)
) # Keep this here, otherwise, the httpx.client closes and streaming is impossible
try:
client = (
HTTPHandler(timeout=timeout) if stream is False else None
) # Keep this here, otherwise, the httpx.client closes and streaming is impossible
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,

35
tests/test_litellm/conftest.py
View file

@ -7,9 +7,12 @@
# 4. Added proper cleanup in fixtures
# 5. Added worker-specific isolation for parallel execution
import base64
import importlib
import os
from pathlib import Path
from types import SimpleNamespace
import httpx
import pytest
import asyncio
@ -595,3 +598,35 @@ def pytest_sessionfinish(session, exitstatus):
_close_handler_if_needed(getattr(litellm, "aclient", None))
_close_handler_if_needed(getattr(litellm, "client", None))
_run_coroutine_if_needed(close_litellm_async_clients())
ONE_PIXEL_PNG = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
)
@pytest.fixture
def async_only_image_fetch(monkeypatch):
from litellm.litellm_core_utils.prompt_templates import image_handling
fetch = SimpleNamespace(
fetched=[],
base64_png=base64.b64encode(ONE_PIXEL_PNG).decode(),
data_url="data:image/png;base64," + base64.b64encode(ONE_PIXEL_PNG).decode(),
)
def forbid_sync_fetch(client, url, **kwargs):
raise litellm.ImageFetchError(f"sync image fetch ran on the event loop: {url}")
async def serve_png(client, url, **kwargs):
fetch.fetched.append(url)
return httpx.Response(
200,
content=ONE_PIXEL_PNG,
headers={"content-type": "image/png"},
request=httpx.Request("GET", url),
)
monkeypatch.setattr(image_handling, "safe_get", forbid_sync_fetch)
monkeypatch.setattr(image_handling, "async_safe_get", serve_png)
return fetch

62
tests/test_litellm/litellm_core_utils/test_image_handling.py
View file

@ -1,3 +1,5 @@
import copy
import uuid
from unittest.mock import patch
import pytest
@ -8,6 +10,7 @@ from litellm import constants
from litellm.litellm_core_utils.prompt_templates import image_handling
from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_convert_url_to_base64,
async_inline_remote_media,
convert_url_to_base64,
)
@ -268,3 +271,62 @@ def test_image_size_limit_disabled(monkeypatch):
assert "Image URL download is disabled" in str(excinfo.value)
assert "MAX_IMAGE_URL_DOWNLOAD_SIZE_MB=0" in str(excinfo.value)
async def test_async_inline_remote_media_inlines_every_remote_part_shape(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
pdf_url = f"http://docs.example/{uuid.uuid4()}.pdf"
messages = [
{"role": "system", "content": "be terse"},
{
"role": "user",
"content": [
{"type": "text", "text": "what is this?"},
{"type": "image_url", "image_url": {"url": image_url, "detail": "low"}},
{"type": "image_url", "image_url": image_url},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="}},
{"type": "file", "file": {"file_id": pdf_url}},
{"type": "file", "file": {"file_id": image_url, "format": "image/png"}},
],
},
]
snapshot = copy.deepcopy(messages)
inlined = await async_inline_remote_media(messages)
data_url = async_only_image_fetch.data_url
assert inlined[0] == {"role": "system", "content": "be terse"}
assert inlined[1]["content"] == [
{"type": "text", "text": "what is this?"},
{"type": "image_url", "image_url": {"url": data_url, "detail": "low"}},
{"type": "image_url", "image_url": data_url},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="}},
{"type": "file", "file": {"format": "application/pdf", "file_data": data_url}},
{"type": "file", "file": {"format": "image/png", "file_data": data_url}},
]
assert sorted(async_only_image_fetch.fetched) == sorted([image_url, pdf_url])
assert messages == snapshot
async def test_async_inline_remote_media_leaves_messages_without_remote_parts_alone(async_only_image_fetch):
messages = [
{"role": "user", "content": [{"type": "text", "text": "hi"}]},
{
"role": "user",
"content": [{"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="}}],
},
]
assert await async_inline_remote_media(messages) is messages
assert async_only_image_fetch.fetched == []
async def test_async_inline_remote_media_raises_image_fetch_error_when_the_fetch_fails(monkeypatch):
async def serve_404(client, url, **kwargs):
return Response(404, request=Request("GET", url))
monkeypatch.setattr(image_handling, "async_safe_get", serve_404)
url = f"http://img.example/{uuid.uuid4()}.png"
with pytest.raises(litellm.ImageFetchError, match="Status code: 404"):
await async_inline_remote_media([{"role": "user", "content": [{"type": "image_url", "image_url": url}]}])

49
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py
View file

@ -1,15 +1,19 @@
import asyncio
import json
import uuid
from unittest.mock import patch
import httpx
import pytest
# Ensure the project root is on the import path so `litellm` can be imported when
# tests are executed from any working directory.
import litellm
from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
AmazonAnthropicClaudeConfig,
)
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
def test_get_supported_params_thinking():
@ -714,3 +718,48 @@ def test_bedrock_chat_invoke_response_format_stub_still_upgrades_legacy_thinking
assert result["thinking"] == {"type": "adaptive"}
assert result["output_config"] == {"effort": "high"}
async def test_bedrock_invoke_claude_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = request.content.decode()
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "us.anthropic.claude-sonnet-5",
"content": [{"type": "text", "text": "Green"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="bedrock/invoke/us.anthropic.claude-sonnet-5",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
assert image_url not in captured["body"]
assert async_only_image_fetch.base64_png in captured["body"]

51
tests/test_litellm/llms/bedrock/chat/mantle/test_bedrock_chat_mantle_transformation.py Normal file
View file

@ -0,0 +1,51 @@
import uuid
import httpx
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
async def test_bedrock_mantle_claude_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = request.content.decode()
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "us.anthropic.claude-sonnet-5",
"content": [{"type": "text", "text": "Green"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="bedrock/mantle/us.anthropic.claude-sonnet-5",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="fake-secret",
aws_region_name="us-east-1",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
assert image_url not in captured["body"]
assert async_only_image_fetch.base64_png in captured["body"]

106
tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py
View file

@ -17,7 +17,7 @@ from litellm.llms.base_llm.audio_transcription.transformation import (
AudioTranscriptionRequestData,
BaseAudioTranscriptionConfig,
)
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.custom_httpx.llm_http_handler import (
BaseLLMHTTPHandler,
@ -30,7 +30,7 @@ from litellm.llms.azure.videos.transformation import AzureVideoConfig
from litellm.llms.openai.videos.transformation import OpenAIVideoConfig
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import TranscriptionResponse
from litellm.types.utils import ModelResponse, TranscriptionResponse
_ACTIVE_KEY = "_code_interpreter_interception_active"
_SANDBOX_KEY = "_code_interpreter_interception_sandbox_key"
@ -3186,3 +3186,105 @@ async def test_async_container_list_handler_transforms_success_response():
assert [container.id for container in response.data] == ["cntr_a"]
assert response.has_more is True
class _TransformRecordingConfig(BaseConfig):
def __init__(self, transform_async: bool):
self.transform_async = transform_async
self.transform_calls = []
@property
def uses_async_transform_request(self) -> bool:
return self.transform_async
def get_supported_openai_params(self, model):
return []
def map_openai_params(self, non_default_params, optional_params, model, drop_params):
return optional_params
def validate_environment(
self, headers, model, messages, optional_params, litellm_params, api_key=None, api_base=None
):
return {}
def transform_request(self, model, messages, optional_params, litellm_params, headers):
self.transform_calls.append("sync")
return {"transformed_by": "sync"}
async def async_transform_request(self, model, messages, optional_params, litellm_params, headers):
self.transform_calls.append("async")
return {"transformed_by": "async"}
def transform_response(
self,
model,
raw_response,
model_response,
logging_obj,
request_data,
messages,
optional_params,
litellm_params,
encoding,
api_key=None,
json_mode=None,
):
model_response.choices[0].message.content = raw_response.json()["transformed_by"]
return model_response
def get_error_class(self, error_message, status_code, headers):
return BaseLLMException(status_code=status_code, message=error_message, headers=headers)
def _start_async_completion(config):
captured = {}
def handle(request):
captured["body"] = json.loads(request.content)
return httpx.Response(200, json=captured["body"])
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
pending = BaseLLMHTTPHandler().completion(
model="stub-model",
messages=[{"role": "user", "content": "hi"}],
api_base="https://llm.example/v1/chat",
custom_llm_provider="openai",
model_response=ModelResponse(),
encoding=None,
logging_obj=Mock(dynamic_success_callbacks=None, model_call_details={}),
optional_params={},
timeout=10.0,
litellm_params={},
acompletion=True,
client=client,
provider_config=config,
)
return pending, captured
async def test_completion_awaits_async_transform_request_when_config_opts_in():
config = _TransformRecordingConfig(transform_async=True)
pending, captured = _start_async_completion(config)
assert config.transform_calls == []
response = await pending
assert config.transform_calls == ["async"]
assert captured["body"] == {"transformed_by": "async"}
assert response.choices[0].message.content == "async"
async def test_completion_keeps_sync_transform_request_before_returning_by_default():
config = _TransformRecordingConfig(transform_async=False)
pending, captured = _start_async_completion(config)
assert config.transform_calls == ["sync"]
response = await pending
assert config.transform_calls == ["sync"]
assert captured["body"] == {"transformed_by": "sync"}
assert response.choices[0].message.content == "sync"

46
tests/test_litellm/llms/snowflake/chat/test_snowflake_chat_transformation.py
View file

@ -6,6 +6,7 @@ Tests tool calling request/response transformations and chat completions
import asyncio
import os
import copy
import uuid
import json
from typing import Any, Dict, List
@ -945,3 +946,48 @@ class TestSnowflakeChatCompletion:
assert len(chunks_received) > 0
content = "".join(c.choices[0].delta.content for c in chunks_received if c.choices[0].delta.content)
async def test_snowflake_claude_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = request.content.decode()
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "Green"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="snowflake/claude-sonnet-4-6",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
api_key="fake-jwt",
account_id="FAKE-ACCOUNT",
api_base=FAKE_API_BASE,
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
assert image_url not in captured["body"]
assert async_only_image_fetch.base64_png in captured["body"]

42
tests/test_litellm/llms/vertex_ai/gemini/test_transformation.py
View file

@ -1,6 +1,10 @@
import uuid
import httpx
import pytest
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.vertex_ai.gemini import transformation
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
@ -338,3 +342,41 @@ def test_map_function_enterprise_web_search_snake_case():
assert len(result) == 1
assert "enterpriseWebSearch" in result[0]
async def test_gemini_ai_studio_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
captured = {}
def handle(request):
captured["body"] = request.content.decode()
return httpx.Response(
200,
json={
"candidates": [{"content": {"parts": [{"text": "Green"}], "role": "model"}, "finishReason": "STOP"}],
"usageMetadata": {"promptTokenCount": 1, "candidatesTokenCount": 1, "totalTokenCount": 2},
},
)
client = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
response = await litellm.acompletion(
model="gemini/gemini-3.8-flash",
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": "What colour is this?"},
{"type": "image_url", "image_url": {"url": image_url}},
],
}
],
api_key="fake-gemini-key",
client=client,
)
assert response.choices[0].message.content == "Green"
assert async_only_image_fetch.fetched == [image_url]
assert image_url not in captured["body"]
assert async_only_image_fetch.base64_png in captured["body"]