feat(python-sdk): move the envd HTTP API client onto pyqwest

Swap the per-thread/per-loop envd httpx clients (file transfers, health
checks) for shared pyqwest-backed ones built from the same transport pieces
as the REST client and envd RPC. Streamed downloads default to a dedicated
transport whose 60s idle read timeout bounds stalls (reqwest's read timer
ticks during body send/TTFB, so it can't live on the shared transport);
explicit request timeouts become whole-transfer deadlines and streamed
uploads carry no client-side timeout, matching the JS SDK. Present httpx's
dual sync/async MultipartStream as sync-only — the stock adapter matches
AsyncByteStream first and kills multipart uploads mid-request.

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

Author: mishushakov

.changeset/python-pyqwest-envd-api.md

@@ -0,0 +1,27 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Move the envd HTTP API client (sandbox file transfers, health checks) onto
+[`pyqwest`](https://pypi.org/project/pyqwest/) via its httpx-compatible
+transport adapter. envd RPC already runs on pyqwest through `connectrpc`, so
+all sandbox traffic now shares one HTTP stack built from the same transport
+pieces (with separate connection pools per use).
+
+The per-thread (sync) and per-loop (async) envd httpx clients are gone: the
+pyqwest transports are thread-safe and loop-independent, so a single client
+per module serves all threads and event loops.
+
+Timeout semantics through the adapter:
+
+- Streamed downloads (`files.read(format="stream")`): a `request_timeout`
+  set explicitly for the call is the deadline for the whole transfer — by
+  default the transfer is unbounded in total, as before. A stalled stream is
+  reclaimed by a 60-second idle read timeout that resets on every chunk.
+  `stream_idle_timeout` keeps working on the async client (applied per
+  read); the sync client cannot interrupt a blocking read, so it relies on
+  the transport-wide idle bound and now ignores the parameter.
+- Uploads: a buffered upload is bounded by `request_timeout` as a
+  whole-request deadline, and a streamed (file-like) upload carries no
+  client-side timeout (a stalled one is bounded server-side by envd's idle
+  read timeout) — both matching the JS SDK.

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

@@ -1,6 +1,5 @@
 import asyncio
 import threading
-import weakref
 from typing import Dict, Optional, Tuple, Union
 
 import httpx
@@ -13,14 +12,12 @@
     AsyncApiClient,
     ProxyConfig,
     connection_retries,
-    limits,
+    make_async_logging_event_hooks,
     pool_idle_timeout,
     pool_max_idle_per_host,
     proxy_to_config,
 )
-from e2b.connection_config import ConnectionConfig, ProxyTypes
-
-TransportKey = Tuple[bool, Optional[ProxyTypes]]
+from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
@@ -69,13 +66,17 @@ def should_retry_response(
 
 
 def retrying_http_transport(
-    proxy: Optional[ProxyConfig],
+    proxy: Optional[ProxyConfig], read_timeout: Optional[float] = None
 ) -> 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).
+    connect-only retries. The REST API, envd RPC, and envd HTTP API stacks
+    each cache their own instances (pool unification is a follow-up).
+
+    ``read_timeout`` bounds every read on the transport's connections; see
+    :func:`get_envd_transport` for when that is (and isn't) appropriate.
+
     Requests are logged by pyqwest itself on the ``pyqwest.access`` and
     ``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
     diagnostics httpcore used to provide. The SDK's own ``logger`` option is
@@ -86,16 +87,17 @@ def retrying_http_transport(
             proxy=proxy.to_pyqwest() if proxy is not None else None,
             pool_idle_timeout=pool_idle_timeout,
             pool_max_idle_per_host=pool_max_idle_per_host,
+            read_timeout=read_timeout,
         ),
         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
-# transports below, the transport is not bound to an event loop and the
-# cache is process-global rather than per-loop.
+# pyqwest's I/O runs on its own Rust runtime, so unlike the httpx transports
+# they replaced, the transports are not bound to an event loop and the
+# caches are process-global rather than per-loop.
 _transports: Dict[Optional[ProxyConfig], "AsyncApiPyqwestTransport"] = {}
 
 
@@ -112,38 +114,54 @@ def get_transport(config: ConnectionConfig) -> "AsyncApiPyqwestTransport":
         return transport
 
 
-class AsyncEnvdTransportWithLogger(httpx.AsyncHTTPTransport):
-    # Keyed weakly by the event loop object itself, not id(loop) — CPython
-    # reuses object ids, so a new loop could otherwise inherit a transport
-    # bound to a previous, closed loop.
-    _instances: weakref.WeakKeyDictionary[
-        asyncio.AbstractEventLoop,
-        Dict[TransportKey, "AsyncEnvdTransportWithLogger"],
-    ] = weakref.WeakKeyDictionary()
-
-    @property
-    def pool(self):
-        return self._pool
+# One transport per (proxy, streaming) pair, separate from the REST API
+# pools — envd traffic goes to per-sandbox hosts.
+_envd_transports: Dict[
+    Tuple[Optional[ProxyConfig], bool], "AsyncApiPyqwestTransport"
+] = {}
 
 
 def get_envd_transport(
-    config: ConnectionConfig, http2: bool = True
-) -> AsyncEnvdTransportWithLogger:
-    loop = asyncio.get_running_loop()
-    loop_instances = AsyncEnvdTransportWithLogger._instances.get(loop)
-    if loop_instances is None:
-        loop_instances = {}
-        AsyncEnvdTransportWithLogger._instances[loop] = loop_instances
-
-    key: TransportKey = (http2, config.proxy)
-    transport = loop_instances.get(key)
-    if transport is None:
-        transport = AsyncEnvdTransportWithLogger(
-            limits=limits,
-            proxy=config.proxy,
-            http2=http2,
-            retries=connection_retries,
-        )
-        loop_instances[key] = transport
-
-    return transport
+    config: ConnectionConfig, *, for_streaming: bool = False
+) -> "AsyncApiPyqwestTransport":
+    """The shared pyqwest-backed httpx transports for the envd HTTP API
+    (file transfers, health checks).
+
+    The streaming transport carries ``read_timeout``, the idle bound on
+    every read: it resets after each successful read, so it caps how long a
+    streamed download may stall without limiting total transfer time. It is
+    fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines. Only streamed downloads use it: reqwest's read
+    timer keeps running while a request body is sent and while waiting for
+    the response head, so on the regular transport it would cut off uploads
+    and slow unary responses longer than the idle bound (those stay bounded
+    by their whole-request deadlines instead).
+    """
+    proxy = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
+    with _transport_lock:
+        transport = _envd_transports.get(key)
+        if transport is None:
+            transport = AsyncApiPyqwestTransport(
+                retrying_http_transport(
+                    proxy,
+                    read_timeout=READ_TIMEOUT if for_streaming else None,
+                )
+            )
+            _envd_transports[key] = transport
+        return transport
+
+
+def get_envd_api(
+    config: ConnectionConfig, base_url: str, *, for_streaming: bool = False
+) -> httpx.AsyncClient:
+    """An httpx client for a sandbox's envd HTTP API (file transfers, health
+    checks) on the shared pyqwest transports. The client itself is a cheap
+    stateless wrapper — one per consumer is fine — while the pooled transport
+    underneath is shared and loop-independent."""
+    return httpx.AsyncClient(
+        base_url=base_url,
+        transport=get_envd_transport(config, for_streaming=for_streaming),
+        headers=config.sandbox_headers,
+        event_hooks=make_async_logging_event_hooks(config.logger),
+    )

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

@@ -11,14 +11,12 @@
     ApiClient,
     ProxyConfig,
     connection_retries,
-    limits,
+    make_logging_event_hooks,
     pool_idle_timeout,
     pool_max_idle_per_host,
     proxy_to_config,
 )
-from e2b.connection_config import ConnectionConfig, ProxyTypes
-
-TransportKey = Tuple[bool, Optional[ProxyTypes]]
+from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> ApiClient:
@@ -65,13 +63,17 @@ def should_retry_response(
 
 
 def retrying_http_transport(
-    proxy: Optional[ProxyConfig],
+    proxy: Optional[ProxyConfig], read_timeout: Optional[float] = None
 ) -> 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).
+    connect-only retries. The REST API, envd RPC, and envd HTTP API stacks
+    each cache their own instances (pool unification is a follow-up).
+
+    ``read_timeout`` bounds every read on the transport's connections; see
+    :func:`get_envd_transport` for when that is (and isn't) appropriate.
+
     Requests are logged by pyqwest itself on the ``pyqwest.access`` and
     ``pyqwest`` loggers at ``DEBUG`` (off unless enabled) — the transport-level
     diagnostics httpcore used to provide. The SDK's own ``logger`` option is
@@ -82,15 +84,16 @@ def retrying_http_transport(
             proxy=proxy.to_pyqwest() if proxy is not None else None,
             pool_idle_timeout=pool_idle_timeout,
             pool_max_idle_per_host=pool_max_idle_per_host,
+            read_timeout=read_timeout,
         ),
         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
-# below, the cache is process-global rather than per-thread.
+# pyqwest transports are thread-safe, so unlike the httpx transports they
+# replaced, the caches are process-global rather than per-thread.
 _transports: Dict[Optional[ProxyConfig], "ApiPyqwestTransport"] = {}
 
 
@@ -107,31 +110,53 @@ def get_transport(config: ConnectionConfig) -> "ApiPyqwestTransport":
         return transport
 
 
-class EnvdTransportWithLogger(httpx.HTTPTransport):
-    _thread_local = threading.local()
-
-    @property
-    def pool(self):
-        return self._pool
+# One transport per (proxy, streaming) pair, separate from the REST API
+# pools — envd traffic goes to per-sandbox hosts.
+_envd_transports: Dict[Tuple[Optional[ProxyConfig], bool], "ApiPyqwestTransport"] = {}
 
 
 def get_envd_transport(
-    config: ConnectionConfig, http2: bool = True
-) -> EnvdTransportWithLogger:
-    instances: Dict[TransportKey, EnvdTransportWithLogger] = getattr(
-        EnvdTransportWithLogger._thread_local, "instances", {}
-    )
-    key: TransportKey = (http2, config.proxy)
-    cached = instances.get(key)
-    if cached is not None:
-        return cached
-
-    transport = EnvdTransportWithLogger(
-        limits=limits,
-        proxy=config.proxy,
-        http2=http2,
-        retries=connection_retries,
+    config: ConnectionConfig, *, for_streaming: bool = False
+) -> "ApiPyqwestTransport":
+    """The shared pyqwest-backed httpx transports for the envd HTTP API
+    (file transfers, health checks).
+
+    The streaming transport carries ``read_timeout``, the idle bound on
+    every read: it resets after each successful read, so it caps how long a
+    streamed download may stall without limiting total transfer time. It is
+    fixed per transport — the adapter's per-request timeouts are
+    whole-request deadlines, and the sync adapter does not bound body reads
+    at all. Only streamed downloads use it: reqwest's read timer keeps
+    running while a request body is sent and while waiting for the response
+    head, so on the regular transport it would cut off uploads and slow
+    unary responses longer than the idle bound (those stay bounded by their
+    whole-request deadlines instead).
+    """
+    proxy = proxy_to_config(config.proxy)
+    key = (proxy, for_streaming)
+    with _transport_lock:
+        transport = _envd_transports.get(key)
+        if transport is None:
+            transport = ApiPyqwestTransport(
+                retrying_http_transport(
+                    proxy,
+                    read_timeout=READ_TIMEOUT if for_streaming else None,
+                )
+            )
+            _envd_transports[key] = transport
+        return transport
+
+
+def get_envd_api(
+    config: ConnectionConfig, base_url: str, *, for_streaming: bool = False
+) -> httpx.Client:
+    """An httpx client for a sandbox's envd HTTP API (file transfers, health
+    checks) on the shared pyqwest transports. The client itself is a cheap
+    stateless wrapper — one per consumer is fine — while the pooled transport
+    underneath is shared and thread-safe."""
+    return httpx.Client(
+        base_url=base_url,
+        transport=get_envd_transport(config, for_streaming=for_streaming),
+        headers=config.sandbox_headers,
+        event_hooks=make_logging_event_hooks(config.logger),
     )
-    instances[key] = transport
-    EnvdTransportWithLogger._thread_local.instances = instances
-    return transport

packages/python-sdk/e2b/connection_config.py

@@ -21,6 +21,13 @@
 
 REQUEST_TIMEOUT: float = 60.0  # 60 seconds
 
+# Idle bound for every read on the streaming envd file-transfer transport:
+# the transfer is aborted when no bytes at all arrive for this long. It
+# resets on each chunk, so it never limits total transfer time — only a
+# fully stalled stream. Matches the previous default stream idle timeout
+# (the request timeout).
+READ_TIMEOUT: float = 60.0  # 60 seconds
+
 KEEPALIVE_PING_INTERVAL_SEC = 50  # 50 seconds
 KEEPALIVE_PING_HEADER = "Keepalive-Ping-Interval"
 

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

@@ -105,9 +105,13 @@ class LoggingInterceptor:
     streamed message at DEBUG — mirroring the httpx event hooks used for the
     REST API and file transfer requests.
 
-    Upstreamed to pyqwest as a logging middleware
-    (https://github.com/curioswitch/pyqwest/pull/192); this interceptor stays
-    until that ships in a pyqwest release the SDK can depend on.
+    pyqwest logs requests itself on the ``pyqwest.access`` and ``pyqwest``
+    loggers, but those are process-wide loggers that cannot carry the
+    per-sandbox logger this option gives callers, and they see only the HTTP
+    exchange — not the streamed messages, nor the Connect error code of a
+    stream that fails inside a ``200 OK`` response. This interceptor is what
+    serves the ``logger`` option; the pyqwest loggers sit below it as
+    transport-level diagnostics.
     """
 
     def __init__(self, logger: logging.Logger, base_url: str):

packages/python-sdk/e2b/sandbox/filesystem/filesystem.py

@@ -1,3 +1,4 @@
+import asyncio
 import gzip
 import re
 from dataclasses import dataclass, field
@@ -181,6 +182,11 @@ def __iter__(self) -> Iterator[bytes]:
     def __next__(self) -> bytes:
         try:
             return next(self._iterator)
+        except TimeoutError as e:
+            # The transport's idle read timeout is the builtin TimeoutError
+            # under pyqwest; keep the documented httpx exception.
+            self.close()
+            raise httpx.ReadTimeout(str(e)) from e
         except BaseException:
             # Covers normal end (StopIteration) and read errors alike.
             self.close()
@@ -219,17 +225,31 @@ class AsyncFileStreamReader(AsyncIterator[bytes]):
                 ...
     """
 
-    def __init__(self, response: httpx.Response):
+    def __init__(self, response: httpx.Response, idle_timeout: Optional[float] = None):
         self._response = response
         self._iterator = response.aiter_bytes()
+        # An explicit per-call idle bound, applied around each read with
+        # `wait_for` (the transport-wide idle read timeout covers the
+        # default case; see the flavor `read` implementations).
+        self._idle_timeout = idle_timeout
         self._closed = False
 
     def __aiter__(self) -> AsyncIterator[bytes]:
         return self
 
     async def __anext__(self) -> bytes:
         try:
-            return await self._iterator.__anext__()
+            read = self._iterator.__anext__()
+            if self._idle_timeout:
+                return await asyncio.wait_for(read, self._idle_timeout)
+            return await read
+        except (TimeoutError, asyncio.TimeoutError) as e:
+            # Both the transport's idle read timeout (the builtin
+            # TimeoutError under pyqwest) and wait_for's expiry (a distinct
+            # asyncio.TimeoutError until 3.11); keep the documented httpx
+            # exception.
+            await self.aclose()
+            raise httpx.ReadTimeout(str(e)) from e
         except BaseException:
             # Covers normal end (StopAsyncIteration) and read errors alike.
             await self.aclose()