refactor(python-sdk): share the pyqwest transport plumbing with the envd RPC stack

With the envd RPC migration (#1558) on main, both stacks carried identical
copies of the transport pieces. Make e2b.api the canonical home: proxy_to_url
(with stack-neutral error messages) and the pool tuning move out of
e2b.envd.client_shared, and the flavor modules gain a shared
retrying_http_transport factory + ConnectionRetryTransport that
e2b.envd.client_sync/client_async now import instead of defining their own.
The stacks keep separate per-proxy pools (unification is a follow-up).

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

Author: mishushakov

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

@@ -8,8 +8,6 @@
 
 import httpx
 from httpx import AsyncBaseTransport, BaseTransport, Limits, Timeout
-from httpx._types import ProxyTypes
-
 from e2b.api.client.client import AuthenticatedClient
 from e2b.api.client.types import Response
 from e2b.api.metadata import default_headers
@@ -69,16 +67,16 @@ async def on_response(response: Response) -> None:
 
 connection_retries = int(os.getenv("E2B_CONNECTION_RETRIES") or "3")
 
-# Mirror the httpx pool tuning above with pyqwest's equivalents for the REST
-# API transports. `pool_max_idle_per_host` is per host rather than httpx's
-# global idle cap, but API traffic goes to a single host, so the values map
-# directly; reqwest has no counterpart to `E2B_MAX_CONNECTIONS` (it does not
-# cap concurrent connections).
+# Mirror the httpx pool tuning above with pyqwest's equivalents, shared by
+# the REST API and envd RPC transports. `pool_max_idle_per_host` is per host
+# rather than httpx's global idle cap, which suits both: API traffic goes to
+# a single host and each sandbox is its own host. reqwest has no counterpart
+# to `E2B_MAX_CONNECTIONS` (it does not cap concurrent connections).
 pool_idle_timeout = float(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300")
 pool_max_idle_per_host = int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20")
 
 
-def proxy_to_url(proxy: Optional[ProxyTypes]) -> Optional[str]:
+def proxy_to_url(proxy: object) -> Optional[str]:
     """Narrow the ``proxy`` connection option to the proxy URL string pyqwest
     transports take (scheme http, https, socks5, or socks5h, credentials in
     the URL userinfo). ``httpx.URL`` and ``httpx.Proxy`` — which the httpx
@@ -94,20 +92,18 @@ def proxy_to_url(proxy: Optional[ProxyTypes]) -> Optional[str]:
     if isinstance(proxy, httpx.Proxy):
         if proxy.headers:
             raise InvalidArgumentException(
-                "E2B API calls don't support httpx.Proxy custom headers; "
+                "httpx.Proxy custom headers are not supported; "
                 "pass credentials in the proxy URL instead, "
                 'e.g. proxy="http://user:pass@localhost:8030"'
             )
         if proxy.ssl_context is not None:
-            raise InvalidArgumentException(
-                "E2B API calls don't support httpx.Proxy ssl_context"
-            )
+            raise InvalidArgumentException("httpx.Proxy ssl_context is not supported")
         url = proxy.url
         if proxy.auth is not None:
             url = url.copy_with(username=proxy.auth[0], password=proxy.auth[1])
         return str(url)
     raise InvalidArgumentException(
-        "E2B API calls support only URL-string proxies, "
+        "Only URL-string proxies are supported, "
         'e.g. proxy="http://user:pass@localhost:8030"'
     )
 

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

@@ -53,19 +53,38 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
 
 
 class ConnectionRetryTransport(RetryTransport):
-    """Retry only failures establishing the connection, matching the
-    connect-only ``retries`` of the httpx transport this replaced: pyqwest
-    raises the builtin ``ConnectionError`` only before the request was
-    written, so these retries can never replay a request the API may have
-    received. The retry middleware's default policy would otherwise also
-    retry I/O errors and 429/5xx responses for idempotent methods."""
+    """Retry only failures establishing the connection — shared by the REST
+    API and envd RPC stacks: pyqwest raises the builtin ``ConnectionError``
+    only before the request was written, so these retries can never replay a
+    request the server may have received (a delivered REST call or unary RPC
+    like ``SendInput``). This matches the connect-only ``retries`` of the
+    httpx transports this replaced; the retry middleware's default policy
+    would otherwise also retry I/O errors and 429/5xx responses for
+    idempotent methods."""
 
     def should_retry_response(
         self, request: Request, response: Union[Response, Exception]
     ) -> bool:
         return isinstance(response, ConnectionError)
 
 
+def retrying_http_transport(proxy_url: Optional[str]) -> ConnectionRetryTransport:
+    """A fresh pyqwest transport (= its own connection pool) with the SDK's
+    shared tuning — system CA certs (without which TLS through an
+    intercepting proxy fails), the httpx-equivalent pool limits, and
+    connect-only retries. The REST API and envd RPC stacks each cache their
+    own instances (pool unification is a follow-up)."""
+    return ConnectionRetryTransport(
+        HTTPTransport(
+            tls_include_system_certs=True,
+            proxy=proxy_url,
+            pool_idle_timeout=pool_idle_timeout,
+            pool_max_idle_per_host=pool_max_idle_per_host,
+        ),
+        max_retries=connection_retries,
+    )
+
+
 _transport_lock = threading.Lock()
 # One transport (= one connection pool) per proxy; None is the direct pool.
 # pyqwest's I/O runs on its own Rust runtime, so unlike the httpx envd
@@ -82,17 +101,7 @@ def get_transport(config: ConnectionConfig) -> "AsyncApiPyqwestTransport":
     with _transport_lock:
         transport = _transports.get(proxy_url)
         if transport is None:
-            transport = AsyncApiPyqwestTransport(
-                ConnectionRetryTransport(
-                    HTTPTransport(
-                        tls_include_system_certs=True,
-                        proxy=proxy_url,
-                        pool_idle_timeout=pool_idle_timeout,
-                        pool_max_idle_per_host=pool_max_idle_per_host,
-                    ),
-                    max_retries=connection_retries,
-                )
-            )
+            transport = AsyncApiPyqwestTransport(retrying_http_transport(proxy_url))
             _transports[proxy_url] = transport
         return transport
 

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

@@ -49,19 +49,38 @@ def handle_request(self, request: httpx.Request) -> httpx.Response:
 
 
 class ConnectionRetryTransport(SyncRetryTransport):
-    """Retry only failures establishing the connection, matching the
-    connect-only ``retries`` of the httpx transport this replaced: pyqwest
-    raises the builtin ``ConnectionError`` only before the request was
-    written, so these retries can never replay a request the API may have
-    received. The retry middleware's default policy would otherwise also
-    retry I/O errors and 429/5xx responses for idempotent methods."""
+    """Retry only failures establishing the connection — shared by the REST
+    API and envd RPC stacks: pyqwest raises the builtin ``ConnectionError``
+    only before the request was written, so these retries can never replay a
+    request the server may have received (a delivered REST call or unary RPC
+    like ``SendInput``). This matches the connect-only ``retries`` of the
+    httpx transports this replaced; the retry middleware's default policy
+    would otherwise also retry I/O errors and 429/5xx responses for
+    idempotent methods."""
 
     def should_retry_response(
         self, request: SyncRequest, response: Union[SyncResponse, Exception]
     ) -> bool:
         return isinstance(response, ConnectionError)
 
 
+def retrying_http_transport(proxy_url: Optional[str]) -> ConnectionRetryTransport:
+    """A fresh pyqwest transport (= its own connection pool) with the SDK's
+    shared tuning — system CA certs (without which TLS through an
+    intercepting proxy fails), the httpx-equivalent pool limits, and
+    connect-only retries. The REST API and envd RPC stacks each cache their
+    own instances (pool unification is a follow-up)."""
+    return ConnectionRetryTransport(
+        SyncHTTPTransport(
+            tls_include_system_certs=True,
+            proxy=proxy_url,
+            pool_idle_timeout=pool_idle_timeout,
+            pool_max_idle_per_host=pool_max_idle_per_host,
+        ),
+        max_retries=connection_retries,
+    )
+
+
 _transport_lock = threading.Lock()
 # One transport (= one connection pool) per proxy; None is the direct pool.
 # pyqwest transports are thread-safe, so unlike the httpx envd transports
@@ -77,17 +96,7 @@ def get_transport(config: ConnectionConfig) -> "ApiPyqwestTransport":
     with _transport_lock:
         transport = _transports.get(proxy_url)
         if transport is None:
-            transport = ApiPyqwestTransport(
-                ConnectionRetryTransport(
-                    SyncHTTPTransport(
-                        tls_include_system_certs=True,
-                        proxy=proxy_url,
-                        pool_idle_timeout=pool_idle_timeout,
-                        pool_max_idle_per_host=pool_max_idle_per_host,
-                    ),
-                    max_retries=connection_retries,
-                )
-            )
+            transport = ApiPyqwestTransport(retrying_http_transport(proxy_url))
             _transports[proxy_url] = transport
         return transport
 

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

@@ -9,25 +9,20 @@
     Callable,
     Optional,
     TypeVar,
-    Union,
     cast,
 )
 
 from connectrpc.code import Code
 from connectrpc.errors import ConnectError
-from pyqwest import Client, HTTPTransport, Request, Response, Transport
-from pyqwest.middleware.retry import RetryTransport
+from pyqwest import Client, Request, Response, Transport
 
-from e2b.api import connection_retries
+from e2b.api import proxy_to_url
+from e2b.api.client_async import retrying_http_transport
 from e2b.connection_config import ConnectionConfig
 from e2b.envd.client_shared import (
     ENVD_JSON_CODEC,
     ENVD_RPC_COMPRESSION,
     plain_http_error,
-    pool_idle_timeout,
-    pool_max_idle_per_host,
-    proxy_to_url,
-    should_retry_connection,
 )
 from e2b.envd.interceptors import build_interceptors
 from e2b.exceptions import TimeoutException
@@ -70,17 +65,6 @@ async def execute(self, request: Request) -> Response:
         raise error
 
 
-class ConnectionRetryTransport(RetryTransport):
-    """Retry only failures establishing the connection; see
-    :func:`e2b.envd.client_shared.should_retry_connection` for the policy
-    rationale."""
-
-    def should_retry_response(
-        self, request: Request, response: Union[Response, Exception]
-    ) -> bool:
-        return should_retry_connection(response)
-
-
 def get_transport(proxy_url: Optional[str]) -> "PlainHTTPErrorTransport":
     with _transport_lock:
         transport = _transports.get(proxy_url)
@@ -89,17 +73,7 @@ def get_transport(proxy_url: Optional[str]) -> "PlainHTTPErrorTransport":
             # retry backoff counts against the request timeout. The plain-
             # error normalization sits outside the retries so it converts
             # the settled response once.
-            transport = PlainHTTPErrorTransport(
-                ConnectionRetryTransport(
-                    HTTPTransport(
-                        tls_include_system_certs=True,
-                        proxy=proxy_url,
-                        pool_idle_timeout=pool_idle_timeout,
-                        pool_max_idle_per_host=pool_max_idle_per_host,
-                    ),
-                    max_retries=connection_retries,
-                )
-            )
+            transport = PlainHTTPErrorTransport(retrying_http_transport(proxy_url))
             _transports[proxy_url] = transport
         return transport
 
@@ -111,9 +85,9 @@ def create_rpc_client(
 ) -> TClient:
     """Build a generated async connectrpc client (e.g. ``ProcessClient``)
     wired with the shared pyqwest transport (which retries failed connects,
-    see :class:`ConnectionRetryTransport`), the envd JSON codec, and the
-    SDK's default-header and logging interceptors. Compression is disabled
-    (see ``ENVD_RPC_COMPRESSION``).
+    see :class:`e2b.api.client_async.ConnectionRetryTransport`), the envd
+    JSON codec, and the SDK's default-header and logging interceptors.
+    Compression is disabled (see ``ENVD_RPC_COMPRESSION``).
     """
     http_client = Client(get_transport(proxy_to_url(config.proxy)))
     return client_cls(

packages/python-sdk/e2b/envd/client_shared.py

@@ -1,11 +1,12 @@
 """envd RPC client plumbing shared by the sync and async flavors.
 
 The envd RPC clients (process, filesystem) run on `connectrpc`, whose HTTP
-layer is `pyqwest` (Rust reqwest/hyper). This is a separate stack from the
-`httpx` transports in `e2b.api`, which keep serving the REST API and the
-multipart file transfer endpoints. Unlike the previous httpcore-based
-transport, hyper sends RST_STREAM when a server stream is closed early, so
-abandoned command/watch streams don't leak on the shared HTTP/2 connection.
+layer is `pyqwest` (Rust reqwest/hyper) — built on the same
+`retrying_http_transport` pieces as the REST API client in `e2b.api`, in a
+separately cached pool. Only the multipart file transfer endpoints stay on
+the `httpx` envd transports. Unlike the previous httpcore-based transport,
+hyper sends RST_STREAM when a server stream is closed early, so abandoned
+command/watch streams don't leak on the shared HTTP/2 connection.
 
 The flavor-specific transports and client factories live in
 :mod:`e2b.envd.client_sync` and :mod:`e2b.envd.client_async`, mirroring the
@@ -15,22 +16,12 @@
 """
 
 import json
-import os
 from typing import Optional, TypedDict, TypeVar
 
-import httpx
 from connectrpc.code import Code
 from connectrpc.errors import ConnectError
 from protobuf import Message
 
-from e2b.exceptions import InvalidArgumentException
-
-# Mirror the httpx pool tuning in `e2b.api.limits` with pyqwest's equivalents.
-# `pool_max_idle_per_host` is per host rather than httpx's global idle cap,
-# which suits envd traffic — each sandbox is its own host.
-pool_idle_timeout = float(os.getenv("E2B_KEEPALIVE_EXPIRY") or "300")
-pool_max_idle_per_host = int(os.getenv("E2B_MAX_KEEPALIVE_CONNECTIONS") or "20")
-
 _MESSAGE = TypeVar("_MESSAGE", bound=Message)
 
 
@@ -133,25 +124,6 @@ def plain_http_error(
     )
 
 
-def should_retry_connection(response: object) -> bool:
-    """Whether a transport result is a retryable connection-establishment
-    failure — the shared policy of the flavor ``ConnectionRetryTransport``s.
-
-    pyqwest raises the builtin ``ConnectionError`` only before the request
-    was written, so retrying exactly these failures can never replay a
-    request envd may have received — which could re-run a command or
-    re-deliver events — for unary and streaming RPCs alike. Anything later
-    (``WriteError``/``ReadError``/``StreamError``, error responses) surfaces
-    to the caller; the retry middleware's default policy would otherwise also
-    retry I/O errors and 429/5xx responses for idempotent methods. This
-    replaces httpcore's transport ``retries`` from the previous stack and
-    deliberately drops the vendored client's retry on connections dropped
-    mid-request, which could re-execute a delivered unary RPC like
-    ``SendInput``.
-    """
-    return isinstance(response, ConnectionError)
-
-
 class _RPCCompression(TypedDict):
     send_compression: None
     accept_compression: "tuple[()]"
@@ -168,38 +140,3 @@ class _RPCCompression(TypedDict):
     "send_compression": None,
     "accept_compression": (),
 }
-
-
-def proxy_to_url(proxy: object) -> Optional[str]:
-    """Narrow the ``proxy`` connection option to the proxy URL string pyqwest
-    transports take (scheme http, https, socks5, or socks5h, credentials in
-    the URL userinfo). ``httpx.URL`` and ``httpx.Proxy`` — which the REST
-    client accepts and the vendored envd client used to — are converted when
-    they reduce to such a URL; ``httpx.Proxy`` extras that don't (custom
-    headers, an ssl_context) are rejected rather than silently dropped.
-    """
-    if proxy is None:
-        return None
-    if isinstance(proxy, str):
-        return proxy
-    if isinstance(proxy, httpx.URL):
-        return str(proxy)
-    if isinstance(proxy, httpx.Proxy):
-        if proxy.headers:
-            raise InvalidArgumentException(
-                "Sandbox RPC calls don't support httpx.Proxy custom headers; "
-                "pass credentials in the proxy URL instead, "
-                'e.g. proxy="http://user:pass@localhost:8030"'
-            )
-        if proxy.ssl_context is not None:
-            raise InvalidArgumentException(
-                "Sandbox RPC calls don't support httpx.Proxy ssl_context"
-            )
-        url = proxy.url
-        if proxy.auth is not None:
-            url = url.copy_with(username=proxy.auth[0], password=proxy.auth[1])
-        return str(url)
-    raise InvalidArgumentException(
-        "Sandbox RPC calls support only URL-string proxies, "
-        'e.g. proxy="http://user:pass@localhost:8030"'
-    )