feat(python-sdk): adopt pyqwest 0.8's proxy object, multipart fix, and request loggers

.changeset/python-pyqwest-0-8.md

@@ -0,0 +1,27 @@
+---
+"@e2b/python-sdk": minor
+---
+
+Require [`pyqwest`](https://pypi.org/project/pyqwest/) 0.8, which lets the SDK
+drop the workarounds it carried for what the HTTP stack could not express
+before:
+
+- `proxy` honors an `httpx.Proxy`'s credentials and custom headers as given,
+  instead of rejecting headers and folding the credentials into the proxy URL
+  userinfo. A per-proxy `ssl_context` remains unsupported.
+- Multipart uploads (`files.write`) no longer need the SDK to rewrap httpx's
+  request stream on the way to the transport.
+- Low-level HTTP logs are available again — the equivalent of the httpcore
+  records that moving off the httpx transports removed. pyqwest logs one line
+  per request on the `pyqwest.access` logger and request lifecycle records on
+  `pyqwest`, both at `DEBUG` and off unless enabled:
+
+  ```python
+  import logging
+
+  logging.basicConfig()
+  logging.getLogger("pyqwest.access").setLevel(logging.DEBUG)
+  # DEBUG pyqwest.access - HTTP Request: POST https://api.e2b.app/sandboxes "HTTP/2 201 Created"
+  ```
+
+  The SDK's own `logger` option is unchanged and independent of these.

.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.

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

@@ -19,13 +19,12 @@ Connection-establishment failures are retried with backoff
 the previous transports. Timeouts keep raising `httpx.ReadTimeout` (an
 `httpx.TimeoutException`), as before.
 
-`proxy` for API calls now takes a URL string (e.g.
+`proxy` for API calls 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.
+socks5h), an `httpx.URL`, or an `httpx.Proxy` — including its credentials
+and custom headers. The one `httpx.Proxy` option pyqwest cannot express, a
+per-proxy `ssl_context`, raises `InvalidArgumentException` rather than being
+silently dropped.
 
 envd traffic is not affected: RPC (commands, PTY, filesystem watch) already
 runs on pyqwest via `connectrpc`, and the envd HTTP API (file transfers,

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

@@ -4,11 +4,11 @@
 import re
 from dataclasses import dataclass
 from types import TracebackType
-from typing import Optional, Protocol, Union
+from typing import NamedTuple, Optional, Protocol, Tuple, Union
 
 import httpx
 from httpx import AsyncBaseTransport, BaseTransport, Limits, Timeout
-from httpx._types import ProxyTypes
+from pyqwest import Proxy
 
 from e2b.api.client.client import AuthenticatedClient
 from e2b.api.client.types import Response
@@ -69,45 +69,56 @@ 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]:
-    """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."""
+class ProxyConfig(NamedTuple):
+    """The ``proxy`` connection option in the shape pyqwest transports take.
+
+    A tuple so it can key the transport caches directly: it is hashable and
+    compares by value, where a ``pyqwest.Proxy`` compares by identity and
+    would hand every call its own connection pool."""
+
+    url: str
+    auth: Optional[Tuple[str, str]] = None
+    headers: Tuple[Tuple[str, str], ...] = ()
+
+    def to_pyqwest(self) -> Proxy:
+        """The ``pyqwest.Proxy`` to hand a transport."""
+        return Proxy(self.url, auth=self.auth, headers=self.headers or None)
+
+
+def proxy_to_config(proxy: object) -> Optional[ProxyConfig]:
+    """Convert the ``proxy`` connection option — a URL string, an
+    ``httpx.URL``, or an ``httpx.Proxy`` — to the proxy configuration pyqwest
+    transports take: a proxy URL (scheme http, https, socks5, or socks5h,
+    credentials allowed in the userinfo), basic-auth credentials, and headers
+    to send to the proxy. An ``httpx.Proxy`` ``ssl_context`` has no pyqwest
+    counterpart and is rejected rather than silently dropped."""
     if proxy is None:
         return None
     if isinstance(proxy, str):
-        return proxy
+        return ProxyConfig(proxy)
     if isinstance(proxy, httpx.URL):
-        return str(proxy)
+        return ProxyConfig(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("httpx.Proxy ssl_context is not supported")
+        # httpx.Proxy splits userinfo out of the URL into `.auth`; pyqwest
+        # takes the credentials the same way, so they pass straight through.
+        return ProxyConfig(
+            str(proxy.url),
+            auth=proxy.auth,
+            headers=tuple(proxy.headers.items()),
+        )
     raise InvalidArgumentException(
-        "E2B API calls support only URL-string proxies, "
+        "Only URL-string, httpx.URL, and httpx.Proxy proxies are supported, "
         'e.g. proxy="http://user:pass@localhost:8030"'
     )
 

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

@@ -1,26 +1,23 @@
 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,
+    ProxyConfig,
     connection_retries,
-    limits,
+    make_async_logging_event_hooks,
     pool_idle_timeout,
     pool_max_idle_per_host,
-    proxy_to_url,
+    proxy_to_config,
 )
-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:
@@ -53,82 +50,118 @@ 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: 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, 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
+    separate and sits above this, on the httpx client."""
+    return ConnectionRetryTransport(
+        HTTPTransport(
+            tls_include_system_certs=True,
+            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.
-_transports: Dict[Optional[str], "AsyncApiPyqwestTransport"] = {}
+# 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"] = {}
 
 
 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)
+    proxy = proxy_to_config(config.proxy)
     with _transport_lock:
-        transport = _transports.get(proxy_url)
+        transport = _transports.get(proxy)
         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
+            transport = AsyncApiPyqwestTransport(retrying_http_transport(proxy))
+            _transports[proxy] = 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, "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),
+    )