feat: allow custom HTTP transport and client configuration

Add `transport` and `http_client` parameters to `Stream` and `AsyncStream`
so users can configure connection pool limits, retries, SSL, HTTP/2, and
other httpx transport-level settings — matching the Java SDK pattern where
a pre-built OkHttpClient can be passed in.

- `transport` (primary): user provides an `httpx.HTTPTransport` /
  `httpx.AsyncHTTPTransport`; the SDK builds its own client with it and
  manages the lifecycle.
- `http_client` (escape hatch): user provides a fully built
  `httpx.Client` / `httpx.AsyncClient`; the SDK layers auth headers and
  params on top. Caller manages lifecycle.

When either is provided, all sub-clients (video, chat, moderation, feeds)
share a single underlying httpx client via the `stream` back-reference
instead of each creating their own.

Made-with: Cursor

Author: itsmeadi

getstream/base.py

@@ -158,12 +158,44 @@ def __init__(
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.client = httpx.Client(
-            base_url=self.base_url or "",
-            headers=self.headers,
-            params=self.params,
-            timeout=httpx.Timeout(self.timeout),
-        )
+        http_client = self._resolve_http_client()
+        if http_client is not None:
+            if not isinstance(http_client, httpx.Client):
+                raise TypeError(
+                    f"http_client must be an httpx.Client instance, "
+                    f"got {type(http_client).__name__}"
+                )
+            http_client.headers.update(self.headers)
+            http_client.params = http_client.params.merge(self.params)
+            http_client.base_url = self.base_url or ""
+            if self.timeout is not None:
+                http_client.timeout = httpx.Timeout(self.timeout)
+            self.client = http_client
+            self._owns_http_client = False
+        else:
+            client_kwargs = dict(
+                base_url=self.base_url or "",
+                headers=self.headers,
+                params=self.params,
+                timeout=httpx.Timeout(self.timeout),
+            )
+            transport = getattr(self, "_transport", None)
+            if transport is not None:
+                client_kwargs["transport"] = transport
+            self.client = httpx.Client(**client_kwargs)
+            self._owns_http_client = True
+
+    def _resolve_http_client(self):
+        """Return a pre-built httpx client if one was provided, checking both
+        the instance attribute (set by BaseStream) and the parent stream
+        back-reference (set by sub-clients like VideoClient)."""
+        client = getattr(self, "_http_client", None)
+        if client is not None:
+            return client
+        stream = getattr(self, "stream", None)
+        if stream is not None:
+            return getattr(stream, "_shared_client", None)
+        return None
 
     def __enter__(self):
         return self
@@ -348,8 +380,13 @@ def _upload_multipart(
     def close(self):
         """
         Close HTTPX client.
+
+        If the client was provided externally via ``http_client``, this is a
+        no-op — the caller that created the client is responsible for closing
+        it.
         """
-        self.client.close()
+        if getattr(self, "_owns_http_client", True):
+            self.client.close()
 
 
 class AsyncBaseClient(TelemetryEndpointMixin, BaseConfig, ResponseParserMixin, ABC):
@@ -368,12 +405,41 @@ def __init__(
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.client = httpx.AsyncClient(
-            base_url=self.base_url or "",
-            headers=self.headers,
-            params=self.params,
-            timeout=httpx.Timeout(self.timeout),
-        )
+        http_client = self._resolve_http_client()
+        if http_client is not None:
+            if not isinstance(http_client, httpx.AsyncClient):
+                raise TypeError(
+                    f"http_client must be an httpx.AsyncClient instance, "
+                    f"got {type(http_client).__name__}"
+                )
+            http_client.headers.update(self.headers)
+            http_client.params = http_client.params.merge(self.params)
+            http_client.base_url = self.base_url or ""
+            if self.timeout is not None:
+                http_client.timeout = httpx.Timeout(self.timeout)
+            self.client = http_client
+            self._owns_http_client = False
+        else:
+            client_kwargs = dict(
+                base_url=self.base_url or "",
+                headers=self.headers,
+                params=self.params,
+                timeout=httpx.Timeout(self.timeout),
+            )
+            transport = getattr(self, "_transport", None)
+            if transport is not None:
+                client_kwargs["transport"] = transport
+            self.client = httpx.AsyncClient(**client_kwargs)
+            self._owns_http_client = True
+
+    def _resolve_http_client(self):
+        client = getattr(self, "_http_client", None)
+        if client is not None:
+            return client
+        stream = getattr(self, "stream", None)
+        if stream is not None:
+            return getattr(stream, "_shared_client", None)
+        return None
 
     async def __aenter__(self):
         return self
@@ -382,8 +448,14 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.aclose()
 
     async def aclose(self):
-        """Close HTTPX async client (closes pools/keep-alives)."""
-        await self.client.aclose()
+        """Close HTTPX async client (closes pools/keep-alives).
+
+        If the client was provided externally via ``http_client``, this is a
+        no-op — the caller that created the client is responsible for closing
+        it.
+        """
+        if getattr(self, "_owns_http_client", True):
+            await self.client.aclose()
 
     async def _upload_multipart(
         self,

getstream/chat/async_client.py

@@ -15,14 +15,14 @@
 
 class ChatClient(ChatRestClient):
     def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=None):
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream
 
     def channel(self, call_type: str, id: str) -> Channel:
         return Channel(self, call_type, id)

getstream/chat/client.py

@@ -15,14 +15,14 @@
 
 class ChatClient(ChatRestClient):
     def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=None):
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream
 
     def channel(self, call_type: str, id: str) -> Channel:
         return Channel(self, call_type, id)

getstream/feeds/client.py

@@ -5,14 +5,14 @@
 
 class FeedsClient(FeedsRestClient):
     def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=None):
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream
 
     def feed(
         self, feed_type: str, feed_id: str, custom_data: Optional[Dict] = None

getstream/moderation/async_client.py

@@ -3,11 +3,11 @@
 
 class ModerationClient(ModerationRestClient):
     def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=None):
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream

getstream/moderation/client.py

@@ -3,11 +3,11 @@
 
 class ModerationClient(ModerationRestClient):
     def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=None):
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream

getstream/stream.py

@@ -47,7 +47,12 @@ def __init__(
         timeout: Optional[float] = 6.0,
         base_url: Optional[str] = BASE_URL,
         user_agent: Optional[str] = None,
+        transport=None,
+        http_client=None,
     ):
+        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
             api_key = api_key or s.api_key
@@ -68,10 +73,19 @@ def __init__(
 
         self.base_url = validate_and_clean_url(base_url)
         self.user_agent = user_agent
+        self._transport = transport
+        self._http_client = http_client
         self.token = self._create_token()
         super().__init__(
             self.api_key, self.base_url, self.token, self.timeout, self.user_agent
         )
+        # After super().__init__(), self.client is fully built and configured.
+        # When the user provided custom HTTP config, sub-clients share this
+        # client instead of each building their own.
+        if transport is not None or http_client is not None:
+            self._shared_client = self.client
+        else:
+            self._shared_client = None
 
     def create_token(
         self,

getstream/video/async_client.py

@@ -12,14 +12,14 @@ def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=No
         :param timeout: A number representing the time limit for a request
         :param user_agent: Optional custom user agent string
         """
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream
 
     def call(self, call_type: str, id: str) -> Call:
         return Call(self, call_type, id)

getstream/video/client.py

@@ -12,14 +12,14 @@ def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=No
         :param timeout: A number representing the time limit for a request
         :param user_agent: Optional custom user agent string
         """
+        self.stream = stream
         super().__init__(
             api_key=api_key,
             base_url=base_url,
             token=token,
             timeout=timeout,
             user_agent=user_agent,
         )
-        self.stream = stream
 
     def call(self, call_type: str, id: str) -> Call:
         return Call(self, call_type, id)