Feature/stream accept tokens (#255)

Author: dangusev

getstream/config.py

@@ -18,6 +18,7 @@ def __init__(
         self.base_url = base_url
         self.params = {"api_key": api_key}
         self.api_key = api_key
+        self.token = token
 
         # Avoid shared mutable defaults and copy any provided headers
         headers_dict = dict(headers) if headers is not None else {}

getstream/stream.py

@@ -31,7 +31,7 @@
 class Settings(BaseSettings):
     # Env names: STREAM_API_KEY, STREAM_API_SECRET, STREAM_BASE_URL, STREAM_TIMEOUT
     api_key: str
-    api_secret: str
+    api_secret: Optional[str] = None
     base_url: Optional[str] = None
     timeout: float = 6.0
 
@@ -50,22 +50,70 @@ def __init__(
         user_agent: Optional[str] = None,
         transport=None,
         http_client=None,
+        token: Optional[str] = None,
     ):
+        """Build a Stream client.
+
+        Pass exactly one of ``api_secret`` or ``token``:
+        - ``api_secret`` enables a server-side client that can mint user tokens
+          and call protected admin endpoints.
+        - ``token`` enables a client-side client authenticated as a single
+          user. Token-only clients cannot mint tokens or call admin endpoints.
+
+        Any of ``api_key``, ``api_secret``, ``base_url`` left as ``None`` are
+        loaded from ``STREAM_*`` env vars; passing ``token`` skips the
+        ``api_secret`` env fallback.
+
+        Args:
+            api_key: Project API key. Falls back to ``STREAM_API_KEY``.
+            api_secret: Project API secret. Mutually exclusive with ``token``.
+                Falls back to ``STREAM_API_SECRET`` only when ``token`` is also
+                ``None``.
+            timeout: HTTP request timeout in seconds; must be > 0.
+            base_url: API base URL. Falls back to ``STREAM_BASE_URL`` then to
+                the SDK default.
+            user_agent: Optional custom ``User-Agent`` string.
+            transport: Optional ``httpx`` transport. Mutually exclusive with
+                ``http_client``.
+            http_client: Optional pre-built ``httpx`` client. Mutually
+                exclusive with ``transport``. When provided, sub-clients
+                (video/chat/moderation) reuse it instead of opening their own.
+            token: Pre-minted user JWT. Mutually exclusive with ``api_secret``.
+
+        Raises:
+            ValueError: If both ``transport`` and ``http_client`` are set; if
+                neither ``api_secret`` nor ``token`` can be resolved; if both
+                are provided; if either is the empty string; if ``api_key`` is
+                missing; or if ``timeout`` is not a positive number.
+        """
         if transport is not None and http_client is not None:
             raise ValueError("Cannot specify both 'transport' and 'http_client'")
 
-        if None in (api_key, api_secret, timeout, base_url):
-            s = Settings()  # loads from env and optional .env
+        # Env fallback for anything not explicitly provided. A caller-supplied
+        # token short-circuits api_secret loading so token-only callers don't
+        # need STREAM_API_SECRET in env.
+        if (
+            api_key is None
+            or base_url is None
+            or (api_secret is None and token is None)
+        ):
+            s = Settings()
             api_key = api_key or s.api_key
-            api_secret = api_secret or s.api_secret
             base_url = base_url or (s.base_url or BASE_URL)
+            if token is None and api_secret is None:
+                api_secret = s.api_secret
 
-        if api_key is None or api_key == "":
+        if not api_key:
             raise ValueError("api_key is required")
-        if api_secret is None or api_secret == "":
-            raise ValueError("api_secret is required")
+        if api_secret and token:
+            raise ValueError("Pass either api_secret or token, not both")
+        if api_secret == "" or token == "":
+            raise ValueError("api_secret and token must not be empty strings")
+        if api_secret is None and token is None:
+            raise ValueError("Either api_secret or token is required")
+
         self.api_key = api_key
-        self.api_secret = api_secret
+        self._api_secret = api_secret or None
 
         if timeout is not None:
             if not isinstance(timeout, (int, float)) or timeout <= 0.0:
@@ -76,7 +124,7 @@ def __init__(
         self.user_agent = user_agent
         self._transport = transport
         self._http_client = http_client
-        self.token = self._create_token()
+        self.token = token or self._create_token()
         super().__init__(
             self.api_key, self.base_url, self.token, self.timeout, self.user_agent
         )
@@ -88,6 +136,25 @@ def __init__(
         else:
             self._shared_client = None
 
+    @property
+    def api_secret(self) -> str:
+        """
+        Get api secret if it's set.
+        Otherwise, raise a ValueError.
+        """
+        if self._api_secret is None:
+            raise ValueError(
+                "api_secret is required; this client was initialized with a token"
+            )
+        return self._api_secret
+
+    @property
+    def has_api_secret(self) -> bool:
+        """
+        Check if api secret is set for this client.
+        """
+        return self._api_secret is not None
+
     def _apply_shared_client(self, sub_client):
         """Replace a sub-client's auto-created httpx client with the shared
         one built from user-provided transport/http_client config."""
@@ -131,6 +198,20 @@ def create_token(
             user_id=user_id, expiration=expiration, iat=int(time.time()) - 5
         )
 
+    def clone_for_token(self, token: str):
+        """Return a sibling client authenticated with the given user token.
+
+        Keeps this client's ``api_key`` and ``base_url``. The clone is
+        token-only; it cannot mint further tokens.
+        """
+        return self.__class__(
+            api_key=self.api_key,
+            token=token,
+            base_url=self.base_url,
+            timeout=self.timeout,
+            user_agent=self.user_agent,
+        )
+
     def create_call_token(
         self,
         user_id: str,
@@ -358,7 +439,8 @@ def from_env(cls, timeout: float = 6.0) -> Stream:
     def as_async(self) -> "AsyncStream":
         return AsyncStream(
             api_key=self.api_key,
-            api_secret=self.api_secret,
+            api_secret=self._api_secret,
+            token=None if self.has_api_secret else self.token,
             timeout=self.timeout,
             base_url=self.base_url,
             user_agent=self.user_agent,

getstream/video/rtc/connection_manager.py

@@ -307,9 +307,11 @@ async def _connect_coordinator_ws(self):
             with telemetry.start_as_current_span(
                 "coordinator-ws-connect",
             ):
+                stream = self.call.client.stream
                 self._coordinator_ws_client = StreamAPIWS(
                     call=self.call,
                     user_details={"id": self.user_id},
+                    user_token=None if stream.has_api_secret else stream.token,
                 )
                 self._coordinator_ws_client.on_wildcard("*", _log_event)
                 self._coordinator_ws_client.on(

getstream/video/rtc/connection_utils.py

@@ -107,33 +107,22 @@ class ConnectionOptions:
     previous_session_id: Optional[str] = None
 
 
-def user_client(call: Call, user_id: str):
-    token = call.client.stream.create_token(user_id=user_id)
-    client = call.client.stream.__class__(
-        api_key=call.client.stream.api_key,
-        api_secret=call.client.stream.api_secret,
-        base_url=call.client.stream.base_url,
-    )
-    # set up authentication
-    client.token = token
-    client.headers["authorization"] = token
-    client.client.headers["authorization"] = token
-    return client
-
-
 async def watch_call(call: Call, user_id: str, connection_id: str):
-    client = user_client(call, user_id)
-
-    # Make the POST request to join the call
-    return await client.post(
-        "/api/v2/video/call/{type}/{id}",
-        JoinCallResponse,
-        path_params={
-            "type": call.call_type,
-            "id": call.id,
-        },
-        query_params=build_query_param(connection_id=connection_id),
+    stream = call.client.stream
+    token = (
+        stream.create_token(user_id=user_id) if stream.has_api_secret else stream.token
     )
+    # Make the POST request to join the call
+    async with stream.clone_for_token(token) as client:
+        return await client.post(
+            "/api/v2/video/call/{type}/{id}",
+            JoinCallResponse,
+            path_params={
+                "type": call.call_type,
+                "id": call.id,
+            },
+            query_params=build_query_param(connection_id=connection_id),
+        )
 
 
 async def join_call(
@@ -195,7 +184,6 @@ async def join_call_coordinator_request(
     Returns:
         A response containing the call information and credentials
     """
-    client = user_client(call, user_id)
 
     # Prepare path parameters for the request
     path_params = {
@@ -218,12 +206,19 @@ async def join_call_coordinator_request(
         json_body["migrating_from_list"] = migrating_from_list
 
     # Make the POST request to join the call
-    return await client.post(
-        "/api/v2/video/call/{type}/{id}/join",
-        JoinCallResponse,
-        path_params=path_params,
-        json=json_body,
+    stream = call.client.stream
+    # Create a new client instance with the token for the given user_id
+    token = (
+        stream.create_token(user_id=user_id) if stream.has_api_secret else stream.token
     )
+    client = stream.clone_for_token(token)
+    async with client:
+        return await client.post(
+            "/api/v2/video/call/{type}/{id}/join",
+            JoinCallResponse,
+            path_params=path_params,
+            json=json_body,
+        )
 
 
 async def create_join_request(token: str, session_id: str) -> events_pb2.JoinRequest:

getstream/video/rtc/coordinator/ws.py

@@ -15,7 +15,6 @@
 
 from websockets import ClientConnection
 
-from getstream import AsyncStream
 from getstream.utils import StreamAsyncIOEventEmitter
 from getstream.video.async_call import Call
 from .errors import (
@@ -192,15 +191,10 @@ async def _open_socket(self) -> dict:
             return message
 
         # make sure we update connection_id to subscribe to events
-        client = AsyncStream(
-            api_key=self.call.client.stream.api_key,
-            api_secret=self.call.client.stream.api_secret,
-            base_url=self.call.client.stream.base_url,
-        )
+        if self.user_token is None:
+            raise ValueError("user_token is required")
 
-        client.token = self.user_token
-        client.headers["authorization"] = self.user_token
-        client.client.headers["authorization"] = self.user_token  # type: ignore[index]
+        client = self.call.client.stream.clone_for_token(self.user_token)
         if self.call.id is None:
             raise ValueError("call.id is required")
         path_params = {

getstream/video/rtc/coordinator_api.py

@@ -2,6 +2,8 @@
 addopts = --doctest-modules -s --timeout=120 --import-mode=importlib
 testpaths = tests getstream
 
+asyncio_mode = auto
+
 log_cli = true
 log_cli_level = INFO
 log_cli_format = %(asctime)s - %(name)s - %(levelname)s - %(message)s