fix(anthropic): stream azure_ai natively on an Azure OpenAI host on /v1/messages (#49216)

* fix(anthropic): stream azure_ai natively on an Azure OpenAI host on /v1/messages

The /v1/messages bridges got the requested model (azure_ai/gpt-6-luna) with the resolved provider (azure), so the Responses bridge faked the stream and the chat bridge called a deployment named azure_ai/gpt-6-luna (404). Pass the resolved provider prefix when the requested prefix is a different known provider.

* test(anthropic): cover the bridged model with respx and a regional bedrock_mantle id

* fix(anthropic): carry azure_ai env credentials into bridged /v1/messages calls

When an azure_ai deployment's api_base and key come only from AZURE_AI_API_BASE
and AZURE_AI_API_KEY, the provider the async wrapper derived from the model alone
replaced the model's own resolution, so the bridged call lost both and failed.
The model's own resolution now wins when it names the same provider, an explicit
provider that disagrees still wins, and the bridge carries the resolved key and base.

* fix(anthropic): keep the caller's key, proxy routing and MCP credentials on bridged /v1/messages calls

* fix(anthropic): keep an azure deployment's Entra credential over the env key on bridged /v1/messages calls

* fix(anthropic): skip the sign-in lookup for chatgpt and github_copilot model prefixes on bridged /v1/messages calls

* fix(anthropic): keep the Azure credential of an azure_ai deployment that sets its own api_base on bridged /v1/messages calls

* test(anthropic): annotate the parameters of the bridged /v1/messages tests

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

litellm/llms/anthropic/messages/native/handler.py

@@ -8,6 +8,7 @@
 import asyncio
 import contextvars
 from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine, Iterator, Mapping, Sequence
+from dataclasses import dataclass
 from functools import partial
 from typing import Final, cast
 
@@ -15,6 +16,7 @@ from pydantic import TypeAdapter
 
 import litellm
 from litellm.anthropic_beta_headers_manager import aload_beta_headers_config
+from litellm.constants import PROVIDERS_THAT_AUTHENTICATE_ON_PROVIDER_INFO
 from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
 from litellm.litellm_core_utils.core_helpers import as_str_mapping
 from litellm.litellm_core_utils.exception_mapping_utils import exception_type
@@ -44,7 +46,7 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
 )
 from litellm.types.router import GenericLiteLLMParams
-from litellm.types.utils import CallTypes, LlmProviders
+from litellm.types.utils import CallTypes, LlmProviders, LlmProvidersSet
 from litellm.utils import ProviderConfigManager, client
 
 from ..adapters.handler import LiteLLMMessagesToCompletionTransformationHandler, request_safeguards_evaluator
@@ -94,6 +96,68 @@ def _responses_mode_is_lost_by_prefix_strip(
     ) and not _bridges_to_responses_api(resolved_model, custom_llm_provider, deployment_model_info)
 
 
+@dataclass(frozen=True, slots=True)
+class _BridgeTarget:
+    model: str
+    api_key: str | None
+    api_base: str | None
+
+
+def _resolve_requested_provider(
+    model: str, custom_llm_provider: str | None, litellm_params: GenericLiteLLMParams
+) -> tuple[str, str, str | None, str | None]:
+    resolved: Final = litellm.get_llm_provider(
+        model=model,
+        custom_llm_provider=custom_llm_provider,
+        api_base=litellm_params.api_base,
+        api_key=litellm_params.api_key,
+        litellm_params=litellm_params,
+    )
+    requested_provider: Final = model.partition("/")[0]
+    if (
+        custom_llm_provider in (None, requested_provider)
+        or requested_provider not in LlmProvidersSet
+        or requested_provider in PROVIDERS_THAT_AUTHENTICATE_ON_PROVIDER_INFO
+    ):
+        return resolved
+    own: Final = _resolve_by_model_prefix(model, litellm_params)
+    return own if own is not None and own[1] == custom_llm_provider else resolved
+
+
+def _resolve_by_model_prefix(
+    model: str, litellm_params: GenericLiteLLMParams
+) -> tuple[str, str, str | None, str | None] | None:
+    try:
+        return litellm.get_llm_provider(
+            model=model,
+            api_base=litellm_params.api_base,
+            api_key=litellm_params.api_key,
+            litellm_params=litellm_params.model_copy(update={"custom_llm_provider": None}),
+        )
+    except litellm.BadRequestError:
+        return None
+
+
+def _sets_own_entra_credential(litellm_params: GenericLiteLLMParams) -> bool:
+    return bool(
+        litellm_params.azure_ad_token
+        or (litellm_params.model_extra or {}).get("azure_ad_token_provider") is not None
+        or (litellm_params.tenant_id and litellm_params.client_id and litellm_params.client_secret)
+        or (litellm_params.azure_username and litellm_params.azure_password and litellm_params.client_id)
+    )
+
+
+def _bridge_target(requested: _BridgeTarget, resolved: _BridgeTarget, custom_llm_provider: str) -> _BridgeTarget:
+    requested_provider: Final = requested.model.partition("/")[0]
+    if requested_provider == custom_llm_provider or requested_provider not in LlmProvidersSet:
+        return requested
+    return _BridgeTarget(
+        model=f"{custom_llm_provider}/{resolved.model}",
+        api_key=requested.api_key if requested.api_base else requested.api_key or resolved.api_key,
+        api_base=resolved.api_base,
+    )
+
+
 def _should_route_to_responses_api(
     custom_llm_provider: str | None,
     requested_model: str | None = None,
@@ -624,13 +688,7 @@ def anthropic_messages_handler(
         custom_llm_provider,
         dynamic_api_key,
         dynamic_api_base,
-    ) = litellm.get_llm_provider(
-        model=model,
-        custom_llm_provider=custom_llm_provider,
-        api_base=litellm_params.api_base,
-        api_key=litellm_params.api_key,
-        litellm_params=litellm_params,
-    )
+    ) = _resolve_requested_provider(model, custom_llm_provider, litellm_params)
 
     # Store agentic loop params in logging object for agentic hooks
     # This provides original request context needed for follow-up calls
@@ -656,6 +714,16 @@ def anthropic_messages_handler(
             mock_response=litellm_params.mock_response,
         )
 
+    bridge: Final = _bridge_target(
+        _BridgeTarget(model=original_model, api_key=api_key, api_base=api_base),
+        _BridgeTarget(
+            model=model,
+            api_key=None if _sets_own_entra_credential(litellm_params) else dynamic_api_key,
+            api_base=dynamic_api_base,
+        ),
+        custom_llm_provider,
+    )
+
     # Expand litellm_proxy MCP references through the MCP gateway before dispatch, so every
     # downstream path (native passthrough and both bridges) gets real tools rather than a
     # reference the provider cannot resolve. Popped from kwargs so it never reaches the provider.
@@ -682,8 +750,8 @@ def anthropic_messages_handler(
             top_k=top_k,
             top_p=top_p,
             container=container,
-            api_key=api_key,
-            api_base=api_base,
+            api_key=bridge.api_key,
+            api_base=bridge.api_base,
             client=client,
             custom_llm_provider=custom_llm_provider,
             **kwargs,
@@ -700,7 +768,7 @@ def anthropic_messages_handler(
             return LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
                 max_tokens=max_tokens,
                 messages=messages,
-                model=original_model,
+                model=bridge.model,
                 metadata=metadata,
                 stop_sequences=stop_sequences,
                 stream=stream,
@@ -716,8 +784,8 @@ def anthropic_messages_handler(
                 top_k=top_k,
                 top_p=top_p,
                 _is_async=is_async,
-                api_key=api_key,
-                api_base=api_base,
+                api_key=bridge.api_key,
+                api_base=bridge.api_base,
                 client=client,
                 custom_llm_provider=custom_llm_provider,
                 **kwargs,
@@ -730,7 +798,7 @@ def anthropic_messages_handler(
         return LiteLLMMessagesToCompletionTransformationHandler.anthropic_messages_handler(
             max_tokens=max_tokens,
             messages=messages,
-            model=original_model,
+            model=bridge.model,
             metadata=metadata,
             stop_sequences=stop_sequences,
             stream=stream,
@@ -744,8 +812,8 @@ def anthropic_messages_handler(
             top_k=top_k,
             top_p=top_p,
             _is_async=is_async,
-            api_key=api_key,
-            api_base=api_base,
+            api_key=bridge.api_key,
+            api_base=bridge.api_base,
             client=client,
             custom_llm_provider=custom_llm_provider,
             **kwargs,

tests/unit/llms/anthropic/messages/native/test_anthropic_messages_handler.py

@@ -3,10 +3,13 @@ import copy
 import json
 import os
 import uuid
+from collections.abc import Mapping
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Dict, Final, List
 
 import httpx
+import openai
 import responses
 import respx
 import pytest
@@ -1136,6 +1139,260 @@ async def test_messages_strips_provider_prefix_exactly_once(requested_model, exp
     assert captured["url"] == expected_url
 
 
+_GET_WEATHER_TOOL: Final = {
+    "name": "get_weather",
+    "description": "Get the current weather for a city",
+    "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "tools, expected_path",
+    [
+        ([_GET_WEATHER_TOOL], "/openai/v1/responses"),
+        (None, "/openai/deployments/gpt-6-luna/chat/completions"),
+    ],
+    ids=["function_tool_bridged_to_responses", "no_tools_on_chat"],
+)
+@pytest.mark.parametrize("credentials_from_env", [False, True], ids=["credentials_in_params", "credentials_from_env"])
+async def test_azure_ai_on_azure_openai_host_streams_natively_with_the_bare_deployment(
+    tools: list[dict[str, object]] | None,
+    expected_path: str,
+    credentials_from_env: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    monkeypatch.delenv("AZURE_API_BASE", raising=False)
+    monkeypatch.delenv("AZURE_API_KEY", raising=False)
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://qa-resource.cognitiveservices.azure.com")
+    monkeypatch.setenv("AZURE_AI_API_KEY", "env-api-key")
+    credentials: Final = (
+        {}
+        if credentials_from_env
+        else {"api_base": "https://qa-resource.cognitiveservices.azure.com", "api_key": "params-api-key"}
+    )
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route(host="qa-resource.cognitiveservices.azure.com").mock(
+            side_effect=httpx.ConnectError("cut at the wire")
+        )
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "What is the weather in Berlin?"}],
+                model="azure_ai/gpt-6-luna",
+                stream=True,
+                tools=tools,
+                **credentials,
+            )
+
+    bodies: Final = [json.loads(call.request.content) for call in upstream.calls]
+    assert {call.request.url.path for call in upstream.calls} == {expected_path}
+    assert {call.request.headers["api-key"] for call in upstream.calls} == {credentials.get("api_key", "env-api-key")}
+    assert {body["model"] for body in bodies} == {"gpt-6-luna"}
+    assert all(body["stream"] is True for body in bodies)
+
+
+@pytest.mark.asyncio
+async def test_bridged_messages_keep_the_region_of_a_regional_bedrock_mantle_model(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(litellm.exceptions.ServiceUnavailableError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="bedrock_mantle/eu-west-1/openai.gpt-oss-120b",
+                api_key="test-api-key",
+            )
+
+    assert {call.request.url.host for call in upstream.calls} == {"bedrock-mantle.eu-west-1.api.aws"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model", ["anthropic/claude-sonnet-5-5", "ragflow/gpt-4o"])
+async def test_bridged_messages_keep_the_callers_explicit_provider_over_the_model_prefix(
+    model: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model=model,
+                custom_llm_provider="openai",
+                api_base="https://gateway.example.com/v1",
+                api_key="test-api-key",
+            )
+
+    assert {call.request.url.host for call in upstream.calls} == {"gateway.example.com"}
+    assert {json.loads(call.request.content)["model"] for call in upstream.calls} == {model}
+
+
+@pytest.mark.parametrize("model", ["chatgpt/gpt-5", "github_copilot/gpt-5"])
+def test_bridged_messages_reach_the_callers_provider_without_signing_in_to_the_prefix_provider(
+    model: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("OPENAI_API_KEY", "server-openai-key")
+    monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path))
+    monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            litellm.anthropic.messages.create(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model=model,
+                custom_llm_provider="openai",
+            )
+
+    assert {call.request.url.host for call in upstream.calls} == {"api.openai.com"}
+    assert {call.request.headers["authorization"] for call in upstream.calls} == {"Bearer server-openai-key"}
+
+
+@pytest.mark.asyncio
+async def test_bridged_messages_send_the_deployments_key_for_azure_models_served_as_openai(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("OPENAI_API_KEY", "server-openai-key")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="azure/mistral-large-latest",
+                api_base="https://example-resource.services.ai.azure.com/models",
+                api_key="deployment-key",
+            )
+
+    assert {call.request.headers["authorization"] for call in upstream.calls} == {"Bearer deployment-key"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "entra_credential",
+    [
+        {"azure_ad_token": "entra-token"},
+        {"azure_ad_token_provider": lambda: "entra-token"},
+        {"tenant_id": "tenant", "client_id": "client", "client_secret": "secret"},
+        {"azure_username": "user", "azure_password": "password", "client_id": "client"},
+    ],
+)
+@pytest.mark.parametrize("api_base_from_env", [False, True], ids=["api_base_in_params", "api_base_from_env"])
+async def test_bridged_messages_keep_the_deployments_entra_credential_over_the_env_key(
+    monkeypatch: pytest.MonkeyPatch, entra_credential: Mapping[str, object], api_base_from_env: bool
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://example-resource.cognitiveservices.azure.com")
+    monkeypatch.setenv("AZURE_AI_API_KEY", "azure-env-key")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+    api_base: Final = {} if api_base_from_env else {"api_base": "https://example-resource.cognitiveservices.azure.com"}
+
+    with (
+        patch("litellm.llms.azure.common_utils.get_azure_ad_token_from_entra_id", return_value=lambda: "entra-token"),  # test-quality-ok: stands in for the Entra token fetch so the test can see which credential reaches Azure
+        patch(
+            "litellm.llms.azure.common_utils.get_azure_ad_token_from_username_password",
+            return_value=lambda: "entra-token",
+        ),  # test-quality-ok: stands in for the Entra token fetch so the test can see which credential reaches Azure
+        respx.mock(assert_all_called=True) as respx_mock,
+    ):
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="azure_ai/gpt-6-luna",
+                tools=[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
+                **api_base,
+                **entra_credential,
+            )
+
+    assert {call.request.headers.get("authorization") for call in upstream.calls} == {"Bearer entra-token"}
+    assert {call.request.headers.get("api-key") for call in upstream.calls} == {None}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "azure_env_credential, expected_header, expected_value",
+    [
+        ({"AZURE_API_KEY": "azure-env-key"}, "api-key", "azure-env-key"),
+        (
+            {"AZURE_TENANT_ID": "tenant", "AZURE_CLIENT_ID": "client", "AZURE_CLIENT_SECRET": "secret"},
+            "authorization",
+            "Bearer entra-token",
+        ),
+    ],
+    ids=["azure_env_key", "azure_env_entra"],
+)
+async def test_bridged_messages_keep_the_azure_env_credential_when_the_deployment_sets_its_own_api_base(
+    azure_env_credential: Mapping[str, str], expected_header: str, expected_value: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    for name in ("AZURE_API_KEY", "AZURE_OPENAI_API_KEY", "AZURE_AD_TOKEN", "AZURE_TENANT_ID", "AZURE_CLIENT_ID"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("AZURE_AI_API_KEY", "key-for-another-resource")
+    for name, value in azure_env_credential.items():
+        monkeypatch.setenv(name, value)
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with (
+        patch("litellm.llms.azure.common_utils.get_azure_ad_token_from_entra_id", return_value=lambda: "entra-token"),  # test-quality-ok: stands in for the Entra token fetch so the test can see which credential reaches Azure
+        respx.mock(assert_all_called=True) as respx_mock,
+    ):
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="azure_ai/gpt-6-luna",
+                api_base="https://example-resource.cognitiveservices.azure.com",
+                tools=[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
+            )
+
+    assert {call.request.headers.get(expected_header) for call in upstream.calls} == {expected_value}
+
+
+@pytest.mark.asyncio
+async def test_bridged_messages_still_route_through_the_litellm_proxy_when_asked(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("LITELLM_PROXY_API_BASE", "https://proxy.example.com")
+    monkeypatch.setenv("LITELLM_PROXY_API_KEY", "proxy-key")
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://example-resource.cognitiveservices.azure.com")
+    monkeypatch.setenv("AZURE_AI_API_KEY", "azure-env-key")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="azure_ai/gpt-6-luna",
+                use_litellm_proxy=True,
+            )
+
+    assert {call.request.url.host for call in upstream.calls} == {"proxy.example.com"}
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "model_info, expected_url",

tests/unit/llms/anthropic/messages/native/test_mcp_handler.py

@@ -2,8 +2,12 @@ from importlib import import_module
 from typing import Final
 from unittest.mock import AsyncMock, patch
 
+import httpx
+import openai
 import pytest
+import respx
 
+import litellm
 from litellm.llms.anthropic.messages.adapters.transformation import (
     LiteLLMAnthropicMessagesAdapter,
 )
@@ -56,6 +60,30 @@ def test_anthropic_messages_handler_routes_litellm_proxy_mcp_to_the_gateway():
     assert result is not None
 
 
+@pytest.mark.asyncio
+async def test_anthropic_messages_with_mcp_reaches_an_env_only_deployment_with_its_env_credentials(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("AZURE_AI_API_BASE", "https://example-resource.cognitiveservices.azure.com")
+    monkeypatch.setenv("AZURE_AI_API_KEY", "azure-env-key")
+    litellm.in_memory_llm_clients_cache.flush_cache()
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        upstream: Final = respx_mock.route().mock(side_effect=httpx.ConnectError("cut at the wire"))
+        with pytest.raises(openai.APIError):
+            await litellm.anthropic.messages.acreate(
+                max_tokens=100,
+                messages=[{"role": "user", "content": "ping"}],
+                model="azure_ai/gpt-6-luna",
+                tools=[{**MCP_REFERENCE, "server_url": "litellm_proxy"}],
+            )
+
+    assert {call.request.url.host for call in upstream.calls} == {"example-resource.cognitiveservices.azure.com"}
+    assert {call.request.headers.get("api-key") for call in upstream.calls} == {"azure-env-key"}
+    assert not any(b"litellm_proxy" in call.request.content for call in upstream.calls)
+
+
 def test_anthropic_messages_handler_skips_the_gateway_on_recursion():
     """The gateway's own follow-up call must not re-enter the gateway."""
     with patch(