feat(python-sdk): move template build-context uploads onto pyqwest

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

@@ -0,0 +1,30 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Move the REST API client (sandbox lifecycle, listing, templates, volumes
+control plane) onto [`pyqwest`](https://pypi.org/project/pyqwest/) (Rust
+reqwest/hyper) via its httpx-compatible transport adapter, replacing the
+httpx-native `HTTPTransport`/`AsyncHTTPTransport`. The generated httpx client
+API is unchanged — only the transport underneath is swapped — so logging
+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), 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.
+
+`proxy` for API calls now takes a URL string (e.g.
+`proxy="http://user:pass@localhost:8030"`, scheme http, https, socks5, or
+socks5h). `httpx.URL` and `httpx.Proxy` keep working when they reduce to
+such a URL (`httpx.Proxy` auth is folded back into the URL userinfo);
+`httpx.Proxy` extras pyqwest can't express — custom headers, an
+`ssl_context` — raise `InvalidArgumentException` rather than being silently
+dropped.
+
+envd traffic (sandbox commands, filesystem, PTY, file transfers) is not
+affected and stays on its existing stack.

.changeset/python-pyqwest-template-uploads.md

@@ -0,0 +1,11 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Move template build-context uploads (to S3 presigned URLs) onto
+[`pyqwest`](https://pypi.org/project/pyqwest/) via its httpx-compatible
+transport adapter. Content-Length framing for the streamed archive body is
+preserved (S3 rejects chunked transfer encoding). The 1-hour upload timeout
+now bounds the entire upload rather than each socket operation, and
+`verify_ssl=False` on the client is no longer honored for uploads (pyqwest
+has no insecure-TLS option).

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

@@ -1,23 +1,22 @@
-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
+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
 from e2b.connection_config import ConnectionConfig
 from e2b.exceptions import (
     AuthenticationException,
+    InvalidArgumentException,
     RateLimitException,
     SandboxException,
 )
@@ -70,6 +69,48 @@ async def on_response(response: Response) -> None:
 
 connection_retries = int(os.getenv("E2B_CONNECTION_RETRIES", "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).
+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]:
+    """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
+    transports accepted — 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(
+                "E2B API 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(
+                "E2B API 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(
+        "E2B API calls support only URL-string proxies, "
+        'e.g. proxy="http://user:pass@localhost:8030"'
+    )
+
 
 @dataclass
 class SandboxCreateResponse:
@@ -154,31 +195,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:
@@ -223,12 +253,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
@@ -249,53 +277,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

@@ -1,50 +1,118 @@
 import asyncio
+import threading
 import weakref
-from typing import Dict, Optional, Tuple
+from typing import Dict, Optional, Tuple, Union
 
 import httpx
 
 from httpx._types import ProxyTypes
-
-from e2b.api import AsyncApiClient, connection_retries, limits
+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,
+    pool_idle_timeout,
+    pool_max_idle_per_host,
+    proxy_to_url,
+)
 from e2b.connection_config import ConnectionConfig
 
 TransportKey = Tuple[bool, Optional[ProxyTypes]]
 
 
 def get_api_client(config: ConnectionConfig, **kwargs) -> AsyncApiClient:
-    return AsyncApiClient(
-        config,
-        async_transport_factory=lambda: get_transport(config),
-        **kwargs,
-    )
-
-
-class AsyncTransportWithLogger(httpx.AsyncHTTPTransport):
+    return AsyncApiClient(config, transport=get_transport(config), **kwargs)
+
+
+class AsyncApiPyqwestTransport(AsyncPyqwestTransport):
+    """Strip the ``Host`` header httpx adds to every request: hyper derives
+    HTTP/1 ``Host`` and HTTP/2 ``:authority`` from the URL itself, and
+    forwarding an explicit ``host`` header on an HTTP/2 connection makes the
+    E2B API edge reset the stream with PROTOCOL_ERROR. (Custom ``Host``
+    overrides are therefore not honored, matching hyper's URL-derived
+    behavior.)"""
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        if "host" in request.headers:
+            del request.headers["host"]
+        return await super().handle_async_request(request)
+
+
+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."""
+
+    def should_retry_response(
+        self, request: Request, response: Union[Response, Exception]
+    ) -> bool:
+        return isinstance(response, ConnectionError)
+
+
+_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.
+_transports: Dict[Optional[str], "AsyncApiPyqwestTransport"] = {}
+
+
+def get_transport(config: ConnectionConfig) -> "AsyncApiPyqwestTransport":
+    """The shared pyqwest-backed httpx transport for REST API calls. For TLS
+    connections ALPN negotiates the HTTP version (HTTP/2 against the E2B
+    API), like the http2-enabled httpx transport this replaced."""
+    proxy_url = proxy_to_url(config.proxy)
+    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,
+                )
+            )
+            _transports[proxy_url] = transport
+        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, "AsyncTransportWithLogger"],
+        Dict[TransportKey, "AsyncEnvdTransportWithLogger"],
     ] = weakref.WeakKeyDictionary()
 
     @property
     def pool(self):
         return self._pool
 
 
-def _get_cached_transport(cls, config: ConnectionConfig, http2: bool):
+def get_envd_transport(
+    config: ConnectionConfig, http2: bool = True
+) -> AsyncEnvdTransportWithLogger:
     loop = asyncio.get_running_loop()
-    loop_instances = cls._instances.get(loop)
+    loop_instances = AsyncEnvdTransportWithLogger._instances.get(loop)
     if loop_instances is None:
         loop_instances = {}
-        cls._instances[loop] = loop_instances
+        AsyncEnvdTransportWithLogger._instances[loop] = loop_instances
 
     key: TransportKey = (http2, config.proxy)
     transport = loop_instances.get(key)
     if transport is None:
-        transport = cls(
+        transport = AsyncEnvdTransportWithLogger(
             limits=limits,
             proxy=config.proxy,
             http2=http2,
@@ -53,22 +121,3 @@ def _get_cached_transport(cls, config: ConnectionConfig, http2: bool):
         loop_instances[key] = transport
 
     return transport
-
-
-def get_transport(
-    config: ConnectionConfig, http2: bool = True
-) -> AsyncTransportWithLogger:
-    return _get_cached_transport(AsyncTransportWithLogger, config, http2)
-
-
-class AsyncEnvdTransportWithLogger(AsyncTransportWithLogger):
-    _instances: weakref.WeakKeyDictionary[
-        asyncio.AbstractEventLoop,
-        Dict[TransportKey, "AsyncEnvdTransportWithLogger"],
-    ] = weakref.WeakKeyDictionary()
-
-
-def get_envd_transport(
-    config: ConnectionConfig, http2: bool = True
-) -> AsyncEnvdTransportWithLogger:
-    return _get_cached_transport(AsyncEnvdTransportWithLogger, config, http2)