feat(sdks/python): resilient SDK transport (OSEP-0016)

Implement the Python side of OSEP-0016 for the sandbox and
code-interpreter SDKs. Adds a shared retry policy, extended
exception hierarchy, and async/sync httpx transport wrappers.

Retry policy
- New `RetryPolicy` value type on `ConnectionConfig` /
  `ConnectionConfigSync`; defaults enable retry for idempotent
  methods only, `POST`/`PATCH` status-code retry is opt-in.
- Default idempotent retryable status set: `{429, 502, 503}`.
- Decorrelated jitter (default), full jitter, and deterministic
  exponential; `Retry-After` honored with a fixed 60s ceiling.
- `RetryPolicy.disabled()` gives end-to-end fast-fail semantics
  and suppresses fresh-connection recovery.
- `HTTPStatus` used at the public API surface so callers get
  type checking and IDE completion for the status sets.

Exception hierarchy
- Extend `SandboxException` with an `is_retryable` accessor
  reflecting the SDK's actual retry decision at emission time.
- New subclasses: `SandboxRateLimitException` (carries
  `retry_after`), `SandboxTimeoutException`,
  `SandboxConnectionException`. Existing classes and their
  public fields are unchanged.

Transport wrappers
- `RetryAsyncTransport` (httpx.AsyncBaseTransport) and
  `RetrySyncTransport` (httpx.BaseTransport) install the retry
  engine at the same layer the SDK already uses for transport
  concerns. Generated OpenAPI code is not touched.
- `ConnectionConfig` exposes a `sse_transport` property so SSE
  bootstraps bypass retry while sharing the underlying pool.
  Command / isolated / code-interpreter SSE clients now use it.
- `lifecycle_metrics` keeps its own short-lived httpx.Client;
  sharing the SDK transport would let the metrics client's
  `close()` tear down every other adapter's connection pool.

Tests
- Retry policy defaults, validation, and `HTTPStatus` usage.
- Pure decision function: status matrix, transport branch,
  budget/deadline/cancellation predicates, opt-in rules,
  jitter algorithms, and `Retry-After` parsing/clamping.
- Async + sync httpx transport integration via scripted mock
  transports covering success, 4xx pass-through, budget
  exhaustion, pre-send vs. post-send classification, opt-in,
  and observability callback invariants.
- `ConnectionConfig` wiring: default wraps, `disabled()` skips
  wrapping, user-supplied transports are used verbatim.
- SSE clients use the raw inner transport, verified end-to-end.

Author: Pangjiping

sdks/code-interpreter/python/pyproject.toml

@@ -43,7 +43,7 @@ classifiers = [
 ]
 dependencies = [
     "pydantic>=2.4.2,<3.0",
-    "opensandbox>=0.1.10,<0.2.0",
+    "opensandbox>=0.1.15,<0.2.0",
 ]
 
 [project.urls]

sdks/code-interpreter/python/src/code_interpreter/adapters/code_adapter.py

@@ -130,6 +130,7 @@ def __init__(
             "Accept": "text/event-stream",
             "Cache-Control": "no-cache",
         }
+        # SSE bootstraps must not be retried; use the raw transport.
         self._sse_client = httpx.AsyncClient(
             headers=sse_headers,
             timeout=httpx.Timeout(
@@ -138,7 +139,7 @@ def __init__(
                 write=timeout_seconds,
                 pool=None,
             ),
-            transport=self.connection_config.transport,
+            transport=self.connection_config.sse_transport,
         )
 
     async def _get_client(self):

sdks/code-interpreter/python/src/code_interpreter/sync/adapters/code_adapter.py

@@ -118,12 +118,13 @@ def __init__(
             "Accept": "text/event-stream",
             "Cache-Control": "no-cache",
         }
+        # SSE bootstraps must not be retried; use the raw transport.
         self._sse_client = httpx.Client(
             headers=sse_headers,
             timeout=httpx.Timeout(
                 connect=timeout_seconds, read=None, write=timeout_seconds, pool=None
             ),
-            transport=self.connection_config.transport,
+            transport=self.connection_config.sse_transport,
         )
 
     def _get_execd_url(self, path: str) -> str:

sdks/sandbox/python/src/opensandbox/adapters/command_adapter.py

@@ -184,6 +184,9 @@ def __init__(
             "Accept": "text/event-stream",
             "Cache-Control": "no-cache",
         }
+        # SSE bootstraps must not be retried: bodies are not replayable
+        # and resume is out of scope here. Share the raw transport so
+        # the underlying connection pool is still shared.
         self._sse_client = httpx.AsyncClient(
             headers=sse_headers,
             timeout=httpx.Timeout(
@@ -192,7 +195,7 @@ def __init__(
                 write=timeout_seconds,
                 pool=None,
             ),
-            transport=self.connection_config.transport,
+            transport=self.connection_config.sse_transport,
         )
 
     async def _get_client(self):

sdks/sandbox/python/src/opensandbox/adapters/converter/response_handler.py

@@ -28,7 +28,12 @@
 from http import HTTPStatus
 from typing import Any, TypeVar
 
-from opensandbox.exceptions import SandboxApiException, SandboxError
+from opensandbox.exceptions import (
+    SandboxApiException,
+    SandboxError,
+    SandboxRateLimitException,
+)
+from opensandbox.transport._decision import parse_retry_after
 
 logger = logging.getLogger(__name__)
 
@@ -98,6 +103,49 @@ def require_parsed(response_obj: Any, expected_type: type[T], operation_name: st
     return parsed
 
 
+def _retry_after_seconds(headers: Any) -> float | None:
+    """Return Retry-After (seconds) from response headers, or None."""
+    if not headers:
+        return None
+    try:
+        raw = headers.get("Retry-After") or headers.get("retry-after")
+    except Exception:
+        return None
+    parsed = parse_retry_after(raw if isinstance(raw, str) else None)
+    return parsed.total_seconds() if parsed is not None else None
+
+
+# Upper bound on the raw response body slice we splice into an
+# exception's ``str()``. The full body is always available on
+# ``exc.response_body`` untruncated.
+_RAW_BODY_MESSAGE_LIMIT = 512
+
+
+def _raw_body_bytes(response_obj: Any) -> bytes | None:
+    """Return the response's raw body as ``bytes`` when available."""
+    content = getattr(response_obj, "content", None)
+    if isinstance(content, bytes):
+        return content
+    if isinstance(content, bytearray):
+        return bytes(content)
+    return None
+
+
+def _raw_body_message_fragment(body: bytes | None) -> str | None:
+    """Best-effort decode of the raw body for splicing into an error message."""
+    if not body:
+        return None
+    try:
+        text = body.decode("utf-8", errors="replace").strip()
+    except Exception:
+        return None
+    if not text:
+        return None
+    if len(text) > _RAW_BODY_MESSAGE_LIMIT:
+        text = text[:_RAW_BODY_MESSAGE_LIMIT] + "…"
+    return text
+
+
 def handle_api_error(response_obj: Any, operation_name: str = "API call") -> None:
     """
     Check API response for errors and raise exception if needed.
@@ -109,10 +157,13 @@ def handle_api_error(response_obj: Any, operation_name: str = "API call") -> Non
         operation_name: Name of the operation for error messages
 
     Raises:
-        SandboxApiException: If the response indicates an error
+        SandboxRateLimitException: On HTTP 429 (Too Many Requests).
+        SandboxApiException: On any other HTTP >= 300.
     """
     status_code = _status_code_to_int(getattr(response_obj, "status_code", 0))
-    request_id = extract_request_id(getattr(response_obj, "headers", None))
+    headers = getattr(response_obj, "headers", None)
+    request_id = extract_request_id(headers)
+    raw_body = _raw_body_bytes(response_obj)
 
     logger.debug(f"{operation_name} response: status={status_code}")
 
@@ -136,9 +187,28 @@ def handle_api_error(response_obj: Any, operation_name: str = "API call") -> Non
                     message=str(parsed_message or ""),
                 )
 
+        # Fall back to the raw body when the SDK could not parse a
+        # structured message/code, so callers do not lose the server's
+        # own explanation.
+        if sandbox_error is None:
+            raw_fragment = _raw_body_message_fragment(raw_body)
+            if raw_fragment:
+                error_message = f"{error_message}: {raw_fragment}"
+
+        if status_code == HTTPStatus.TOO_MANY_REQUESTS:
+            raise SandboxRateLimitException(
+                message=error_message,
+                status_code=status_code,
+                request_id=request_id,
+                retry_after=_retry_after_seconds(headers),
+                error=sandbox_error,
+                response_body=raw_body,
+            )
+
         raise SandboxApiException(
             message=error_message,
             status_code=status_code,
             request_id=request_id,
             error=sandbox_error,
+            response_body=raw_body,
         )

sdks/sandbox/python/src/opensandbox/adapters/isolated_adapter.py

@@ -231,6 +231,7 @@ def __init__(
             "Accept": "text/event-stream",
             "Cache-Control": "no-cache",
         }
+        # SSE bootstraps must not be retried; use the raw transport.
         self._sse_client = httpx.AsyncClient(
             headers=sse_headers,
             timeout=httpx.Timeout(
@@ -239,7 +240,7 @@ def __init__(
                 write=timeout_seconds,
                 pool=None,
             ),
-            transport=self.connection_config.transport,
+            transport=self.connection_config.sse_transport,
         )
 
     def _get_url(self, path: str) -> str:

sdks/sandbox/python/src/opensandbox/config/connection.py

@@ -29,6 +29,8 @@
     field_validator,
 )
 
+from opensandbox.transport import RetryAsyncTransport, RetryPolicy
+
 
 class ConnectionConfig(BaseModel):
     """
@@ -49,6 +51,7 @@ class ConnectionConfig(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     _owns_transport: bool = PrivateAttr(default=True)
+    _sse_transport: httpx.AsyncBaseTransport | None = PrivateAttr(default=None)
 
     api_key: str | None = Field(
         default=None, description="API key for authentication with sandbox service"
@@ -73,9 +76,17 @@ class ConnectionConfig(BaseModel):
     transport: httpx.AsyncBaseTransport | None = Field(
         default=None,
         description=(
-            "Shared httpx transport instance used by all HTTP clients within a "
-            "Sandbox/Manager instance. Pass a custom transport (e.g. AsyncHTTPTransport "
-            "with custom settings) to control connection pooling, proxies, retries, etc."
+            "Shared httpx transport for non-SSE clients. When unset the SDK "
+            "builds an AsyncHTTPTransport wrapped by RetryAsyncTransport "
+            "honoring `retry_policy`. Pass a custom transport to override "
+            "the whole stack (retry wrapping is then the caller's job)."
+        ),
+    )
+    retry_policy: RetryPolicy = Field(
+        default_factory=RetryPolicy,
+        description=(
+            "Retry policy applied to non-SSE requests. "
+            "Pass RetryPolicy.disabled() for fast-fail semantics."
         ),
     )
     use_server_proxy: bool = Field(
@@ -114,6 +125,10 @@ class ConnectionConfig(BaseModel):
     def model_post_init(self, __context: object) -> None:
         # If the user explicitly provided `transport`, the SDK must not close it.
         self._owns_transport = "transport" not in self.model_fields_set
+        # A caller-supplied transport is used verbatim for both normal
+        # and SSE clients; we cannot know whether it retries already.
+        if self.transport is not None:
+            self._sse_transport = self.transport
         # Best-effort: attach the SDK host's own IP so the server can see the
         # client's self-reported address. Never overrides a user-supplied value
         # and is skipped silently when the IP cannot be determined.
@@ -125,30 +140,56 @@ def with_transport_if_missing(self) -> "ConnectionConfig":
         """
         Ensure a transport exists for this SDK resource.
 
-        If `transport` is missing, return a copy with a default transport and
-        mark it as SDK-owned. If present, return self unchanged.
+        When `transport` is missing, return a copy whose `transport` is
+        retry-wrapped and whose `sse_transport` is the raw inner
+        transport. When present, return self unchanged.
         """
         if self.transport is not None:
             return self
-        transport = httpx.AsyncHTTPTransport(
+        inner = httpx.AsyncHTTPTransport(
             limits=httpx.Limits(
                 max_connections=100,
                 max_keepalive_connections=20,
                 keepalive_expiry=30.0,
             ),
         )
-        config = self.model_copy(update={"transport": transport})
+        wrapped: httpx.AsyncBaseTransport
+        if self.retry_policy.wraps_transport():
+            # Wrapper is required either for retries or purely to enforce
+            # per_attempt_timeout / overall_deadline / on_retry hooks.
+            wrapped = RetryAsyncTransport(
+                inner, self.retry_policy, owns_inner=False
+            )
+        else:
+            # Fully disabled policy with no timeout/deadline knobs: skip
+            # wrapping so fast-fail is truly end-to-end.
+            wrapped = inner
+        config = self.model_copy(update={"transport": wrapped})
         config._owns_transport = True
+        config._sse_transport = inner
         return config
 
+    @property
+    def sse_transport(self) -> httpx.AsyncBaseTransport | None:
+        """Transport for SSE clients. Never retry-wrapped, falls back to `transport`."""
+        if self._sse_transport is not None:
+            return self._sse_transport
+        return self.transport
+
     async def close_transport_if_owned(self) -> None:
         """Close the transport only if it was created by default_factory."""
-        if self.transport is None or not self._owns_transport:
+        if not self._owns_transport:
+            return
+        # The connection pool lives on the inner (SSE) transport. The
+        # retry wrapper's aclose is a no-op when it does not own its
+        # inner, so closing whichever we own is enough.
+        target = self._sse_transport or self.transport
+        if target is None:
             return
         try:
-            await self.transport.aclose()
+            await target.aclose()
         except Exception:
-            # Avoid raising during cleanup paths
+            # Avoid raising during cleanup paths.
             pass
 
     @field_validator("protocol")

sdks/sandbox/python/src/opensandbox/config/connection_sync.py

@@ -25,6 +25,8 @@
 import httpx
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator
 
+from opensandbox.transport import RetryPolicy, RetrySyncTransport
+
 
 class ConnectionConfigSync(BaseModel):
     """
@@ -39,6 +41,7 @@ class ConnectionConfigSync(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     _owns_transport: bool = PrivateAttr(default=True)
+    _sse_transport: httpx.BaseTransport | None = PrivateAttr(default=None)
 
     api_key: str | None = Field(
         default=None, description="API key for authentication with sandbox service"
@@ -60,9 +63,16 @@ class ConnectionConfigSync(BaseModel):
     transport: httpx.BaseTransport | None = Field(
         default=None,
         description=(
-            "Shared httpx transport instance used by all HTTP clients within a "
-            "Sandbox/Manager instance. Pass a custom transport (e.g. HTTPTransport "
-            "with custom limits/proxies) to control connection pooling, proxies, retries, etc."
+            "Shared httpx transport for non-SSE clients. When unset the SDK "
+            "builds an HTTPTransport wrapped by RetrySyncTransport honoring "
+            "`retry_policy`."
+        ),
+    )
+    retry_policy: RetryPolicy = Field(
+        default_factory=RetryPolicy,
+        description=(
+            "Retry policy applied to non-SSE requests. "
+            "Pass RetryPolicy.disabled() for fast-fail semantics."
         ),
     )
     use_server_proxy: bool = Field(
@@ -99,6 +109,8 @@ class ConnectionConfigSync(BaseModel):
 
     def model_post_init(self, __context: object) -> None:
         self._owns_transport = "transport" not in self.model_fields_set
+        if self.transport is not None:
+            self._sse_transport = self.transport
         # Best-effort: attach the SDK host's own IP (see async ConnectionConfig).
         from opensandbox.config import client_ip
 
@@ -108,28 +120,47 @@ def with_transport_if_missing(self) -> "ConnectionConfigSync":
         """
         Ensure a transport exists for this SDK resource.
 
-        If `transport` is missing, return a copy with a default transport and
-        mark it as SDK-owned. If present, return self unchanged.
+        When `transport` is missing, return a copy whose `transport` is
+        retry-wrapped and whose `sse_transport` is the raw inner
+        transport. When present, return self unchanged.
         """
         if self.transport is not None:
             return self
-        transport = httpx.HTTPTransport(
+        inner = httpx.HTTPTransport(
             limits=httpx.Limits(
                 max_connections=100,
                 max_keepalive_connections=20,
                 keepalive_expiry=30.0,
             ),
         )
-        config = self.model_copy(update={"transport": transport})
+        wrapped: httpx.BaseTransport
+        if self.retry_policy.wraps_transport():
+            wrapped = RetrySyncTransport(
+                inner, self.retry_policy, owns_inner=False
+            )
+        else:
+            wrapped = inner
+        config = self.model_copy(update={"transport": wrapped})
         config._owns_transport = True
+        config._sse_transport = inner
         return config
 
+    @property
+    def sse_transport(self) -> httpx.BaseTransport | None:
+        """Transport for SSE clients. Never retry-wrapped, falls back to `transport`."""
+        if self._sse_transport is not None:
+            return self._sse_transport
+        return self.transport
+
     def close_transport_if_owned(self) -> None:
         """Close the transport only if it was created by default_factory."""
-        if self.transport is None or not self._owns_transport:
+        if not self._owns_transport:
+            return
+        target = self._sse_transport or self.transport
+        if target is None:
             return
         try:
-            self.transport.close()
+            target.close()
         except Exception:
             pass