feat: async Python client for non-blocking sandbox operations (#510)

* feat: add async Python sandbox client

Add AsyncSandboxClient and supporting async modules so users can
await sandbox creation, command execution, and file I/O without
blocking the event loop. This is needed for async frameworks like
FastAPI, aiohttp, and async agent orchestrators.

Key design decisions:
- Explicit retry logic matching sync client's status-code retries
- asyncio.Lock guards on shared state for coroutine safety
- BaseException handling to catch CancelledError during create
- try/finally on all K8s watch streams to prevent leaked connections
- Required connection_config (no LocalTunnel support in async)
- Lazy __getattr__ import with actionable ImportError message

Optional deps: pip install k8s-agent-sandbox[async]

Supersedes #256.

* fix: clear sandbox registry on close, set httpx default timeout

- Clear _active_connection_sandboxes after closing connections to
  prevent stale references
- Set httpx.AsyncClient timeout to 60s matching sync client behavior
  (httpx defaults to 5s)

Made-with: Cursor

* remove internal plan doc from PR

Made-with: Cursor

* fix: address code review feedback

- Remove dead retry branch in async connector (retryable status codes
  are handled before raise_for_status, so the except handler never
  sees them)
- Narrow except BaseException to except (Exception, CancelledError)
  to avoid catching KeyboardInterrupt/SystemExit
- Replace logging.basicConfig() with logging.getLogger(__name__) so
  library code doesn't hijack the root logger
- Use built-in generics (list, dict, tuple) instead of typing imports
  since project requires Python 3.10+
- Replace __getattr__ lazy import with global try/except pattern
- Add test for close() clearing the sandbox registry

Made-with: Cursor

* fix: address second round of review feedback

- Make fallback AsyncSandboxClient a class so isinstance() works
- Use logger = logging.getLogger(__name__) in async_connector and
  async_k8s_helper (was still using root logger)
- Add None event guard to resolve_sandbox_name watch loop to prevent
  TypeError crash on None events
- Document no-atexit behavior in AsyncSandboxClient docstring

Made-with: Cursor

* fix: add transport-level retries for connection errors

Add AsyncHTTPTransport(retries=3) so connection-level failures (TCP,
DNS) are retried at the transport layer, matching the sync client's
urllib3.Retry behavior. The manual retry loop handles status-code
retries (500/502/503/504) which the transport doesn't cover.

Made-with: Cursor

* fix: retry watch streams on premature connection drops

Watch streams can close prematurely due to network blips or API server
restarts, causing an erroneous TimeoutError even when only seconds have
elapsed. Track a deadline and re-establish the watch with remaining time
instead of raising immediately.

Fixed in all 6 watch methods (3 sync + 3 async): resolve_sandbox_name,
wait_for_sandbox_ready, wait_for_gateway_ip.

Made-with: Cursor

Author: AndyJCai

clients/python/agentic-sandbox-client/README.md

@@ -189,6 +189,40 @@ client = SandboxClient(
 sandbox = client.create_sandbox(template="node-sandbox-template", namespace="default").
 ```
 
+### 5. Async Client
+
+For async applications (FastAPI, aiohttp, async agent orchestrators), use the `AsyncSandboxClient`.
+Install the async extras first:
+
+```bash
+pip install k8s-agent-sandbox[async]
+```
+
+The async client requires an explicit connection config — `LocalTunnel` mode is not supported
+because it relies on a synchronous `kubectl port-forward` subprocess. Use `DirectConnection` or
+`GatewayConnection` instead.
+
+```python
+import asyncio
+from k8s_agent_sandbox import AsyncSandboxClient
+from k8s_agent_sandbox.models import SandboxDirectConnectionConfig
+
+async def main():
+    config = SandboxDirectConnectionConfig(
+        api_url="http://sandbox-router-svc.default.svc.cluster.local:8080"
+    )
+
+    async with AsyncSandboxClient(connection_config=config) as client:
+        sandbox = await client.create_sandbox(
+            template="python-sandbox-template",
+            namespace="default",
+        )
+        result = await sandbox.commands.run("echo 'Hello from async!'")
+        print(result.stdout)
+
+asyncio.run(main())
+```
+
 ## Testing
 
 A test script is included to verify the full lifecycle (Creation -> Execution -> File I/O -> Cleanup).

clients/python/agentic-sandbox-client/k8s_agent_sandbox/__init__.py

@@ -20,3 +20,15 @@
     SandboxPortForwardError,
     SandboxRequestError,
 )
+
+
+try:
+    from .async_sandbox_client import AsyncSandboxClient
+except ImportError:
+    class AsyncSandboxClient:  # type: ignore[no-redef]
+        """Placeholder that raises ImportError when async extras are missing."""
+        def __init__(self, *args, **kwargs):
+            raise ImportError(
+                "AsyncSandboxClient requires the 'async' extras. "
+                "Install with: pip install k8s-agent-sandbox[async]"
+            )

clients/python/agentic-sandbox-client/k8s_agent_sandbox/async_connector.py

@@ -0,0 +1,141 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+import logging
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+from .async_k8s_helper import AsyncK8sHelper
+from .exceptions import SandboxRequestError
+from .models import (
+    SandboxConnectionConfig,
+    SandboxDirectConnectionConfig,
+    SandboxGatewayConnectionConfig,
+    SandboxLocalTunnelConnectionConfig,
+)
+
+RETRYABLE_STATUS_CODES = {500, 502, 503, 504}
+MAX_RETRIES = 5
+BACKOFF_FACTOR = 0.5
+
+
+class AsyncSandboxConnector:
+    """
+    Async connector for communicating with a Sandbox over HTTP using httpx.
+
+    Supports DirectConnection and GatewayConnection modes. LocalTunnel mode
+    is not supported because it relies on a long-running subprocess; use the
+    sync SandboxConnector for local development.
+    """
+
+    def __init__(
+        self,
+        sandbox_id: str,
+        namespace: str,
+        connection_config: SandboxConnectionConfig,
+        k8s_helper: AsyncK8sHelper,
+    ):
+        if isinstance(connection_config, SandboxLocalTunnelConnectionConfig):
+            raise ValueError(
+                "AsyncSandboxConnector does not support SandboxLocalTunnelConnectionConfig. "
+                "Use SandboxDirectConnectionConfig or SandboxGatewayConnectionConfig instead. "
+                "For local development, use the synchronous SandboxClient."
+            )
+
+        self.id = sandbox_id
+        self.namespace = namespace
+        self.connection_config = connection_config
+        self.k8s_helper = k8s_helper
+
+        self._base_url: str | None = None
+        transport = httpx.AsyncHTTPTransport(retries=3)
+        self.client = httpx.AsyncClient(
+            transport=transport, timeout=httpx.Timeout(60.0)
+        )
+
+    async def _resolve_base_url(self) -> str:
+        if self._base_url:
+            return self._base_url
+
+        if isinstance(self.connection_config, SandboxDirectConnectionConfig):
+            self._base_url = self.connection_config.api_url
+        elif isinstance(self.connection_config, SandboxGatewayConnectionConfig):
+            ip_address = await self.k8s_helper.wait_for_gateway_ip(
+                self.connection_config.gateway_name,
+                self.connection_config.gateway_namespace,
+                self.connection_config.gateway_ready_timeout,
+            )
+            self._base_url = f"http://{ip_address}"
+        else:
+            raise ValueError(
+                f"AsyncSandboxConnector does not support {type(self.connection_config).__name__}."
+            )
+
+        return self._base_url
+
+    async def send_request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
+        base_url = await self._resolve_base_url()
+        url = f"{base_url.rstrip('/')}/{endpoint.lstrip('/')}"
+
+        headers = kwargs.pop("headers", {}).copy()
+        headers["X-Sandbox-ID"] = self.id
+        headers["X-Sandbox-Namespace"] = self.namespace
+        headers["X-Sandbox-Port"] = str(self.connection_config.server_port)
+
+        last_response: httpx.Response | None = None
+        for attempt in range(MAX_RETRIES + 1):
+            try:
+                response = await self.client.request(
+                    method, url, headers=headers, **kwargs
+                )
+                if response.status_code in RETRYABLE_STATUS_CODES and attempt < MAX_RETRIES:
+                    delay = BACKOFF_FACTOR * (2 ** attempt)
+                    logger.warning(
+                        f"Retryable status {response.status_code} from {url}, "
+                        f"attempt {attempt + 1}/{MAX_RETRIES + 1}, retrying in {delay:.1f}s"
+                    )
+                    last_response = response
+                    await asyncio.sleep(delay)
+                    continue
+                response.raise_for_status()
+                return response
+            except httpx.HTTPStatusError as e:
+                logger.error(f"Request to sandbox failed: {e}")
+                raise SandboxRequestError(
+                    f"Failed to communicate with the sandbox at {url}.",
+                    status_code=e.response.status_code,
+                    response=e.response,
+                ) from e
+            except httpx.HTTPError as e:
+                logger.error(f"Request to sandbox failed: {e}")
+                self._base_url = None
+                raise SandboxRequestError(
+                    f"Failed to communicate with the sandbox at {url}.",
+                    status_code=None,
+                    response=None,
+                ) from e
+
+        logger.error(f"All {MAX_RETRIES + 1} attempts failed for {url}")
+        raise SandboxRequestError(
+            f"Failed to communicate with the sandbox at {url} after {MAX_RETRIES + 1} attempts.",
+            status_code=last_response.status_code if last_response else None,
+            response=last_response,
+        )
+
+    async def close(self):
+        await self.client.aclose()
+        self._base_url = None