refactor(python-sdk): drop per-thread/per-loop client caching from ApiClient

With the pyqwest transports thread-safe and loop-independent, the
transport_factory/async_transport_factory plumbing, the thread-local
httpx.Client cache, and the per-loop WeakKeyDictionary of AsyncClients are
dead weight: a single lazily-created httpx client (the generated base
behavior, same shape the volume client already uses) serves all threads and
event loops over the shared per-proxy pool.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: mishushakov

.changeset/python-pyqwest-rest-transport.md

@@ -11,7 +11,9 @@ event hooks, per-request timeouts, and headers behave as before.
 
 Because pyqwest transports are thread-safe and loop-independent (I/O runs on
 a Rust runtime), the API connection pool is now shared process-wide per
-proxy, instead of one pool per thread (sync) or per event loop (async).
+proxy, instead of one pool per thread (sync) or per event loop (async), and
+`ApiClient` no longer maintains per-thread/per-loop httpx client caches — a
+single httpx client serves all threads and event loops.
 Connection-establishment failures are retried with backoff
 (`E2B_CONNECTION_RETRIES`, default 3), matching the connect-only retries of
 the previous transports.

packages/python-sdk/e2b/api/__init__.py

@@ -1,13 +1,10 @@
-import asyncio
 import json
 import logging
 import os
 import re
-import threading
-import weakref
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Callable, Optional, Protocol, Union
+from typing import Optional, Protocol, Union
 
 import httpx
 from httpx import AsyncBaseTransport, BaseTransport, Limits, Timeout
@@ -182,31 +179,20 @@ def validate_api_key(api_key: str) -> None:
 class ApiClient(AuthenticatedClient):
     """
     The client for interacting with the E2B API.
+
+    A single lazily-created httpx client (see the generated
+    ``AuthenticatedClient``) serves all threads and event loops: the pyqwest
+    transports it delegates to are thread-safe and loop-independent, and
+    ``httpx.Client`` is documented thread-safe.
     """
 
     def __init__(
         self,
         config: ConnectionConfig,
         transport: Optional[Union[BaseTransport, AsyncBaseTransport]] = None,
-        transport_factory: Optional[Callable[[], BaseTransport]] = None,
-        async_transport_factory: Optional[Callable[[], AsyncBaseTransport]] = None,
         *args,
         **kwargs,
     ):
-        if transport is not None and (
-            transport_factory is not None or async_transport_factory is not None
-        ):
-            raise ValueError("Use either transport or transport_factory, not both")
-
-        self._transport_factory = transport_factory
-        self._async_transport_factory = async_transport_factory
-        self._thread_local = threading.local()
-        # Keyed weakly by the event loop object itself, not id(loop) —
-        # CPython reuses object ids, so a new loop could otherwise inherit
-        # a client bound to a previous, closed loop.
-        self._async_clients: weakref.WeakKeyDictionary[
-            asyncio.AbstractEventLoop, httpx.AsyncClient
-        ] = weakref.WeakKeyDictionary()
         self._proxy = config.proxy
 
         if config.api_key is None:
@@ -251,12 +237,10 @@ def __init__(
             "event_hooks": self._logging_event_hooks(),
         }
         if transport is not None:
+            # The proxy lives in the transport; passing `proxy` here too
+            # would mount a fresh, never-closed proxy transport per client.
             httpx_args["transport"] = transport
-        if (
-            transport is None
-            and transport_factory is None
-            and async_transport_factory is None
-        ):
+        else:
             httpx_args["proxy"] = config.proxy
 
         # config.request_timeout is None when the timeout is explicitly
@@ -277,53 +261,6 @@ def __init__(
     def _logging_event_hooks(self) -> dict:
         return make_logging_event_hooks(self._logger)
 
-    def _headers_with_auth(self) -> dict:
-        return {
-            **self._headers,
-            self.auth_header_name: (
-                f"{self.prefix} {self.token}" if self.prefix else self.token
-            ),
-        }
-
-    def get_httpx_client(self) -> httpx.Client:
-        if self._client is not None or self._transport_factory is None:
-            return super().get_httpx_client()
-
-        client = getattr(self._thread_local, "client", None)
-        if client is None:
-            client = httpx.Client(
-                base_url=self._base_url,
-                cookies=self._cookies,
-                headers=self._headers_with_auth(),
-                timeout=self._timeout,
-                verify=self._verify_ssl,
-                follow_redirects=self._follow_redirects,
-                event_hooks=self._httpx_args.get("event_hooks"),
-                transport=self._transport_factory(),
-            )
-            self._thread_local.client = client
-        return client
-
-    def get_async_httpx_client(self) -> httpx.AsyncClient:
-        if self._async_client is not None or self._async_transport_factory is None:
-            return super().get_async_httpx_client()
-
-        loop = asyncio.get_running_loop()
-        client = self._async_clients.get(loop)
-        if client is None:
-            client = httpx.AsyncClient(
-                base_url=self._base_url,
-                cookies=self._cookies,
-                headers=self._headers_with_auth(),
-                timeout=self._timeout,
-                verify=self._verify_ssl,
-                follow_redirects=self._follow_redirects,
-                event_hooks=self._httpx_args.get("event_hooks"),
-                transport=self._async_transport_factory(),
-            )
-            self._async_clients[loop] = client
-        return client
-
 
 # We need to override the logging hooks for the async usage
 class AsyncApiClient(ApiClient):

packages/python-sdk/e2b/api/client_async/__init__.py

@@ -24,11 +24,7 @@
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
-    return AsyncApiClient(
-        config,
-        async_transport_factory=lambda: get_transport(config),
-        **kwargs,
-    )
+    return AsyncApiClient(config, transport=get_transport(config), **kwargs)
 
 
 class AsyncApiPyqwestTransport(AsyncPyqwestTransport):

packages/python-sdk/e2b/api/client_sync/__init__.py

@@ -22,11 +22,7 @@
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> ApiClient:
-    return ApiClient(
-        config,
-        transport_factory=lambda: get_transport(config),
-        **kwargs,
-    )
+    return ApiClient(config, transport=get_transport(config), **kwargs)
 
 
 class ApiPyqwestTransport(PyqwestTransport):

packages/python-sdk/tests/test_api_client_transport.py

@@ -233,38 +233,20 @@ def test_sync_api_transport_cache_is_shared_across_threads(test_api_key):
         reset_sync_api_transports()
 
 
-def test_sync_api_client_reused_within_thread_and_shares_transport_across_threads(
-    test_api_key,
-):
+def test_sync_api_client_is_shared_across_threads(test_api_key):
+    # httpx.Client is thread-safe and the pyqwest transport underneath is
+    # too, so a single client (and its pool) serves all threads — the
+    # per-thread client caching this replaced is gone.
     reset_sync_api_transports()
     config = ConnectionConfig(api_key=test_api_key)
     api_client = get_sync_api_client(config)
 
-    def get_worker_client_ids():
-        httpx_client = api_client.get_httpx_client()
-        try:
-            return (
-                id(httpx_client),
-                id(api_client.get_httpx_client()),
-                id(httpx_client._transport),
-            )
-        finally:
-            httpx_client.close()
-
     try:
         main_client = api_client.get_httpx_client()
-        (
-            worker_client_id,
-            worker_cached_client_id,
-            worker_transport_id,
-        ) = run_in_worker_thread(get_worker_client_ids)
+        worker_client = run_in_worker_thread(api_client.get_httpx_client)
 
         assert api_client.get_httpx_client() is main_client
-        # The httpx client wrapper stays per-thread, but every thread's
-        # client delegates to the same shared pyqwest transport.
-        assert worker_client_id == worker_cached_client_id
-        assert worker_client_id != id(main_client)
-        assert worker_transport_id == id(main_client._transport)
+        assert worker_client is main_client
     finally:
         main_client.close()
         reset_sync_api_transports()
@@ -345,42 +327,27 @@ async def test_async_api_client_applies_request_timeout(test_api_key):
 
 
 @pytest.mark.asyncio
-async def test_async_api_client_cache_reuses_within_loop_and_shares_transport(
-    test_api_key,
-):
+async def test_async_api_client_is_shared_across_loops(test_api_key):
+    # pyqwest's I/O runs on its own Rust runtime, so neither the transport
+    # nor the httpx client wrapper is bound to an event loop — a single
+    # client serves all loops (the per-loop client caching this replaced is
+    # gone).
     reset_async_api_transports()
     config = ConnectionConfig(api_key=test_api_key)
     api_client = get_async_api_client(config)
 
-    async def get_client_ids():
-        httpx_client = api_client.get_async_httpx_client()
-        try:
-            return (
-                id(httpx_client),
-                id(api_client.get_async_httpx_client()),
-                id(httpx_client._transport),
-            )
-        finally:
-            await httpx_client.aclose()
+    async def get_client():
+        return api_client.get_async_httpx_client()
 
     try:
         main_client = api_client.get_async_httpx_client()
-        (
-            worker_client_id,
-            worker_cached_client_id,
-            worker_transport_id,
-        ) = await asyncio.get_running_loop().run_in_executor(
+        other_loop_client = await asyncio.get_running_loop().run_in_executor(
             None,
-            lambda: asyncio.run(get_client_ids()),
+            lambda: asyncio.run(get_client()),
         )
 
         assert api_client.get_async_httpx_client() is main_client
-        # The httpx client wrapper stays per-loop, but pyqwest's I/O runs on
-        # its own Rust runtime, so every loop's client delegates to the same
-        # shared transport.
-        assert worker_client_id == worker_cached_client_id
-        assert worker_client_id != id(main_client)
-        assert worker_transport_id == id(main_client._transport)
+        assert other_loop_client is main_client
     finally:
         await main_client.aclose()
         reset_async_api_transports()
@@ -415,7 +382,9 @@ async def get_transport():
         reset_async_api_transports()
 
 
-def test_async_api_client_not_reused_across_sequential_loops(test_api_key):
+def test_async_api_client_reused_across_sequential_loops(test_api_key):
+    # A closed loop doesn't strand the client: nothing in it is loop-bound,
+    # so a later loop reuses the same client and shared transport.
     reset_async_api_transports()
     config = ConnectionConfig(api_key=test_api_key)
     api_client = get_async_api_client(config)
@@ -429,22 +398,18 @@ async def get_client():
         loop_a = asyncio.new_event_loop()
         try:
             client_a = loop_a.run_until_complete(get_client())
-            loop_a.run_until_complete(client_a.aclose())
         finally:
             loop_a.close()
         del loop_a
         gc.collect()
 
-        assert len(api_client._async_clients) == 0
-
         loop_b = asyncio.new_event_loop()
         try:
             client_b = loop_b.run_until_complete(get_client())
-            loop_b.run_until_complete(client_b.aclose())
         finally:
             loop_b.close()
 
-        assert client_b is not client_a
+        assert client_b is client_a
     finally:
         reset_async_api_transports()
 
@@ -517,6 +482,47 @@ def test_sync_api_client_round_trips_through_pyqwest(test_api_key, echo_server):
         reset_sync_api_transports()
 
 
+def test_sync_api_client_serves_concurrent_threads(test_api_key, echo_server):
+    # The scenario the removed per-thread client caching used to guard: one
+    # client, one shared pyqwest pool, many threads at once.
+    reset_sync_api_transports()
+    config = ConnectionConfig(api_key=test_api_key, api_url=echo_server)
+    api_client = get_sync_api_client(config)
+    httpx_client = api_client.get_httpx_client()
+
+    def request(i: int) -> tuple[int, str]:
+        response = httpx_client.request("GET", f"/sandboxes/{i}")
+        return response.status_code, response.json()["path"]
+
+    try:
+        with ThreadPoolExecutor(max_workers=16) as executor:
+            results = list(executor.map(request, range(32)))
+
+        assert results == [(200, f"/sandboxes/{i}") for i in range(32)]
+    finally:
+        httpx_client.close()
+        reset_sync_api_transports()
+
+
+@pytest.mark.asyncio
+async def test_async_api_client_serves_concurrent_requests(test_api_key, echo_server):
+    reset_async_api_transports()
+    config = ConnectionConfig(api_key=test_api_key, api_url=echo_server)
+    api_client = get_async_api_client(config)
+    httpx_client = api_client.get_async_httpx_client()
+
+    async def request(i: int) -> tuple[int, str]:
+        response = await httpx_client.request("GET", f"/sandboxes/{i}")
+        return response.status_code, response.json()["path"]
+
+    try:
+        results = await asyncio.gather(*(request(i) for i in range(32)))
+        assert list(results) == [(200, f"/sandboxes/{i}") for i in range(32)]
+    finally:
+        await httpx_client.aclose()
+        reset_async_api_transports()
+
+
 @pytest.mark.asyncio
 async def test_async_api_client_round_trips_through_pyqwest(test_api_key, echo_server):
     reset_async_api_transports()