Route Vertex-vs-Gemini API branches in GoogleModel by client transport instead of provider name (#7280)

Signed-off-by: James Gore <[email protected]>
Co-authored-by: David SF <[email protected]>

pydantic_ai_slim/pydantic_ai/models/google.py

@@ -519,13 +519,34 @@ class GoogleModel(Model[Client]):
         """The model provider."""
         return self._provider.name
 
+    @property
+    def _is_google_cloud(self) -> bool:
+        """Whether requests go to Google Cloud (Vertex) rather than the Gemini Developer API.
+
+        Derived from the client's transport rather than the provider name, because either provider
+        accepts a pre-built `client=` and stores it as-is, so the two can disagree in both
+        directions: a Vertex-backed client in `GoogleProvider` keeps `name` `'google'`, and a
+        Gemini-API client in `GoogleCloudProvider` keeps `name` `'google-cloud'`. The name cannot be
+        reconciled against the client instead — it is persisted in `ModelMessage.provider_name` and
+        checked when replaying history, so changing it breaks replay of anything captured under the
+        old name. `AnthropicModel` reads its transport off the client for the same reason.
+        """
+        return bool(self.client.vertexai)
+
     @property
     def _matching_provider_names(self) -> frozenset[str]:
-        if self.system in _GOOGLE_CLOUD_PROVIDER_NAMES:
-            return _GOOGLE_CLOUD_PROVIDER_NAMES
-        if self.system in _GEMINI_API_PROVIDER_NAMES:
-            return _GEMINI_API_PROVIDER_NAMES
-        return frozenset({self.system})  # pragma: no cover
+        # Keyed on the transport, because the pre-v2 names this set exists to accept were themselves
+        # transport-derived: `GoogleProvider.name` used to return `google-vertex`/`google-gla` off the
+        # client. The union with `self.system` is load-bearing, not cosmetic — a Vertex-backed
+        # `GoogleProvider` stamps `provider_name='google'`, which the Google Cloud set does not contain,
+        # so without it this model would reject its own history. Both consumers depend on this set:
+        # `_content_model_response` replays thinking signatures and native tool parts through it, and
+        # `_validate_uploaded_file` checks `UploadedFile.provider_name` against it.
+        if self.system in _GOOGLE_CLOUD_PROVIDER_NAMES or self.system in _GEMINI_API_PROVIDER_NAMES:
+            names = _GOOGLE_CLOUD_PROVIDER_NAMES if self._is_google_cloud else _GEMINI_API_PROVIDER_NAMES
+            return names | {self.system}
+        # A provider outside both families has no pre-v2 alias to accept, so it only matches itself.
+        return frozenset({self.system})
 
     @cached_property
     def profile(self) -> GoogleModelProfile:
@@ -597,7 +618,7 @@ class GoogleModel(Model[Client]):
         config = CountTokensConfigDict(
             http_options=generation_config.get('http_options'),
         )
-        if self._provider.name not in _GEMINI_API_PROVIDER_NAMES:
+        if self._is_google_cloud:
             # The fields are not supported by the Gemini API per https://github.com/googleapis/python-genai/blob/7e4ec284dc6e521949626f3ed54028163ef9121d/google/genai/models.py#L1195-L1214
             # The Vertex `countTokens` endpoint accepts native/server-side tools (e.g. Google Search grounding), so we
             # forward `tools` as-is to mirror the real request for an accurate count. This intentionally differs from
@@ -673,7 +694,7 @@ class GoogleModel(Model[Client]):
                 )
             image_config['image_size'] = tool.size
 
-        if self.system in _GOOGLE_CLOUD_PROVIDER_NAMES:
+        if self._is_google_cloud:
             if tool.output_format is not None:
                 if tool.output_format not in _GOOGLE_IMAGE_OUTPUT_FORMATS:
                     raise UserError(
@@ -809,7 +830,7 @@ class GoogleModel(Model[Client]):
         if (
             emits_tool_call_invocations
             and self.profile.get('google_supports_server_side_tool_invocations', False)
-            and self.system not in _GOOGLE_CLOUD_PROVIDER_NAMES
+            and not self._is_google_cloud
         ):
             tool_config['include_server_side_tool_invocations'] = True
 
@@ -941,7 +962,7 @@ class GoogleModel(Model[Client]):
             headers.update(extra_headers)
 
         gla_service_tier: _GlaServiceTier | None = None
-        if self.system in _GOOGLE_CLOUD_PROVIDER_NAMES:
+        if self._is_google_cloud:
             headers.update(_google_cloud_service_tier_headers(_resolve_google_cloud_service_tier(model_settings)))
         else:
             gla_service_tier = _resolve_gla_service_tier(model_settings)
@@ -1231,7 +1252,7 @@ class GoogleModel(Model[Client]):
                 f'UploadedFile with `provider_name={file.provider_name!r}` cannot be used with GoogleModel. '
                 f'Expected `provider_name` to be one of {sorted(self._matching_provider_names)!r}.'
             )
-        if self.system in _GOOGLE_CLOUD_PROVIDER_NAMES:
+        if self._is_google_cloud:
             if not file.file_id.startswith('gs://'):
                 raise UserError(
                     f'UploadedFile for GoogleModel (Google Cloud) must use a GCS URI (gs://bucket/path), got: {file.file_id}'
@@ -1256,12 +1277,12 @@ class GoogleModel(Model[Client]):
             file_uri, mime_type = self._validate_uploaded_file(file)
             return ('file', file_uri, mime_type)
         elif isinstance(file, VideoUrl) and (
-            file.is_youtube or (file.url.startswith('gs://') and self.system in _GOOGLE_CLOUD_PROVIDER_NAMES)
+            file.is_youtube or (file.url.startswith('gs://') and self._is_google_cloud)
         ):
             return ('file', file.url, file.media_type)
         elif isinstance(file, FileUrl):
             if file.force_download or (
-                self.system in _GEMINI_API_PROVIDER_NAMES
+                not self._is_google_cloud
                 and not file.url.startswith(r'https://generativelanguage.googleapis.com/v1beta/files')
             ):
                 downloaded_item = await download_item(file, data_format='bytes')

pydantic_ai_slim/pydantic_ai/providers/google.py

@@ -131,7 +131,10 @@ class GoogleProvider(BaseGoogleProvider):
         Args:
             api_key: The [API key](https://ai.google.dev/gemini-api/docs/api-key) to
                 use for authentication. It can also be set via the `GOOGLE_API_KEY` environment variable.
-            client: A pre-initialized client to use.
+            client: A pre-initialized client to use. Stored as-is, so a client built for Google Cloud
+                (`Client(vertexai=True, ...)`, or its current spelling `Client(enterprise=True, ...)`)
+                is used on that transport even though `name` stays `'google'`. `GoogleModel` reads the
+                transport off the client rather than the provider name, so it routes correctly either way.
             http_client: An existing `httpx2.AsyncClient` or legacy `httpx.AsyncClient` to use for making HTTP requests.
             base_url: The base URL for the Gemini API.
             retry_options: HTTP retry options for transient errors (429, 5xx, etc.).

pydantic_ai_slim/pydantic_ai/providers/google_cloud.py

@@ -60,7 +60,10 @@ class GoogleCloudProvider(BaseGoogleProvider):
                 `global` — see the
                 [Vertex AI locations docs](https://cloud.google.com/vertex-ai/generative-ai/docs/learn/locations#available-regions).
                 Can be obtained from the `GOOGLE_CLOUD_LOCATION` environment variable.
-            client: A pre-initialized client to use.
+            client: A pre-initialized client to use. Stored as-is without forcing `vertexai=True`, so a
+                client built for the Gemini Developer API is used on that transport even though `name`
+                stays `'google-cloud'`. `GoogleModel` reads the transport off the client rather than the
+                provider name, so it routes correctly either way.
             http_client: An existing `httpx2.AsyncClient` or legacy `httpx.AsyncClient` to use for making HTTP requests.
             base_url: The base URL for the Google Cloud API.
             retry_options: HTTP retry options for transient errors (429, 5xx, etc.).

tests/models/conftest.py

@@ -9,12 +9,22 @@ import httpx2
 import pytest
 from pydantic import TypeAdapter
 
+from ..conftest import try_import
+
 if TYPE_CHECKING:
     from vcr.cassette import Cassette
 
     from pydantic_ai.models.anthropic import AnthropicModel
+    from pydantic_ai.providers.google import GoogleProvider
+    from pydantic_ai.providers.google_cloud import GoogleCloudProvider
     from tests.cassette_utils import CassetteContext
 
+with try_import() as google_imports:
+    from google.genai import Client
+
+    from pydantic_ai.providers.google import GoogleProvider
+    from pydantic_ai.providers.google_cloud import GoogleCloudProvider
+
 # `validate_json` parses through pydantic-core rather than the stdlib, and types the result without a cast.
 _REQUEST_BODY_ADAPTER = TypeAdapter(dict[str, Any])
 
@@ -174,3 +184,30 @@ def cassette_ctx(request: pytest.FixtureRequest, vcr: Cassette) -> CassetteConte
         test_module=test_module,  # pyright: ignore[reportUnknownArgumentType]
         test_dir=test_dir,
     )
+
+
+@pytest.fixture
+def vertex_client_google_provider() -> GoogleProvider:
+    """A Vertex-backed `genai.Client` wrapped in `GoogleProvider`, the construction from #6792.
+
+    `system` stays `'google'` while the transport is Google Cloud (Vertex), so transport
+    (not the provider name) must drive Vertex-vs-Gemini-API behavior.
+    """
+    if not google_imports():  # pragma: lax no cover
+        pytest.skip('google is not installed')
+
+    return GoogleProvider(client=Client(vertexai=True, project='test-project', location='us-central1'))
+
+
+@pytest.fixture
+def gla_client_google_cloud_provider() -> GoogleCloudProvider:
+    """A Gemini-Developer-API `genai.Client` wrapped in `GoogleCloudProvider`, the mirror of #6792.
+
+    `system` stays `'google-cloud'` while the transport is the Gemini Developer API. `__init__`
+    short-circuits on `client=` before it would force `vertexai=True`, so the two disagree in this
+    direction too and every transport branch has to follow the client rather than the name.
+    """
+    if not google_imports():  # pragma: lax no cover
+        pytest.skip('google is not installed')
+
+    return GoogleCloudProvider(client=Client(vertexai=False, api_key='mock-api-key'))

tests/models/google/test_media.py

@@ -51,6 +51,13 @@ def mapping_model() -> GoogleModel:
     return GoogleModel('gemini-1.5-flash', provider=GoogleProvider(api_key='test-key'))
 
 
+@pytest.fixture
+def vertex_mapping_model(vertex_client_google_provider: GoogleProvider) -> GoogleModel:
+    """Like `mapping_model`, but Google Cloud (Vertex) — built the way #6792 reports,
+    so transport (not the provider name) drives the mapping."""
+    return GoogleModel('gemini-1.5-flash', provider=vertex_client_google_provider)
+
+
 # =============================================================================
 # Per-Part `media_resolution` forwarding via `vendor_metadata`
 # =============================================================================
@@ -61,8 +68,8 @@ class MediaResolutionCase:
     id: str
     content: BinaryContent | ImageUrl | DocumentUrl | UploadedFile
     expected: dict[str, object]
-    system: str | None = None
-    """When set, patch `GoogleModel.system` (e.g. 'google-cloud' for gs:// URIs)."""
+    google_cloud: bool = False
+    """When True, run against the Vertex-backed model (needed for gs:// URIs)."""
 
 
 MEDIA_RESOLUTION_CASES = [
@@ -117,7 +124,7 @@ MEDIA_RESOLUTION_CASES = [
             'file_data': {'file_uri': 'gs://bucket/image.png', 'mime_type': 'image/png'},
             'media_resolution': {'level': 'MEDIA_RESOLUTION_ULTRA_HIGH'},
         },
-        system='google-cloud',
+        google_cloud=True,
     ),
     MediaResolutionCase(
         id='document_url_media_resolution',
@@ -129,7 +136,7 @@ MEDIA_RESOLUTION_CASES = [
             'file_data': {'file_uri': 'gs://bucket/report.pdf', 'mime_type': 'application/pdf'},
             'media_resolution': {'level': 'MEDIA_RESOLUTION_HIGH'},
         },
-        system='google-cloud',
+        google_cloud=True,
     ),
     MediaResolutionCase(
         id='uploaded_file_media_resolution_and_video_metadata',
@@ -157,18 +164,17 @@ MEDIA_RESOLUTION_CASES = [
 
 @pytest.mark.parametrize('case', [pytest.param(c, id=c.id) for c in MEDIA_RESOLUTION_CASES])
 async def test_media_resolution_forwarding(
-    case: MediaResolutionCase, mapping_model: GoogleModel, mocker: MockerFixture
+    case: MediaResolutionCase, mapping_model: GoogleModel, vertex_mapping_model: GoogleModel
 ):
     """`vendor_metadata['media_resolution']` is lifted to the per-Part `media_resolution`
     field for every file type, remaining keys still route to `video_metadata`, and the
     user's `vendor_metadata` dict is never mutated (the mapper works on a copy).
     """
-    if case.system is not None:
-        mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value=case.system)
+    model = vertex_mapping_model if case.google_cloud else mapping_model
 
     original_vendor_metadata = deepcopy(case.content.vendor_metadata)
 
-    content = await mapping_model._map_user_prompt(UserPromptPart(content=[case.content]))  # pyright: ignore[reportPrivateUsage]
+    content = await model._map_user_prompt(UserPromptPart(content=[case.content]))  # pyright: ignore[reportPrivateUsage]
 
     assert content == [case.expected]
     assert case.content.vendor_metadata == original_vendor_metadata
@@ -218,13 +224,12 @@ async def test_uploaded_file_invalid_file_id(allow_model_requests: None, mapping
         await agent.run(['Analyze this file', UploadedFile(file_id='file-abc123', provider_name='google')])
 
 
-async def test_uploaded_file_vertex_requires_gs_uri(mapping_model: GoogleModel, mocker: MockerFixture):
+async def test_uploaded_file_vertex_requires_gs_uri(vertex_mapping_model: GoogleModel):
     """Vertex `UploadedFile` must use a gs:// URI (not Files API https URLs)."""
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
 
     https_files_api = 'https://generativelanguage.googleapis.com/v1beta/files/abc123'
     with pytest.raises(UserError, match='must use a GCS URI'):
-        await mapping_model._map_user_prompt(  # pyright: ignore[reportPrivateUsage]
+        await vertex_mapping_model._map_user_prompt(  # pyright: ignore[reportPrivateUsage]
             UserPromptPart(
                 content=[UploadedFile(file_id=https_files_api, provider_name='google-cloud')],
             )
@@ -273,19 +278,18 @@ async def test_youtube_video_url_without_vendor_metadata(mapping_model: GoogleMo
 # =============================================================================
 
 
-async def test_gcs_video_url_with_vendor_metadata_on_google_cloud(mapping_model: GoogleModel, mocker: MockerFixture):
+async def test_gcs_video_url_with_vendor_metadata_on_google_cloud(vertex_mapping_model: GoogleModel):
     """GCS URIs use file_uri with video_metadata on google-cloud (Vertex).
 
     This is the main fix - GCS URIs were previously falling through to FileUrl
     handling which doesn't pass vendor_metadata as video_metadata.
     """
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
 
     video = VideoUrl(
         url='gs://bucket/video.mp4',
         vendor_metadata={'start_offset': '300s', 'end_offset': '330s'},
     )
-    content = await mapping_model._map_user_prompt(UserPromptPart(content=[video]))  # pyright: ignore[reportPrivateUsage]
+    content = await vertex_mapping_model._map_user_prompt(UserPromptPart(content=[video]))  # pyright: ignore[reportPrivateUsage]
 
     assert len(content) == 1
     assert content[0] == {
@@ -341,15 +345,14 @@ async def test_http_video_url_downloads_on_google(mapping_model: GoogleModel, mo
     ]
 
 
-async def test_http_video_url_uses_file_uri_on_google_cloud(mapping_model: GoogleModel, mocker: MockerFixture):
+async def test_http_video_url_uses_file_uri_on_google_cloud(vertex_mapping_model: GoogleModel):
     """HTTP VideoUrls use file_uri directly on google-cloud (Vertex) with video_metadata."""
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
 
     video = VideoUrl(
         url='https://example.com/video.mp4',
         vendor_metadata={'start_offset': '10s', 'end_offset': '20s'},
     )
-    content = await mapping_model._map_user_prompt(UserPromptPart(content=[video]))  # pyright: ignore[reportPrivateUsage]
+    content = await vertex_mapping_model._map_user_prompt(UserPromptPart(content=[video]))  # pyright: ignore[reportPrivateUsage]
 
     assert len(content) == 1
     assert content[0] == {
@@ -388,20 +391,18 @@ async def test_http_video_url_uses_file_uri_on_google_cloud(mapping_model: Googl
     ],
 )
 async def test_file_url_in_tool_return_on_vertex(
-    mocker: MockerFixture, file_url: VideoUrl | ImageUrl, expected: dict[str, object]
+    vertex_client_google_provider: GoogleProvider, file_url: VideoUrl | ImageUrl, expected: dict[str, object]
 ):
     """Test file URLs use file_data (not download) in tool returns on Vertex."""
-    model = GoogleModel('gemini-3-flash-preview', provider=GoogleProvider(api_key='test-key'))
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-flash-preview', provider=vertex_client_google_provider)
 
     result = await model._map_file_to_function_response_part(file_url)  # pyright: ignore[reportPrivateUsage]
 
     assert result == expected
 
 
-async def test_map_user_prompt_with_text_content(mapping_model: GoogleModel, mocker: MockerFixture):
+async def test_map_user_prompt_with_text_content(mapping_model: GoogleModel):
     """Test that _map_user_prompt correctly handles a mix of text content and str."""
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google')
 
     user_prompt_part = UserPromptPart(
         content=['Hi', TextContent(content='This is some context', metadata={'source': 'user'})]

tests/models/google/test_output.py

@@ -4,7 +4,7 @@ On supported models (Gemini 2.5+), `VALIDATED` is the default — it enforces th
 schema rewrites, so it's a safe silent improvement — and a caller opts a tool out with `strict=False`.
 
 Test organization:
-1. Mode resolution (unit, against a `MagicMock` client)
+1. Mode resolution (unit, against an offline `genai.Client`)
 2. `strict` resolution via `GoogleJsonSchemaTransformer`
 3. End-to-end wire contract (live recording)
 """
@@ -14,7 +14,6 @@ from __future__ import annotations as _annotations
 import json
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
-from unittest.mock import MagicMock
 
 import httpx2
 import pytest
@@ -28,6 +27,8 @@ from ..._inline_snapshot import snapshot
 from ...conftest import try_import
 
 with try_import() as imports_successful:
+    from google.genai import Client
+
     from pydantic_ai.models.google import GoogleModel
     from pydantic_ai.providers.google import GoogleProvider
 
@@ -130,7 +131,7 @@ def test_google_strict_tools_upgrade_auto_to_validated(case: dict[str, Any]):
     Asserted on the request shape directly rather than via VCR: a cassette replay can't catch the mode we send,
     since it replays a recorded response without re-validating the request against the API.
     """
-    m = GoogleModel(case['model'], provider=GoogleProvider(client=MagicMock()))
+    m = GoogleModel(case['model'], provider=GoogleProvider(client=Client(vertexai=False, api_key='mock-api-key')))
     params = ModelRequestParameters(
         function_tools=case['function_tools'],
         output_tools=case.get('output_tools', []),
@@ -152,7 +153,7 @@ def test_google_strict_resolution_via_transformer():
     """`GoogleJsonSchemaTransformer` treats every schema as `VALIDATED`-compatible (the mode needs no schema
     rewrites): `strict=None` resolves to `True` (VALIDATED-eligible), and an explicit `strict=False` is
     preserved as the per-tool opt-out."""
-    m = GoogleModel('gemini-2.5-flash', provider=GoogleProvider(client=MagicMock()))
+    m = GoogleModel('gemini-2.5-flash', provider=GoogleProvider(client=Client(vertexai=False, api_key='mock-api-key')))
 
     # `strict` left as `None` resolves to `True`: default-on, VALIDATED-eligible.
     params = m.customize_request_parameters(

tests/models/google/test_transport.py

@@ -0,0 +1,255 @@
+"""Vertex-vs-Gemini-API routing, driven by the client's transport rather than the provider name.
+
+Either provider accepts a pre-built `client=` and stores it as-is, so `name` and transport can
+disagree in both directions: a Google Cloud client in `GoogleProvider` keeps `name` `'google'`
+(#6792), and a Gemini Developer API client in `GoogleCloudProvider` keeps `name` `'google-cloud'`.
+`GoogleModel` reads the transport off the client, so both route by where the bytes actually go.
+
+None of these are VCR tests. The routed fields and headers are decided before the request is built,
+the cassette serializer does not persist provider request headers, and the default matchers do not
+inspect the body — so a recording would replay green through a regression in either direction.
+"""
+
+from __future__ import annotations as _annotations
+
+from dataclasses import dataclass
+from types import SimpleNamespace
+
+import pytest
+from pytest_mock import MockerFixture
+
+from pydantic_ai import UploadedFile
+from pydantic_ai.exceptions import UserError
+from pydantic_ai.messages import ModelRequest, UploadedFileProviderName, UserPromptPart
+from pydantic_ai.models import ModelRequestParameters
+from pydantic_ai.native_tools import WebSearchTool
+from pydantic_ai.tools import ToolDefinition
+
+from ...conftest import try_import
+
+with try_import() as imports_successful:
+    from google.genai import Client
+
+    from pydantic_ai.models.google import GoogleModel, GoogleModelSettings
+    from pydantic_ai.providers import Provider
+    from pydantic_ai.providers.google import GoogleProvider
+    from pydantic_ai.providers.google_cloud import GoogleCloudProvider
+
+pytestmark = [
+    pytest.mark.anyio,
+    pytest.mark.skipif(not imports_successful(), reason='google-genai not installed'),
+]
+
+
+async def test_google_cloud_client_in_google_provider_uses_cloud_service_tier_headers(
+    allow_model_requests: None, vertex_client_google_provider: GoogleProvider
+) -> None:
+    """A Google Cloud client wrapped in `GoogleProvider` gets Cloud service-tier handling (#6792)."""
+    m = GoogleModel('gemini-2.5-flash', provider=vertex_client_google_provider)
+    assert m.system == 'google'
+
+    _, config = await m._build_content_and_config(  # pyright: ignore[reportPrivateUsage]
+        messages=[ModelRequest(parts=[UserPromptPart(content='Hello')])],
+        model_settings=GoogleModelSettings(google_cloud_service_tier='pt_only'),
+        model_request_parameters=ModelRequestParameters(),
+    )
+
+    http_options = config.get('http_options')  # pyright: ignore[reportUnknownMemberType]
+    assert http_options is not None
+    headers = http_options.get('headers')
+    assert headers is not None
+    assert headers['X-Vertex-AI-LLM-Request-Type'] == 'dedicated'
+
+
+async def test_google_cloud_service_tier_is_dropped_on_a_gemini_api_transport(
+    allow_model_requests: None, gla_client_google_cloud_provider: GoogleCloudProvider
+) -> None:
+    """`google_cloud_service_tier` has no Gemini API equivalent, so it is dropped on that transport.
+
+    Newly reachable: a Gemini API client in `GoogleCloudProvider` keeps `system == 'google-cloud'`,
+    which used to send Vertex routing headers. Pins that the setting is ignored silently rather than
+    raising, and that it does not leak into the Gemini API's own `service_tier` config field.
+    """
+    m = GoogleModel('gemini-2.5-flash', provider=gla_client_google_cloud_provider)
+    assert m.system == 'google-cloud'
+
+    _, config = await m._build_content_and_config(  # pyright: ignore[reportPrivateUsage]
+        messages=[ModelRequest(parts=[UserPromptPart(content='Hello')])],
+        model_settings=GoogleModelSettings(google_cloud_service_tier='pt_only'),
+        model_request_parameters=ModelRequestParameters(),
+    )
+
+    http_options = config.get('http_options')  # pyright: ignore[reportUnknownMemberType]
+    assert http_options is not None
+    headers = http_options.get('headers')
+    assert headers is not None
+    assert not any(header.startswith('X-Vertex-AI') for header in headers)
+    assert 'service_tier' not in config
+
+
+def test_gemini_api_sets_include_server_side_tool_invocations_on_a_google_cloud_provider(
+    gla_client_google_cloud_provider: GoogleCloudProvider,
+) -> None:
+    """The Gemini Developer API flag is set even when `system` is `'google-cloud'`.
+
+    Newly reachable: a Gemini API client in `GoogleCloudProvider` used to skip the flag because
+    `system == 'google-cloud'`. Pins that transport, not name, drives the SET. Twin of
+    `test_google_gemini_api_sets_include_server_side_tool_invocations` on the agreeing construction.
+
+    Not a VCR test: the field is decided before the request is sent, and cassette matchers do not
+    inspect the body.
+    """
+    model = GoogleModel('gemini-3-pro-preview', provider=gla_client_google_cloud_provider)
+    assert model.system == 'google-cloud'
+    params = ModelRequestParameters(function_tools=[ToolDefinition(name='search')], native_tools=[WebSearchTool()])
+    _tools, tool_config, _image_config = model._get_tool_config(params, GoogleModelSettings())  # pyright: ignore[reportPrivateUsage]
+    assert tool_config is not None
+    assert tool_config.get('include_server_side_tool_invocations') is True
+
+
+async def test_count_tokens_forwards_tools_on_a_google_cloud_transport(
+    allow_model_requests: None,
+    vertex_client_google_provider: GoogleProvider,
+    mocker: MockerFixture,
+) -> None:
+    """Vertex `countTokens` extras follow the client, not `system == 'google'`.
+
+    Twin of `test_google_vertexai_count_tokens_forwards_native_tools` on the agreeing construction.
+    Not a VCR test: tools-forwarding is decided before the request is sent.
+    """
+    m = GoogleModel('gemini-2.5-flash', provider=vertex_client_google_provider)
+    assert m.system == 'google'
+    fake = mocker.AsyncMock(return_value=SimpleNamespace(total_tokens=1))
+    mocker.patch.object(m.client.aio.models, 'count_tokens', fake)
+
+    await m.count_tokens(
+        [ModelRequest(parts=[UserPromptPart(content='Hello')])],
+        None,
+        ModelRequestParameters(native_tools=[WebSearchTool()]),
+    )
+
+    config = fake.call_args.kwargs['config']
+    assert config.get('tools') == [{'google_search': {}}]
+
+
+async def test_count_tokens_omits_tools_on_a_gemini_api_transport(
+    allow_model_requests: None,
+    gla_client_google_cloud_provider: GoogleCloudProvider,
+    mocker: MockerFixture,
+) -> None:
+    """Gemini API `countTokens` does not take Vertex extras, even when `system` is `'google-cloud'`."""
+    m = GoogleModel('gemini-2.5-flash', provider=gla_client_google_cloud_provider)
+    assert m.system == 'google-cloud'
+    fake = mocker.AsyncMock(return_value=SimpleNamespace(total_tokens=1))
+    mocker.patch.object(m.client.aio.models, 'count_tokens', fake)
+
+    await m.count_tokens(
+        [ModelRequest(parts=[UserPromptPart(content='Hello')])],
+        None,
+        ModelRequestParameters(native_tools=[WebSearchTool()]),
+    )
+
+    config = fake.call_args.kwargs['config']
+    assert config.get('tools') is None
+
+
+GCS_URI = 'gs://bucket/doc.pdf'
+FILES_API_URI = 'https://generativelanguage.googleapis.com/v1beta/files/abc'
+
+
+@dataclass(frozen=True)
+class UploadedFileCase:
+    """One construction where the provider name and the client's transport disagree."""
+
+    id: str
+    provider_fixture: str
+    canonical_name: UploadedFileProviderName
+    """The name this construction stamps on its own files — also the one used for the rejection."""
+    also_accepted: UploadedFileProviderName
+    valid_file_id: str
+    """A file id the transport can actually serve."""
+    rejected_file_id: str
+    rejection_match: str
+
+
+UPLOADED_FILE_CASES = [
+    UploadedFileCase(
+        id='google_cloud_client_in_google_provider',
+        provider_fixture='vertex_client_google_provider',
+        canonical_name='google-cloud',
+        also_accepted='google',
+        valid_file_id=GCS_URI,
+        rejected_file_id=FILES_API_URI,
+        rejection_match='must use a GCS URI',
+    ),
+    UploadedFileCase(
+        id='gemini_api_client_in_google_cloud_provider',
+        provider_fixture='gla_client_google_cloud_provider',
+        canonical_name='google',
+        also_accepted='google-cloud',
+        valid_file_id=FILES_API_URI,
+        rejected_file_id=GCS_URI,
+        rejection_match='must use a file URI from the Google Files API',
+    ),
+]
+
+
+@pytest.mark.parametrize('case', [pytest.param(c, id=c.id) for c in UPLOADED_FILE_CASES])
+def test_uploaded_file_validation_follows_the_client_transport(
+    request: pytest.FixtureRequest, case: UploadedFileCase
+) -> None:
+    """`UploadedFile` validation follows the client's transport, not the provider name (#6792).
+
+    Both disagreeing constructions run through one test so the pair cannot drift: each accepts the
+    file id its transport can serve and rejects the other's, whichever way `name` points.
+    """
+    m = GoogleModel('gemini-2.5-flash', provider=request.getfixturevalue(case.provider_fixture))
+
+    for provider_name in (case.canonical_name, case.also_accepted):
+        file = UploadedFile(file_id=case.valid_file_id, provider_name=provider_name, media_type='application/pdf')
+        assert m._validate_uploaded_file(file) == (case.valid_file_id, 'application/pdf')  # pyright: ignore[reportPrivateUsage]
+
+    wrong_transport_file = UploadedFile(
+        file_id=case.rejected_file_id, provider_name=case.canonical_name, media_type='application/pdf'
+    )
+    with pytest.raises(UserError, match=case.rejection_match):
+        m._validate_uploaded_file(wrong_transport_file)  # pyright: ignore[reportPrivateUsage]
+
+
+@pytest.mark.parametrize('vertexai', [True, False])
+def test_provider_outside_both_name_families_matches_only_itself(vertexai: bool) -> None:
+    """A custom `Provider[Client]` keeps its own name as the only accepted one, on either transport.
+
+    `GoogleModel` accepts any `Provider[Client]`, so a third-party provider can carry a name in
+    neither Google family. It has no pre-v2 alias to accept, and inheriting a family's aliases would
+    replay another provider's thinking signatures and native tool parts. Its transport branches still
+    follow its client — that part is what the two directions above also rely on.
+    """
+
+    class MyProxyProvider(Provider[Client]):
+        def __init__(self, client: Client) -> None:
+            self._client = client
+
+        @property
+        def name(self) -> str:
+            return 'my-google-proxy'
+
+        @property
+        def base_url(self) -> str:
+            return 'https://proxy.example.invalid'
+
+        @property
+        def client(self) -> Client:
+            return self._client
+
+    client = (
+        Client(vertexai=True, project='test-project', location='us-central1')
+        if vertexai
+        else Client(vertexai=False, api_key='mock-api-key')
+    )
+    m = GoogleModel('gemini-2.5-flash', provider=MyProxyProvider(client))
+
+    assert m.system == 'my-google-proxy'
+    assert m.base_url == 'https://proxy.example.invalid'
+    assert m._matching_provider_names == frozenset({'my-google-proxy'})  # pyright: ignore[reportPrivateUsage]
+    assert m._is_google_cloud is vertexai  # pyright: ignore[reportPrivateUsage]

tests/models/test_google.py

@@ -3899,12 +3899,9 @@ async def test_google_image_generation_auto_size_raises_error(google_provider: G
         model._get_native_tools(params)  # pyright: ignore[reportPrivateUsage]
 
 
-async def test_google_image_generation_tool_output_format(
-    mocker: MockerFixture, google_provider: GoogleProvider
-) -> None:
+async def test_google_image_generation_tool_output_format(vertex_client_google_provider: GoogleProvider) -> None:
     """Test that ImageGenerationTool.output_format is mapped to ImageConfigDict.output_mime_type on Vertex AI."""
-    model = GoogleModel('gemini-3-pro-image-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-image-preview', provider=vertex_client_google_provider)
     params = ModelRequestParameters(native_tools=[ImageGenerationTool(output_format='png')])
 
     tools, image_config = model._get_native_tools(params)  # pyright: ignore[reportPrivateUsage]
@@ -3913,11 +3910,10 @@ async def test_google_image_generation_tool_output_format(
 
 
 async def test_google_image_generation_tool_unsupported_format_raises_error(
-    mocker: MockerFixture, google_provider: GoogleProvider
+    vertex_client_google_provider: GoogleProvider,
 ) -> None:
     """Test that unsupported output_format values raise an error on Vertex AI."""
-    model = GoogleModel('gemini-3-pro-image-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-image-preview', provider=vertex_client_google_provider)
     # 'gif' is not supported by Google
     params = ModelRequestParameters(native_tools=[ImageGenerationTool(output_format='gif')])  # pyright: ignore[reportArgumentType]
 
@@ -3926,11 +3922,10 @@ async def test_google_image_generation_tool_unsupported_format_raises_error(
 
 
 async def test_google_image_generation_tool_output_compression(
-    mocker: MockerFixture, google_provider: GoogleProvider
+    vertex_client_google_provider: GoogleProvider,
 ) -> None:
     """Test that ImageGenerationTool.output_compression is mapped to ImageConfigDict.output_compression_quality on Vertex AI."""
-    model = GoogleModel('gemini-3-pro-image-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-image-preview', provider=vertex_client_google_provider)
 
     # Test explicit value
     params = ModelRequestParameters(native_tools=[ImageGenerationTool(output_compression=85)])
@@ -3945,11 +3940,10 @@ async def test_google_image_generation_tool_output_compression(
 
 
 async def test_google_image_generation_tool_compression_validation(
-    mocker: MockerFixture, google_provider: GoogleProvider
+    vertex_client_google_provider: GoogleProvider,
 ) -> None:
     """Test compression validation on Vertex AI: range and JPEG-only."""
-    model = GoogleModel('gemini-3-pro-image-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-image-preview', provider=vertex_client_google_provider)
 
     # Invalid range: > 100
     with pytest.raises(UserError, match='`output_compression` must be between 0 and 100'):
@@ -4011,10 +4005,9 @@ async def test_google_vertexai_image_generation_with_output_format(
     assert result.output.media_type == 'image/jpeg'
 
 
-async def test_google_image_generation_tool_all_fields(mocker: MockerFixture, google_provider: GoogleProvider) -> None:
+async def test_google_image_generation_tool_all_fields(vertex_client_google_provider: GoogleProvider) -> None:
     """Test that all ImageGenerationTool fields are mapped correctly on Vertex AI."""
-    model = GoogleModel('gemini-3-pro-image-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-image-preview', provider=vertex_client_google_provider)
     params = ModelRequestParameters(
         native_tools=[ImageGenerationTool(aspect_ratio='16:9', size='2K', output_format='jpeg', output_compression=90)]
     )
@@ -4030,15 +4023,19 @@ async def test_google_image_generation_tool_all_fields(mocker: MockerFixture, go
 
 
 def test_google_vertex_skips_include_server_side_tool_invocations(
-    mocker: MockerFixture, google_provider: GoogleProvider
+    vertex_client_google_provider: GoogleProvider,
 ) -> None:
     """Vertex rejects `include_server_side_tool_invocations`, so it must not be set on Gemini 3+ via Vertex.
 
+    The model is built the way #6792 reports: a
+    Vertex-backed `genai.Client` wrapped in `GoogleProvider`, whose `system` stays `'google'` —
+    the transport, not the provider name, must drive the skip.
+
     Not a VCR test: the field is dropped before the request is sent, and our cassette matchers don't
     inspect the request body, so a recording would stay green if it were reintroduced.
     """
-    model = GoogleModel('gemini-3-pro-preview', provider=google_provider)
-    mocker.patch.object(GoogleModel, 'system', new_callable=mocker.PropertyMock, return_value='google-cloud')
+    model = GoogleModel('gemini-3-pro-preview', provider=vertex_client_google_provider)
+    assert model.system == 'google'
     # A function tool is included so `tool_config` is non-empty on both paths; the only field that
     # should differ is `include_server_side_tool_invocations`.
     params = ModelRequestParameters(function_tools=[ToolDefinition(name='search')], native_tools=[WebSearchTool()])

tests/models/test_multimodal_tool_returns.py

@@ -14,7 +14,6 @@ The SUPPORT_MATRIX determines expected behavior for each (provider, file_type) p
 from __future__ import annotations
 
 import os
-import unittest.mock
 from dataclasses import dataclass
 from itertools import count
 from pathlib import Path
@@ -954,6 +953,7 @@ UPLOADED_FILE_ERROR_CASES: list[UploadedFileErrorCase] = [
 async def test_uploaded_file_validation_error_in_tool_return(
     case: UploadedFileErrorCase,
     bedrock_provider: Any,
+    vertex_client_google_provider: GoogleProvider,
 ) -> None:
     """Test that invalid UploadedFile in a tool return raises UserError before the API call."""
     provider = case.provider
@@ -977,12 +977,9 @@ async def test_uploaded_file_validation_error_in_tool_return(
         with pytest.raises(UserError, match=case.match):
             await m_bedrock._map_messages(messages, params, None)  # pyright: ignore[reportPrivateUsage]
     elif provider == 'google_vertex':
-        m_google = GoogleModel('gemini-3-flash-preview', provider=GoogleProvider(api_key='test-key'))
+        m_google = GoogleModel('gemini-3-flash-preview', provider=vertex_client_google_provider)
         with pytest.raises(UserError, match=case.match):
-            with unittest.mock.patch.object(
-                type(m_google), 'system', new_callable=lambda: property(lambda self: 'google-vertex')
-            ):
-                await m_google._map_messages(messages, params)  # pyright: ignore[reportPrivateUsage]
+            await m_google._map_messages(messages, params)  # pyright: ignore[reportPrivateUsage]
     elif provider == 'openai_responses':
         m_openai = OpenAIResponsesModel('gpt-5-mini', provider=OpenAIProvider(api_key='test-key'))
         with pytest.raises(UserError, match=case.match):
@@ -992,13 +989,15 @@ async def test_uploaded_file_validation_error_in_tool_return(
 
 
 @pytest.mark.skipif(not google_available(), reason='google dependencies not installed')
-async def test_uploaded_file_vertex_valid_gcs_uri() -> None:
-    """Test that a valid Vertex UploadedFile with gs:// URI maps correctly."""
-    model = GoogleModel('gemini-3-flash-preview', provider=GoogleProvider(api_key='test-key'))
+async def test_uploaded_file_vertex_valid_gcs_uri(vertex_client_google_provider: GoogleProvider) -> None:
+    """Test that a valid Vertex UploadedFile with gs:// URI maps correctly.
+
+    The model is Vertex-backed via the client transport (not the provider name), matching #6792.
+    """
+    model = GoogleModel('gemini-3-flash-preview', provider=vertex_client_google_provider)
     file = UploadedFile(file_id='gs://bucket/path/file.pdf', provider_name='google-cloud', media_type='application/pdf')
     messages: list[ModelMessage] = [
         ModelRequest(parts=[ToolReturnPart(tool_name='get_file', content=file, tool_call_id='1')]),
     ]
-    with unittest.mock.patch.object(type(model), 'system', new_callable=lambda: property(lambda self: 'google-vertex')):
-        _, contents = await model._map_messages(messages, ModelRequestParameters())  # pyright: ignore[reportPrivateUsage]
+    _, contents = await model._map_messages(messages, ModelRequestParameters())  # pyright: ignore[reportPrivateUsage]
     assert len(contents) == 1

tests/models/test_tool_choice_unit.py

@@ -47,6 +47,8 @@ with try_import() as openai_available:
     from pydantic_ai.providers.openai import OpenAIProvider
 
 with try_import() as google_available:
+    from google.genai import Client
+
     from pydantic_ai.models.google import GoogleModel
     from pydantic_ai.providers.google import GoogleProvider
 
@@ -770,8 +772,7 @@ def test_bedrock_prepare_request_thinking_auto_output_mode(supports_json_schema:
 @skip_if_no_google
 def test_google_auto_tuple_filters_tool_defs():
     """When resolve_tool_choice returns ('auto', [...]), Google filters tool_defs to only include allowed tools."""
-    mock_client = MagicMock()
-    provider = GoogleProvider(client=mock_client)
+    provider = GoogleProvider(client=Client(vertexai=False, api_key='mock-api-key'))
     m = GoogleModel('gemini-2.0-flash', provider=provider)
     params = ModelRequestParameters(
         function_tools=[make_tool('func')],
@@ -790,7 +791,7 @@ def test_google_auto_tuple_filters_tool_defs():
 
 @skip_if_no_google
 def test_google_allowed_function_names_ignore_unavailable_tools():
-    m = GoogleModel('gemini-2.0-flash', provider=GoogleProvider(client=MagicMock()))
+    m = GoogleModel('gemini-2.0-flash', provider=GoogleProvider(client=Client(vertexai=False, api_key='mock-api-key')))
     params = ModelRequestParameters(function_tools=[make_tool('get_time')], allow_text_output=False)
 
     with pytest.warns(UserWarning, match='not currently available'):
@@ -845,7 +846,7 @@ def test_google_native_tool_only_omits_function_calling_config(case: dict[str, A
     Asserted on the request shape directly rather than via VCR: a cassette replay can't catch a
     malformed request, since it replays a recorded response without re-validating against the API.
     """
-    m = GoogleModel(case['model'], provider=GoogleProvider(client=MagicMock()))
+    m = GoogleModel(case['model'], provider=GoogleProvider(client=Client(vertexai=False, api_key='mock-api-key')))
 
     _, tool_config, _ = m._get_tool_config(case['request_parameters'], {})  # pyright: ignore[reportPrivateUsage]
 

tests/v2/test_google_history_replay.py

@@ -6,6 +6,11 @@ model class accepts both via the `_GEMINI_API_PROVIDER_NAMES` / `_GOOGLE_CLOUD_P
 replay still routes thinking signatures and built-in tool parts correctly. These tests pin
 that contract.
 
+The pre-v2 names were transport-derived, so the set follows the client's transport rather than the
+current `name`, and the two can disagree when a pre-built client is passed to either provider. Each
+model therefore replays its transport's alias and drops the other transport's, whose signatures it
+could not use anyway.
+
 TODO: generalize this into a cross-module history-replay test suite (V2-RULES rule 21).
 """
 
@@ -21,6 +26,8 @@ from pydantic_ai.native_tools import CodeExecutionTool
 from ..conftest import try_import
 
 with try_import() as imports_successful:
+    from google.genai import Client
+
     from pydantic_ai.models.google import GoogleModel, _content_model_response  # pyright: ignore[reportPrivateUsage]
     from pydantic_ai.providers.google import GoogleProvider
     from pydantic_ai.providers.google_cloud import GoogleCloudProvider
@@ -42,11 +49,28 @@ def _google_cloud_model() -> GoogleModel:
     return GoogleModel('gemini-2.5-flash', provider=GoogleCloudProvider(project='p', location='us-central1'))
 
 
+def _vertex_client_in_google_provider_model() -> GoogleModel:
+    """A Google Cloud client handed to `GoogleProvider`, so `name` stays `'google'` (#6792)."""
+    return GoogleModel(
+        'gemini-2.5-flash',
+        provider=GoogleProvider(client=Client(vertexai=True, project='p', location='us-central1')),
+    )
+
+
+def _gla_client_in_google_cloud_provider_model() -> GoogleModel:
+    """A Gemini Developer API client handed to `GoogleCloudProvider`, so `name` stays `'google-cloud'`."""
+    return GoogleModel('gemini-2.5-flash', provider=GoogleCloudProvider(client=Client(vertexai=False, api_key='k')))
+
+
 @pytest.mark.parametrize(
     ('model_factory', 'historical_provider_name', 'current_provider_name'),
     [
         (_gemini_api_model, 'google-gla', 'google'),
         (_google_cloud_model, 'google-vertex', 'google-cloud'),
+        # Name and transport disagree: the accepted set follows the transport, so the pre-v2 alias
+        # that replays is the transport's one, not the one the current `name` would suggest.
+        (_vertex_client_in_google_provider_model, 'google-vertex', 'google'),
+        (_gla_client_in_google_cloud_provider_model, 'google-gla', 'google-cloud'),
     ],
 )
 def test_history_with_legacy_provider_name_still_routes_thinking_signature(
@@ -81,6 +105,10 @@ def test_history_with_legacy_provider_name_still_routes_thinking_signature(
     [
         (_gemini_api_model, 'google-gla', 'google'),
         (_google_cloud_model, 'google-vertex', 'google-cloud'),
+        # Name and transport disagree: the accepted set follows the transport, so the pre-v2 alias
+        # that replays is the transport's one, not the one the current `name` would suggest.
+        (_vertex_client_in_google_provider_model, 'google-vertex', 'google'),
+        (_gla_client_in_google_cloud_provider_model, 'google-gla', 'google-cloud'),
     ],
 )
 def test_history_with_legacy_provider_name_still_replays_builtin_tool_parts(
@@ -112,3 +140,49 @@ def test_history_with_legacy_provider_name_still_replays_builtin_tool_parts(
     parts = content.get('parts') or []
     assert any('executable_code' in p for p in parts)
     assert any('code_execution_result' in p for p in parts)
+
+
+@pytest.mark.parametrize(
+    ('model_factory', 'foreign_provider_name', 'current_provider_name'),
+    [
+        (_gemini_api_model, 'google-vertex', 'google'),
+        (_google_cloud_model, 'google-gla', 'google-cloud'),
+        (_vertex_client_in_google_provider_model, 'google-gla', 'google'),
+        (_gla_client_in_google_cloud_provider_model, 'google-vertex', 'google-cloud'),
+    ],
+)
+def test_history_from_the_other_transport_does_not_replay_its_signature(
+    model_factory: type, foreign_provider_name: str, current_provider_name: str
+) -> None:
+    """A signature captured on the *other* transport is dropped rather than replayed.
+
+    `ThinkingPart.provider_name` documents that signatures are only sent back to the same provider,
+    and a signature minted by the Gemini Developer API is not valid on Google Cloud or vice versa.
+    The positive cases above pin which alias replays; this pins that the other transport's alias
+    does not, so re-widening the set back to both families can't go unnoticed.
+
+    What this cannot pin: the v2 names carry no transport, so a `'google'`-stamped history minted on
+    the Gemini Developer API still replays on a Vertex-backed `GoogleProvider` (and the mirror for
+    `'google-cloud'`). That ambiguity is inherent to the v2 names, not introduced by keying on the
+    transport — the pre-v2 aliases are the only part of the stamp that identifies one.
+    """
+    model = model_factory()
+    assert model.system == current_provider_name
+
+    signature = base64.b64encode(b'sig').decode('ascii')
+    response = ModelResponse(
+        parts=[
+            ThinkingPart(content='reasoning', provider_name=foreign_provider_name, signature=signature),
+            TextPart(content='final'),
+        ],
+        provider_name=foreign_provider_name,
+    )
+
+    accepted = model._matching_provider_names
+    assert foreign_provider_name not in accepted
+
+    content = _content_model_response(response, accepted)
+    assert content is not None
+    parts = content.get('parts') or []
+    text_part = next(p for p in parts if p.get('text') == 'final')
+    assert 'thought_signature' not in text_part