test(e2e): cover anthropic, azure and vertex chat and responses cells

tests/e2e/llm_translation/test_chat_completions_provider_matrix_e2e.py

@@ -0,0 +1,718 @@
+"""Live /chat/completions coverage for the provider routes a customer reaches by
+registering their own deployment: Anthropic's first-party API, Azure OpenAI,
+Vertex AI Gemini, and Azure AI Foundry.
+
+Each class is one route's translation contract. The gateway speaks the
+OpenAI-compatible request shape to the caller and the provider's native shape
+upstream, so every capability here (tool calls, vision, streaming, extended
+thinking, prompt caching, schema-constrained output) is a translation that can
+break per provider while the other routes keep passing. Each test registers the
+deployment it needs through /model/new with `os.environ/...` credential
+references the proxy resolves at call time, and deletes it on teardown.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Literal
+
+import pytest
+from pydantic import BaseModel
+
+from e2e_config import unique_marker
+from e2e_http import StreamingResponse, unwrap
+from lifecycle import ResourceManager
+from models import (
+    ChatBody,
+    ChatMessage,
+    ChatResponse,
+    ChatTool,
+    ChatToolFunction,
+    ImageContentPart,
+    ImageUrl,
+    LiteLLMParamsBody,
+    TextContentPart,
+    ThinkingParam,
+)
+from passthrough_client import PassthroughClient
+
+pytestmark = pytest.mark.e2e
+
+ANTHROPIC_BACKEND = "anthropic/claude-haiku-4-5"
+AZURE_OPENAI_BACKEND = "azure/gpt-5.6-sol-e2e"
+VERTEX_BACKEND = "vertex_ai/gemini-2.5-flash"
+AZURE_FOUNDRY_BACKEND = "azure_ai/claude-haiku-4-5"
+
+CACHE_READ_DEADLINE_SECONDS = 30.0
+CACHE_READ_INTERVAL_SECONDS = 3.0
+
+SPLIT_COLOR_IMAGE = (
+    "data:image/png;base64,"
+    "iVBORw0KGgoAAAANSUhEUgAAAGAAAABgCAIAAABt+uBvAAAAmklEQVR42u3QQQkAAAgEMJNcEvtjLFuIj8ES"
+    "rCZ5JT2vlCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBB"
+    "ggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBBggQJEiRIkCBB"
+    "dxaTpa47LOh2vwAAAABJRU5ErkJggg=="
+)
+
+
+def _anthropic_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY")
+
+
+def _azure_openai_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(
+        model=AZURE_OPENAI_BACKEND,
+        api_base="os.environ/AZURE_API_BASE",
+        api_key="os.environ/AZURE_API_KEY",
+    )
+
+
+def _vertex_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(
+        model=VERTEX_BACKEND,
+        vertex_project="os.environ/VERTEXAI_PROJECT",
+        vertex_credentials="os.environ/VERTEXAI_CREDENTIALS",
+    )
+
+
+def _azure_foundry_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(
+        model=AZURE_FOUNDRY_BACKEND,
+        api_base="os.environ/AZURE_AI_API_BASE",
+        api_key="os.environ/AZURE_AI_API_KEY",
+    )
+
+
+def _register(
+    client: PassthroughClient,
+    resources: ResourceManager,
+    prefix: str,
+    params: LiteLLMParamsBody,
+) -> tuple[str, str]:
+    """Register a deployment for one test and hand back its model name plus a key
+    that may call it. Both are torn down by the resources fixture."""
+    model = f"{prefix}-{unique_marker()}"
+    model_id = client.proxy.create_model(model, params)
+    resources.defer(lambda: client.proxy.delete_model(model_id))
+    return model, resources.key()
+
+
+class _StreamToolCallFunction(BaseModel):
+    name: str | None = None
+    arguments: str | None = None
+
+
+class _StreamToolCall(BaseModel):
+    function: _StreamToolCallFunction = _StreamToolCallFunction()
+
+
+class _StreamDelta(BaseModel):
+    content: str | None = None
+    tool_calls: list[_StreamToolCall] | None = None
+
+
+class _StreamChoice(BaseModel):
+    delta: _StreamDelta = _StreamDelta()
+
+
+class _StreamChunk(BaseModel):
+    choices: list[_StreamChoice] = []
+
+
+def _streamed_text(events: list[str]) -> str:
+    """Concatenate the delta content across streamed chunks. Parsing every event as
+    JSON also fails loudly on a truncated or garbled chunk, so an incomplete stream
+    cannot pass as content."""
+    chunks = [_StreamChunk.model_validate_json(event) for event in events]
+    return "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices)
+
+
+def _streamed_tool_call(events: list[str]) -> tuple[str, str]:
+    """Reassemble the tool call streamed across chunks: the name arrives once and the
+    arguments arrive as fragments, so concatenating both and parsing the arguments as
+    JSON catches a stream that never completes the call or splits its argument JSON."""
+    chunks = [_StreamChunk.model_validate_json(event) for event in events]
+    calls = [call for chunk in chunks for choice in chunk.choices for call in (choice.delta.tool_calls or [])]
+    name = "".join(call.function.name or "" for call in calls)
+    arguments = "".join(call.function.arguments or "" for call in calls)
+    return name, arguments
+
+
+def _assert_streamed_completion(result: StreamingResponse) -> None:
+    """A streamed /chat/completions must deliver real content, not a clean-but-empty
+    stream."""
+    assert result.ok and result.is_streaming, f"stream was not established: {result}"
+    assert result.stream_error is None, f"stream carried an error event: {result.stream_error}"
+    assert len(result.stream_events) > 1, f"stream did not deliver multiple data events: {result}"
+    assert _streamed_text(result.stream_events).strip(), (
+        f"stream completed with no content deltas: {result.stream_events[:3]}"
+    )
+
+
+class _WeatherArgs(BaseModel):
+    location: str
+
+
+_WEATHER_TOOL = ChatTool(
+    function=ChatToolFunction(
+        name="get_weather",
+        description="Get the current weather for a location",
+        parameters={
+            "type": "object",
+            "properties": {"location": {"type": "string"}},
+            "required": ["location"],
+        },
+    )
+)
+
+_WEATHER_PROMPT = "What is the weather in San Francisco? Use the get_weather tool."
+
+_AZURE_HOSTED_PROMPT = "Reply with the single word pong."
+"""Plain prose, with none of the random hex uniquifier the other routes append:
+Azure runs Microsoft's prompt shields in front of both its OpenAI and its Foundry
+deployments, and a random token in an otherwise trivial prompt intermittently
+trips the jailbreak classifier into a 400. Each test registers its own uniquely
+named deployment, which is what actually keeps runs from sharing a cached
+response."""
+
+
+def _assert_weather_tool_call(response: ChatResponse) -> None:
+    """The model, forced to call the tool, must return a get_weather call whose
+    arguments parse as JSON and carry a location. A translation that drops tool_calls
+    or emits malformed argument JSON fails here rather than passing on a 200."""
+    assert response.choices, f"chat returned no choices: {response}"
+    message = response.choices[0].message
+    calls = message.tool_calls if message else None
+    assert calls, f"model returned no tool call for a tool-forced prompt: {response}"
+    weather = next((call for call in calls if call.function.name == "get_weather"), None)
+    assert weather is not None, f"expected a get_weather call, got {[c.function.name for c in calls]}"
+    assert weather.function.arguments, f"get_weather call carried no arguments: {weather}"
+    args = _WeatherArgs.model_validate_json(weather.function.arguments)
+    assert args.location.strip(), f"get_weather arguments missing location: {weather.function.arguments}"
+
+
+def _vision_messages() -> list[ChatMessage]:
+    """A base64 data URI rather than a hosted image: Anthropic and Vertex fetch a
+    remote image_url from their own side, and both are blocked by the usual public
+    image hosts, so a link makes the test measure the host's crawler policy instead
+    of the gateway's image translation."""
+    return [
+        ChatMessage(
+            role="user",
+            content=[
+                TextContentPart(
+                    text=(
+                        "This image is split into two halves of different solid colors. "
+                        "Name the two colors and nothing else."
+                    )
+                ),
+                ImageContentPart(image_url=ImageUrl(url=SPLIT_COLOR_IMAGE)),
+            ],
+        )
+    ]
+
+
+def _assert_reads_split_colors(response: ChatResponse) -> None:
+    assert response.choices, f"vision returned no choices: {response}"
+    message = response.choices[0].message
+    content = ((message.content if message else None) or "").lower()
+    assert "red" in content and "blue" in content, (
+        f"vision response did not read the two halves of the image: {content[:200]}"
+    )
+
+
+def _assert_answered(response: ChatResponse) -> None:
+    assert response.model, f"response carried no model name: {response}"
+    assert response.choices, f"response had no choices: {response}"
+    message = response.choices[0].message
+    assert message and message.content and message.content.strip(), (
+        f"200 with an empty completion: {response}"
+    )
+
+
+class _Person(BaseModel):
+    name: str
+    age: int
+
+
+_PERSON_SCHEMA: dict[str, object] = {
+    "type": "json_schema",
+    "json_schema": {
+        "name": "person",
+        "strict": True,
+        "schema": {
+            "type": "object",
+            "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
+            "required": ["name", "age"],
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+class _CacheControl(BaseModel):
+    type: Literal["ephemeral"] = "ephemeral"
+
+
+class _CacheableTextPart(BaseModel):
+    """A text content part carrying Anthropic's cache_control breakpoint. Anthropic
+    caches only what a breakpoint marks, so the OpenAI-shaped body has to carry it
+    through the translation; the shared TextContentPart has no such field."""
+
+    type: Literal["text"] = "text"
+    text: str
+    cache_control: _CacheControl | None = None
+
+
+class _CacheableMessage(BaseModel):
+    role: str
+    content: list[_CacheableTextPart]
+
+
+class _CacheableChatBody(BaseModel):
+    model: str
+    messages: list[_CacheableMessage]
+    max_tokens: int = 16
+
+
+def _cacheable_prefix(marker: str) -> str:
+    """A system prompt past Anthropic's 2048-token minimum cacheable size for Haiku,
+    unique per run so no other run's cache entry can satisfy the read."""
+    return " ".join(f"Standing instruction {index} for run {marker}: stay concise." for index in range(400))
+
+
+def _cached_prefix_body(model: str, prefix: str, turn: str) -> _CacheableChatBody:
+    return _CacheableChatBody(
+        model=model,
+        messages=[
+            _CacheableMessage(role="system", content=[_CacheableTextPart(text=prefix, cache_control=_CacheControl())]),
+            _CacheableMessage(role="user", content=[_CacheableTextPart(text=turn)]),
+        ],
+    )
+
+
+class _ThinkingBlock(BaseModel):
+    """One entry of `thinking_blocks`: Anthropic's own thinking representation as
+    LiteLLM carries it onto the OpenAI-shaped response. `signature` is the token
+    Anthropic issues for a genuine thinking block, so its presence is what separates
+    a translated thinking block from prose the model happened to write."""
+
+    type: str
+    thinking: str | None = None
+    signature: str | None = None
+
+
+class _ThinkingMessage(BaseModel):
+    content: str | None = None
+    reasoning_content: str | None = None
+    thinking_blocks: list[_ThinkingBlock] | None = None
+
+
+class _ThinkingChoice(BaseModel):
+    message: _ThinkingMessage = _ThinkingMessage()
+
+
+class _ThinkingUsageDetails(BaseModel):
+    reasoning_tokens: int | None = None
+
+
+class _ThinkingUsage(BaseModel):
+    completion_tokens_details: _ThinkingUsageDetails = _ThinkingUsageDetails()
+
+
+class _ThinkingChatResponse(BaseModel):
+    """The slice of a /chat/completions answer that carries extended thinking: the
+    answer text, Anthropic's reasoning content and thinking blocks, and the reasoning
+    token accounting. The shared ChatResponse models neither thinking_blocks nor the
+    reasoning-token count, and both are needed to prove thinking actually happened."""
+
+    choices: list[_ThinkingChoice] = []
+    usage: _ThinkingUsage = _ThinkingUsage()
+
+    @property
+    def message(self) -> _ThinkingMessage:
+        assert self.choices, f"chat returned no choices: {self}"
+        return self.choices[0].message
+
+    @property
+    def reasoning_tokens(self) -> int:
+        return self.usage.completion_tokens_details.reasoning_tokens or 0
+
+    @property
+    def blocks(self) -> tuple[_ThinkingBlock, ...]:
+        return tuple(self.message.thinking_blocks or ())
+
+
+_THINKING_PROMPT = "What is 17 times 23? Think it through step by step."
+
+
+class TestAnthropicChatCompletions:
+    """Anthropic's first-party API through the OpenAI-compatible /chat/completions
+    translation - the path a customer keeps when they point an OpenAI SDK at the
+    gateway but bill Claude."""
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.basic.stream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_streams_real_content(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-anthropic-stream", _anthropic_params())
+
+        result = client.proxy.chat_stream(
+            key,
+            ChatBody(
+                model=model,
+                messages=[
+                    ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")
+                ],
+                max_tokens=64,
+                stream=True,
+            ),
+        )
+        _assert_streamed_completion(result)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.tool_use.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_returns_tool_call(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-anthropic-tool", _anthropic_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content=_WEATHER_PROMPT)],
+                    tools=[_WEATHER_TOOL],
+                    tool_choice="required",
+                    max_tokens=256,
+                ),
+            )
+        )
+        _assert_weather_tool_call(response)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.tool_use.stream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_streams_tool_call(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-anthropic-tool-stream", _anthropic_params())
+
+        result = client.proxy.chat_stream(
+            key,
+            ChatBody(
+                model=model,
+                messages=[ChatMessage(role="user", content=_WEATHER_PROMPT)],
+                tools=[_WEATHER_TOOL],
+                tool_choice="required",
+                max_tokens=256,
+                stream=True,
+            ),
+        )
+        assert result.ok and result.is_streaming, f"tool stream was not established: {result}"
+        assert result.stream_error is None, f"tool stream carried an error event: {result.stream_error}"
+        name, arguments = _streamed_tool_call(result.stream_events)
+        assert name == "get_weather", f"streamed tool call named {name!r}: {result.stream_events[:5]}"
+        args = _WeatherArgs.model_validate_json(arguments)
+        assert args.location.strip(), f"streamed tool call arguments missing location: {arguments!r}"
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.vision.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_vision_describes_image(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-anthropic-vision", _anthropic_params())
+
+        response = unwrap(client.proxy.chat(key, ChatBody(model=model, messages=_vision_messages(), max_tokens=32)))
+        _assert_reads_split_colors(response)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.prompt_cache_5m.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_prompt_cache_hits_on_repeat(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        """A cache_control breakpoint on the system prefix must survive the
+        translation, so the first call writes the prefix to Anthropic's cache and a
+        later call reads it back instead of re-billing it at write price.
+
+        Every call carries a fresh user turn after the breakpoint: that is both how a
+        customer actually uses a cached prefix and what keeps the gateway's own
+        response cache from replaying the first answer, which would hand the test
+        turn one's write usage again and hide whether a read ever happened.
+        """
+        model, key = _register(client, resources, "e2e-anthropic-cache", _anthropic_params())
+        prefix = _cacheable_prefix(unique_marker())
+
+        def call() -> ChatResponse:
+            return unwrap(
+                client.proxy.transport.post(
+                    "/chat/completions",
+                    headers=client.proxy.transport.bearer(key),
+                    json=_cached_prefix_body(
+                        model, prefix, f"Reply with the single word pong. {unique_marker()}"
+                    ),
+                    response_type=ChatResponse,
+                )
+            )
+
+        first = call()
+        written = first.usage.cache_creation_input_tokens if first.usage else None
+        assert written and written > 0, (
+            f"a cache_control breakpoint must write the prefix to the cache, got usage={first.usage}"
+        )
+
+        deadline = time.monotonic() + CACHE_READ_DEADLINE_SECONDS
+        while True:
+            usage = call().usage
+            read = usage.cache_read_input_tokens if usage else None
+            if read and read >= written:
+                return
+            if time.monotonic() >= deadline:
+                pytest.fail(
+                    f"a later call must read the whole {written}-token cached prefix back, but the "
+                    f"cache never became readable within {CACHE_READ_DEADLINE_SECONDS}s (last usage: {usage})"
+                )
+            time.sleep(CACHE_READ_INTERVAL_SECONDS)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.structured_output.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_structured_output_conforms_to_schema(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-anthropic-schema", _anthropic_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content="Extract the person. John Doe is 42 years old.")],
+                    response_format=_PERSON_SCHEMA,
+                    max_tokens=256,
+                ),
+            )
+        )
+        assert response.choices, f"structured output returned no choices: {response}"
+        content = response.choices[0].message.content if response.choices[0].message else None
+        assert content, f"structured output returned empty content: {response}"
+        person = _Person.model_validate_json(content)
+        assert person.name.strip() and person.age == 42, (
+            f"schema-constrained extraction was wrong: {person}"
+        )
+
+    @pytest.mark.covers(
+        "llm.chat_completions.anthropic.thinking.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_anthropic_chat_thinking_changes_the_translated_response(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        """Enabling extended thinking has to change what comes back, so the same
+        question is asked twice: once plain and once with a thinking budget.
+
+        Asserting only that a thinking-enabled call returned some text would pass
+        against a gateway that dropped the thinking parameter entirely, so each half
+        of the contrast is pinned to Anthropic's own representation as translated
+        onto the OpenAI-shaped response: the reasoning text, a thinking block
+        carrying the signature Anthropic issues for it, and the reasoning-token
+        accounting. The plain call must carry none of the three.
+        """
+        model, key = _register(client, resources, "e2e-anthropic-thinking", _anthropic_params())
+
+        def ask(thinking: ThinkingParam | None) -> _ThinkingChatResponse:
+            return unwrap(
+                client.proxy.transport.post(
+                    "/chat/completions",
+                    headers=client.proxy.transport.bearer(key),
+                    json=ChatBody(
+                        model=model,
+                        messages=[
+                            ChatMessage(role="user", content=f"{_THINKING_PROMPT} {unique_marker()}")
+                        ],
+                        thinking=thinking,
+                        max_tokens=2048,
+                    ),
+                    response_type=_ThinkingChatResponse,
+                )
+            )
+
+        plain = ask(None)
+        thought = ask(ThinkingParam(type="enabled", budget_tokens=1024))
+
+        assert plain.message.content and plain.message.content.strip(), (
+            f"the baseline call returned no answer: {plain}"
+        )
+        assert thought.message.content and thought.message.content.strip(), (
+            f"the thinking call returned no answer: {thought}"
+        )
+
+        assert not (plain.message.reasoning_content or "").strip(), (
+            f"thinking was never requested, so reasoning_content must be empty: {plain.message}"
+        )
+        assert not plain.blocks, f"thinking was never requested, so no thinking blocks may come back: {plain.blocks}"
+        assert plain.reasoning_tokens == 0, (
+            f"thinking was never requested, so no reasoning tokens may be billed: {plain.usage}"
+        )
+
+        assert (thought.message.reasoning_content or "").strip(), (
+            "thinking was enabled but no reasoning_content came back on the Anthropic path"
+        )
+        signed = [block for block in thought.blocks if block.type == "thinking" and block.signature]
+        assert signed, (
+            f"thinking was enabled but no signed Anthropic thinking block survived the "
+            f"translation: {thought.blocks}"
+        )
+        assert thought.reasoning_tokens > 0, (
+            f"thinking was enabled but no reasoning tokens were reported: {thought.usage}"
+        )
+
+
+class TestAzureOpenAIChatCompletions:
+    """Azure OpenAI, where the model lives behind a per-resource deployment name and
+    the gateway authenticates with the resource's api_base plus api_key."""
+
+    @pytest.mark.covers(
+        "llm.chat_completions.azure_openai.basic.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_azure_openai_chat_returns_content(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-azure-openai-chat", _azure_openai_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content=_AZURE_HOSTED_PROMPT)],
+                    max_tokens=512,
+                ),
+            )
+        )
+        _assert_answered(response)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.azure_openai.tool_use.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_azure_openai_chat_returns_tool_call(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-azure-openai-tool", _azure_openai_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content=_WEATHER_PROMPT)],
+                    tools=[_WEATHER_TOOL],
+                    tool_choice="required",
+                    max_tokens=1024,
+                    reasoning_effort="low",
+                ),
+            )
+        )
+        _assert_weather_tool_call(response)
+
+
+class TestVertexChatCompletions:
+    """Vertex AI Gemini, where the gateway mints a Google token from the project's
+    service account instead of sending an API key."""
+
+    @pytest.mark.covers(
+        "llm.chat_completions.vertex.basic.stream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_vertex_chat_streams_real_content(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-vertex-stream", _vertex_params())
+
+        result = client.proxy.chat_stream(
+            key,
+            ChatBody(
+                model=model,
+                messages=[
+                    ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")
+                ],
+                max_tokens=512,
+                stream=True,
+            ),
+        )
+        _assert_streamed_completion(result)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.vertex.tool_use.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_vertex_chat_returns_tool_call(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-vertex-tool", _vertex_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content=_WEATHER_PROMPT)],
+                    tools=[_WEATHER_TOOL],
+                    tool_choice="required",
+                    max_tokens=512,
+                ),
+            )
+        )
+        _assert_weather_tool_call(response)
+
+    @pytest.mark.covers(
+        "llm.chat_completions.vertex.vision.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_vertex_chat_vision_describes_image(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-vertex-vision", _vertex_params())
+
+        response = unwrap(client.proxy.chat(key, ChatBody(model=model, messages=_vision_messages(), max_tokens=512)))
+        _assert_reads_split_colors(response)
+
+
+class TestAzureFoundryChatCompletions:
+    """Azure AI Foundry (azure_ai), which serves Claude on Azure's own contract; the
+    OpenAI-compatible route has to translate to it rather than to Anthropic's."""
+
+    @pytest.mark.covers(
+        "llm.chat_completions.azure_foundry.basic.nonstream.works",
+        exercised_on=["chat_completions"],
+    )
+    def test_azure_foundry_chat_returns_content(
+        self, client: PassthroughClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(client, resources, "e2e-azure-foundry-chat", _azure_foundry_params())
+
+        response = unwrap(
+            client.proxy.chat(
+                key,
+                ChatBody(
+                    model=model,
+                    messages=[ChatMessage(role="user", content=_AZURE_HOSTED_PROMPT)],
+                    max_tokens=512,
+                ),
+            )
+        )
+        _assert_answered(response)

tests/e2e/llm_translation/test_responses_provider_matrix_e2e.py

@@ -0,0 +1,135 @@
+"""Live /v1/responses coverage for the cloud providers that have no native
+Responses API of their own: Azure OpenAI and Vertex AI Gemini.
+
+The gateway accepts the OpenAI Responses request shape and translates it to
+whatever the deployment's provider speaks, so a customer can keep one client
+across clouds. Each test registers its deployment through /model/new with
+`os.environ/...` credential references the proxy resolves at call time, drives
+the endpoint, and deletes the deployment on teardown.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import cast
+
+import pytest
+from pydantic import BaseModel
+
+from e2e_config import unique_marker
+from e2e_http import require_successful_call
+from endpoints_client import (
+    EndpointsClient,
+    FunctionParameterProperty,
+    FunctionParameters,
+    ResponsesFunctionTool,
+    ResponsesResult,
+)
+from lifecycle import ResourceManager
+from models import LiteLLMParamsBody
+
+pytestmark = pytest.mark.e2e
+
+AZURE_OPENAI_BACKEND = "azure/gpt-5.6-sol-e2e"
+VERTEX_BACKEND = "vertex_ai/gemini-2.5-flash"
+
+WEATHER_TOOL = ResponsesFunctionTool(
+    name="get_weather",
+    description="Get the weather for a location",
+    parameters=FunctionParameters(
+        properties={"location": FunctionParameterProperty(type="string")},
+        required=["location"],
+    ),
+)
+
+WEATHER_PROMPT = "What is the weather in San Francisco? Use the get_weather tool."
+
+BASIC_PROMPT = "Reply with one word."
+
+
+class WeatherArguments(BaseModel):
+    location: str
+
+
+def _azure_openai_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(
+        model=AZURE_OPENAI_BACKEND,
+        api_base="os.environ/AZURE_API_BASE",
+        api_key="os.environ/AZURE_API_KEY",
+    )
+
+
+def _vertex_params() -> LiteLLMParamsBody:
+    return LiteLLMParamsBody(
+        model=VERTEX_BACKEND,
+        vertex_project="os.environ/VERTEXAI_PROJECT",
+        vertex_credentials="os.environ/VERTEXAI_CREDENTIALS",
+    )
+
+
+def _register(
+    endpoints_client: EndpointsClient,
+    resources: ResourceManager,
+    prefix: str,
+    params: LiteLLMParamsBody,
+) -> tuple[str, str]:
+    model = f"{prefix}-{unique_marker()}"
+    model_id = endpoints_client.create_model(model, params)
+    resources.defer(lambda: endpoints_client.delete_model(model_id))
+    return model, resources.key()
+
+
+def _assert_weather_function_call(body: str) -> None:
+    parsed = ResponsesResult.model_validate_json(body)
+    function_call = next((call for call in parsed.function_calls if call.name == "get_weather"), None)
+    assert function_call is not None, f"no get_weather function call: {body[:500]}"
+    assert function_call.arguments is not None
+    raw_arguments = cast(object, json.loads(function_call.arguments))
+    arguments = WeatherArguments.model_validate(raw_arguments)
+    assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
+
+
+class TestAzureOpenAIResponses:
+    @pytest.mark.covers("llm.responses.azure_openai.basic.nonstream.works")
+    def test_azure_openai_responses_returns_completion(
+        self, endpoints_client: EndpointsClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(endpoints_client, resources, "e2e-responses-azure", _azure_openai_params())
+
+        result = endpoints_client.responses(key, model, BASIC_PROMPT)
+        require_successful_call(result)
+        parsed = ResponsesResult.model_validate_json(result.body)
+        assert parsed.text.strip(), f"/responses over azure openai returned no output text: {result.body[:300]}"
+
+    @pytest.mark.covers("llm.responses.azure_openai.tool_use.nonstream.works")
+    def test_azure_openai_responses_returns_function_call(
+        self, endpoints_client: EndpointsClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(endpoints_client, resources, "e2e-responses-azure-tool", _azure_openai_params())
+
+        result = endpoints_client.responses_with_tools(key, model, WEATHER_PROMPT, [WEATHER_TOOL])
+        require_successful_call(result)
+        _assert_weather_function_call(result.body)
+
+
+class TestVertexResponses:
+    @pytest.mark.covers("llm.responses.vertex.basic.nonstream.works")
+    def test_vertex_responses_returns_completion(
+        self, endpoints_client: EndpointsClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(endpoints_client, resources, "e2e-responses-vertex", _vertex_params())
+
+        result = endpoints_client.responses(key, model, BASIC_PROMPT)
+        require_successful_call(result)
+        parsed = ResponsesResult.model_validate_json(result.body)
+        assert parsed.text.strip(), f"/responses over vertex returned no output text: {result.body[:300]}"
+
+    @pytest.mark.covers("llm.responses.vertex.tool_use.nonstream.works")
+    def test_vertex_responses_returns_function_call(
+        self, endpoints_client: EndpointsClient, resources: ResourceManager
+    ) -> None:
+        model, key = _register(endpoints_client, resources, "e2e-responses-vertex-tool", _vertex_params())
+
+        result = endpoints_client.responses_with_tools(key, model, WEATHER_PROMPT, [WEATHER_TOOL])
+        require_successful_call(result)
+        _assert_weather_function_call(result.body)