@@ -121,7 +121,7 @@ class Request(transport.Request):
     .. automethod:: __call__
     """
 
-    def __init__(self, session: aiohttp.ClientSession = None):
+    def __init__(self, session: Optional[aiohttp.ClientSession] = None):
         self._session = session
         self._closed = False
 
 
@@ -17,7 +17,12 @@
 """
 
 import asyncio
+import contextlib
 import logging
+import os
+import ssl
+import tempfile
+from typing import Optional
 
 from google.auth import exceptions
 import google.auth.transport._mtls_helper
@@ -26,6 +31,61 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+@contextlib.contextmanager
+def _create_temp_file(content: bytes):
+    """Creates a temporary file with the given content.
+
+    Args:
+        content (bytes): The content to write to the file.
+
+    Yields:
+        str: The path to the temporary file.
+    """
+    # Create a temporary file that is readable only by the owner.
+    fd, file_path = tempfile.mkstemp()
+    try:
+        with os.fdopen(fd, "wb") as f:
+            f.write(content)
+        yield file_path
+    finally:
+        # Securely delete the file after use.
+        if os.path.exists(file_path):
+            os.remove(file_path)
+
+
+def make_client_cert_ssl_context(
+    cert_bytes: bytes, key_bytes: bytes, passphrase: Optional[bytes] = None
+) -> ssl.SSLContext:
+    """Creates an SSLContext with the given client certificate and key.
+    This function writes the certificate and key to temporary files so that
+    ssl.create_default_context can load them, as the ssl module requires
+    file paths for client certificates. These temporary files are deleted
+    immediately after the SSL context is created.
+    Args:
+        cert_bytes (bytes): The client certificate content in PEM format.
+        key_bytes (bytes): The client private key content in PEM format.
+        passphrase (Optional[bytes]): The passphrase for the private key, if any.
+    Returns:
+        ssl.SSLContext: The configured SSL context with client certificate.
+
+    Raises:
+        google.auth.exceptions.TransportError: If there is an error loading the certificate.
+    """
+    with _create_temp_file(cert_bytes) as cert_path, _create_temp_file(
+        key_bytes
+    ) as key_path:
+        try:
+            context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
+            context.load_cert_chain(
+                certfile=cert_path, keyfile=key_path, password=passphrase
+            )
+            return context
+        except (ssl.SSLError, OSError, IOError, ValueError, RuntimeError) as exc:
+            raise exceptions.TransportError(
+                "Failed to load client certificate and key for mTLS."
+            ) from exc
+
+
 async def _run_in_executor(func, *args):
     """Run a blocking function in an executor to avoid blocking the event loop.
 
@@ -44,7 +104,7 @@ def default_client_cert_source():
     """Get a callback which returns the default client SSL credentials.
 
     Returns:
-        Awaitable[Callable[[], [bytes, bytes]]]: A callback which returns the default
+        Awaitable[Callable[[], Tuple[bytes, bytes]]]: A callback which returns the default
             client certificate bytes and private key bytes, both in PEM format.
 
     Raises:
 
@@ -21,17 +21,22 @@
 from google.auth import _exponential_backoff, exceptions
 from google.auth.aio import transport
 from google.auth.aio.credentials import Credentials
+from google.auth.aio.transport import mtls
 from google.auth.exceptions import TimeoutError
+import google.auth.transport._mtls_helper
 
 if TYPE_CHECKING:  # pragma: NO COVER
+    import aiohttp
     from aiohttp import ClientTimeout  # type: ignore
 
 else:
     try:
+        import aiohttp
         from aiohttp import ClientTimeout
     except (ImportError, AttributeError):
         ClientTimeout = None
 
+# Tracks the internal aiohttp installation and usage
 try:
     from google.auth.aio.transport.aiohttp import Request as AiohttpRequest
 
@@ -133,12 +138,88 @@ def __init__(
         _auth_request = auth_request
         if not _auth_request and AIOHTTP_INSTALLED:
             _auth_request = AiohttpRequest()
+        self._is_mtls = False
+        self._mtls_init_task = None
+        self._cached_cert = None
         if _auth_request is None:
             raise exceptions.TransportError(
                 "`auth_request` must either be configured or the external package `aiohttp` must be installed to use the default value."
             )
         self._auth_request = _auth_request
 
+    async def configure_mtls_channel(self, client_cert_callback=None):
+        """Configure the client certificate and key for SSL connection.
+
+        The function does nothing unless `GOOGLE_API_USE_CLIENT_CERTIFICATE` is
+        explicitly set to `true`. In this case if client certificate and key are
+        successfully obtained (from the given client_cert_callback or from application
+        default SSL credentials), the underlying transport will be reconfigured
+        to use mTLS.
+        Note: This function does nothing if the `aiohttp` library is not
+        installed.
+        Important: Calling this method will close any ongoing API requests associated
+        with the current session. To ensure a smooth transition, it is recommended
+        to call this during session initialization.
+
+        Args:
+            client_cert_callback (Optional[Callable[[], (bytes, bytes)]]):
+                The optional callback returns the client certificate and private
+                key bytes both in PEM format.
+                If the callback is None, application default SSL credentials
+                will be used.
+
+        Raises:
+            google.auth.exceptions.MutualTLSChannelError: If mutual TLS channel
+                creation failed for any reason.
+        """
+        if self._mtls_init_task is None:
+
+            async def _do_configure():
+                # Run the blocking check in an executor
+                use_client_cert = await mtls._run_in_executor(
+                    google.auth.transport._mtls_helper.check_use_client_cert
+                )
+                if not use_client_cert:
+                    self._is_mtls = False
+                    return
+
+                try:
+                    (
+                        self._is_mtls,
+                        cert,
+                        key,
+                    ) = await mtls.get_client_cert_and_key(client_cert_callback)
+
+                    if self._is_mtls:
+                        self._cached_cert = cert
+                        ssl_context = await mtls._run_in_executor(
+                            mtls.make_client_cert_ssl_context, cert, key
+                        )
+
+                        # Re-create the auth request with the new SSL context
+                        if AIOHTTP_INSTALLED and isinstance(
+                            self._auth_request, AiohttpRequest
+                        ):
+                            connector = aiohttp.TCPConnector(ssl=ssl_context)
+                            new_session = aiohttp.ClientSession(connector=connector)
+
+                            old_auth_request = self._auth_request
+                            self._auth_request = AiohttpRequest(session=new_session)
+
+                            await old_auth_request.close()
+
+                except (
+                    exceptions.ClientCertError,
+                    ImportError,
+                    OSError,
+                ) as caught_exc:
+                    new_exc = exceptions.MutualTLSChannelError(caught_exc)
+                    raise new_exc from caught_exc
+
+            self._mtls_init_task = asyncio.create_task(_do_configure())
+
+        return await self._mtls_init_task
+
     async def request(
         self,
         method: str,
@@ -182,22 +263,37 @@ async def request(
                 the configured `max_allowed_time` or the request exceeds the configured
                 `timeout`.
         """
-
+        if self._mtls_init_task:
+            try:
+                await self._mtls_init_task
+            except Exception:
+                # Suppress all exceptions from the background mTLS initialization task,
+                # allowing the request to fail naturally elsewhere.
+                pass
         retries = _exponential_backoff.AsyncExponentialBackoff(
             total_attempts=total_attempts,
         )
+        if headers is None:
+            headers = {}
         async with timeout_guard(max_allowed_time) as with_timeout:
             await with_timeout(
                 # Note: before_request will attempt to refresh credentials if expired.
                 self._credentials.before_request(
                     self._auth_request, method, url, headers
                 )
             )
+            actual_timeout: float = 0.0
+            if ClientTimeout is not None and isinstance(timeout, ClientTimeout):
+                actual_timeout = timeout.total if timeout.total is not None else 0.0
+            elif isinstance(timeout, (int, float)):
+                actual_timeout = float(timeout)
             # Workaround issue in python 3.9 related to code coverage by adding `# pragma: no branch`
             # See https://github.com/googleapis/gapic-generator-python/pull/1174#issuecomment-1025132372
             async for _ in retries:  # pragma: no branch
                 response = await with_timeout(
-                    self._auth_request(url, method, data, headers, timeout, **kwargs)
+                    self._auth_request(
+                        url, method, data, headers, actual_timeout, **kwargs
+                    )
                 )
                 if response.status_code not in transport.DEFAULT_RETRYABLE_STATUS_CODES:
                     break
@@ -468,6 +564,11 @@ async def delete(
             **kwargs,
         )
 
+    @property
+    def is_mtls(self):
+        """Indicates if mutual TLS is enabled."""
+        return self._is_mtls
+
     async def close(self) -> None:
         """
         Close the underlying auth request session.
 
@@ -91,7 +91,7 @@ def blacken(session):
 @nox.session(python=DEFAULT_PYTHON_VERSION)
 def mypy(session):
     """Verify type hints are mypy compatible."""
-    session.install("-e", ".")
+    session.install("-e", ".[aiohttp]")
     session.install(
         "mypy",
         "types-certifi",
 
@@ -0,0 +1,123 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import os
+import ssl
+from unittest import mock
+
+import pytest
+
+from google.auth import exceptions
+from google.auth.aio import credentials
+from google.auth.aio.transport import sessions
+
+# This is the valid "workload" format the library expects
+VALID_WORKLOAD_CONFIG = {
+    "version": 1,
+    "cert_configs": {
+        "workload": {"cert_path": "/tmp/mock_cert.pem", "key_path": "/tmp/mock_key.pem"}
+    },
+}
+
+
+class TestSessionsMtls:
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel(self):
+        """
+        Tests that the mTLS channel configures correctly when a
+        valid workload config is mocked.
+        """
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists, mock.patch(
+            "builtins.open", mock.mock_open(read_data=json.dumps(VALID_WORKLOAD_CONFIG))
+        ), mock.patch(
+            "google.auth.aio.transport.mtls.get_client_cert_and_key"
+        ) as mock_helper, mock.patch(
+            "google.auth.aio.transport.mtls.make_client_cert_ssl_context"
+        ) as mock_make_context:
+            mock_exists.return_value = True
+            mock_helper.return_value = (True, b"fake_cert_data", b"fake_key_data")
+
+            mock_context = mock.Mock(spec=ssl.SSLContext)
+            mock_make_context.return_value = mock_context
+
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            await session.configure_mtls_channel()
+
+            assert session._is_mtls is True
+            mock_make_context.assert_called_once_with(
+                b"fake_cert_data", b"fake_key_data"
+            )
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_disabled(self):
+        """
+        Tests behavior when the config file does not exist.
+        """
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists:
+            mock_exists.return_value = False
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            await session.configure_mtls_channel()
+
+            # If the file doesn't exist, it shouldn't error; it just won't use mTLS
+            assert session._is_mtls is False
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_invalid_format(self):
+        """
+        Verifies that the MutualTLSChannelError is raised for bad formats.
+        """
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch("os.path.exists") as mock_exists, mock.patch(
+            "builtins.open", mock.mock_open(read_data='{"invalid": "format"}')
+        ):
+            mock_exists.return_value = True
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                await session.configure_mtls_channel()
+
+    @pytest.mark.asyncio
+    async def test_configure_mtls_channel_mock_callback(self):
+        """
+        Tests mTLS configuration using bytes-returning callback.
+        """
+
+        def mock_callback():
+            return (b"fake_cert_bytes", b"fake_key_bytes")
+
+        with mock.patch.dict(
+            os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}
+        ), mock.patch(
+            "google.auth.transport.mtls.has_default_client_cert_source",
+            return_value=True,
+        ), mock.patch(
+            "ssl.SSLContext.load_cert_chain"
+        ):
+            mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+            session = sessions.AsyncAuthorizedSession(mock_creds)
+
+            await session.configure_mtls_channel(client_cert_callback=mock_callback)
+
+            assert session._is_mtls is True