Merge pull request #39863 from BerriAI/litellm_lit_7022_azure_ai_passthrough_config

fix(azure_ai): add passthrough config so router-model relays reach the deployment's own endpoint

litellm/litellm_core_utils/litellm_logging.py

@@ -120,7 +120,6 @@ from litellm.types.utils import (
     CachingDetails,
     CallTypes,
     CostBreakdown,
-    CostResponseTypes,
     CustomPricingLiteLLMParams,
     DynamicPromptManagementParamLiteral,
     EmbeddingResponse,
@@ -204,7 +203,7 @@ if TYPE_CHECKING:
     from litellm.integrations.otel.logger import OpenTelemetryV2
     from litellm.integrations.otel.model.config import ExporterSpec, OpenTelemetryV2Config
     from litellm.litellm_core_utils.llm_cost_calc.utils import BilledTokenRates
-    from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig
+    from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig, LoggedRelayResponse
 try:
     from litellm_enterprise.enterprise_callbacks.callback_controls import (
         EnterpriseCallbackControls,
@@ -2381,7 +2380,7 @@ class Logging(LiteLLMLoggingBaseClass):
         self,
         raw_bytes: list[bytes],
         provider_config: "BasePassthroughConfig",
-    ) -> Optional["CostResponseTypes"]:
+    ) -> Optional["LoggedRelayResponse"]:
         all_chunks: Final = provider_config._convert_raw_bytes_to_str_lines(raw_bytes)
         complete_streaming_response: Final = provider_config.handle_logging_collected_chunks(
             all_chunks=all_chunks,

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -1,6 +1,6 @@
 import base64
 import time
-from collections.abc import Iterator, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from itertools import groupby
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypedDict, Union, cast
@@ -210,7 +210,7 @@ def apply_grounding_request_counts(
 
 
 class ChunkProcessor:
-    def __init__(self, chunks: list, messages: list | None = None):
+    def __init__(self, chunks: list, messages: Sequence | None = None):
         self.chunks = self._sort_chunks(chunks)
         self.messages = messages
         self.first_chunk = chunks[0]
@@ -1004,8 +1004,9 @@ class ChunkProcessor:
         chunks: Sequence["_UsageBearingChunk | ModelResponse"],
         model: str,
         completion_output: str,
-        messages: list | None = None,
+        messages: Sequence | None = None,
         reasoning_tokens: int | None = None,
+        count_prompt_tokens: Callable[[], int] | None = None,
     ) -> Usage:
         """
         Calculate usage for the given chunks.
@@ -1030,7 +1031,9 @@ class ChunkProcessor:
         cost: Final[float | None] = calculated_usage_per_chunk["cost"]
 
         try:
-            returned_usage.prompt_tokens = prompt_tokens or token_counter(model=model, messages=messages)
+            returned_usage.prompt_tokens = prompt_tokens or (
+                count_prompt_tokens() if count_prompt_tokens else token_counter(model=model, messages=messages)
+            )
         except Exception:  # don't allow this failing to block a complete streaming response from being returned
             print_verbose("token_counter failed, assuming prompt tokens is 0")
             returned_usage.prompt_tokens = 0

litellm/litellm_core_utils/token_counter.py

@@ -179,6 +179,13 @@ def calculate_tiles_needed(
     return total_tiles
 
 
+def high_detail_image_token_upper_bound(base_tokens: int = 85) -> int:
+    largest_tile_count: Final = calculate_tiles_needed(
+        MAX_LONG_SIDE_FOR_IMAGE_HIGH_RES, MAX_SHORT_SIDE_FOR_IMAGE_HIGH_RES
+    )
+    return base_tokens + (base_tokens * 2) * largest_tile_count
+
+
 def _unpack_ints(fmt: str, buffer: bytes) -> tuple[int, ...]:
     return struct.unpack(fmt, buffer)
 

litellm/llms/azure/passthrough/transformation.py

@@ -1,24 +1,102 @@
+import re
+from collections.abc import Callable, Collection, Mapping, Sequence
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Final, Optional
 
 import httpx
 from httpx import Response
+from pydantic import BaseModel, ValidationError
 
 from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.llms.azure.common_utils import BaseAzureLLM
-from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig
+from litellm.llms.base_llm.passthrough.transformation import (
+    BasePassthroughConfig,
+    RelayShape,
+    logged_relay_shape,
+    replace_path_segment,
+    strip_leading_model_segment,
+)
 from litellm.secret_managers.main import get_secret_str
-from litellm.types.llms.openai import AllMessageValues
+from litellm.types.llms.openai import AllMessageValues, ResponsesAPIResponse, ResponsesTerminalEvent
 from litellm.types.router import GenericLiteLLMParams
+from litellm.types.utils import CallTypes, EmbeddingResponse, ImageResponse
 
 if TYPE_CHECKING:
     from httpx import URL
 
-    from litellm.types.utils import CostResponseTypes
+    from litellm.llms.base_llm.passthrough.transformation import LoggedRelayResponse
+
+
+class RelayedChatRequest(BaseModel):
+    messages: Sequence[Mapping[str, object]] | None = None
+
+
+class RelayedCallDetails(BaseModel):
+    request_data: RelayedChatRequest | None = None
+
+
+def _relayed_messages(litellm_logging_obj: Logging) -> Sequence[Mapping[str, object]] | None:
+    try:
+        details: Final = RelayedCallDetails.model_validate(litellm_logging_obj.model_call_details)
+    except ValidationError:
+        return None
+    return details.request_data.messages if details.request_data else None
+
+
+RESPONSES_RELAY_SHAPE: Final = RelayShape("/responses", CallTypes.aresponses, ResponsesAPIResponse.model_validate)
+
+OPENAI_RELAY_SHAPES: Final = (
+    RelayShape("/embeddings", CallTypes.aembedding, EmbeddingResponse.model_validate),
+    RESPONSES_RELAY_SHAPE,
+    RelayShape("/images/generations", CallTypes.aimage_generation, ImageResponse.model_validate),
+)
+
+
+def logged_responses_stream(all_chunks: Sequence[str], logging_obj: Logging) -> ResponsesTerminalEvent | None:
+    """A streaming logging object assembles the logged response from the terminal event, not from its body."""
+    from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
+
+    terminal_event: Final = OpenAIResponsesAPIConfig.parse_terminal_event_from_stream_chunks(all_chunks=all_chunks)
+    if terminal_event is None:
+        return None
+    logging_obj.call_type = (
+        RESPONSES_RELAY_SHAPE.call_type.value
+    )  # rebind-ok: routes cost calculation to the relayed shape's pricing path
+    return terminal_event
+
+
+AZURE_DEPLOYMENT_SEGMENT: Final = re.compile(r"(?<![^/])openai/deployments/([^/]+)")
+
+
+def azure_router_model_in_endpoint(endpoint: str, router_models: Collection[str]) -> str | None:
+    parts: Final = endpoint.split("/")
+    if len(parts) < 2:
+        return None
+    return next((part for part in parts if part in router_models), None)
+
+
+def foreign_azure_deployment(
+    endpoint: str, model_group: str, served_models: Callable[[], Collection[str]]
+) -> str | None:
+    match: Final = AZURE_DEPLOYMENT_SEGMENT.search(endpoint)
+    if match is None:
+        return None
+    deployment: Final = match.group(1)
+    if deployment == model_group:
+        return None
+    served: Final = frozenset(name.casefold() for name in served_models())
+    return None if deployment.casefold() in served else deployment
+
+
+def without_api_version(api_base: str) -> str:
+    url: Final = httpx.URL(api_base)
+    kept_params: Final = tuple((key, value) for key, value in url.params.multi_items() if key != "api-version")
+    return str(url.copy_with(params=httpx.QueryParams(kept_params)))
 
 
 class AzurePassthroughConfig(BasePassthroughConfig):
     def is_streaming_request(self, endpoint: str, request_data: dict) -> bool:
-        return "stream" in request_data
+        return bool(request_data.get("stream"))
 
     def get_complete_url(
         self,
@@ -36,14 +114,17 @@ class AzurePassthroughConfig(BasePassthroughConfig):
 
         litellm_metadata: Final = litellm_params.get("litellm_metadata") or {}
         model_group: Final = litellm_metadata.get("model_group")
-        if model_group and model_group in endpoint:
-            endpoint = endpoint.replace(model_group, model)
+        routed_endpoint: Final = replace_path_segment(endpoint, model_group, model) if model_group else endpoint
+        native_endpoint: Final = strip_leading_model_segment(routed_endpoint, (model,))
 
+        caller_api_version: Final = request_query_params.get("api-version") if request_query_params else None
+        relay_base: Final = without_api_version(base_target_url) if caller_api_version else base_target_url
         complete_url: Final = BaseAzureLLM._get_base_azure_url(
-            api_base=base_target_url,
-            litellm_params=litellm_params,
-            route=endpoint,
-            default_api_version=litellm_params.get("api_version"),
+            api_base=relay_base,
+            litellm_params=MappingProxyType(
+                {**litellm_params, "api_version": caller_api_version or litellm_params.get("api_version")}
+            ),
+            route=native_endpoint,
         )
         return (
             httpx.URL(complete_url),
@@ -92,13 +173,13 @@ class AzurePassthroughConfig(BasePassthroughConfig):
         request_data: dict,
         logging_obj: Logging,
         endpoint: str,
-    ) -> Optional["CostResponseTypes"]:
+    ) -> Optional["LoggedRelayResponse"]:
         from litellm import encoding
         from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
         from litellm.types.utils import ModelResponse
 
         if "chat/completions" not in endpoint:
-            return None
+            return logged_relay_shape(OPENAI_RELAY_SHAPES, httpx_response, logging_obj, endpoint)
 
         openai_chat_config: Final = OpenAIGPTConfig()
 
@@ -116,3 +197,27 @@ class AzurePassthroughConfig(BasePassthroughConfig):
         )
 
         return litellm_model_response
+
+    def handle_logging_collected_chunks(
+        self,
+        all_chunks: Sequence[str],
+        litellm_logging_obj: Logging,
+        model: str,
+        custom_llm_provider: str,
+        endpoint: str,
+    ) -> Optional["LoggedRelayResponse"]:
+        from litellm.proxy.pass_through_endpoints.llm_provider_handlers.openai_passthrough_logging_handler import (
+            OpenAIPassthroughLoggingHandler,
+        )
+
+        if f"/{endpoint.strip('/')}".endswith(RESPONSES_RELAY_SHAPE.path_suffix):
+            return logged_responses_stream(all_chunks, litellm_logging_obj)
+        if "chat/completions" not in endpoint:
+            return None
+
+        return OpenAIPassthroughLoggingHandler()._build_complete_streaming_response(  # pyright: ignore[reportPrivateUsage]  # the only OpenAI SSE-to-ModelResponse assembler; reimplementing it would fork the parser
+            all_chunks=all_chunks,
+            litellm_logging_obj=litellm_logging_obj,
+            model=model,
+            messages=_relayed_messages(litellm_logging_obj),
+        )

litellm/llms/azure_ai/chat/transformation.py

@@ -2,7 +2,6 @@ import copy
 import enum
 import re
 from typing import TYPE_CHECKING, Final, cast
-from urllib.parse import urlparse
 
 import httpx
 from httpx import Response
@@ -15,7 +14,10 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
     filter_value_from_dict,
 )
 from litellm.llms.azure.common_utils import BaseAzureLLM
-from litellm.llms.azure_ai.common_utils import is_foundry_model_inference_base
+from litellm.llms.azure_ai.common_utils import (
+    api_key_header_for_base,
+    is_foundry_model_inference_base,
+)
 from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj
 from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config
 from litellm.llms.openai.common_utils import drop_params_from_unprocessable_entity_error
@@ -146,11 +148,7 @@ class AzureAIStudioConfig(OpenAIConfig):
         """
         Returns True if the request should use `api-key` header for authentication.
         """
-        parsed_url: Final = urlparse(api_base)
-        host: Final = parsed_url.hostname
-        if host and (host.endswith(".services.ai.azure.com") or host.endswith(".openai.azure.com")):
-            return True
-        return False
+        return api_key_header_for_base(api_base) == "api-key"
 
     def get_complete_url(
         self,

litellm/llms/azure_ai/common_utils.py

@@ -19,6 +19,13 @@ def is_foundry_model_inference_base(api_base: str) -> bool:
     return "/openai/deployments" not in parsed.path
 
 
+def api_key_header_for_base(api_base: str | None) -> AzureAIApiKeyHeader:
+    host: Final = urlparse(api_base).hostname if api_base else None
+    if host and (host.endswith(".services.ai.azure.com") or host.endswith(".openai.azure.com")):
+        return "api-key"
+    return "Authorization"
+
+
 def get_azure_ai_entra_token(litellm_params: Mapping[str, object] | None = None) -> str | None:
     """
     Resolve an Entra ID / OAuth access token for an Azure AI Foundry deployment.

litellm/llms/azure_ai/passthrough/transformation.py

@@ -0,0 +1,232 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final
+
+import httpx
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
+
+from litellm._logging import verbose_logger
+from litellm.llms.azure_ai.common_utils import (
+    AzureFoundryModelInfo,
+    api_key_header_for_base,
+    get_azure_ai_auth_headers,
+)
+from litellm.llms.azure_ai.ocr.common_utils import get_azure_ai_ocr_config
+from litellm.llms.base_llm.passthrough.transformation import (
+    BasePassthroughConfig,
+    RelayShape,
+    logged_relay_shape,
+    strip_leading_model_segment,
+)
+from litellm.types.llms.openai import AllMessageValues
+from litellm.types.rerank import RerankResponse
+from litellm.types.utils import CallTypes, ImageResponse, StandardPassThroughResponseObject
+
+if TYPE_CHECKING:
+    from httpx import URL, Response
+
+    from litellm.litellm_core_utils.litellm_logging import Logging
+    from litellm.llms.base_llm.ocr.transformation import BaseOCRConfig, OCRResponse
+    from litellm.llms.base_llm.passthrough.transformation import LoggedRelayResponse
+
+
+EMPTY_QUERY: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+class PassthroughMetadata(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    model_group: str = ""
+
+
+def model_group_from(litellm_params: Mapping[str, object]) -> str:
+    try:
+        return PassthroughMetadata.model_validate(litellm_params.get("litellm_metadata")).model_group
+    except ValidationError:
+        return ""
+
+
+def api_version_from(litellm_params: Mapping[str, object]) -> str | None:
+    try:
+        return TypeAdapter(str | None).validate_python(litellm_params.get("api_version"))
+    except ValidationError:
+        return None
+
+
+def foundry_root(api_base: str) -> str:
+    url: Final = httpx.URL(api_base)
+    segments: Final = tuple(segment for segment in url.path.split("/") if segment)
+    root_segments: Final = segments[: segments.index("models")] if "models" in segments else segments
+    return str(url.copy_with(path="/" + "/".join(root_segments), query=None)).rstrip("/")
+
+
+def is_repeated_native_prefix(native_segments: tuple[str, ...], overlap: int) -> bool:
+    return overlap == len(native_segments) or native_segments[0] == "openai"
+
+
+def without_repeated_native_prefix(root: str, native_endpoint: str) -> str:
+    url: Final = httpx.URL(root)
+    root_segments: Final = tuple(segment for segment in url.path.split("/") if segment)
+    native_segments: Final = tuple(segment.casefold() for segment in native_endpoint.split("/") if segment)
+    overlap: Final = next(
+        (
+            length
+            for length in range(min(len(root_segments), len(native_segments)), 0, -1)
+            if tuple(segment.casefold() for segment in root_segments[-length:]) == native_segments[:length]
+            and is_repeated_native_prefix(native_segments, length)
+        ),
+        0,
+    )
+    kept_segments: Final = root_segments[: len(root_segments) - overlap]
+    return str(url.copy_with(path="/" + "/".join(kept_segments), query=None)).rstrip("/")
+
+
+def relay_query_params(
+    request_query_params: Mapping[str, object] | None,
+    deployment_api_version: str | None,
+    api_base: str,
+) -> Mapping[str, object] | None:
+    if request_query_params and "api-version" in request_query_params:
+        return request_query_params
+    api_version: Final = deployment_api_version or httpx.URL(api_base).params.get("api-version")
+    if api_version is None:
+        return request_query_params
+    return MappingProxyType({**(request_query_params or EMPTY_QUERY), "api-version": api_version})
+
+
+def relayed_body(httpx_response: Response) -> str | dict:
+    try:
+        body: Final[object] = httpx_response.json()
+    except ValueError:
+        return httpx_response.text
+    return body if isinstance(body, dict) else httpx_response.text
+
+
+FOUNDRY_RELAY_SHAPES: Final = (
+    RelayShape("/rerank", CallTypes.arerank, RerankResponse.model_validate),
+    RelayShape("/providers/blackforestlabs/v1/flux-2-pro", CallTypes.aimage_generation, ImageResponse.model_validate),
+)
+
+
+class AzureAIPassthroughConfig(AzureFoundryModelInfo, BasePassthroughConfig):
+    def __init__(self, ocr_config_for: Callable[[str], BaseOCRConfig | None] = get_azure_ai_ocr_config) -> None:
+        super().__init__()
+        self.ocr_config_for: Final = ocr_config_for
+
+    def is_streaming_request(self, endpoint: str, request_data: Mapping[str, object]) -> bool:
+        return bool(request_data.get("stream"))
+
+    def get_complete_url(
+        self,
+        api_base: str | None,
+        api_key: str | None,
+        model: str,
+        endpoint: str,
+        request_query_params: Mapping[str, object] | None,
+        litellm_params: Mapping[str, object],
+    ) -> tuple[URL, str]:
+        base_target_url: Final = self.get_api_base(api_base)
+        if base_target_url is None:
+            raise ValueError("Azure AI api base not found: set `api_base` on the deployment or AZURE_AI_API_BASE")
+
+        native_endpoint: Final = strip_leading_model_segment(endpoint, (model, model_group_from(litellm_params)))
+        root: Final = without_repeated_native_prefix(foundry_root(base_target_url), native_endpoint)
+        query_params: Final = relay_query_params(
+            request_query_params, api_version_from(litellm_params), base_target_url
+        )
+        return (self.format_url(native_endpoint, root, query_params), root)
+
+    def validate_environment(
+        self,
+        headers: Mapping[str, str],
+        model: str,
+        messages: Sequence[AllMessageValues],
+        optional_params: Mapping[str, object],
+        litellm_params: Mapping[str, object],
+        api_key: str | None = None,
+        api_base: str | None = None,
+    ) -> dict[str, str]:  # mutable-ok: base class contract returns dict for httpx
+        auth_headers: Final = get_azure_ai_auth_headers(
+            api_key=api_key,
+            litellm_params=litellm_params,
+            api_key_header=api_key_header_for_base(api_base),
+        )
+        return {**headers, **auth_headers}  # mutable-ok: base class contract returns dict for httpx
+
+    def logging_non_streaming_response(
+        self,
+        model: str,
+        custom_llm_provider: str,
+        httpx_response: Response,
+        request_data: Mapping[str, object],
+        logging_obj: Logging,
+        endpoint: str,
+    ) -> LoggedRelayResponse | OCRResponse | StandardPassThroughResponseObject | None:
+        from litellm.llms.azure.passthrough.transformation import AzurePassthroughConfig
+
+        chat_result: Final = AzurePassthroughConfig().logging_non_streaming_response(  # pyright: ignore[reportUnknownMemberType]  # the Azure config still types request_data as a bare dict
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            httpx_response=httpx_response,
+            request_data=dict(request_data),  # mutable-ok: AzurePassthroughConfig wants a dict
+            logging_obj=logging_obj,
+            endpoint=endpoint,
+        )
+        if chat_result is not None:
+            return chat_result
+        ocr_result: Final = self.logged_ocr_response(model, httpx_response, logging_obj, endpoint)
+        if ocr_result is not None:
+            return ocr_result
+        foundry_result: Final = logged_relay_shape(FOUNDRY_RELAY_SHAPES, httpx_response, logging_obj, endpoint)
+        if foundry_result is not None:
+            return foundry_result
+        return StandardPassThroughResponseObject(response=relayed_body(httpx_response))
+
+    def logged_ocr_response(
+        self, model: str, httpx_response: Response, logging_obj: Logging, endpoint: str
+    ) -> OCRResponse | None:
+        ocr_config: Final = self.ocr_config_for(model)
+        if ocr_config is None or httpx_response.status_code != 200:
+            return None
+        relayed_url: Final = httpx_response.request.url
+        relayed_origin: Final = str(relayed_url.copy_with(path="/", query=None, fragment=None)).rstrip("/")
+        ocr_url: Final = httpx.URL(
+            ocr_config.get_complete_url(
+                api_base=relayed_origin,
+                model=model,
+                optional_params={},  # mutable-ok: BaseOCRConfig wants a dict
+            )
+        )
+        known_prefixes: Final = (model, model_group_from(logging_obj.litellm_params))
+        native_endpoint: Final = strip_leading_model_segment(endpoint, known_prefixes)
+        if f"/{native_endpoint.strip('/')}" != ocr_url.path:
+            return None
+        try:
+            ocr_response: Final = ocr_config.transform_ocr_response(
+                model=model, raw_response=httpx_response, logging_obj=logging_obj
+            )
+        except (ValueError, AttributeError) as error:
+            verbose_logger.warning("azure_ai passthrough: OCR body from %s is not costable: %s", ocr_url, error)
+            return None
+        logging_obj.call_type = CallTypes.aocr.value  # rebind-ok: routes cost calculation to the per-page OCR path
+        return ocr_response
+
+    def handle_logging_collected_chunks(
+        self,
+        all_chunks: Sequence[str],
+        litellm_logging_obj: Logging,
+        model: str,
+        custom_llm_provider: str,
+        endpoint: str,
+    ) -> LoggedRelayResponse | None:
+        from litellm.llms.azure.passthrough.transformation import AzurePassthroughConfig
+
+        return AzurePassthroughConfig().handle_logging_collected_chunks(
+            all_chunks=all_chunks,
+            litellm_logging_obj=litellm_logging_obj,
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            endpoint=endpoint,
+        )

litellm/llms/base_llm/passthrough/transformation.py

@@ -1,5 +1,14 @@
+from __future__ import annotations
+
+import re
 from abc import abstractmethod
-from typing import TYPE_CHECKING, Final, Optional, Union
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final, TypeAlias
+
+from pydantic import TypeAdapter, ValidationError
+
+from litellm.types.utils import CallTypes
 
 from ..base_utils import BaseLLMModelInfo
 
@@ -7,9 +16,68 @@ if TYPE_CHECKING:
     from httpx import URL, Headers, Response
 
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
-    from litellm.types.utils import CostResponseTypes
+    from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesTerminalEvent
+    from litellm.types.rerank import RerankResponse
+    from litellm.types.utils import CostResponseTypes, StandardPassThroughResponseObject
 
     from ..chat.transformation import BaseLLMException
+    from ..ocr.transformation import OCRResponse
+
+    LoggedRelayResponse: TypeAlias = CostResponseTypes | RerankResponse | ResponsesAPIResponse | ResponsesTerminalEvent
+
+
+RELAYED_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
+
+
+def strip_leading_model_segment(endpoint: str, model_names: tuple[str, ...]) -> str:
+    path: Final = endpoint.lstrip("/")
+    for model_name in model_names:
+        if not model_name:
+            continue
+        if path == model_name:
+            return ""
+        if path.startswith(f"{model_name}/"):
+            return path[len(model_name) + 1 :]
+    return path
+
+
+def replace_path_segment(endpoint: str, segment: str, replacement: str) -> str:
+    bounded_segment: Final = re.compile(rf"(?<![^/]){re.escape(segment)}(?![^/:])")
+    return bounded_segment.sub(lambda _: replacement, endpoint)
+
+
+def relayed_json_object(httpx_response: Response) -> Mapping[str, object] | None:
+    if httpx_response.status_code != 200:
+        return None
+    try:
+        return RELAYED_JSON_OBJECT.validate_python(httpx_response.json())
+    except (ValueError, ValidationError):
+        return None
+
+
+@dataclass(frozen=True, slots=True)
+class RelayShape:
+    path_suffix: str
+    call_type: CallTypes
+    parse: Callable[[Mapping[str, object]], LoggedRelayResponse]
+
+
+def logged_relay_shape(
+    shapes: Sequence[RelayShape], httpx_response: Response, logging_obj: LiteLLMLoggingObj, endpoint: str
+) -> LoggedRelayResponse | None:
+    relayed_path: Final = f"/{endpoint.strip('/')}"
+    shape: Final = next((candidate for candidate in shapes if relayed_path.endswith(candidate.path_suffix)), None)
+    body: Final = relayed_json_object(httpx_response) if shape else None
+    if shape is None or body is None:
+        return None
+    try:
+        parsed: Final = shape.parse(body)
+    except ValidationError:
+        return None
+    logging_obj.call_type = (
+        shape.call_type.value
+    )  # rebind-ok: routes cost calculation to the relayed shape's pricing path
+    return parsed
 
 
 class BasePassthroughConfig(BaseLLMModelInfo):
@@ -23,8 +91,8 @@ class BasePassthroughConfig(BaseLLMModelInfo):
         self,
         endpoint: str,
         base_target_url: str,
-        request_query_params: dict | None,
-    ) -> "URL":
+        request_query_params: Mapping[str, object] | None,
+    ) -> URL:
         """
         Helper function to add query params to the url
         Args:
@@ -58,7 +126,7 @@ class BasePassthroughConfig(BaseLLMModelInfo):
         endpoint: str,
         request_query_params: dict | None,
         litellm_params: dict,
-    ) -> tuple["URL", str]:
+    ) -> tuple[URL, str]:
         """
         Get the complete url for the request
         Returns:
@@ -88,9 +156,7 @@ class BasePassthroughConfig(BaseLLMModelInfo):
         """
         return headers, None
 
-    def get_error_class(
-        self, error_message: str, status_code: int, headers: Union[dict, "Headers"]
-    ) -> "BaseLLMException":
+    def get_error_class(self, error_message: str, status_code: int, headers: dict | Headers) -> BaseLLMException:
         from litellm.llms.base_llm.chat.transformation import BaseLLMException
 
         return BaseLLMException(status_code=status_code, message=error_message, headers=headers)
@@ -99,21 +165,21 @@ class BasePassthroughConfig(BaseLLMModelInfo):
         self,
         model: str,
         custom_llm_provider: str,
-        httpx_response: "Response",
+        httpx_response: Response,
         request_data: dict,
-        logging_obj: "LiteLLMLoggingObj",
+        logging_obj: LiteLLMLoggingObj,
         endpoint: str,
-    ) -> Optional["CostResponseTypes"]:
+    ) -> LoggedRelayResponse | OCRResponse | StandardPassThroughResponseObject | None:
         pass
 
     def handle_logging_collected_chunks(
         self,
         all_chunks: list[str],
-        litellm_logging_obj: "LiteLLMLoggingObj",
+        litellm_logging_obj: LiteLLMLoggingObj,
         model: str,
         custom_llm_provider: str,
         endpoint: str,
-    ) -> Optional["CostResponseTypes"]:
+    ) -> LoggedRelayResponse | None:
         return None
 
     def _convert_raw_bytes_to_str_lines(self, raw_bytes: list[bytes]) -> list[str]:

litellm/llms/openai/responses/transformation.py

@@ -620,15 +620,20 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
             return event_pydantic_model.model_construct(**parsed_chunk)
 
     @staticmethod
-    def parse_terminal_response_from_stream_chunks(all_chunks: list[str]) -> ResponsesAPIResponse | None:
+    def parse_terminal_event_from_stream_chunks(all_chunks: Sequence[str]) -> ResponsesTerminalEvent | None:
         for chunk_str in reversed(all_chunks):
             for event_model in (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent):
                 try:
-                    return event_model.model_validate_json(chunk_str.removeprefix("data: ")).response
+                    return event_model.model_validate_json(chunk_str.removeprefix("data: "))
                 except ValueError:
                     continue
         return None
 
+    @staticmethod
+    def parse_terminal_response_from_stream_chunks(all_chunks: list[str]) -> ResponsesAPIResponse | None:
+        terminal_event: Final = OpenAIResponsesAPIConfig.parse_terminal_event_from_stream_chunks(all_chunks)
+        return None if terminal_event is None else terminal_event.response
+
     @staticmethod
     def get_event_model_class(event_type: str) -> type[BaseLiteLLMOpenAIResponseObject]:
         """

litellm/main.py

@@ -19,7 +19,7 @@ import random
 import sys
 import time
 import traceback
-from collections.abc import AsyncIterator, Coroutine, Iterable, Mapping, Sequence
+from collections.abc import AsyncIterator, Callable, Coroutine, Iterable, Mapping, Sequence
 from concurrent import futures
 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
 from copy import deepcopy
@@ -8595,7 +8595,7 @@ def config_completion(**kwargs):
         )
 
 
-def stream_chunk_builder_text_completion(chunks: list, messages: list | None = None) -> TextCompletionResponse:
+def stream_chunk_builder_text_completion(chunks: list, messages: Sequence | None = None) -> TextCompletionResponse:
     id: Final = chunks[0]["id"]
     object: Final = chunks[0]["object"]
     created: Final = chunks[0]["created"]
@@ -8712,10 +8712,11 @@ def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_o
 
 def stream_chunk_builder(
     chunks: list,
-    messages: list | None = None,
+    messages: Sequence | None = None,
     start_time=None,
     end_time=None,
     logging_obj: Optional["Logging"] = None,
+    count_prompt_tokens: Callable[[], int] | None = None,
 ) -> ModelResponse | TextCompletionResponse | None:
     try:
         if chunks is None:
@@ -8789,6 +8790,7 @@ def stream_chunk_builder(
                 completion_output=completion_output,
                 messages=messages,
                 reasoning_tokens=0,
+                count_prompt_tokens=count_prompt_tokens,
             )
             setattr(response, "usage", usage)
 
@@ -8966,6 +8968,7 @@ def stream_chunk_builder(
             completion_output=completion_output,
             messages=messages,
             reasoning_tokens=reasoning_tokens,
+            count_prompt_tokens=count_prompt_tokens,
         )
 
         setattr(response, "usage", usage)

litellm/proxy/auth/auth_utils.py

@@ -27,6 +27,7 @@ from litellm.litellm_core_utils.url_utils import (
     provider_url_destination_candidates,
     validate_url,
 )
+from litellm.llms.azure.passthrough.transformation import azure_router_model_in_endpoint
 from litellm.proxy._types import *
 from litellm.proxy.common_utils.http_parsing_utils import extract_nested_form_metadata
 from litellm.types.passthrough_endpoints.pass_through_endpoints import (
@@ -2003,9 +2004,20 @@ def get_model_from_request(
         bedrock_model: Final = _model_from_bedrock_route(route)
         return model if bedrock_model is None else bedrock_model
 
+    if route.lower().startswith(("/azure/", "/azure_ai/")):
+        azure_model: Final = _router_model_from_azure_route(route, llm_router)
+        return model if azure_model is None else azure_model
+
     return model
 
 
+def _router_model_from_azure_route(route: str, llm_router: Router | None) -> str | None:
+    if llm_router is None:
+        return None
+    endpoint: Final = re.sub(r"^/azure(?:_ai)?/", "", route, flags=re.IGNORECASE)
+    return azure_router_model_in_endpoint(endpoint, frozenset(llm_router.get_model_names()))
+
+
 def _model_from_bedrock_route(route: str) -> str | None:
     from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
         _extract_model_from_bedrock_endpoint,

litellm/proxy/common_utils/http_parsing_utils.py

@@ -39,7 +39,7 @@ def _is_form_content_type(content_type: str) -> bool:
     return _normalize_media_type(content_type) in _FORM_CONTENT_TYPES
 
 
-def _is_json_content_type(content_type: str) -> bool:
+def is_json_content_type(content_type: str) -> bool:
     """True iff the body should be parsed as JSON."""
     return _normalize_media_type(content_type) == "application/json"
 
@@ -406,7 +406,7 @@ async def get_request_body(request: Request) -> dict[str, Any]:
     """
     if request.method == "POST":
         content_type: Final = request.headers.get("content-type", "")
-        if _is_json_content_type(content_type):
+        if is_json_content_type(content_type):
             return await _read_request_body(request)
         elif _is_form_content_type(content_type):
             return await get_form_data(request)

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -34,6 +34,7 @@ from litellm.constants import (
 )
 from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
 from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+from litellm.llms.azure.passthrough.transformation import foreign_azure_deployment
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
 from litellm.passthrough.main import AsyncPassthroughStreamingResponse
@@ -53,6 +54,7 @@ from litellm.proxy.common_utils.http_parsing_utils import (
     _safe_set_request_parsed_body,
     get_form_data,
     get_request_body,
+    is_json_content_type,
 )
 from litellm.proxy.common_utils.sse_keepalive import (
     wrap_passthrough_sse_bytes_with_keepalive_pings,
@@ -78,6 +80,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import (
     LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY,
 )
 from litellm.types.passthrough_endpoints.vertex_ai import VertexPassThroughCredentials
+from litellm.types.router import LiteLLMParamsTypedDict
 from litellm.types.utils import LlmProviders
 from litellm.types.vector_stores import LiteLLM_ManagedVectorStore
 from litellm.utils import ProviderConfigManager
@@ -120,6 +123,24 @@ def is_passthrough_request_using_router_model(request_body: dict, llm_router: li
         return False
 
 
+class RelayRejection(TypedDict):
+    error: ReadOnly[str]
+
+
+def _deployment_model_name(litellm_params: LiteLLMParamsTypedDict) -> str:
+    model: Final = litellm_params.get("model", "")
+    try:
+        return get_llm_provider(model=model, custom_llm_provider=litellm_params.get("custom_llm_provider"))[0]
+    except litellm.BadRequestError:
+        return model
+
+
+def _models_served_by_group(llm_router: litellm.Router, model_group: str) -> frozenset[str]:
+    return frozenset(
+        _deployment_model_name(row["litellm_params"]) for row in llm_router.get_model_list(model_name=model_group) or ()
+    )
+
+
 def is_passthrough_request_streaming(request_body: object) -> bool:
     """
     Returns True if the request is streaming.
@@ -412,7 +433,7 @@ async def vllm_proxy_route(
                 content=None,
                 data=None,
                 files=None,
-                json=(request_body if request.headers.get("content-type") == "application/json" else None),
+                json=(request_body if is_json_content_type(request.headers.get("content-type", "")) else None),
                 params=None,
                 headers=None,
                 cookies=None,
@@ -1499,6 +1520,14 @@ async def _relay_upstream_bytes(upstream: AsyncGenerator[bytes, bytes]) -> Async
         await upstream.aclose()
 
 
+async def _relay_upstream_response(upstream: httpx.Response) -> Response:
+    return Response(
+        content=await upstream.aread(),
+        status_code=upstream.status_code,
+        headers=HttpPassThroughEndpointHelpers.get_response_headers(headers=upstream.headers, custom_headers=None),
+    )
+
+
 async def _relay_azure_router_model(
     llm_router: litellm.Router,
     model: str,
@@ -1508,30 +1537,37 @@ async def _relay_azure_router_model(
     is_streaming_request: bool,
     user_api_key_dict: UserAPIKeyAuth,
 ) -> Response:
-    result: Final = await llm_router.allm_passthrough_route(
-        model=model,
-        method=request.method,
-        endpoint=endpoint,
-        request_query_params=request.query_params,
-        request_headers=_safe_get_request_headers(request),
-        stream=is_streaming_request,
-        content=None,
-        data=None,
-        files=None,
-        json=(request_body if request.headers.get("content-type") == "application/json" else None),
-        params=None,
-        headers=None,
-        cookies=None,
-        litellm_metadata=get_passthrough_router_request_metadata(user_api_key_dict),
+    foreign_deployment: Final = foreign_azure_deployment(
+        endpoint, model, lambda: _models_served_by_group(llm_router, model)
     )
+    if foreign_deployment is not None:
+        rejection: Final[RelayRejection] = {
+            "error": f"deployment '{foreign_deployment}' in the path is not served by model group '{model}'; "
+            "put the model group name in the deployments segment"
+        }
+        raise HTTPException(status_code=400, detail=rejection)
+    try:
+        result: Final = await llm_router.allm_passthrough_route(
+            model=model,
+            method=request.method,
+            endpoint=endpoint,
+            request_query_params=request.query_params,
+            request_headers=_safe_get_request_headers(request),
+            stream=is_streaming_request,
+            content=None,
+            data=None,
+            files=None,
+            json=(request_body if is_json_content_type(request.headers.get("content-type", "")) else None),
+            params=None,
+            headers=None,
+            cookies=None,
+            litellm_metadata=get_passthrough_router_request_metadata(user_api_key_dict),
+        )
+    except httpx.HTTPStatusError as upstream_error:
+        return await _relay_upstream_response(upstream_error.response)
 
     if not is_streaming_request:
-        upstream: Final = cast(httpx.Response, result)
-        return Response(
-            content=await upstream.aread(),
-            status_code=upstream.status_code,
-            headers=HttpPassThroughEndpointHelpers.get_response_headers(headers=upstream.headers, custom_headers=None),
-        )
+        return await _relay_upstream_response(cast(httpx.Response, result))
 
     if inspect.isasyncgen(result):
         sse_headers: Final = {"content-type": "text/event-stream"}

litellm/proxy/pass_through_endpoints/llm_provider_handlers/openai_passthrough_logging_handler.py

@@ -4,6 +4,7 @@ OpenAI Passthrough Logging Handler
 Handles cost tracking and logging for OpenAI passthrough endpoints, specifically /chat/completions.
 """
 
+from collections.abc import Mapping, Sequence
 from datetime import datetime
 from typing import Final
 from urllib.parse import urlparse
@@ -16,6 +17,7 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 from litellm.litellm_core_utils.litellm_logging import (
     get_standard_logging_object_payload,
 )
+from litellm.litellm_core_utils.token_counter import high_detail_image_token_upper_bound
 from litellm.llms.openai.openai import OpenAIConfig
 from litellm.llms.openai.openai import OpenAIConfig as OpenAIConfigType
 from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
@@ -96,6 +98,47 @@ def _is_openai_compatible_url(url_route: str | None) -> bool:
     return False
 
 
+def _is_remote_high_detail_image(part: object) -> bool:
+    if not isinstance(part, Mapping) or part.get("type") != "image_url":
+        return False
+    image_url: Final = part.get("image_url")
+    if not isinstance(image_url, Mapping):
+        return False
+    url: Final = image_url.get("url")
+    return (
+        isinstance(url, str) and url.lower().startswith(("http://", "https://")) and image_url.get("detail") == "high"
+    )
+
+
+def _content_parts(message: Mapping[str, object]) -> Sequence[object]:
+    content: Final = message.get("content")
+    return content if isinstance(content, list) else ()
+
+
+def _without_remote_high_detail_images(message: Mapping[str, object]) -> Mapping[str, object]:
+    if not isinstance(message.get("content"), list):
+        return message
+    kept_parts: Final = [  # mutable-ok: token_counter reads message content only when it is a list
+        part for part in _content_parts(message) if not _is_remote_high_detail_image(part)
+    ]
+    return {**message, "content": kept_parts}  # mutable-ok: token_counter rejects any message that is not a dict
+
+
+def count_relayed_prompt_tokens(model: str, messages: Sequence[Mapping[str, object]] | None) -> int:
+    if messages is None:
+        return 0
+    remote_high_detail_images: Final = sum(
+        1 for message in messages for part in _content_parts(message) if _is_remote_high_detail_image(part)
+    )
+    local_messages: Final = [  # mutable-ok: token_counter takes a list of messages
+        _without_remote_high_detail_images(message) for message in messages
+    ]
+    return (
+        litellm.token_counter(model=model, messages=local_messages)
+        + high_detail_image_token_upper_bound() * remote_high_detail_images
+    )
+
+
 class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
     """
     OpenAI-specific passthrough logging handler that provides cost tracking for /chat/completions endpoints.
@@ -512,9 +555,10 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
 
     def _build_complete_streaming_response(
         self,
-        all_chunks: list[str],
+        all_chunks: Sequence[str],
         litellm_logging_obj: LiteLLMLoggingObj,
         model: str,
+        messages: Sequence[Mapping[str, object]] | None = None,
     ) -> ModelResponse | TextCompletionResponse | None:
         """
         Builds complete response from raw chunks for OpenAI streaming responses.
@@ -558,7 +602,11 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
                 return None
 
             # Build complete response from chunks
-            complete_streaming_response: Final = litellm.stream_chunk_builder(chunks=all_openai_chunks)
+            complete_streaming_response: Final = litellm.stream_chunk_builder(
+                chunks=all_openai_chunks,
+                messages=messages,
+                count_prompt_tokens=lambda: count_relayed_prompt_tokens(model, messages),
+            )
 
             return complete_streaming_response
 

litellm/router.py

@@ -99,6 +99,7 @@ from litellm.litellm_core_utils.sensitive_data_masker import (
     mask_sensitive_structure,
 )
 from litellm.litellm_core_utils.token_counter import offload_token_count
+from litellm.llms.base_llm.passthrough.transformation import replace_path_segment
 from litellm.llms.base_llm.vector_store.transformation import (
     RouterVectorStoreEmbeddingExecutor,
     vector_store_request_metadata,
@@ -150,6 +151,7 @@ from litellm.router_utils.common_utils import (
     filter_team_based_models,
     filter_web_search_deployments,
     get_request_team_id,
+    provider_for_generic_call,
     resolve_model_group_alias,
     truncate_fallback_error_detail,
     warn_on_provider_credential_mismatch,
@@ -5199,7 +5201,7 @@ class Router:
                 # If get_llm_provider fails, fall back to using model_name as-is
                 replacement_model_name = model_name
 
-            kwargs["endpoint"] = kwargs["endpoint"].replace(model, replacement_model_name)
+            kwargs["endpoint"] = replace_path_segment(kwargs["endpoint"], model, replacement_model_name)
         return kwargs
 
     async def _ageneric_api_call_with_fallbacks_helper(self, model: str, original_generic_function: Callable, **kwargs):
@@ -5235,16 +5237,7 @@ class Router:
                 kwargs=kwargs, model=model, model_name=model_name
             )
 
-            # Get custom_llm_provider from deployment params
-            try:
-                custom_llm_provider = data.get("custom_llm_provider")
-                _, inferred_custom_llm_provider, _, _ = get_llm_provider(
-                    model=data["model"],
-                    custom_llm_provider=custom_llm_provider,
-                )
-                custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider
-            except Exception:
-                custom_llm_provider = None
+            custom_llm_provider: Final = provider_for_generic_call(data)
 
             response_kwargs: Final = {
                 **data,
@@ -5755,15 +5748,7 @@ class Router:
             # Perform pre-call checks for routing strategy
             self.routing_strategy_pre_call_checks(deployment=deployment)
 
-            try:
-                custom_llm_provider = data.get("custom_llm_provider")
-                _, inferred_custom_llm_provider, _, _ = get_llm_provider(
-                    model=data["model"],
-                    custom_llm_provider=custom_llm_provider,
-                )
-                custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider
-            except Exception:
-                custom_llm_provider = None
+            custom_llm_provider: Final = provider_for_generic_call(data)
 
             response: Final = original_function(
                 **{

litellm/router_utils/common_utils.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Final
 if TYPE_CHECKING:
     from litellm.types.llms.openai import OpenAIFileObject
 
+import litellm
 from litellm._logging import verbose_logger, verbose_router_logger
 from litellm.constants import ROUTER_FALLBACK_ERROR_DETAIL_MAX_CHARS
 from litellm.exceptions import BadRequestError
@@ -256,6 +257,32 @@ PROVIDER_SCOPED_CREDENTIAL_PARAMS: Final[Mapping[str, frozenset[str]]] = Mapping
 )
 
 
+def provider_for_generic_call(litellm_params: Mapping[str, object]) -> str | None:
+    """
+    The provider the router hands a deployment's generic SDK call, or None when it cannot be resolved.
+
+    A model that carries its own provider prefix keeps that prefix even where get_llm_provider
+    would resolve it to a sibling provider (azure_ai/<openai model> on an Azure OpenAI host
+    resolves to azure): the SDK call still receives the prefixed model, and an explicit provider
+    that contradicts the prefix makes get_llm_provider re-prefix it into a deployment name that
+    does not exist upstream.
+    """
+    declared: Final = litellm_params.get("custom_llm_provider")
+    if isinstance(declared, str) and declared:
+        return declared
+    model: Final = litellm_params.get("model")
+    if not isinstance(model, str) or not model:
+        return None
+    prefix: Final = model.split("/", 1)[0]
+    if "/" in model and prefix in litellm.provider_list:
+        return prefix
+    try:
+        _, inferred, _, _ = get_llm_provider(model=model)
+    except BadRequestError:
+        return None
+    return inferred
+
+
 def warn_on_provider_credential_mismatch(model_name: str, litellm_params: Mapping[str, object]) -> str | None:
     """
     Warn when a deployment carries one provider's credentials but resolves to another.

litellm/types/llms/openai.py

@@ -1564,6 +1564,9 @@ class ResponseIncompleteEvent(BaseLiteLLMOpenAIResponseObject):
     response: ResponsesAPIResponse
 
 
+ResponsesTerminalEvent: TypeAlias = ResponseCompletedEvent | ResponseIncompleteEvent | ResponseFailedEvent
+
+
 class ResponsePartAddedEvent(BaseLiteLLMOpenAIResponseObject):
     type: Literal[ResponsesAPIStreamEvents.RESPONSE_PART_ADDED]
     item_id: str

litellm/utils.py

@@ -8850,6 +8850,12 @@ class ProviderConfigManager:
             )
 
             return AzurePassthroughConfig()
+        elif LlmProviders.AZURE_AI == provider:
+            from litellm.llms.azure_ai.passthrough.transformation import (
+                AzureAIPassthroughConfig,
+            )
+
+            return AzureAIPassthroughConfig()
         elif LlmProviders.GIGACHAT == provider:
             from litellm.llms.gigachat.passthrough.transformation import (
                 GigaChatPassthroughConfig,

tests/test_litellm/litellm_core_utils/test_token_counter.py

@@ -1,6 +1,7 @@
 #### What this tests ####
 #    This tests litellm.token_counter.token_counter() function
 import asyncio
+import base64
 import importlib
 import threading
 import time
@@ -25,6 +26,8 @@ from litellm.litellm_core_utils.token_counter import (
     _get_exact_count_function,
     _get_extrapolating_count_function,
     _get_tiktoken_count_function,
+    calculate_img_tokens,
+    high_detail_image_token_upper_bound,
     offload_token_count,
 )
 from litellm.litellm_core_utils.token_counter import token_counter as token_counter_new
@@ -1558,3 +1561,18 @@ def test_openai_file_block_without_inline_bytes_counts_what_it_carries():
     assert _count_user_content([prompt, named]) == _count_user_content(
         [prompt, {"type": "text", "text": "report.pdf"}]
     )
+
+
+def _png_data_url(width: int, height: int) -> str:
+    ihdr = b"\x89PNG\r\n\x1a\n" + (13).to_bytes(4, "big") + b"IHDR" + width.to_bytes(4, "big") + height.to_bytes(4, "big")
+    return "data:image/png;base64," + base64.b64encode(ihdr + b"\x08\x06\x00\x00\x00").decode()
+
+
+@pytest.mark.parametrize(("width", "height"), [(1, 1), (768, 768), (2000, 768), (768, 2000), (4096, 4096), (8000, 3072)])
+def test_high_detail_image_token_upper_bound_covers_every_image_size(width: int, height: int) -> None:
+    assert calculate_img_tokens(_png_data_url(width, height), mode="high") <= high_detail_image_token_upper_bound()
+
+
+def test_high_detail_image_token_upper_bound_is_reached_by_the_largest_high_res_image() -> None:
+    assert calculate_img_tokens(_png_data_url(2000, 768), mode="high") == high_detail_image_token_upper_bound()
+    assert calculate_img_tokens(_png_data_url(1, 1), mode="high") < high_detail_image_token_upper_bound()

tests/test_litellm/llms/azure/passthrough/test_azure_passthrough_transformation.py

@@ -1,11 +1,20 @@
 import json
+from datetime import datetime
 from unittest.mock import MagicMock
 
 import httpx
+import pytest
 
-
-from litellm.llms.azure.passthrough.transformation import AzurePassthroughConfig
-from litellm.types.utils import ModelResponse
+import litellm
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.token_counter import high_detail_image_token_upper_bound
+from litellm.llms.azure.passthrough.transformation import (
+    AzurePassthroughConfig,
+    azure_router_model_in_endpoint,
+    foreign_azure_deployment,
+)
+from litellm.types.llms.openai import ResponseCompletedEvent, ResponsesAPIResponse
+from litellm.types.utils import EmbeddingResponse, ModelResponse
 
 
 def _azure_chat_completion_body():
@@ -73,22 +82,408 @@ def test_azure_passthrough_logging_non_streaming_response_chat_completions():
     assert result.usage.total_tokens == 18
 
 
-def test_azure_passthrough_logging_non_streaming_response_unknown_endpoint_returns_none():
-    """
-    Endpoints other than chat/completions (responses, messages, images) fall
-    through to None — matches base-class behavior and Bedrock's "unknown
-    endpoint" handling. Not a regression; just scoping.
-    """
-    config = AzurePassthroughConfig()
-    logging_obj = MagicMock()
-
-    result = config.logging_non_streaming_response(
-        model="gpt-4.1-mini",
+def _relay_logging_obj(model: str) -> Logging:
+    logging_obj = Logging(
+        model=model,
+        messages=[],
+        stream=False,
+        call_type="allm_passthrough_route",
+        start_time=datetime.now(),
+        litellm_call_id="call-1",
+        function_id="fn-1",
+    )
+    logging_obj.update_environment_variables(
+        model=model,
+        litellm_params={"api_base": "https://my-resource.openai.azure.com", "custom_llm_provider": "azure"},
+        optional_params={},
         custom_llm_provider="azure",
-        httpx_response=_make_httpx_response(_azure_chat_completion_body()),
+    )
+    return logging_obj
+
+
+def _relay_logging_result(model: str, endpoint: str, body, status_code: int = 200):
+    logging_obj = _relay_logging_obj(model)
+    response = httpx.Response(
+        status_code=status_code,
+        headers={"content-type": "application/json"},
+        content=json.dumps(body).encode("utf-8"),
+        request=httpx.Request(
+            "POST", f"https://my-resource.openai.azure.com/{endpoint}?api-version=2025-04-01-preview"
+        ),
+    )
+    result = AzurePassthroughConfig().logging_non_streaming_response(
+        model=model,
+        custom_llm_provider="azure",
+        httpx_response=response,
         request_data={},
         logging_obj=logging_obj,
-        endpoint="openai/responses",
+        endpoint=endpoint,
+    )
+    return result, logging_obj
+
+
+EMBEDDINGS_BODY = {
+    "object": "list",
+    "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
+    "model": "text-embedding-3-small",
+    "usage": {"prompt_tokens": 1000, "total_tokens": 1000},
+}
+
+RESPONSES_BODY = {
+    "id": "resp_1",
+    "object": "response",
+    "created_at": 1,
+    "status": "completed",
+    "model": "gpt-4.1-mini",
+    "output": [
+        {
+            "type": "message",
+            "id": "msg_1",
+            "role": "assistant",
+            "status": "completed",
+            "content": [{"type": "output_text", "text": "hi", "annotations": []}],
+        }
+    ],
+    "usage": {"input_tokens": 1000, "output_tokens": 100, "total_tokens": 1100},
+}
+
+
+def test_azure_passthrough_embeddings_relay_is_costed_per_input_token():
+    result, logging_obj = _relay_logging_result(
+        "text-embedding-3-small", "openai/deployments/text-embedding-3-small/embeddings", EMBEDDINGS_BODY
+    )
+    per_token = litellm.get_model_info("azure/text-embedding-3-small")["input_cost_per_token"]
+
+    assert isinstance(result, EmbeddingResponse)
+    assert logging_obj.call_type == "aembedding"
+    assert per_token > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(1000 * per_token)
+
+
+def test_azure_passthrough_responses_relay_is_costed_per_token():
+    result, logging_obj = _relay_logging_result("gpt-4.1-mini", "openai/responses", RESPONSES_BODY)
+    info = litellm.get_model_info("azure/gpt-4.1-mini")
+
+    assert isinstance(result, ResponsesAPIResponse)
+    assert logging_obj.call_type == "aresponses"
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(
+        1000 * info["input_cost_per_token"] + 100 * info["output_cost_per_token"]
+    )
+
+
+def test_azure_passthrough_failed_embeddings_relay_is_not_costed():
+    result, logging_obj = _relay_logging_result(
+        "text-embedding-3-small",
+        "openai/deployments/text-embedding-3-small/embeddings",
+        {"error": {"code": "429", "message": "rate limited"}},
+        status_code=429,
     )
 
     assert result is None
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def test_azure_passthrough_logging_non_streaming_response_unknown_endpoint_returns_none():
+    result, logging_obj = _relay_logging_result(
+        "gpt-4o-mini-tts", "openai/deployments/gpt-4o-mini-tts/audio/speech", {"audio": "..."}
+    )
+
+    assert result is None
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def _sse_line(payload: dict) -> str:
+    return "data: " + json.dumps(payload)
+
+
+def _azure_chat_completion_chunks() -> list[str]:
+    head = {"id": "chatcmpl-abc123", "object": "chat.completion.chunk", "created": 1700000000, "model": "gpt-4.1-mini"}
+    return [
+        _sse_line(
+            {
+                **head,
+                "choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hello!"}, "finish_reason": None}],
+            }
+        ),
+        _sse_line(
+            {**head, "choices": [{"index": 0, "delta": {"content": " How can I assist?"}, "finish_reason": None}]}
+        ),
+        _sse_line({**head, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}),
+        _sse_line({**head, "choices": [], "usage": {"prompt_tokens": 10, "completion_tokens": 8, "total_tokens": 18}}),
+        "data: [DONE]",
+    ]
+
+
+def test_azure_passthrough_streaming_chat_chunks_build_the_complete_response():
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=_azure_chat_completion_chunks(),
+        litellm_logging_obj=MagicMock(),
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/deployments/gpt-4.1-mini/chat/completions",
+    )
+
+    assert isinstance(response, ModelResponse)
+    assert response.choices[0].message.content == "Hello! How can I assist?"
+    assert response.usage.prompt_tokens == 10
+    assert response.usage.completion_tokens == 8
+
+
+def test_azure_passthrough_streaming_chunks_without_usage_count_prompt_tokens_from_the_relayed_request():
+    messages = [{"role": "user", "content": "Say hi in three words"}]
+    logging_obj = MagicMock()
+    logging_obj.model_call_details = {"request_data": {"messages": messages, "stream": True}}
+
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=[chunk for chunk in _azure_chat_completion_chunks() if '"usage"' not in chunk],
+        litellm_logging_obj=logging_obj,
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/deployments/gpt-4.1-mini/chat/completions",
+    )
+
+    assert isinstance(response, ModelResponse)
+    assert response.choices[0].message.content == "Hello! How can I assist?"
+    assert response.usage.prompt_tokens > 0
+    assert response.usage.prompt_tokens == litellm.token_counter(model="gpt-4.1-mini", messages=messages)
+    assert response.usage.completion_tokens > 0
+
+
+def test_azure_passthrough_streaming_chunks_count_remote_image_prompt_tokens_without_fetching_the_image():
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "Describe this"},
+                {"type": "image_url", "image_url": {"url": "http://127.0.0.1:9/doc.png", "detail": "high"}},
+            ],
+        }
+    ]
+    logging_obj = MagicMock()
+    logging_obj.model_call_details = {"request_data": {"messages": messages, "stream": True}}
+
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=[chunk for chunk in _azure_chat_completion_chunks() if '"usage"' not in chunk],
+        litellm_logging_obj=logging_obj,
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/deployments/gpt-4.1-mini/chat/completions",
+    )
+
+    text_only_messages = [{"role": "user", "content": [{"type": "text", "text": "Describe this"}]}]
+    assert isinstance(response, ModelResponse)
+    assert response.usage.prompt_tokens == (
+        litellm.token_counter(model="gpt-4.1-mini", messages=text_only_messages) + high_detail_image_token_upper_bound()
+    )
+
+
+def test_azure_passthrough_streaming_chunks_for_unknown_endpoint_return_none():
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=_azure_chat_completion_chunks(),
+        litellm_logging_obj=MagicMock(),
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/deployments/gpt-4.1-mini/embeddings",
+    )
+
+    assert response is None
+
+
+def _azure_responses_stream_chunks(terminal_event: str | None = "response.completed") -> list[str]:
+    in_progress = {**RESPONSES_BODY, "status": "in_progress", "output": [], "usage": None}
+    events = [
+        ("response.created", {"type": "response.created", "sequence_number": 0, "response": in_progress}),
+        (
+            "response.output_text.delta",
+            {"type": "response.output_text.delta", "sequence_number": 1, "item_id": "msg_1", "delta": "hi"},
+        ),
+    ] + (
+        [(terminal_event, {"type": terminal_event, "sequence_number": 2, "response": RESPONSES_BODY})]
+        if terminal_event
+        else []
+    )
+    return [line for name, payload in events for line in (f"event: {name}", _sse_line(payload))]
+
+
+def test_azure_passthrough_streaming_responses_chunks_are_costed_per_token():
+    logging_obj = _relay_logging_obj("gpt-4.1-mini")
+
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=_azure_responses_stream_chunks(),
+        litellm_logging_obj=logging_obj,
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/responses",
+    )
+    info = litellm.get_model_info("azure/gpt-4.1-mini")
+
+    assert isinstance(response, ResponseCompletedEvent)
+    assert response.response.usage.input_tokens == 1000
+    assert logging_obj.call_type == "aresponses"
+    assert logging_obj._response_cost_calculator(result=response.response) == pytest.approx(
+        1000 * info["input_cost_per_token"] + 100 * info["output_cost_per_token"]
+    )
+
+
+def test_azure_passthrough_streaming_responses_without_a_terminal_event_are_not_costed():
+    logging_obj = _relay_logging_obj("gpt-4.1-mini")
+
+    response = AzurePassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=_azure_responses_stream_chunks(terminal_event=None),
+        litellm_logging_obj=logging_obj,
+        model="gpt-4.1-mini",
+        custom_llm_provider="azure",
+        endpoint="openai/responses",
+    )
+
+    assert response is None
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def _complete_url(request_query_params: dict, litellm_params: dict) -> httpx.URL:
+    url, _ = AzurePassthroughConfig().get_complete_url(
+        api_base="https://my-resource.openai.azure.com",
+        api_key="key",
+        model="gpt-4.1-mini",
+        endpoint="openai/deployments/gpt-4.1-mini/chat/completions",
+        request_query_params=request_query_params,
+        litellm_params=litellm_params,
+    )
+    return url
+
+
+def test_azure_passthrough_url_forwards_the_callers_api_version():
+    url = _complete_url(request_query_params={"api-version": "2025-04-01-preview"}, litellm_params={})
+
+    assert url.path == "/openai/deployments/gpt-4.1-mini/chat/completions"
+    assert url.params["api-version"] == "2025-04-01-preview"
+
+
+def test_azure_passthrough_url_prefers_the_callers_api_version_over_the_deployments():
+    url = _complete_url(
+        request_query_params={"api-version": "2025-04-01-preview"}, litellm_params={"api_version": "2024-10-21"}
+    )
+
+    assert url.params["api-version"] == "2025-04-01-preview"
+
+
+def test_azure_passthrough_url_fills_in_the_deployments_api_version_when_the_caller_sends_none():
+    url = _complete_url(request_query_params={}, litellm_params={"api_version": "2024-10-21"})
+
+    assert url.params["api-version"] == "2024-10-21"
+
+
+FULL_URL_API_BASE = (
+    "https://my-resource.openai.azure.com/openai/deployments/gpt-4.1-mini/chat/completions?api-version=2024-10-21"
+)
+
+
+def _full_url_complete_url(request_query_params: dict) -> httpx.URL:
+    url, _ = AzurePassthroughConfig().get_complete_url(
+        api_base=FULL_URL_API_BASE,
+        api_key="key",
+        model="gpt-4.1-mini",
+        endpoint="chat/completions",
+        request_query_params=request_query_params,
+        litellm_params={},
+    )
+    return url
+
+
+def test_azure_passthrough_url_prefers_the_callers_api_version_over_a_full_url_api_bases():
+    url = _full_url_complete_url(request_query_params={"api-version": "2025-04-01-preview"})
+
+    assert str(url) == (
+        "https://my-resource.openai.azure.com/openai/deployments/gpt-4.1-mini/chat/completions"
+        "?api-version=2025-04-01-preview"
+    )
+
+
+def test_azure_passthrough_url_keeps_a_full_url_api_bases_api_version_when_the_caller_sends_none():
+    url = _full_url_complete_url(request_query_params={})
+
+    assert url.params["api-version"] == "2024-10-21"
+
+
+def test_azure_passthrough_url_strips_the_leading_router_model_segment():
+    url, _ = AzurePassthroughConfig().get_complete_url(
+        api_base="https://my-resource.openai.azure.com",
+        api_key="key",
+        model="gpt-4.1-mini",
+        endpoint="gpt-4.1-mini/openai/deployments/gpt-4.1-mini/chat/completions",
+        request_query_params={"api-version": "2024-10-21"},
+        litellm_params={},
+    )
+
+    assert (
+        str(url)
+        == "https://my-resource.openai.azure.com/openai/deployments/gpt-4.1-mini/chat/completions?api-version=2024-10-21"
+    )
+
+
+def test_azure_passthrough_url_rewrites_the_model_group_only_as_a_whole_segment():
+    url, _ = AzurePassthroughConfig().get_complete_url(
+        api_base="https://my-resource.openai.azure.com",
+        api_key="key",
+        model="gpt-4.1-mini",
+        endpoint="gpt/openai/deployments/gpt-4.1-mini/chat/completions",
+        request_query_params={"api-version": "2024-10-21"},
+        litellm_params={"litellm_metadata": {"model_group": "gpt"}},
+    )
+
+    assert (
+        str(url)
+        == "https://my-resource.openai.azure.com/openai/deployments/gpt-4.1-mini/chat/completions?api-version=2024-10-21"
+    )
+
+
+@pytest.mark.parametrize(
+    "request_data, expected",
+    [({"stream": True}, True), ({"stream": 1}, True), ({"stream": False}, False), ({}, False)],
+)
+def test_azure_passthrough_is_streaming_request_reads_the_stream_flag(request_data, expected):
+    assert (
+        AzurePassthroughConfig().is_streaming_request(
+            endpoint="openai/deployments/x/chat/completions", request_data=request_data
+        )
+        is expected
+    )
+
+
+@pytest.mark.parametrize(
+    "endpoint, expected",
+    [
+        ("gpt/openai/deployments/gpt/chat/completions", None),
+        ("openai/deployments/gpt/chat/completions", None),
+        ("gpt/openai/deployments/gpt-5.4-mini/chat/completions", None),
+        ("gpt/openai/deployments/GPT-5.4-MINI/chat/completions", None),
+        ("gpt/openai/deployments/Gpt/chat/completions", "Gpt"),
+        ("gpt/models/chat/completions", None),
+        ("gpt/openai/deployments/gpt-5.4/chat/completions", "gpt-5.4"),
+        ("gpt/openai/deployments/other-group/chat/completions", "other-group"),
+        ("openai/deployments/victim/gpt/chat/completions", "victim"),
+        ("gpt/openai/deployments/GPT-5.4/chat/completions", "GPT-5.4"),
+    ],
+)
+def test_foreign_azure_deployment_names_a_segment_outside_the_group(endpoint, expected):
+    assert foreign_azure_deployment(endpoint, "gpt", lambda: frozenset({"gpt-5.4-mini"})) == expected
+
+
+def test_foreign_azure_deployment_skips_the_router_when_the_segment_is_the_group_itself():
+    def served_models():
+        raise AssertionError("the router must not be consulted for the group's own name")
+
+    assert foreign_azure_deployment("gpt/openai/deployments/gpt/chat/completions", "gpt", served_models) is None
+
+
+@pytest.mark.parametrize(
+    "endpoint, expected",
+    [
+        ("other-group/openai/deployments/other-group/chat/completions", "other-group"),
+        ("openai/deployments/gpt/chat/completions", "gpt"),
+        ("openai/deployments/my-azure-deployment/chat/completions", None),
+        ("gpt", None),
+    ],
+)
+def test_azure_router_model_in_endpoint_picks_the_first_router_model_segment(endpoint, expected):
+    assert azure_router_model_in_endpoint(endpoint, frozenset({"gpt", "other-group"})) == expected

tests/test_litellm/llms/azure_ai/passthrough/test_azure_ai_passthrough_transformation.py

@@ -0,0 +1,617 @@
+import json
+from datetime import datetime
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.llms.azure_ai.passthrough.transformation import AzureAIPassthroughConfig
+from litellm.llms.base_llm.ocr.transformation import OCRResponse
+from litellm.types.rerank import RerankResponse
+from litellm.types.utils import EmbeddingResponse, ImageResponse, LlmProviders, ModelResponse
+from litellm.utils import ProviderConfigManager
+
+FOUNDRY_BASE = "https://my-resource.services.ai.azure.com"
+RESPONSES_COMPLETED_EVENT = {
+    "type": "response.completed",
+    "sequence_number": 2,
+    "response": {
+        "id": "resp_1",
+        "object": "response",
+        "created_at": 1,
+        "status": "completed",
+        "model": "gpt-5.4-mini",
+        "output": [
+            {
+                "type": "message",
+                "id": "msg_1",
+                "role": "assistant",
+                "status": "completed",
+                "content": [{"type": "output_text", "text": "hi", "annotations": []}],
+            }
+        ],
+        "usage": {"input_tokens": 1000, "output_tokens": 100, "total_tokens": 1100},
+    },
+}
+
+
+class _SpendProbe(CustomLogger):
+    logged_call_type: str | None = None
+    logged_cost: float | None = None
+
+    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        self.logged_call_type = kwargs["call_type"]
+        self.logged_cost = kwargs["response_cost"]
+
+
+@pytest.fixture(autouse=True)
+def clear_azure_ai_env(monkeypatch):
+    for env_var in ("AZURE_AI_API_BASE", "AZURE_AI_API_KEY", "AZURE_AD_TOKEN", "AZURE_API_KEY"):
+        monkeypatch.delenv(env_var, raising=False)
+    monkeypatch.setattr(litellm, "api_base", None)
+    monkeypatch.setattr(litellm, "api_key", None)
+
+
+def test_provider_config_manager_resolves_azure_ai_passthrough_config():
+    config = ProviderConfigManager.get_provider_passthrough_config(
+        model="Cohere-parse-v5", provider=LlmProviders.AZURE_AI
+    )
+
+    assert isinstance(config, AzureAIPassthroughConfig)
+
+
+def test_router_model_prefix_is_stripped_and_native_path_kept_verbatim():
+    url, base = AzureAIPassthroughConfig().get_complete_url(
+        api_base=FOUNDRY_BASE,
+        api_key=None,
+        model="Cohere-parse-v5",
+        endpoint="Cohere-parse-v5/providers/cohere/v2/parse",
+        request_query_params=None,
+        litellm_params={},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/providers/cohere/v2/parse"
+    assert base == FOUNDRY_BASE
+
+
+def test_model_group_prefix_is_stripped_when_router_metadata_names_it():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=FOUNDRY_BASE,
+        api_key=None,
+        model="Cohere-parse-v5",
+        endpoint="/parse-alias/providers/cohere/v2/parse",
+        request_query_params=None,
+        litellm_params={"litellm_metadata": {"model_group": "parse-alias"}},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/providers/cohere/v2/parse"
+
+
+def test_model_inside_the_path_stays_and_query_params_are_forwarded():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=f"{FOUNDRY_BASE}/",
+        api_key=None,
+        model="gpt-5.4-mini",
+        endpoint="openai/deployments/gpt-5.4-mini/chat/completions",
+        request_query_params={"api-version": "2024-10-21"},
+        litellm_params={},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/openai/deployments/gpt-5.4-mini/chat/completions?api-version=2024-10-21"
+
+
+def test_api_base_that_already_ends_in_models_is_cut_back_to_the_foundry_root():
+    url, base = AzureAIPassthroughConfig().get_complete_url(
+        api_base=f"{FOUNDRY_BASE}/models",
+        api_key="key",
+        model="gpt-5.4-mini",
+        endpoint="gpt-5.4-mini/models/chat/completions",
+        request_query_params={"api-version": "2024-05-01-preview"},
+        litellm_params={},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/models/chat/completions?api-version=2024-05-01-preview"
+    assert base == FOUNDRY_BASE
+
+
+def test_full_url_api_base_that_already_ends_with_the_native_path_is_not_doubled():
+    model_router_url = (
+        "https://my-resource.cognitiveservices.azure.com/openai/deployments/model-router/chat/completions"
+    )
+
+    url, base = AzureAIPassthroughConfig().get_complete_url(
+        api_base=f"{model_router_url}?api-version=2025-01-01-preview",
+        api_key="key",
+        model="model_router/model-router",
+        endpoint="model-router/chat/completions",
+        request_query_params=None,
+        litellm_params={"litellm_metadata": {"model_group": "model-router"}},
+    )
+
+    assert str(url) == f"{model_router_url}?api-version=2025-01-01-preview"
+    assert base == "https://my-resource.cognitiveservices.azure.com/openai/deployments/model-router"
+
+
+@pytest.mark.parametrize("relayed_deployment", ["gpt-4o", "GPT-4o"])
+def test_deployment_root_api_base_is_not_repeated_when_the_relay_carries_the_deployment_path(relayed_deployment):
+    url, base = AzureAIPassthroughConfig().get_complete_url(
+        api_base="https://my-resource.openai.azure.com/openai/deployments/gpt-4o",
+        api_key="key",
+        model="gpt-4o",
+        endpoint=f"aoai-gpt-4o/openai/deployments/{relayed_deployment}/chat/completions",
+        request_query_params={"api-version": "2024-10-21"},
+        litellm_params={"litellm_metadata": {"model_group": "aoai-gpt-4o"}},
+    )
+
+    assert str(url) == (
+        f"https://my-resource.openai.azure.com/openai/deployments/{relayed_deployment}/chat/completions"
+        "?api-version=2024-10-21"
+    )
+    assert base == "https://my-resource.openai.azure.com"
+
+
+def test_deployment_named_like_the_first_native_segment_keeps_its_deployment_root():
+    url, base = AzureAIPassthroughConfig().get_complete_url(
+        api_base="https://my-resource.openai.azure.com/openai/deployments/chat",
+        api_key="key",
+        model="chat",
+        endpoint="aoai-chat/chat/completions",
+        request_query_params={"api-version": "2024-10-21"},
+        litellm_params={"litellm_metadata": {"model_group": "aoai-chat"}},
+    )
+
+    assert str(url) == "https://my-resource.openai.azure.com/openai/deployments/chat/chat/completions?api-version=2024-10-21"
+    assert base == "https://my-resource.openai.azure.com/openai/deployments/chat"
+
+
+def test_parse_relay_under_a_models_api_base_targets_the_foundry_root():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=f"{FOUNDRY_BASE}/models",
+        api_key="key",
+        model="Cohere-parse-v5",
+        endpoint="Cohere-parse-v5/providers/cohere/v2/parse",
+        request_query_params=None,
+        litellm_params={},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/providers/cohere/v2/parse"
+
+
+def test_deployment_api_version_fills_in_when_the_caller_sends_none():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=FOUNDRY_BASE,
+        api_key="key",
+        model="gpt-5.4-mini",
+        endpoint="gpt-5.4-mini/models/chat/completions",
+        request_query_params=None,
+        litellm_params={"api_version": "2024-05-01-preview"},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/models/chat/completions?api-version=2024-05-01-preview"
+
+
+def test_callers_api_version_beats_the_deployments():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=FOUNDRY_BASE,
+        api_key="key",
+        model="gpt-5.4-mini",
+        endpoint="gpt-5.4-mini/models/chat/completions",
+        request_query_params={"api-version": "2025-04-01-preview"},
+        litellm_params={"api_version": "2024-05-01-preview"},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/models/chat/completions?api-version=2025-04-01-preview"
+
+
+def test_api_version_on_the_configured_api_base_is_the_last_fallback():
+    url, _ = AzureAIPassthroughConfig().get_complete_url(
+        api_base=f"{FOUNDRY_BASE}/models/chat/completions?api-version=2024-05-01-preview",
+        api_key="key",
+        model="gpt-5.4-mini",
+        endpoint="gpt-5.4-mini/models/chat/completions",
+        request_query_params=None,
+        litellm_params={},
+    )
+
+    assert str(url) == f"{FOUNDRY_BASE}/models/chat/completions?api-version=2024-05-01-preview"
+
+
+def test_missing_api_base_raises_instead_of_building_a_relative_url():
+    with pytest.raises(ValueError, match="AZURE_AI_API_BASE"):
+        AzureAIPassthroughConfig().get_complete_url(
+            api_base=None,
+            api_key=None,
+            model="Cohere-parse-v5",
+            endpoint="Cohere-parse-v5/providers/cohere/v2/parse",
+            request_query_params=None,
+            litellm_params={},
+        )
+
+
+def _auth_headers(api_key: str | None, api_base: str, litellm_params: dict | None = None) -> dict:
+    return AzureAIPassthroughConfig().validate_environment(
+        headers={"content-type": "application/json"},
+        model="Cohere-parse-v5",
+        messages=[],
+        optional_params={},
+        litellm_params=litellm_params or {},
+        api_key=api_key,
+        api_base=api_base,
+    )
+
+
+def test_foundry_host_gets_the_api_key_header():
+    headers = _auth_headers(api_key="deployment-key", api_base=FOUNDRY_BASE)
+
+    assert headers == {"content-type": "application/json", "api-key": "deployment-key"}
+
+
+def test_serverless_host_gets_a_bearer_token():
+    headers = _auth_headers(api_key="deployment-key", api_base="https://cohere-parse.eastus.models.ai.azure.com")
+
+    assert headers["Authorization"] == "Bearer deployment-key"
+    assert "api-key" not in headers
+
+
+def test_entra_token_is_used_when_the_deployment_has_no_api_key():
+    headers = _auth_headers(api_key=None, api_base=FOUNDRY_BASE, litellm_params={"azure_ad_token": "entra-token"})
+
+    assert headers["Authorization"] == "Bearer entra-token"
+
+
+def test_no_credentials_at_all_raises():
+    with pytest.raises(ValueError, match="Missing Azure AI credentials"):
+        _auth_headers(api_key=None, api_base=FOUNDRY_BASE)
+
+
+@pytest.mark.parametrize(
+    "request_data, expected",
+    [({"stream": True}, True), ({"stream": 1}, True), ({"stream": False}, False), ({}, False)],
+)
+def test_is_streaming_request_reads_the_stream_flag(request_data, expected):
+    assert (
+        AzureAIPassthroughConfig().is_streaming_request(endpoint="models/chat/completions", request_data=request_data)
+        is expected
+    )
+
+
+def _chat_completion_response() -> httpx.Response:
+    body = {
+        "id": "chatcmpl-1",
+        "object": "chat.completion",
+        "created": 1700000000,
+        "model": "gpt-5.4-mini",
+        "choices": [{"index": 0, "message": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
+        "usage": {"prompt_tokens": 10, "completion_tokens": 8, "total_tokens": 18},
+    }
+    return httpx.Response(
+        status_code=200,
+        headers={"content-type": "application/json"},
+        content=json.dumps(body).encode("utf-8"),
+        request=httpx.Request("POST", f"{FOUNDRY_BASE}/models/chat/completions"),
+    )
+
+
+def test_chat_completions_relay_yields_a_model_response_for_cost_tracking():
+    result = AzureAIPassthroughConfig().logging_non_streaming_response(
+        model="gpt-5.4-mini",
+        custom_llm_provider="azure_ai",
+        httpx_response=_chat_completion_response(),
+        request_data={"model": "gpt-5.4-mini", "messages": [{"role": "user", "content": "hi"}]},
+        logging_obj=MagicMock(),
+        endpoint="models/chat/completions",
+    )
+
+    assert isinstance(result, ModelResponse)
+    assert result.choices[0].message.content == "hi"
+    assert result.usage.prompt_tokens == 10
+    assert result.usage.completion_tokens == 8
+
+
+def _non_chat_logging_result(content: bytes, content_type: str):
+    parse_response = httpx.Response(
+        status_code=200,
+        headers={"content-type": content_type},
+        content=content,
+        request=httpx.Request("POST", f"{FOUNDRY_BASE}/providers/cohere/v2/parse"),
+    )
+    return AzureAIPassthroughConfig().logging_non_streaming_response(
+        model="Cohere-parse-v5",
+        custom_llm_provider="azure_ai",
+        httpx_response=parse_response,
+        request_data={"model": "Cohere-parse-v5"},
+        logging_obj=MagicMock(),
+        endpoint="providers/cohere/v2/parse",
+    )
+
+
+def test_non_chat_relay_with_a_non_json_body_logs_the_raw_text():
+    assert _non_chat_logging_result(b"page one", "text/plain") == {"response": "page one"}
+
+
+def _relay_logging_obj(
+    model: str,
+    api_base: str,
+    stream: bool = False,
+    callbacks: list[CustomLogger] | None = None,
+    endpoint: str = "",
+) -> Logging:
+    logging_obj = Logging(
+        model=model,
+        messages=[],
+        stream=stream,
+        call_type="allm_passthrough_route",
+        start_time=datetime.now(),
+        litellm_call_id="call-1",
+        function_id="fn-1",
+        dynamic_async_success_callbacks=callbacks,
+    )
+    logging_obj.update_environment_variables(
+        model=model,
+        litellm_params={"api_base": api_base, "custom_llm_provider": "azure_ai"},
+        optional_params={},
+        custom_llm_provider="azure_ai",
+        endpoint=endpoint,
+    )
+    return logging_obj
+
+
+def _relay_logging_result(
+    config: AzureAIPassthroughConfig,
+    model: str,
+    native_path: str,
+    body,
+    api_base: str = FOUNDRY_BASE,
+    status_code: int = 200,
+):
+    relayed_url = f"{FOUNDRY_BASE}/{native_path}?api-version=2024-05-01-preview"
+    logging_obj = _relay_logging_obj(model, api_base)
+    response = httpx.Response(
+        status_code=status_code,
+        headers={"content-type": "application/json"},
+        content=json.dumps(body).encode("utf-8"),
+        request=httpx.Request("POST", relayed_url),
+    )
+    result = config.logging_non_streaming_response(
+        model=model,
+        custom_llm_provider="azure_ai",
+        httpx_response=response,
+        request_data={"model": model},
+        logging_obj=logging_obj,
+        endpoint=f"{model}/{native_path}",
+    )
+    return result, logging_obj
+
+
+MISTRAL_OCR_BODY = {
+    "pages": [{"index": 0, "markdown": "page one"}, {"index": 1, "markdown": "page two"}],
+    "model": "mistral-document-ai-2512",
+    "usage_info": {"pages_processed": 2, "doc_size_bytes": 4321},
+}
+
+
+def test_mistral_document_ai_relay_is_costed_per_page():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "mistral-document-ai-2512", "providers/mistral/azure/ocr", MISTRAL_OCR_BODY
+    )
+    per_page = litellm.get_model_info("azure_ai/mistral-document-ai-2512")["ocr_cost_per_page"]
+
+    assert isinstance(result, OCRResponse)
+    assert result.usage_info.pages_processed == 2
+    assert per_page > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(2 * per_page)
+
+
+def test_ocr_route_under_a_models_api_base_is_still_recognised():
+    result, _ = _relay_logging_result(
+        AzureAIPassthroughConfig(),
+        "mistral-document-ai-2512",
+        "providers/mistral/azure/ocr",
+        MISTRAL_OCR_BODY,
+        api_base=f"{FOUNDRY_BASE}/models",
+    )
+
+    assert isinstance(result, OCRResponse)
+
+
+def test_relay_to_a_non_ocr_route_keeps_the_passthrough_object_and_call_type():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "mistral-document-ai-2512", "models/info", {"name": "mistral-document-ai-2512"}
+    )
+
+    assert result == {"response": {"name": "mistral-document-ai-2512"}}
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+COHERE_PARSE_BODY = {"id": "parse-1", "pages": [], "meta": {"billed_units": {"pages": 3}}}
+
+
+def test_cohere_parse_relay_is_costed_per_billed_page():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "Cohere-parse-v5", "providers/cohere/v2/parse", COHERE_PARSE_BODY
+    )
+    per_page = litellm.get_model_info("azure_ai/Cohere-parse-v5")["ocr_cost_per_page"]
+
+    assert isinstance(result, OCRResponse)
+    assert result.usage_info.pages_processed == 3
+    assert logging_obj.call_type == "aocr"
+    assert per_page > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(3 * per_page)
+
+
+def test_deployment_without_an_ocr_config_is_never_costed_as_ocr():
+    config = AzureAIPassthroughConfig(ocr_config_for=lambda model: None)
+    result, logging_obj = _relay_logging_result(
+        config, "mistral-document-ai-2512", "providers/mistral/azure/ocr", MISTRAL_OCR_BODY
+    )
+
+    assert result == {"response": MISTRAL_OCR_BODY}
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def test_accepted_ocr_job_without_a_result_body_is_not_costed():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(),
+        "mistral-document-ai-2512",
+        "providers/mistral/azure/ocr",
+        {"status": "running"},
+        status_code=202,
+    )
+
+    assert result == {"response": {"status": "running"}}
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def test_unparseable_ocr_body_falls_back_to_the_passthrough_object():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(),
+        "mistral-document-ai-2512",
+        "providers/mistral/azure/ocr",
+        ["not", "an", "ocr", "body"],
+    )
+
+    assert result == {"response": '["not", "an", "ocr", "body"]'}
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+EMBEDDINGS_BODY = {
+    "object": "list",
+    "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
+    "model": "embed-v-4-0",
+    "usage": {"prompt_tokens": 1200, "total_tokens": 1200},
+}
+
+RERANK_BODY = {
+    "id": "rerank-1",
+    "results": [{"index": 1, "relevance_score": 0.9}, {"index": 0, "relevance_score": 0.2}],
+    "meta": {"api_version": {"version": "2"}, "billed_units": {"search_units": 2}},
+}
+
+IMAGE_BODY = {"created": 1, "data": [{"b64_json": "AAAA"}]}
+
+
+def test_foundry_embeddings_relay_is_costed_per_input_token():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "embed-v-4-0", "models/embeddings", EMBEDDINGS_BODY
+    )
+    per_token = litellm.get_model_info("azure_ai/embed-v-4-0")["input_cost_per_token"]
+
+    assert isinstance(result, EmbeddingResponse)
+    assert logging_obj.call_type == "aembedding"
+    assert per_token > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(1200 * per_token)
+
+
+def test_cohere_rerank_relay_is_costed_per_search_unit():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "cohere-rerank-v4.0-fast", "providers/cohere/v2/rerank", RERANK_BODY
+    )
+    per_query = litellm.get_model_info("azure_ai/cohere-rerank-v4.0-fast")["input_cost_per_query"]
+
+    assert isinstance(result, RerankResponse)
+    assert logging_obj.call_type == "arerank"
+    assert per_query > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(2 * per_query)
+
+
+def test_image_generation_relay_is_costed_per_image():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "FLUX.2-pro", "openai/deployments/FLUX.2-pro/images/generations", IMAGE_BODY
+    )
+    per_image = litellm.get_model_info("azure_ai/FLUX.2-pro")["output_cost_per_image"]
+
+    assert isinstance(result, ImageResponse)
+    assert logging_obj.call_type == "aimage_generation"
+    assert per_image > 0
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(per_image)
+
+
+def test_flux_2_relay_through_the_provider_route_is_costed_per_image():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(), "FLUX.2-pro", "providers/blackforestlabs/v1/flux-2-pro", IMAGE_BODY
+    )
+    per_image = litellm.get_model_info("azure_ai/FLUX.2-pro")["output_cost_per_image"]
+
+    assert isinstance(result, ImageResponse)
+    assert logging_obj.call_type == "aimage_generation"
+    assert logging_obj._response_cost_calculator(result=result) == pytest.approx(per_image)
+
+
+def test_rejected_rerank_relay_keeps_the_passthrough_object_and_call_type():
+    result, logging_obj = _relay_logging_result(
+        AzureAIPassthroughConfig(),
+        "cohere-rerank-v4.0-fast",
+        "providers/cohere/v2/rerank",
+        {"message": "invalid request"},
+        status_code=400,
+    )
+
+    assert result == {"response": {"message": "invalid request"}}
+    assert logging_obj.call_type == "allm_passthrough_route"
+
+
+def test_streaming_chat_completion_chunks_are_costed_like_azure():
+    head = {"id": "chatcmpl-1", "object": "chat.completion.chunk", "created": 1, "model": "gpt-5.4-mini"}
+    chunks = [
+        "data: "
+        + json.dumps(
+            {
+                **head,
+                "choices": [{"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": "stop"}],
+            }
+        ),
+        "data: "
+        + json.dumps({**head, "choices": [], "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4}}),
+        "data: [DONE]",
+    ]
+
+    response = AzureAIPassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=chunks,
+        litellm_logging_obj=MagicMock(),
+        model="gpt-5.4-mini",
+        custom_llm_provider="azure_ai",
+        endpoint="chat/completions",
+    )
+
+    assert isinstance(response, ModelResponse)
+    assert response.choices[0].message.content == "hi"
+    assert response.usage.total_tokens == 4
+
+
+def test_streaming_responses_chunks_through_a_router_relay_are_costed_like_azure():
+    logging_obj = _relay_logging_obj("gpt-5.4-mini", FOUNDRY_BASE)
+
+    response = AzureAIPassthroughConfig().handle_logging_collected_chunks(
+        all_chunks=["event: response.completed", "data: " + json.dumps(RESPONSES_COMPLETED_EVENT)],
+        litellm_logging_obj=logging_obj,
+        model="gpt-5.4-mini",
+        custom_llm_provider="azure_ai",
+        endpoint="gpt/openai/responses",
+    )
+    info = litellm.get_model_info("azure_ai/gpt-5.4-mini")
+
+    assert response is not None
+    assert response.response.usage.output_tokens == 100
+    assert logging_obj.call_type == "aresponses"
+    assert logging_obj._response_cost_calculator(result=response.response) == pytest.approx(
+        1000 * info["input_cost_per_token"] + 100 * info["output_cost_per_token"]
+    )
+
+
+async def test_streaming_responses_relay_flush_reaches_the_success_callbacks_with_a_price():
+    probe = _SpendProbe()
+    logging_obj = _relay_logging_obj(
+        "gpt-5.4-mini", FOUNDRY_BASE, stream=True, callbacks=[probe], endpoint="gpt/openai/responses"
+    )
+    stream = "event: response.completed\ndata: " + json.dumps(RESPONSES_COMPLETED_EVENT) + "\n\n"
+
+    await logging_obj.async_flush_passthrough_collected_chunks(
+        raw_bytes=[stream.encode()], provider_config=AzureAIPassthroughConfig()
+    )
+    info = litellm.get_model_info("azure_ai/gpt-5.4-mini")
+
+    assert probe.logged_call_type == "allm_passthrough_route"
+    assert probe.logged_cost == pytest.approx(1000 * info["input_cost_per_token"] + 100 * info["output_cost_per_token"])

tests/test_litellm/passthrough/test_passthrough_main.py

@@ -873,3 +873,118 @@ def test_llm_passthrough_route_propagates_allm_passthrough_route_to_logging_obj(
 
     assert captured_litellm_params.get("allm_passthrough_route") is True
     assert LitellmLogging._is_sync_litellm_request(captured_litellm_params) is False
+
+
+FOUNDRY_BASE = "https://my-resource.services.ai.azure.com"
+
+
+def _foundry_parse_response() -> httpx.Response:
+    return httpx.Response(
+        status_code=200,
+        headers={"content-type": "application/json"},
+        content=b'{"id":"parse-1","pages":[]}',
+        request=httpx.Request("POST", f"{FOUNDRY_BASE}/providers/cohere/v2/parse"),
+    )
+
+
+def test_azure_ai_relay_reaches_the_deployment_with_its_own_credential():
+    """
+    Regression for LIT-7022: azure_ai had no passthrough config, so every
+    /azure_ai/<router-model>/<native-path> relay raised "Provider azure_ai not found"
+    before a request was built.
+    """
+    client = HTTPHandler()
+
+    with patch.object(client.client, "send", return_value=_foundry_parse_response()) as mock_send:
+        response = llm_passthrough_route(
+            model="azure_ai/Cohere-parse-v5",
+            endpoint="Cohere-parse-v5/providers/cohere/v2/parse",
+            method="POST",
+            custom_llm_provider="azure_ai",
+            api_base=FOUNDRY_BASE,
+            api_key="deployment-key",
+            json={"model": "Cohere-parse-v5", "document": {"type": "image_url", "image_url": "https://x/y.png"}},
+            client=client,
+            litellm_logging_obj=MagicMock(),
+        )
+
+    sent = mock_send.call_args.kwargs["request"]
+    assert str(sent.url) == f"{FOUNDRY_BASE}/providers/cohere/v2/parse"
+    assert sent.headers["api-key"] == "deployment-key"
+    assert json.loads(sent.content)["model"] == "Cohere-parse-v5"
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_router_relays_azure_ai_model_through_the_deployment_api_base():
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "foundry-parse",
+                "litellm_params": {
+                    "model": "azure_ai/Cohere-parse-v5",
+                    "api_base": FOUNDRY_BASE,
+                    "api_key": "deployment-key",
+                },
+            }
+        ]
+    )
+    async_client = AsyncHTTPHandler()
+
+    with patch.object(async_client.client, "send", AsyncMock(return_value=_foundry_parse_response())) as mock_send:
+        response = await router.allm_passthrough_route(
+            model="foundry-parse",
+            method="POST",
+            endpoint="foundry-parse/providers/cohere/v2/parse",
+            json={"model": "foundry-parse", "document": {"type": "image_url", "image_url": "https://x/y.png"}},
+            client=async_client,
+        )
+
+    sent = mock_send.call_args.kwargs["request"]
+    assert str(sent.url) == f"{FOUNDRY_BASE}/providers/cohere/v2/parse"
+    assert sent.headers["api-key"] == "deployment-key"
+    assert json.loads(sent.content)["model"] == "Cohere-parse-v5"
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_router_relays_an_openai_model_on_a_foundry_base_as_azure_ai(monkeypatch):
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://unrelated.openai.azure.com")
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "foundry-gpt",
+                "litellm_params": {
+                    "model": "azure_ai/gpt-5.4-mini",
+                    "api_base": FOUNDRY_BASE,
+                    "api_key": "deployment-key",
+                },
+            }
+        ]
+    )
+    async_client = AsyncHTTPHandler()
+    upstream = httpx.Response(
+        status_code=200,
+        headers={"content-type": "application/json"},
+        content=(
+            b'{"id":"chatcmpl-1","object":"chat.completion","model":"gpt-5.4-mini",'
+            b'"choices":[{"index":0,"finish_reason":"stop","message":{"role":"assistant","content":"hi"}}],'
+            b'"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}'
+        ),
+        request=httpx.Request("POST", f"{FOUNDRY_BASE}/models/chat/completions"),
+    )
+
+    with patch.object(async_client.client, "send", AsyncMock(return_value=upstream)) as mock_send:
+        await router.allm_passthrough_route(
+            model="foundry-gpt",
+            method="POST",
+            endpoint="foundry-gpt/models/chat/completions",
+            request_query_params={"api-version": "2024-05-01-preview"},
+            json={"model": "foundry-gpt", "messages": [{"role": "user", "content": "hi"}]},
+            client=async_client,
+        )
+
+    sent = mock_send.call_args.kwargs["request"]
+    assert str(sent.url) == f"{FOUNDRY_BASE}/models/chat/completions?api-version=2024-05-01-preview"
+    assert sent.headers["api-key"] == "deployment-key"
+    assert json.loads(sent.content)["model"] == "gpt-5.4-mini"

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -565,6 +565,38 @@ def test_get_model_from_request_bedrock_unparseable_endpoint_keeps_body_model():
     )
 
 
+def _azure_relay_router():
+    from litellm.router import Router
+
+    return Router(
+        model_list=[
+            {
+                "model_name": "gpt",
+                "litellm_params": {"model": "azure_ai/gpt-5.4-mini", "api_base": "https://a.services.ai.azure.com", "api_key": "k"},
+            },
+            {
+                "model_name": "other-group",
+                "litellm_params": {"model": "azure/gpt-5.4", "api_base": "https://b.openai.azure.com", "api_key": "k"},
+            },
+        ]
+    )
+
+
+@pytest.mark.parametrize(
+    "route, request_data, expected",
+    [
+        ("/azure_ai/other-group/openai/deployments/other-group/chat/completions", {"model": "gpt"}, "other-group"),
+        ("/azure_ai/other-group/models/chat/completions", {}, "other-group"),
+        ("/azure/openai/deployments/gpt/chat/completions", {"model": "other-group"}, "gpt"),
+        ("/azure/openai/deployments/gpt/chat/completions", {}, "gpt"),
+        ("/azure/openai/deployments/my-azure-deployment/chat/completions", {"model": "gpt"}, "gpt"),
+        ("/azure_ai/gpt", {"model": "other-group"}, "other-group"),
+    ],
+)
+def test_get_model_from_request_azure_relay_routes_use_the_model_group_in_the_path(route, request_data, expected):
+    assert get_model_from_request(request_data=request_data, route=route, llm_router=_azure_relay_router()) == expected
+
+
 def test_get_model_from_request_includes_file_endpoint_header_model():
     assert (
         get_model_from_request(

tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_openai_passthrough_logging_handler.py

@@ -9,8 +9,10 @@ import pytest
 
 import litellm
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.litellm_core_utils.token_counter import high_detail_image_token_upper_bound
 from litellm.proxy.pass_through_endpoints.llm_provider_handlers.openai_passthrough_logging_handler import (
     OpenAIPassthroughLoggingHandler,
+    count_relayed_prompt_tokens,
 )
 from litellm.proxy.pass_through_endpoints.success_handler import (
     PassThroughEndpointLogging,
@@ -2037,3 +2039,56 @@ class TestOpenAIPassthroughEmbeddingsSpendLog:
 
 if __name__ == "__main__":
     pytest.main([__file__])
+
+
+ONE_PIXEL_PNG_DATA_URL = (
+    "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
+)
+UNREACHABLE_IMAGE_URL = "http://127.0.0.1:9/doc.png"
+TEXT_ONLY_MESSAGES = [{"role": "user", "content": [{"type": "text", "text": "Describe this"}]}]
+
+
+def _image_messages(url: str, detail: str) -> list[dict]:
+    return [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "Describe this"},
+                {"type": "image_url", "image_url": {"url": url, "detail": detail}},
+            ],
+        }
+    ]
+
+
+def test_count_relayed_prompt_tokens_counts_a_data_url_image_exactly():
+    messages = _image_messages(ONE_PIXEL_PNG_DATA_URL, "high")
+
+    assert count_relayed_prompt_tokens("gpt-4.1-mini", messages) == litellm.token_counter(
+        model="gpt-4.1-mini", messages=messages
+    )
+
+
+def test_count_relayed_prompt_tokens_keeps_a_low_detail_remote_image_at_the_base_count():
+    messages = _image_messages(UNREACHABLE_IMAGE_URL, "low")
+
+    assert count_relayed_prompt_tokens("gpt-4.1-mini", messages) == litellm.token_counter(
+        model="gpt-4.1-mini", messages=messages
+    )
+    assert count_relayed_prompt_tokens("gpt-4.1-mini", messages) < high_detail_image_token_upper_bound()
+
+
+def test_count_relayed_prompt_tokens_charges_only_the_remote_high_detail_image_at_the_upper_bound():
+    messages = _image_messages(UNREACHABLE_IMAGE_URL, "high")
+
+    assert count_relayed_prompt_tokens("gpt-4.1-mini", messages) == (
+        litellm.token_counter(model="gpt-4.1-mini", messages=TEXT_ONLY_MESSAGES) + high_detail_image_token_upper_bound()
+    )
+
+
+@pytest.mark.parametrize("scheme", ["HTTPS://", "Http://"])
+def test_count_relayed_prompt_tokens_charges_an_uppercase_scheme_remote_high_detail_image_at_the_upper_bound(scheme):
+    messages = _image_messages(scheme + UNREACHABLE_IMAGE_URL.split("://", 1)[1], "high")
+
+    assert count_relayed_prompt_tokens("gpt-4.1-mini", messages) == (
+        litellm.token_counter(model="gpt-4.1-mini", messages=TEXT_ONLY_MESSAGES) + high_detail_image_token_upper_bound()
+    )

tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -5022,7 +5022,11 @@ class TestPassthroughRouterModelBudgetReservation:
 
         monkeypatch.setattr(proxy_server, "llm_router", RecordingRouter())
         monkeypatch.setattr(ep, "get_request_body", fake_get_request_body)
-        monkeypatch.setattr(ep, "is_passthrough_request_using_router_model", lambda *a, **k: True)
+        monkeypatch.setattr(
+            ep,
+            "is_passthrough_request_using_router_model",
+            lambda request_body, llm_router=None: request_body.get("model") in ("gpt-5", "router-model"),
+        )
         return captured
 
     def _assert_metadata_carries_attribution(self, captured: list[dict], user_api_key_dict: UserAPIKeyAuth) -> None:
@@ -5121,7 +5125,11 @@ class TestAzureRouterModelStreamingDispatch:
 
         monkeypatch.setattr(proxy_server, "llm_router", StreamingRouter())
         monkeypatch.setattr(ep, "get_request_body", fake_get_request_body)
-        monkeypatch.setattr(ep, "is_passthrough_request_using_router_model", lambda *a, **k: True)
+        monkeypatch.setattr(
+            ep,
+            "is_passthrough_request_using_router_model",
+            lambda request_body, llm_router=None: request_body.get("model") in ("gpt-5", "router-model"),
+        )
 
         request = MagicMock(spec=Request)
         request.method = "POST"
@@ -5181,7 +5189,11 @@ class TestAzureRouterModelStreamingKeepalive:
 
         monkeypatch.setattr(proxy_server, "llm_router", StreamingRouter())
         monkeypatch.setattr(ep, "get_request_body", fake_get_request_body)
-        monkeypatch.setattr(ep, "is_passthrough_request_using_router_model", lambda *a, **k: True)
+        monkeypatch.setattr(
+            ep,
+            "is_passthrough_request_using_router_model",
+            lambda request_body, llm_router=None: request_body.get("model") in ("gpt-5", "router-model"),
+        )
 
         request = MagicMock(spec=Request)
         request.method = "POST"
@@ -5231,6 +5243,97 @@ class TestAzureRouterModelStreamingKeepalive:
         assert chunks == [b"data: hello\n\n"]
 
 
+class TestRouterModelRelayUpstreamContract:
+    def _request(self, content_type: str) -> MagicMock:
+        request = MagicMock(spec=Request)
+        request.method = "POST"
+        request.headers = {"content-type": content_type}
+        request.query_params = {}
+        return request
+
+    def _install_router(self, monkeypatch, router, body: dict) -> None:
+        import litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints as ep
+        import litellm.proxy.proxy_server as proxy_server
+
+        async def fake_get_request_body(_request):
+            return body
+
+        monkeypatch.setattr(proxy_server, "llm_router", router)
+        monkeypatch.setattr(ep, "get_request_body", fake_get_request_body)
+        monkeypatch.setattr(
+            ep,
+            "is_passthrough_request_using_router_model",
+            lambda request_body, llm_router=None: request_body.get("model") in ("gpt-5", "router-model"),
+        )
+
+    def _recording_router(self, captured: list[dict]):
+        class RecordingRouter:
+            async def allm_passthrough_route(self, **kwargs):
+                captured.append(kwargs)
+                return httpx.Response(200, json={"ok": True})
+
+        return RecordingRouter()
+
+    @pytest.mark.asyncio
+    async def test_azure_relay_keeps_the_json_body_when_the_content_type_carries_a_charset(self, monkeypatch):
+        body = {"model": "gpt-5", "messages": [{"role": "user", "content": "hi"}]}
+        captured: list[dict] = []
+        self._install_router(monkeypatch, self._recording_router(captured), body)
+
+        await azure_proxy_route(
+            endpoint="openai/deployments/gpt-5/chat/completions",
+            request=self._request("application/json; charset=utf-8"),
+            fastapi_response=MagicMock(spec=Response),
+            user_api_key_dict=UserAPIKeyAuth(api_key="hashed-token"),
+        )
+
+        assert captured[0]["json"] == body
+
+    @pytest.mark.asyncio
+    async def test_vllm_relay_keeps_the_json_body_when_the_content_type_carries_a_charset(self, monkeypatch):
+        body = {"model": "router-model", "messages": [{"role": "user", "content": "hi"}]}
+        captured: list[dict] = []
+        self._install_router(monkeypatch, self._recording_router(captured), body)
+
+        await vllm_proxy_route(
+            endpoint="/chat/completions",
+            request=self._request("application/json; charset=utf-8"),
+            fastapi_response=MagicMock(spec=Response),
+            user_api_key_dict=UserAPIKeyAuth(api_key="hashed-token"),
+        )
+
+        assert captured[0]["json"] == body
+
+    @pytest.mark.asyncio
+    async def test_azure_relay_returns_the_upstream_status_and_body_when_the_deployment_rejects_the_call(
+        self, monkeypatch
+    ):
+        upstream_body = {"error": {"code": "DeploymentNotFound", "message": "The API deployment does not exist."}}
+
+        class RejectingRouter:
+            async def allm_passthrough_route(self, **kwargs):
+                upstream_request = httpx.Request(
+                    "POST", "https://my-azure.openai.azure.com/openai/deployments/gpt-5/chat/completions"
+                )
+                upstream = httpx.Response(
+                    404, json=upstream_body, headers={"x-ms-request-id": "req-1"}, request=upstream_request
+                )
+                raise httpx.HTTPStatusError("404", request=upstream_request, response=upstream)
+
+        self._install_router(monkeypatch, RejectingRouter(), {"model": "gpt-5", "stream": False})
+
+        result = await azure_proxy_route(
+            endpoint="openai/deployments/gpt-5/chat/completions",
+            request=self._request("application/json"),
+            fastapi_response=MagicMock(spec=Response),
+            user_api_key_dict=UserAPIKeyAuth(api_key="hashed-token"),
+        )
+
+        assert result.status_code == 404
+        assert json.loads(result.body) == upstream_body
+        assert result.headers["x-ms-request-id"] == "req-1"
+
+
 @pytest.mark.asyncio
 async def test_bedrock_count_tokens_error_forwards_provider_headers():
     """The count tokens route converts BedrockError into an HTTPException, and dropping the
@@ -5263,3 +5366,88 @@ async def test_bedrock_count_tokens_error_forwards_provider_headers():
 
     assert exc_info.value.status_code == 500
     assert exc_info.value.headers["llm_provider-x-amzn-requestid"] == "req-count-tokens-500"
+
+
+class _AzureGroupRouter:
+    def __init__(self, captured: list[dict]) -> None:
+        self.captured = captured
+
+    def get_model_names(self, team_id=None):
+        return ["gpt", "other-group"]
+
+    def get_model_list(self, model_name=None, team_id=None):
+        rows = [
+            {"model_name": "gpt", "litellm_params": {"model": "azure_ai/gpt-5.4-mini", "api_key": "k"}},
+            {"model_name": "other-group", "litellm_params": {"model": "azure/gpt-5.4", "api_key": "k"}},
+        ]
+        return [row for row in rows if model_name is None or row["model_name"] == model_name]
+
+    async def allm_passthrough_route(self, **kwargs):
+        self.captured.append(kwargs)
+        return httpx.Response(200, json={"ok": True})
+
+
+class TestAzureRelayDeploymentSegment:
+    """A key allowed one model group must not reach another deployment by naming it in the
+    ``openai/deployments/<x>`` segment while the group segment picks the credential."""
+
+    def test_models_served_by_group_resolves_each_deployment_to_its_model_name(self):
+        from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import _models_served_by_group
+
+        assert _models_served_by_group(_AzureGroupRouter([]), "gpt") == frozenset({"gpt-5.4-mini"})
+        assert _models_served_by_group(_AzureGroupRouter([]), "missing-group") == frozenset()
+
+    def _install(self, monkeypatch, body: dict) -> list[dict]:
+        import litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints as ep
+        import litellm.proxy.proxy_server as proxy_server
+
+        captured: list[dict] = []
+
+        async def fake_get_request_body(_request):
+            return body
+
+        monkeypatch.setattr(proxy_server, "llm_router", _AzureGroupRouter(captured))
+        monkeypatch.setattr(ep, "get_request_body", fake_get_request_body)
+        return captured
+
+    def _request(self) -> Request:
+        request = MagicMock(spec=Request)
+        request.method = "POST"
+        request.headers = {"content-type": "application/json"}
+        request.query_params = {}
+        return request
+
+    @pytest.mark.asyncio
+    async def test_azure_relay_rejects_a_deployment_the_group_does_not_serve(self, monkeypatch):
+        from fastapi import HTTPException
+
+        captured = self._install(monkeypatch, {"model": "gpt", "messages": []})
+
+        with pytest.raises(HTTPException) as exc_info:
+            await azure_proxy_route(
+                endpoint="gpt/openai/deployments/gpt-5.4/chat/completions",
+                request=self._request(),
+                fastapi_response=MagicMock(spec=Response),
+                user_api_key_dict=UserAPIKeyAuth(api_key="hashed-token", models=["gpt"]),
+            )
+
+        assert exc_info.value.status_code == 400
+        assert "gpt-5.4" in exc_info.value.detail["error"]
+        assert captured == []
+
+    @pytest.mark.asyncio
+    async def test_azure_relay_dispatches_the_group_and_its_own_deployment_name(self, monkeypatch):
+        captured = self._install(monkeypatch, {"model": "gpt", "messages": []})
+
+        for endpoint in (
+            "gpt/openai/deployments/gpt/chat/completions",
+            "gpt/openai/deployments/gpt-5.4-mini/chat/completions",
+        ):
+            await azure_proxy_route(
+                endpoint=endpoint,
+                request=self._request(),
+                fastapi_response=MagicMock(spec=Response),
+                user_api_key_dict=UserAPIKeyAuth(api_key="hashed-token", models=["gpt"]),
+            )
+
+        assert [call["model"] for call in captured] == ["gpt", "gpt"]

tests/test_litellm/router_utils/test_router_utils_common_utils.py

@@ -12,6 +12,7 @@ from litellm.router_utils.common_utils import (
     add_model_file_id_mappings,
     filter_team_based_models,
     filter_web_search_deployments,
+    provider_for_generic_call,
     resolve_model_group_alias,
     truncate_fallback_error_detail,
     PROVIDER_SCOPED_CREDENTIAL_PARAMS,
@@ -756,3 +757,20 @@ class TestWarnOnProviderCredentialMismatch:
             )
             is None
         )
+
+
+@pytest.mark.parametrize(
+    ("litellm_params", "expected"),
+    [
+        ({"model": "azure_ai/gpt-5.4-mini", "custom_llm_provider": "azure"}, "azure"),
+        ({"model": "azure_ai/gpt-5.4-mini", "api_base": "https://my-resource.openai.azure.com"}, "azure_ai"),
+        ({"model": "cohere/command-r"}, "cohere"),
+        ({"model": "gpt-5.4-mini"}, "openai"),
+        ({"model": "no-provider-knows-this-model"}, None),
+        ({"api_base": "https://my-resource.openai.azure.com"}, None),
+    ],
+    ids=["declared_wins", "prefix_beats_host_flip", "prefix_beats_cohere_chat_flip", "unprefixed_inferred", "unknown", "no_model"],
+)
+def test_provider_for_generic_call(litellm_params, expected, monkeypatch):
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://unrelated.openai.azure.com")
+    assert provider_for_generic_call(litellm_params) == expected

tests/test_litellm/test_router.py

@@ -5049,6 +5049,41 @@ def test_get_deployment_model_info_base_model_merge_priority():
     print("✓ Base model merge priority test passed!")
 
 
+@pytest.mark.parametrize(
+    "model, litellm_params, endpoint, expected",
+    [
+        (
+            "gpt",
+            {"model": "azure_ai/gpt-5.4-mini", "api_base": "https://my-resource.services.ai.azure.com", "api_key": "key"},
+            "gpt/openai/deployments/gpt-5.4-mini/chat/completions",
+            "gpt-5.4-mini/openai/deployments/gpt-5.4-mini/chat/completions",
+        ),
+        (
+            "aws/anthropic/bedrock-claude",
+            {"model": "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"},
+            "/model/aws/anthropic/bedrock-claude/invoke",
+            "/model/us.anthropic.claude-haiku-4-5-20251001-v1:0/invoke",
+        ),
+        (
+            "my-gemini",
+            {"model": "gemini/gemini-3.1-pro-preview", "api_key": "key"},
+            "v1beta/models/my-gemini:streamGenerateContent",
+            "v1beta/models/gemini-3.1-pro-preview:streamGenerateContent",
+        ),
+    ],
+)
+def test_add_deployment_model_to_endpoint_rewrites_the_model_group_only_as_whole_path_segments(
+    model, litellm_params, endpoint, expected
+):
+    router = litellm.Router(model_list=[{"model_name": model, "litellm_params": litellm_params}])
+
+    result = router._add_deployment_model_to_endpoint_for_llm_passthrough_route(
+        kwargs={"endpoint": endpoint}, model=model, model_name=litellm_params["model"]
+    )
+
+    assert result["endpoint"] == expected
+
+
 def test_add_deployment_model_to_endpoint_for_llm_passthrough_route():
     """
     Test that _add_deployment_model_to_endpoint_for_llm_passthrough_route correctly strips bedrock provider prefix
@@ -14137,6 +14172,49 @@ async def test_router_retry_policy_controls_upstream_attempt_count(
     assert upstream.call_count == expected_upstream_calls
 
 
+@pytest.mark.asyncio
+async def test_generic_call_keeps_the_deployment_name_of_an_azure_ai_model_on_an_azure_openai_host(monkeypatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": "aoai-gpt",
+                "litellm_params": {
+                    "model": "azure_ai/gpt-5.4-mini",
+                    "api_base": "https://my-resource.openai.azure.com",
+                    "api_key": "deployment-key",
+                },
+            }
+        ]
+    )
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream = respx_mock.post(host="my-resource.openai.azure.com", path__regex=r"^/openai/.*responses$").mock(
+            return_value=httpx.Response(
+                200,
+                json={
+                    "id": "resp_1",
+                    "object": "response",
+                    "created_at": 1,
+                    "status": "completed",
+                    "model": "gpt-5.4-mini",
+                    "output": [
+                        {
+                            "type": "message",
+                            "id": "msg_1",
+                            "role": "assistant",
+                            "status": "completed",
+                            "content": [{"type": "output_text", "text": "hi", "annotations": []}],
+                        }
+                    ],
+                },
+            )
+        )
+        await router.aresponses(model="aoai-gpt", input="hi")
+
+    assert json.loads(upstream.calls.last.request.content)["model"] == "gpt-5.4-mini"
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "retry_policy,upstream_error",