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,26 +1,22 @@
 import asyncio
 import threading
-import weakref
 from typing import Dict, Optional, Tuple, Union
 
 import httpx
 
-from httpx._types import ProxyTypes
 from pyqwest import HTTPTransport, Request, Response
 from pyqwest.httpx import AsyncPyqwestTransport
 from pyqwest.middleware.retry import RetryTransport
 
 from e2b.api import (
     AsyncApiClient,
     connection_retries,
-    limits,
+    make_async_logging_event_hooks,
     pool_idle_timeout,
     pool_max_idle_per_host,
     proxy_to_url,
 )
-from e2b.connection_config import ConnectionConfig
-
-TransportKey = Tuple[bool, Optional[ProxyTypes]]
+from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
@@ -68,28 +64,34 @@ def should_retry_response(
         return isinstance(response, ConnectionError)
 
 
-def retrying_http_transport(proxy_url: Optional[str]) -> ConnectionRetryTransport:
+def retrying_http_transport(
+    proxy_url: Optional[str], 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."""
     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,
+            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[str], "AsyncApiPyqwestTransport"] = {}
 
 
@@ -106,38 +108,52 @@ 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[str], 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_url = proxy_to_url(config.proxy)
+    key = (proxy_url, for_streaming)
+    with _transport_lock:
+        transport = _envd_transports.get(key)
+        if transport is None:
+            transport = AsyncApiPyqwestTransport(
+                retrying_http_transport(
+                    proxy_url,
+                    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

@@ -3,28 +3,44 @@
 import httpx
 import threading
 
-from httpx._types import ProxyTypes
 from pyqwest import SyncHTTPTransport, SyncRequest, SyncResponse
 from pyqwest.httpx import PyqwestTransport
 from pyqwest.middleware.retry import SyncRetryTransport
 
 from e2b.api import (
     ApiClient,
     connection_retries,
-    limits,
+    make_logging_event_hooks,
     pool_idle_timeout,
     pool_max_idle_per_host,
     proxy_to_url,
 )
-from e2b.connection_config import ConnectionConfig
-
-TransportKey = Tuple[bool, Optional[ProxyTypes]]
+from e2b.connection_config import READ_TIMEOUT, ConnectionConfig
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> ApiClient:
     return ApiClient(config, transport=get_transport(config), **kwargs)
 
 
+class _SyncOnlyStream(httpx.SyncByteStream):
+    """Present a dual sync/async httpx request stream as sync-only.
+
+    httpx's ``MultipartStream`` (``files=`` uploads) implements both
+    ``SyncByteStream`` and ``AsyncByteStream``; the stock adapter's content
+    conversion matches ``AsyncByteStream`` first and raises
+    ``TypeError("unreachable")`` from inside the body iterator, which
+    surfaces as a ``WriteError`` mid-request."""
+
+    def __init__(self, stream: httpx.SyncByteStream):
+        self._stream = stream
+
+    def __iter__(self):
+        return iter(self._stream)
+
+    def close(self) -> None:
+        self._stream.close()
+
+
 class ApiPyqwestTransport(PyqwestTransport):
     """The SDK's tweaks on the stock pyqwest httpx adapter.
 
@@ -42,6 +58,13 @@ class ApiPyqwestTransport(PyqwestTransport):
     def handle_request(self, request: httpx.Request) -> httpx.Response:
         if "host" in request.headers:
             del request.headers["host"]
+        stream = request.stream
+        if (
+            isinstance(stream, httpx.SyncByteStream)
+            and isinstance(stream, httpx.AsyncByteStream)
+            and not isinstance(stream, httpx.ByteStream)
+        ):
+            request.stream = _SyncOnlyStream(stream)
         try:
             return super().handle_request(request)
         except TimeoutError as e:
@@ -64,27 +87,33 @@ def should_retry_response(
         return isinstance(response, ConnectionError)
 
 
-def retrying_http_transport(proxy_url: Optional[str]) -> ConnectionRetryTransport:
+def retrying_http_transport(
+    proxy_url: Optional[str], 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."""
     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,
+            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[str], "ApiPyqwestTransport"] = {}
 
 
@@ -101,31 +130,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[str], 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_url = proxy_to_url(config.proxy)
+    key = (proxy_url, for_streaming)
+    with _transport_lock:
+        transport = _envd_transports.get(key)
+        if transport is None:
+            transport = ApiPyqwestTransport(
+                retrying_http_transport(
+                    proxy_url,
+                    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

@@ -11,6 +11,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"