feat: add Databricks sandbox backend for the native executor

examples/databricks/runner.py

@@ -0,0 +1,53 @@
+"""Run an agent against a Databricks sandbox through the native executor.
+
+Prerequisites (see src/agents/extensions/sandbox/databricks/README.md):
+  - the SDK preview wheel plus the ``databricks`` extra
+  - a running Databricks sandbox whose image carries the ``codex`` exec-server
+  - Databricks auth (a CLI profile) and model-provider auth
+
+Environment variables:
+  DATABRICKS_CONFIG_PROFILE  Databricks CLI profile (default: DEFAULT)
+  SANDBOX_NAME               the target sandbox name/id (required)
+  SANDBOX_CWD                remote workspace path (default: /home/sandbox-agent/work)
+  OPENAI_API_KEY             model-provider key for NativeProvider (required)
+  OPENAI_BASE_URL            optional model-provider base URL
+  OPENAI_MODEL               model name (default: gpt-5.5)
+
+Run:
+    python -m examples.databricks.runner "Inspect the project and list next steps"
+"""
+
+import asyncio
+import os
+import sys
+
+from databricks.sdk import WorkspaceClient
+
+from agents import Agent, NativeProvider
+from agents.extensions.sandbox.databricks import databricks_sandbox_environment
+
+
+async def main(task: str) -> None:
+    workspace_client = WorkspaceClient(profile=os.getenv("DATABRICKS_CONFIG_PROFILE", "DEFAULT"))
+    async with databricks_sandbox_environment(
+        workspace_client=workspace_client,
+        sandbox_name=os.environ["SANDBOX_NAME"],
+        cwd=os.getenv("SANDBOX_CWD", "/home/sandbox-agent/work"),
+    ) as environment:
+        async with NativeProvider(
+            base_url=os.getenv("OPENAI_BASE_URL"),
+            api_key=os.environ["OPENAI_API_KEY"],
+            supports_websockets=False,
+        ) as provider:
+            agent = Agent(
+                instructions="Inspect the assigned workspace and report findings.",
+                model=os.getenv("OPENAI_MODEL", "gpt-5.5"),
+                provider=provider,
+            )
+            async with agent.session(environment=environment) as session:
+                result = await (await session.submit(task)).get_result()
+                print(result.final_output)
+
+
+if __name__ == "__main__":
+    asyncio.run(main(" ".join(sys.argv[1:]) or "Inspect the assigned workspace"))

pyproject.toml

@@ -25,6 +25,14 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
 ]
 
+[project.optional-dependencies]
+databricks = [
+    "databricks-sdk[sandbox]>=0.153.0",
+    "websockets>=13",
+    "cryptography>=42",
+    "certifi",
+]
+
 [project.urls]
 Homepage = "https://github.com/openai/openai-agents-python/tree/feat/codex-harness"
 Repository = "https://github.com/openai/openai-agents-python"
@@ -140,6 +148,10 @@ disallow_incomplete_defs = false
 disallow_untyped_defs = false
 disallow_untyped_calls = false
 
+[[tool.mypy.overrides]]
+module = ["databricks.*"]
+ignore_missing_imports = true
+
 [tool.coverage.run]
 source = ["src/agents"]
 omit = ["tests/*"]

src/agents/extensions/__init__.py

@@ -0,0 +1 @@
+"""Optional, separately installed integrations for the Agents SDK."""

src/agents/extensions/sandbox/__init__.py

@@ -0,0 +1,6 @@
+"""Sandbox integrations that provision executors for the native ``Environment``.
+
+Each provider lives in its own subpackage and depends on an extra, so importing
+``agents`` never requires a provider SDK. Import the provider explicitly, e.g.
+``from agents.extensions.sandbox.databricks import databricks_sandbox_environment``.
+"""

src/agents/extensions/sandbox/databricks/README.md

@@ -0,0 +1,106 @@
+# Databricks sandbox backend
+
+Back a native `Environment` with a Databricks sandbox, so an agent driven by
+this SDK executes `exec_command` / `write_stdin` / `apply_patch` inside the
+sandbox.
+
+## How it works
+
+The native `Environment` dials an authenticated `wss://` executor and relays the
+exec-server protocol over it. A Databricks sandbox runs that exec-server in
+stdio mode; the `databricks-sdk` streaming command API
+(`WorkspaceClient.sandbox.execute_command`) gives bidirectional stdio to it.
+Nothing in the SDK speaks stdio and nothing in the sandbox speaks WebSocket, so
+`SandboxExecServerGateway` bridges them: it terminates the WebSocket,
+authenticates the bearer token, opens the exec-server's stdio, and relays bytes
+both ways. It parses neither protocol.
+
+Because `execute_command` reaches the sandbox outbound through the Databricks
+control plane, the gateway runs on `localhost` next to the application and the
+sandbox needs no inbound ingress. The agent loop runs in your process (via
+`NativeProvider`), which is what makes the `Environment` approval / transform
+callbacks and `NetworkPolicy` available - the trade against the managed
+Agents API path, where OpenAI runs the loop and the sandbox exec-server dials
+out instead.
+
+## Install
+
+The v1.0 SDK is a GitHub preview wheel (off PyPI). Install it, then this extra:
+
+```sh
+pip install "<openai-agents preview wheel url>"
+pip install "openai-agents[databricks]"   # from this source tree
+```
+
+The extra pulls `databricks-sdk[sandbox]`, `websockets`, `cryptography`, and
+`certifi`.
+
+## Usage
+
+```python
+from databricks.sdk import WorkspaceClient
+from agents import Agent, NativeProvider
+from agents.extensions.sandbox.databricks import databricks_sandbox_environment
+
+async with databricks_sandbox_environment(
+    workspace_client=WorkspaceClient(profile="my-workspace"),
+    sandbox_name="openai-agent-sandbox",
+    cwd="/home/sandbox-agent/work",
+) as environment:
+    async with NativeProvider(api_key=..., supports_websockets=False) as provider:
+        agent = Agent(model="gpt-5.5", provider=provider, instructions="...")
+        async with agent.session(environment=environment) as session:
+            print((await (await session.submit("...")).get_result()).final_output)
+```
+
+A runnable version is in `examples/databricks/runner.py`.
+
+## Provisioning the sandbox
+
+This package wires an `Environment` to an *existing* sandbox; it does not create
+one, to avoid depending on sandbox-spec classes it cannot pin. Provision with
+the usual API, then ensure the exec-server is reachable, e.g.:
+
+```python
+from databricks.sdk import WorkspaceClient
+
+w = WorkspaceClient(profile="my-workspace")
+w.sandbox.create_sandbox(..., sandbox_id="openai-agent-sandbox")
+```
+
+`exec-server` is killed when the sandbox stops, so a new connection after a
+stop relaunches it.
+
+## TLS
+
+`Environment` requires `wss`, so the gateway needs TLS even on loopback. By
+default `databricks_sandbox_environment` generates an ephemeral self-signed
+certificate and, for the session, points `SSL_CERT_FILE` at a bundle of the
+public CAs (`certifi`) plus that certificate, then restores it on exit - so the
+native runtime trusts the gateway without breaking model-provider TLS. Pass your
+own `ssl_context` and set `trust_gateway_ca=False` to manage trust yourself.
+
+## Unverified prerequisites (draft)
+
+Two assumptions this draft is built on still need confirmation against a live
+sandbox before it can be relied on:
+
+1. **exec-server stdio mode in the image.** The default command is
+   `("codex", "exec-server", "stdio")`. The SDK's own `offline_docker.py`
+   example proves `exec-server stdio` is valid; the Databricks cookbook only
+   exercised the outbound `codex exec-server --remote` mode. Confirm with
+   `codex exec-server --help` in a sandbox, and override `exec_server_command`
+   if the binary path or subcommand differ.
+2. **Streaming `execute_command` suitability.** The relay assumes
+   `CommandProcess` gives a usable bidirectional binary stdio stream (`stdin`
+   writes, `stdout.read1`) for the exec-server protocol. The v0.x provider used
+   only the buffered `execute_command_sync`, so this streaming path has not been
+   exercised for this purpose. Validate interactive `write_stdin` under load and
+   how `CommandProcess` reconnect interacts with the exec-server session.
+3. **WebSocket framing.** The gateway relays an opaque byte stream: it forwards
+   each inbound frame to stdin as-is and sends arbitrary `read1` chunks of stdout
+   back as frames. This is correct only if the native client reassembles the
+   byte stream rather than requiring one protocol message per frame (the
+   reference `offline_docker.py` bridges the socket straight to byte pipes, which
+   indicates byte-stream semantics). If the executor is message-framed, the relay
+   must preserve message boundaries instead.

src/agents/extensions/sandbox/databricks/__init__.py

@@ -0,0 +1,17 @@
+"""Databricks sandbox backend for the native ``Environment``.
+
+Install with the ``databricks`` extra. The application provisions a Databricks
+sandbox whose image carries the ``codex`` exec-server; this package runs a
+localhost WSS gateway that relays that exec-server's stdio to the SDK, and
+builds the ``Environment`` pointing at it.
+"""
+
+from .environment import databricks_sandbox_environment
+from .gateway import GatewayHandle, SandboxExecServerGateway, self_signed_server_context
+
+__all__ = [
+    "databricks_sandbox_environment",
+    "GatewayHandle",
+    "SandboxExecServerGateway",
+    "self_signed_server_context",
+]

src/agents/extensions/sandbox/databricks/environment.py

@@ -0,0 +1,135 @@
+"""Provision a native ``Environment`` backed by a Databricks sandbox.
+
+This async context manager starts a :class:`SandboxExecServerGateway` for an
+existing sandbox and yields an ``Environment`` wired to it. The SDK's
+``EnvironmentSource`` accepts an async context manager, so the whole
+start-gateway / tear-down lifecycle fits one ``async with``:
+
+    async with databricks_sandbox_environment(
+        workspace_client=WorkspaceClient(profile="my-workspace"),
+        sandbox_name="openai-agent-sandbox",
+        cwd="/home/sandbox-agent/work",
+    ) as environment:
+        async with agent.session(environment=environment) as session:
+            ...
+
+Provisioning the sandbox itself (``WorkspaceClient.sandbox.create_sandbox``) and
+ensuring the ``codex`` exec-server is present are left to the caller, so this
+code takes no dependency on sandbox-spec classes it cannot pin. See the package
+README for the create/ensure pattern.
+"""
+
+from __future__ import annotations
+
+import os
+import secrets
+import ssl
+import tempfile
+from collections.abc import AsyncIterator, Callable, Sequence
+from contextlib import asynccontextmanager
+from typing import TYPE_CHECKING
+
+from ....executor import Environment, ExecutorToolName
+from .gateway import SandboxExecServerGateway, self_signed_server_context
+
+if TYPE_CHECKING:
+    from databricks.sdk import WorkspaceClient
+
+_DEFAULT_TOOLS: tuple[ExecutorToolName, ...] = ("exec_command", "write_stdin", "apply_patch")
+
+
+@asynccontextmanager
+async def databricks_sandbox_environment(
+    *,
+    workspace_client: WorkspaceClient,
+    sandbox_name: str,
+    cwd: str,
+    tools: Sequence[ExecutorToolName] = _DEFAULT_TOOLS,
+    exec_server_command: Sequence[str] = ("codex", "exec-server", "stdio"),
+    token: str | None = None,
+    ssl_context: ssl.SSLContext | None = None,
+    trust_gateway_ca: bool = True,
+) -> AsyncIterator[Environment]:
+    """Yield an ``Environment`` connected to ``sandbox_name`` through a gateway.
+
+    Args:
+        workspace_client: An authenticated ``databricks.sdk.WorkspaceClient``.
+        sandbox_name: An existing, running sandbox with the exec-server available.
+        cwd: Absolute remote workspace path the executor operates in.
+        tools: Native executor tools to expose. ``write_stdin`` requires
+            ``exec_command``.
+        exec_server_command: Argv that starts the exec-server in stdio mode in
+            the sandbox. The default assumes the image's ``codex`` binary serves
+            ``exec-server stdio`` - confirm this before relying on it.
+        token: Bearer secret shared between the gateway and the SDK. A random
+            value is generated when omitted.
+        ssl_context: Server TLS context for the gateway. A self-signed loopback
+            certificate is generated when omitted.
+        trust_gateway_ca: When a self-signed context is generated, add its CA to
+            a combined bundle and point ``SSL_CERT_FILE`` at it for the session,
+            so the native runtime trusts the gateway while keeping public CAs
+            (via ``certifi``) for the model provider. Restored on exit. Disable
+            if you manage trust yourself.
+    """
+    secret = token or secrets.token_urlsafe(32)
+    generated_ca: str | None = None
+    if ssl_context is None:
+        ssl_context, generated_ca = self_signed_server_context()
+
+    gateway = SandboxExecServerGateway(
+        workspace_client=workspace_client,
+        sandbox_name=sandbox_name,
+        token=secret,
+        ssl_context=ssl_context,
+        exec_server_command=exec_server_command,
+        ca_cert_path=generated_ca,
+    )
+    handle = await gateway.start()
+
+    restore: Callable[[], None] | None = None
+    if trust_gateway_ca and handle.ca_cert_path is not None:
+        restore = _install_trust_bundle(handle.ca_cert_path)
+    try:
+        yield Environment(
+            url=handle.url,
+            cwd=cwd,
+            tools=tuple(tools),
+            headers={"Authorization": f"Bearer {secret}"},
+        )
+    finally:
+        if restore is not None:
+            restore()
+        await gateway.aclose()
+
+
+def _install_trust_bundle(ca_cert_path: str) -> Callable[[], None]:
+    """Point ``SSL_CERT_FILE`` at ``certifi`` plus the gateway CA; return a restorer.
+
+    The native runtime shares one trust store for the model provider and the
+    executor connection, so the gateway CA is appended to the public bundle
+    rather than replacing it.
+    """
+    import certifi
+
+    with open(certifi.where(), "rb") as public_bundle:
+        public = public_bundle.read()
+    with open(ca_cert_path, "rb") as gateway_ca:
+        extra = gateway_ca.read()
+
+    combined = tempfile.NamedTemporaryFile(prefix="dbx-trust-", suffix=".pem", delete=False)
+    with combined:
+        combined.write(public)
+        if not public.endswith(b"\n"):
+            combined.write(b"\n")
+        combined.write(extra)
+
+    previous = os.environ.get("SSL_CERT_FILE")
+    os.environ["SSL_CERT_FILE"] = combined.name
+
+    def restore() -> None:
+        if previous is None:
+            os.environ.pop("SSL_CERT_FILE", None)
+        else:
+            os.environ["SSL_CERT_FILE"] = previous
+
+    return restore