feat(auth): add third party auth support through access_token

src/supabase/src/supabase/_async/client.py

@@ -70,6 +70,7 @@ def __init__(
             URL(supabase_url) if supabase_url.endswith("/") else URL(supabase_url + "/")
         )
         self.supabase_key = supabase_key
+        self.access_token = options.access_token
         self.options = copy.copy(options)
         self.options.headers = {
             **options.headers,
@@ -84,20 +85,15 @@ def __init__(
         self.storage_url = self.supabase_url.joinpath("storage", "v1", "")
         self.functions_url = self.supabase_url.joinpath("functions", "v1")
 
-        # Instantiate clients.
-        self.auth = self._init_supabase_auth_client(
-            auth_url=str(self.auth_url),
-            client_options=self.options,
-        )
         self.realtime = self._init_realtime_client(
             realtime_url=self.realtime_url,
             supabase_key=self.supabase_key,
             options=self.options.realtime if self.options else None,
         )
+        self._auth: Optional[AsyncSupabaseAuthClient] = None
         self._postgrest: Optional[AsyncPostgrestClient] = None
         self._storage: Optional[AsyncStorageClient] = None
         self._functions: Optional[AsyncFunctionsClient] = None
-        self.auth.on_auth_state_change(self._listen_to_auth_events)
 
     @classmethod
     async def create(
@@ -110,16 +106,18 @@ async def create(
         client = cls(supabase_url, supabase_key, options)
 
         if auth_header is None:
-            try:
-                session = await client.auth.get_session()
-                session_access_token = (
-                    client._create_auth_header(session.access_token)
-                    if session
-                    else None
-                )
-            except Exception:
-                session_access_token = None
-
+            if client.access_token:
+                session_access_token: Optional[str] = await client.access_token()
+            else:
+                try:
+                    session = await client.auth.get_session()
+                    session_access_token = (
+                        client._create_auth_header(session.access_token)
+                        if session
+                        else None
+                    )
+                except Exception:
+                    session_access_token = None
             client.options.headers.update(
                 client._get_auth_headers(session_access_token)
             )
@@ -179,6 +177,21 @@ def rpc(
             params = {}
         return self.postgrest.rpc(fn, params, count, head, get)
 
+    @property
+    def auth(self) -> AsyncSupabaseAuthClient:
+        if not self.access_token:
+            if self._auth is None:
+                self._auth = self._init_supabase_auth_client(
+                    auth_url=str(self.auth_url),
+                    client_options=self.options,
+                )
+                self.auth.on_auth_state_change(self._listen_to_auth_events)
+            return self._auth
+        else:
+            raise SupabaseException(
+                "supabase_auth cannot be used when 'access_token' option is set."
+            )
+
     @property
     def postgrest(self) -> AsyncPostgrestClient:
         if self._postgrest is None:

src/supabase/src/supabase/_sync/client.py

@@ -69,6 +69,7 @@ def __init__(
             URL(supabase_url) if supabase_url.endswith("/") else URL(supabase_url + "/")
         )
         self.supabase_key = supabase_key
+        self.access_token = options.access_token
         self.options = copy.copy(options)
         self.options.headers = {
             **options.headers,
@@ -83,20 +84,15 @@ def __init__(
         self.storage_url = self.supabase_url.joinpath("storage", "v1", "")
         self.functions_url = self.supabase_url.joinpath("functions", "v1")
 
-        # Instantiate clients.
-        self.auth = self._init_supabase_auth_client(
-            auth_url=str(self.auth_url),
-            client_options=self.options,
-        )
         self.realtime = self._init_realtime_client(
             realtime_url=self.realtime_url,
             supabase_key=self.supabase_key,
             options=self.options.realtime if self.options else None,
         )
+        self._auth: Optional[SyncSupabaseAuthClient] = None
         self._postgrest: Optional[SyncPostgrestClient] = None
         self._storage: Optional[SyncStorageClient] = None
         self._functions: Optional[SyncFunctionsClient] = None
-        self.auth.on_auth_state_change(self._listen_to_auth_events)
 
     @classmethod
     def create(
@@ -109,16 +105,18 @@ def create(
         client = cls(supabase_url, supabase_key, options)
 
         if auth_header is None:
-            try:
-                session = client.auth.get_session()
-                session_access_token = (
-                    client._create_auth_header(session.access_token)
-                    if session
-                    else None
-                )
-            except Exception:
-                session_access_token = None
-
+            if client.access_token:
+                session_access_token: Optional[str] = client.access_token()
+            else:
+                try:
+                    session = client.auth.get_session()
+                    session_access_token = (
+                        client._create_auth_header(session.access_token)
+                        if session
+                        else None
+                    )
+                except Exception:
+                    session_access_token = None
             client.options.headers.update(
                 client._get_auth_headers(session_access_token)
             )
@@ -178,6 +176,21 @@ def rpc(
             params = {}
         return self.postgrest.rpc(fn, params, count, head, get)
 
+    @property
+    def auth(self) -> SyncSupabaseAuthClient:
+        if not self.access_token:
+            if self._auth is None:
+                self._auth = self._init_supabase_auth_client(
+                    auth_url=str(self.auth_url),
+                    client_options=self.options,
+                )
+                self.auth.on_auth_state_change(self._listen_to_auth_events)
+            return self._auth
+        else:
+            raise SupabaseException(
+                "supabase_auth cannot be used when 'access_token' option is set."
+            )
+
     @property
     def postgrest(self) -> SyncPostgrestClient:
         if self._postgrest is None:

src/supabase/src/supabase/lib/client_options.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field
-from typing import Dict, Optional, Union
+from typing import Awaitable, Callable, Dict, Optional, Union
 
 from httpx import AsyncClient as AsyncHttpxClient
 from httpx import Client as SyncHttpxClient
@@ -65,6 +65,18 @@ class AsyncClientOptions(ClientOptions):
     httpx_client: Optional[AsyncHttpxClient] = None
     """httpx client instance to be used by the PostgREST, functions, auth and storage clients."""
 
+    access_token: Optional[Callable[[], Awaitable[str]]] = None
+    """
+    An async function for using a third party authentication system with Supabase.
+    The function should return an access token or ID token (JWT) by
+    obtaining it from the third-party auth client library. Note that this
+    function may be called concurrently and many times.
+
+    When set, the `auth` namespace of the Supabase client cannot be used.
+    Create another client if you wish to use Supabase Auth and third-party
+    authentications concurrently in the same application.
+    """
+
     def replace(
         self,
         schema: Optional[str] = None,
@@ -79,6 +91,7 @@ def replace(
         ] = DEFAULT_POSTGREST_CLIENT_TIMEOUT,
         storage_client_timeout: int = DEFAULT_STORAGE_CLIENT_TIMEOUT,
         flow_type: Optional[AuthFlowType] = None,
+        access_token: Optional[Callable[[], Awaitable[str]]] = None,
     ) -> "AsyncClientOptions":
         """Create a new SupabaseClientOptions with changes"""
         client_options = AsyncClientOptions()
@@ -98,6 +111,7 @@ def replace(
             storage_client_timeout or self.storage_client_timeout
         )
         client_options.flow_type = flow_type or self.flow_type
+        client_options.access_token = access_token or self.access_token
         return client_options
 
 
@@ -108,6 +122,18 @@ class SyncClientOptions(ClientOptions):
     httpx_client: Optional[SyncHttpxClient] = None
     """httpx client instance to be used by the PostgREST, functions, auth and storage clients."""
 
+    access_token: Optional[Callable[[], str]] = None
+    """
+    An async function for using a third party authentication system with Supabase.
+    The function should return an access token or ID token (JWT) by
+    obtaining it from the third-party auth client library. Note that this
+    function may be called concurrently and many times.
+
+    When set, the `auth` namespace of the Supabase client cannot be used.
+    Create another client if you wish to use Supabase Auth and third-party
+    authentications concurrently in the same application.
+    """
+
     def replace(
         self,
         schema: Optional[str] = None,
@@ -122,6 +148,7 @@ def replace(
         ] = DEFAULT_POSTGREST_CLIENT_TIMEOUT,
         storage_client_timeout: int = DEFAULT_STORAGE_CLIENT_TIMEOUT,
         flow_type: Optional[AuthFlowType] = None,
+        access_token: Optional[Callable[[], str]] = None,
     ) -> "SyncClientOptions":
         """Create a new SupabaseClientOptions with changes"""
         client_options = SyncClientOptions()
@@ -141,4 +168,5 @@ def replace(
             storage_client_timeout or self.storage_client_timeout
         )
         client_options.flow_type = flow_type or self.flow_type
+        client_options.access_token = access_token or self.access_token
         return client_options

src/supabase/tests/_async/test_client.py

@@ -241,6 +241,27 @@ async def test_custom_headers_immutable() -> None:
     assert client2.options.headers.get("x-app-name") == "apple"
 
 
+async def test_access_token() -> None:
+    url = os.environ["SUPABASE_TEST_URL"]
+    key = os.environ["SUPABASE_TEST_KEY"]
+
+    options = AsyncClientOptions(
+        headers={
+            "x-app-name": "apple",
+            "x-version": "1.0",
+        }
+    )
+
+    client1 = await create_async_client(url, key, options)
+    client2 = await create_async_client(url, key, options)
+
+    client1.options.headers["x-app-name"] = "grapes"
+
+    assert client1.options.headers.get("x-app-name") == "grapes"
+    assert client1.options.headers.get("x-version") == "1.0"
+    assert client2.options.headers.get("x-app-name") == "apple"
+
+
 async def test_httpx_client_base_url_isolation() -> None:
     """Test that shared httpx_client doesn't cause base_url mutation between services.
     This test reproduces the issue where accessing PostgREST after Storage causes

src/supabase/tests/_sync/test_client.py

@@ -241,6 +241,27 @@ def test_custom_headers_immutable() -> None:
     assert client2.options.headers.get("x-app-name") == "apple"
 
 
+def test_access_token() -> None:
+    url = os.environ["SUPABASE_TEST_URL"]
+    key = os.environ["SUPABASE_TEST_KEY"]
+
+    options = ClientOptions(
+        headers={
+            "x-app-name": "apple",
+            "x-version": "1.0",
+        }
+    )
+
+    client1 = create_client(url, key, options)
+    client2 = create_client(url, key, options)
+
+    client1.options.headers["x-app-name"] = "grapes"
+
+    assert client1.options.headers.get("x-app-name") == "grapes"
+    assert client1.options.headers.get("x-version") == "1.0"
+    assert client2.options.headers.get("x-app-name") == "apple"
+
+
 def test_httpx_client_base_url_isolation() -> None:
     """Test that shared httpx_client doesn't cause base_url mutation between services.
     This test reproduces the issue where accessing PostgREST after Storage causes