refactor(azure): move the passthrough deployment-segment helpers under llms

litellm/llms/azure/passthrough/transformation.py

@@ -1,4 +1,5 @@
-from collections.abc import Mapping, Sequence
+import re
+from collections.abc import Callable, Collection, Mapping, Sequence
 from typing import TYPE_CHECKING, Final, Optional
 
 import httpx
@@ -65,6 +66,26 @@ def logged_responses_stream(all_chunks: Sequence[str], logging_obj: Logging) ->
     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)
+    return None if deployment == model_group or deployment in served_models() 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")

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 (
@@ -2011,9 +2012,10 @@ def get_model_from_request(
 
 
 def _router_model_from_azure_route(route: str, llm_router: Router | None) -> str | None:
-    from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import azure_router_model_in_endpoint
-
-    return azure_router_model_in_endpoint(re.sub(r"^/azure(?:_ai)?/", "", route, flags=re.IGNORECASE), llm_router)
+    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:

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -33,6 +33,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
@@ -121,16 +122,6 @@ def is_passthrough_request_using_router_model(request_body: dict, llm_router: li
         return False
 
 
-def azure_router_model_in_endpoint(endpoint: str, llm_router: litellm.Router | None) -> str | None:
-    parts: Final = endpoint.split("/")
-    if len(parts) < 2:
-        return None
-    return next((part for part in parts if is_known_model(part, llm_router)), None)
-
-
-AZURE_DEPLOYMENT_SEGMENT: Final = re.compile(r"(?<![^/])openai/deployments/([^/]+)")
-
-
 def _deployment_model_name(litellm_params: LiteLLMParamsTypedDict) -> str:
     model: Final = litellm_params.get("model", "")
     try:
@@ -139,17 +130,10 @@ def _deployment_model_name(litellm_params: LiteLLMParamsTypedDict) -> str:
         return model
 
 
-def foreign_azure_deployment(endpoint: str, model_group: str, llm_router: litellm.Router) -> 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(
+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 ()
     )
-    return None if deployment in served else deployment
 
 
 def is_passthrough_request_streaming(request_body: object) -> bool:
@@ -1555,7 +1539,9 @@ async def _relay_azure_router_model(
     is_streaming_request: bool,
     user_api_key_dict: UserAPIKeyAuth,
 ) -> Response:
-    foreign_deployment: Final = foreign_azure_deployment(endpoint, model, llm_router)
+    foreign_deployment: Final = foreign_azure_deployment(
+        endpoint, model, lambda: _models_served_by_group(llm_router, model)
+    )
     if foreign_deployment is not None:
         raise HTTPException(
             status_code=400,

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

@@ -8,7 +8,11 @@ import pytest
 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
+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
 
@@ -444,3 +448,32 @@ def test_azure_passthrough_is_streaming_request_reads_the_stream_flag(request_da
         )
         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/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"),
+    ],
+)
+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
+
+
+@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/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -5368,36 +5368,11 @@ 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."""
 
-    @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/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"),
-        ],
-    )
-    def test_foreign_azure_deployment_names_a_segment_outside_the_group(self, endpoint, expected):
-        from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import foreign_azure_deployment
+    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 foreign_azure_deployment(endpoint, "gpt", _AzureGroupRouter([])) == expected
-
-    @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_matches_the_relay_decision(self, endpoint, expected):
-        from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import azure_router_model_in_endpoint
-
-        assert azure_router_model_in_endpoint(endpoint, _AzureGroupRouter([])) == expected
+        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