Add support for image= to sandbox.create(...)

Author: scotttrinh

src/vercel/_internal/unstable/sandbox/api_client.py

@@ -19,6 +19,7 @@
     ValidationError,
     field_serializer,
     field_validator,
+    model_validator,
 )
 
 from vercel._internal.http import (
@@ -131,6 +132,7 @@ class _CreateSandboxRequest(_ApiRequestModel):
     project_id: str = Field(serialization_alias="projectId")
     name: str | None = None
     runtime: str | None = None
+    image: str | None = None
     source: SandboxSource | None = None
     ports: list[int] | None = None
     timeout: timedelta | None = None
@@ -162,6 +164,12 @@ def _serialize_retention(self, value: SnapshotRetention | None) -> JSONObject |
     def _serialize_network_policy(self, value: NetworkPolicy | None) -> JSONObject | None:
         return None if value is None else _serialize_network_policy(value)
 
+    @model_validator(mode="after")
+    def _validate_boot_selectors(self) -> "_CreateSandboxRequest":
+        if self.runtime is not None and self.image is not None:
+            raise ValueError("runtime and image are mutually exclusive")
+        return self
+
     @field_validator("network_policy", mode="before")
     @classmethod
     def _validate_network_policy(cls, value: object) -> NetworkPolicy | None:
@@ -879,6 +887,7 @@ async def create_sandbox(
         project_id: str | None = None,
         name: str | None = None,
         runtime: str | None = None,
+        image: str | None = None,
         source: SandboxSource | None = None,
         ports: list[int] | None = None,
         execution_time_limit: timedelta | None = None,
@@ -895,6 +904,7 @@ async def create_sandbox(
             project_id=project_id or credentials.project_id,
             name=name,
             runtime=runtime,
+            image=image,
             source=source,
             ports=ports,
             timeout=execution_time_limit,

src/vercel/_internal/unstable/sandbox/async_runtime.py

@@ -1074,6 +1074,7 @@ class _CreateSandboxParams:
     project_id: str | None = None
     name: str | None = None
     runtime: str | None = None
+    image: str | None = None
     source: SandboxSource | None = None
     ports: list[int] | None = None
     execution_time_limit: timedelta | None = None
@@ -1120,6 +1121,7 @@ async def _run_once(self) -> Sandbox:
             project_id=self._params.project_id,
             name=self._params.name,
             runtime=self._params.runtime,
+            image=self._params.image,
             source=self._params.source,
             ports=self._params.ports,
             execution_time_limit=self._params.execution_time_limit,
@@ -1262,6 +1264,7 @@ def create_sandbox_operation(
     project_id: str | None = None,
     name: str | None = None,
     runtime: str | None = None,
+    image: str | None = None,
     source: SandboxSource | None = None,
     ports: list[int] | None = None,
     execution_time_limit: DurationInput = None,
@@ -1280,6 +1283,7 @@ def create_sandbox_operation(
             project_id=project_id,
             name=name,
             runtime=runtime,
+            image=image,
             source=source,
             ports=ports,
             execution_time_limit=parse_duration_seconds(execution_time_limit),

src/vercel/_internal/unstable/sandbox/service.py

@@ -247,6 +247,7 @@ async def create_sandbox(
         project_id: str | None = None,
         name: str | None = None,
         runtime: str | None = None,
+        image: str | None = None,
         source: SandboxSource | None = None,
         ports: list[int] | None = None,
         execution_time_limit: timedelta | None = None,
@@ -263,6 +264,7 @@ async def create_sandbox(
             project_id=project_id,
             name=name,
             runtime=runtime,
+            image=image,
             source=source,
             ports=ports,
             execution_time_limit=execution_time_limit,

src/vercel/_internal/unstable/sandbox/sync_runtime.py

@@ -1184,6 +1184,7 @@ def create_sandbox(
     project_id: str | None = None,
     name: str | None = None,
     runtime: str | None = None,
+    image: str | None = None,
     source: SandboxSource | None = None,
     ports: list[int] | None = None,
     execution_time_limit: DurationInput = None,
@@ -1202,6 +1203,7 @@ def create_sandbox(
                 project_id=project_id,
                 name=name,
                 runtime=runtime,
+                image=image,
                 source=source,
                 ports=ports,
                 execution_time_limit=parse_duration_seconds(execution_time_limit),

src/vercel/unstable/sandbox/__init__.py

@@ -87,6 +87,7 @@ def create_sandbox(
     project_id: str | None = None,
     name: str | None = None,
     runtime: str | None = None,
+    image: str | None = None,
     source: SandboxSource | None = None,
     ports: list[int] | None = None,
     execution_time_limit: DurationInput = None,
@@ -109,7 +110,9 @@ def create_sandbox(
         project_id: Project that owns the sandbox. Uses the active credentials
             when omitted.
         name: Requested sandbox name. The service generates one when omitted.
-        runtime: Runtime image or runtime identifier.
+        runtime: Runtime identifier. Mutually exclusive with ``image``.
+        image: Vercel Container Registry image to start the sandbox from.
+            Mutually exclusive with ``runtime``.
         source: Git, tarball, or snapshot source used to initialize the sandbox.
         ports: Ports to expose from the sandbox.
         execution_time_limit: Maximum session runtime in seconds or as a
@@ -137,6 +140,7 @@ def create_sandbox(
         project_id=project_id,
         name=name,
         runtime=runtime,
+        image=image,
         source=source,
         ports=ports,
         execution_time_limit=execution_time_limit,

src/vercel/unstable/sandbox/sync.py

@@ -84,6 +84,7 @@ def create_sandbox(
     project_id: str | None = None,
     name: str | None = None,
     runtime: str | None = None,
+    image: str | None = None,
     source: SandboxSource | None = None,
     ports: list[int] | None = None,
     execution_time_limit: DurationInput = None,
@@ -106,7 +107,9 @@ def create_sandbox(
         project_id: Project that owns the sandbox. Uses the active credentials
             when omitted.
         name: Requested sandbox name. The service generates one when omitted.
-        runtime: Runtime image or runtime identifier.
+        runtime: Runtime identifier. Mutually exclusive with ``image``.
+        image: Vercel Container Registry image to start the sandbox from.
+            Mutually exclusive with ``runtime``.
         source: Git, tarball, or snapshot source used to initialize the sandbox.
         ports: Ports to expose from the sandbox.
         execution_time_limit: Maximum session runtime in seconds or as a
@@ -133,6 +136,7 @@ def create_sandbox(
         project_id=project_id,
         name=name,
         runtime=runtime,
+        image=image,
         source=source,
         ports=ports,
         execution_time_limit=execution_time_limit,

tests/unstable/test_sandbox_public_flow.py

@@ -437,6 +437,60 @@ def update_handler(request: httpx.Request) -> httpx.Response:
     assert not hasattr(handle, "model_dump")
 
 
+@respx.mock
+async def test_public_create_sandbox_serializes_image_without_runtime(
+    mock_env_clear: None,
+) -> None:
+    route = respx.post("https://sandbox.test/v2/sandboxes").mock(
+        return_value=httpx.Response(200, json=_sandbox_response())
+    )
+
+    async with vercel.session(service_options=_session_options()):
+        await sandbox.create_sandbox(
+            name="preview",
+            image="acme/worker:latest",
+            source=SnapshotSource(snapshot_id="snap_123"),
+        )
+
+    body = json.loads(route.calls.last.request.content)
+    assert body["image"] == "acme/worker:latest"
+    assert body["source"] == {"type": "snapshot", "snapshotId": "snap_123"}
+    assert "runtime" not in body
+
+
+@respx.mock
+def test_sync_create_sandbox_serializes_image_without_runtime(mock_env_clear: None) -> None:
+    route = respx.post("https://sandbox.test/v2/sandboxes").mock(
+        return_value=httpx.Response(200, json=_sandbox_response())
+    )
+
+    with vercel.session(service_options=_session_options()):
+        sandbox_sync.create_sandbox(name="preview", image="acme/worker:latest")
+
+    body = json.loads(route.calls.last.request.content)
+    assert body["image"] == "acme/worker:latest"
+    assert "runtime" not in body
+
+
+@respx.mock
+async def test_public_create_sandbox_rejects_runtime_and_image(
+    mock_env_clear: None,
+) -> None:
+    route = respx.post("https://sandbox.test/v2/sandboxes").mock(
+        return_value=httpx.Response(200, json=_sandbox_response())
+    )
+
+    async with vercel.session(service_options=_session_options()):
+        with pytest.raises(ValidationError, match="runtime and image"):
+            await sandbox.create_sandbox(
+                name="preview",
+                runtime="python3.13",
+                image="acme/worker:latest",
+            )
+
+    assert not route.called
+
+
 @respx.mock
 async def test_network_policy_async_public_flow(mock_env_clear: None) -> None:
     create_route = respx.post("https://sandbox.test/v2/sandboxes").mock(