Remote API

Client and service for sandboxes that live in another process. See Remote Sandboxes for the concepts and the security model.

RemoteSandbox

Needs the remote extra (httpx).

pydantic_ai_backends.remote.client.RemoteSandbox

Bases: BaseSandbox

Sandbox backed by a sandboxd service over HTTP.

Implements the same synchronous surface as :class:~pydantic_ai_backends.DockerSandbox, so it drops into :class:~pydantic_ai_backends.SessionManager or a console toolset unchanged. Every file operation is served by the remote side rather than derived from shell commands, so read of a large file transfers only the requested slice.

The session — and therefore the container behind it — is opened on the first operation, not on construction. An agent granted a sandbox it never uses costs nothing: no session, no container, not even a round trip. Call :meth:start explicitly only to pre-warm one.

Failures are returned, never raised: a transport error surfaces the same way a missing file does (b"", [], an Error: ... string), matching LocalBackend and DockerSandbox. A tool call must not take down an agent run because a socket blipped.

Parameters:

Name	Type	Description	Default
service_url	str	Base URL of the service, e.g. "http://sandboxd:8080". Ignored when client is supplied.	'http://localhost:8080'
token	str	Service token, sent as the X-Sandbox-Token header. Used to open a session; afterwards the session's own token is used.	''
session_id	str | None	Identifier to open or re-attach to. Generated when omitted.	None
reuse	bool	Attach to the session under session_id when the service already has one open, instead of failing. Off by default, so a colliding id is reported rather than silently sharing another caller's sandbox. Turn it on for a sandbox that spans several runs — one conversation, say — where a later run has to reach the files an earlier one wrote.	False
runtime	str | None	Alias of a server-allowed runtime. None takes the server default. The server rejects anything not on its allowlist.	None
tenant	str | None	Whoever this sandbox is for, against which the service applies its per-tenant session ceiling. A capacity label only — it grants nothing and authorizes nothing.	None
timeout	float	Request timeout in seconds for operations without their own.	DEFAULT_TIMEOUT_SECONDS
client	Client | None	Pre-built httpx.Client. Supply one to share a connection pool, set custom transport or TLS, or drive an in-process ASGI app.	None

Example

Python

from pydantic_ai_backends.remote import RemoteSandbox

sandbox = RemoteSandbox("http://sandboxd:8080", token="...")
sandbox.start()
print(sandbox.execute("python -c 'print(1+1)'").output)
sandbox.stop()

Source code in src/pydantic_ai_backends/remote/client.py

class RemoteSandbox(BaseSandbox):
    """Sandbox backed by a `sandboxd` service over HTTP.

    Implements the same synchronous surface as
    :class:`~pydantic_ai_backends.DockerSandbox`, so it drops into
    :class:`~pydantic_ai_backends.SessionManager` or a console toolset
    unchanged. Every file operation is served by the remote side rather than
    derived from shell commands, so `read` of a large file transfers only the
    requested slice.

    The session — and therefore the container behind it — is opened on the first
    operation, not on construction. An agent granted a sandbox it never uses
    costs nothing: no session, no container, not even a round trip. Call
    :meth:`start` explicitly only to pre-warm one.

    Failures are returned, never raised: a transport error surfaces the same way
    a missing file does (`b""`, `[]`, an `Error: ...` string), matching
    `LocalBackend` and `DockerSandbox`. A tool call must not take down an
    agent run because a socket blipped.

    Args:
        service_url: Base URL of the service, e.g. `"http://sandboxd:8080"`.
            Ignored when `client` is supplied.
        token: Service token, sent as the `X-Sandbox-Token` header. Used to open
            a session; afterwards the session's own token is used.
        session_id: Identifier to open or re-attach to. Generated when omitted.
        reuse: Attach to the session under `session_id` when the service already
            has one open, instead of failing. Off by default, so a colliding id
            is reported rather than silently sharing another caller's sandbox.
            Turn it on for a sandbox that spans several runs — one conversation,
            say — where a later run has to reach the files an earlier one wrote.
        runtime: Alias of a server-allowed runtime. `None` takes the server
            default. The server rejects anything not on its allowlist.
        tenant: Whoever this sandbox is for, against which the service applies
            its per-tenant session ceiling. A capacity label only — it grants
            nothing and authorizes nothing.
        timeout: Request timeout in seconds for operations without their own.
        client: Pre-built `httpx.Client`. Supply one to share a connection pool,
            set custom transport or TLS, or drive an in-process ASGI app.

    Example:
        ```python
        from pydantic_ai_backends.remote import RemoteSandbox

        sandbox = RemoteSandbox("http://sandboxd:8080", token="...")
        sandbox.start()
        print(sandbox.execute("python -c 'print(1+1)'").output)
        sandbox.stop()
        ```
    """

    def __init__(
        self,
        service_url: str = "http://localhost:8080",
        *,
        token: str = "",
        session_id: str | None = None,
        runtime: str | None = None,
        tenant: str | None = None,
        reuse: bool = False,
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        client: httpx.Client | None = None,
    ) -> None:
        super().__init__(session_id or f"remote-{uuid.uuid4().hex[:12]}")
        self._service_token = token
        self._session_token = token
        # Built here rather than in `start()` so an id or tenant the service
        # would reject fails at construction, where the caller is looking, and
        # not out of whichever tool call happens to open the session.
        self._create_request = wire.CreateSessionRequest(
            session_id=self._id, runtime=runtime, tenant=tenant, reuse=reuse
        )
        self._timeout = timeout
        # Learned from the service when the session opens. Until then the local
        # default stands, which is the right answer for a service we cannot ask.
        self._server_timeout = timeout
        self._started = False
        # Operations run on a thread pool, so two of them can arrive at an
        # unopened session at once; without this both would POST /sessions and
        # one would lose to its own 409.
        self._open_lock = threading.Lock()
        self.touch()

        self._service_url = service_url.rstrip("/")
        if client is not None:
            self._http = client
            self._owns_client = False
        else:
            self._http = self._new_client()
            self._owns_client = True

    def _new_client(self) -> Any:
        """Build the HTTP client this sandbox owns."""
        httpx_module = load("httpx", purpose="RemoteSandbox")
        return httpx_module.Client(
            base_url=self._service_url,
            timeout=httpx_module.Timeout(self._timeout),
        )

    @property
    def session_id(self) -> str:
        """Alias for the sandbox id, for parity with `DockerSandbox`."""
        return self._id

    def _url(self, operation: str) -> str:
        return f"/sessions/{self._id}/{operation}"

    def _post(self, url: str, payload: dict[str, Any], timeout: float | None = None) -> Any:
        """POST JSON with the session token, returning the response or `None`.

        Opens the session first if this is the first operation.

        Args:
            url: Operation path.
            payload: JSON body.
            timeout: Transport timeout. `None` waits out the service's own
                ceiling plus slack, because every operation here runs a shell
                command on the far side and giving up first reports a running
                command as an unreachable service.

        Returns:
            The `httpx.Response`, or `None` when the request could not be made
            or the service answered with an error status.
        """
        if not self._ensure_session():
            return None

        self.touch()
        try:
            response = self._http.post(
                url,
                json=payload,
                headers={wire.TOKEN_HEADER: self._session_token},
                timeout=(
                    timeout
                    if timeout is not None
                    else self._server_timeout + TRANSPORT_SLACK_SECONDS
                ),
            )
        except Exception:
            return None
        return None if response.status_code >= 400 else response

    def _ensure_session(self) -> bool:
        """Open the session on first use.

        Returns:
            Whether a session is available. A failure here is reported the same
            way a failed operation is — the caller degrades instead of raising,
            because a tool call must not end an agent run.
        """
        if self._started:
            return True
        with self._open_lock:
            if self._started:
                return True
            try:
                self.start()
            except RuntimeError:
                return False
        return True

    def start(self) -> None:
        """Open the remote session, or attach to it when `reuse` is set.

        Idempotent within one instance: repeated calls do nothing once a session
        is open. With `reuse`, a *new* instance naming an existing session id
        attaches to it rather than failing — which is what makes a sandbox
        outlive the run that created it.

        Usable again after :meth:`stop`, which opens a fresh session — the same
        way `DockerSandbox` starts a new container after being stopped.

        Raises:
            RuntimeError: If the service refuses to open a session. Unlike the
                file operations this does raise, because a caller that cannot
                get a sandbox at all needs to know why.
        """
        if self._started:
            return

        if self._owns_client and self._http.is_closed:
            # `stop()` closed the client it owned. Rebuilt rather than left
            # broken, so a stopped sandbox behaves like `DockerSandbox` — usable
            # again — instead of silently reporting the service as unreachable
            # for the rest of its life.
            self._http = self._new_client()

        try:
            response = self._http.post(
                "/sessions",
                json=self._create_request.model_dump(mode="json"),
                headers={wire.TOKEN_HEADER: self._service_token},
                timeout=self._timeout,
            )
        except Exception as exc:
            raise RuntimeError(f"Could not reach the sandbox service: {exc}") from exc

        if response.status_code >= 400:
            raise RuntimeError(
                f"Sandbox service refused to open a session "
                f"(HTTP {response.status_code}): {response.text}"
            )

        created = _parse(response, wire.SessionCreated)
        if created is None:
            raise RuntimeError(
                "Sandbox service answered the open with something that is not a "
                "session. Is the URL pointing at sandboxd rather than a proxy?"
            )
        self._id = created.session.session_id
        self._session_token = created.token
        self._started = True
        self._server_timeout = self._fetch_server_timeout()

    @property
    def server_timeout(self) -> float:
        """How long the service lets an operation run, once the session is open.

        Read from `GET /policy` when the session opens, falling back to this
        sandbox's own `timeout` when the service will not say. It exists because
        the two numbers are one contract and were set independently: the client
        gave up after 60s while the service happily ran a command for its default
        300s, so anything in between was reported to the agent as "sandbox
        service unavailable" while it was in fact still running — and typically
        retried, starting a second one.
        """
        return self._server_timeout

    def _fetch_server_timeout(self) -> float:
        """The service's own operation ceiling, or this sandbox's default."""
        try:
            response = self._http.get(
                "/policy",
                headers={wire.TOKEN_HEADER: self._service_token},
                timeout=self._timeout,
            )
        except Exception:
            return self._timeout
        policy = _parse(response if response.status_code < 400 else None, wire.ServicePolicy)
        if policy is None or policy.execute_timeout <= 0:
            return self._timeout
        return float(policy.execute_timeout)

    def is_alive(self) -> bool:
        """Whether the remote session exists and its sandbox is running."""
        try:
            response = self._http.get(
                f"/sessions/{self._id}",
                headers={wire.TOKEN_HEADER: self._session_token},
                timeout=self._timeout,
            )
        except Exception:
            return False
        info = _parse(response if response.status_code < 400 else None, wire.SessionInfo)
        return info is not None and info.alive

    def resource_usage(self) -> SandboxUsage | None:
        """Sample the remote sandbox's resource usage."""
        try:
            response = self._http.get(
                f"/sessions/{self._id}",
                params={"usage": "true"},
                headers={wire.TOKEN_HEADER: self._session_token},
                timeout=self._timeout,
            )
        except Exception:
            return None
        info = _parse(response if response.status_code < 400 else None, wire.SessionInfo)
        usage = info.usage if info is not None else None
        if usage is None:
            return None
        return SandboxUsage(
            memory_bytes=usage.memory_bytes,
            memory_limit_bytes=usage.memory_limit_bytes,
            cpu_percent=usage.cpu_percent,
            pids=usage.pids,
        )

    def stop(self, purge: bool = False) -> None:
        """Delete the remote session and close an owned HTTP client.

        Args:
            purge: Also discard what the session accumulated — its container's
                filesystem and its host workspace. Leave it off to end a turn
                while keeping the files for the next one; turn it on when the
                thing the session belonged to is gone for good.
        """
        if self._started:
            # Best effort: the service reaps idle sessions on its own, so a
            # failed teardown costs a timeout, not a leak.
            with contextlib.suppress(Exception):
                self._http.delete(
                    f"/sessions/{self._id}",
                    params={"purge": "true"} if purge else None,
                    headers={wire.TOKEN_HEADER: self._session_token},
                    timeout=self._timeout,
                )
            self._started = False

        if self._owns_client:
            with contextlib.suppress(Exception):
                self._http.close()

    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a command in the remote sandbox.

        Without an explicit `timeout` the command runs under the service's
        ceiling, and the transport waits that long plus slack — the two are one
        contract, and the client giving up first turned a slow-but-successful
        command into a reported outage.
        """
        body = wire.ExecRequest(command=command, timeout_seconds=timeout)
        effective = float(timeout) if timeout is not None else self._server_timeout
        transport_timeout = effective + TRANSPORT_SLACK_SECONDS
        response = self._post(self._url("exec"), body.model_dump(mode="json"), transport_timeout)
        parsed = _parse(response, wire.ExecResponse)
        if parsed is None:
            return ExecuteResponse(output="Error: sandbox service unavailable", exit_code=1)
        return ExecuteResponse(
            output=parsed.output,
            exit_code=parsed.exit_code,
            truncated=parsed.truncated,
        )

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a text file."""
        body = wire.ReadRequest(path=path, offset=offset, limit=limit)
        response = self._post(self._url("read"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.ReadResponse)
        if parsed is None:
            return f"Error: could not read '{path}'"
        return parsed.content

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file as bytes, or `b""` when unreadable."""
        body = wire.ReadBytesRequest(path=path)
        response = self._post(self._url("read_bytes"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.ReadBytesResponse)
        if parsed is None:
            return b""
        try:
            return base64.b64decode(parsed.content_b64, validate=True)
        except Exception:
            return b""

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file."""
        raw = content if isinstance(content, bytes) else content.encode("utf-8")
        body = wire.WriteRequest(path=path, content_b64=base64.b64encode(raw).decode("ascii"))
        response = self._post(self._url("write"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.WriteResponse)
        if parsed is None:
            return WriteResult(error=f"Error: could not write '{path}'")
        return WriteResult(path=parsed.path, error=parsed.error)

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Replace a string inside a file."""
        body = wire.EditRequest(
            path=path,
            old_string=old_string,
            new_string=new_string,
            replace_all=replace_all,
        )
        response = self._post(self._url("edit"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.EditResponse)
        if parsed is None:
            return EditResult(error=f"Error: could not edit '{path}'")
        return EditResult(path=parsed.path, error=parsed.error, occurrences=parsed.occurrences)

    def exists(self, path: str) -> bool:
        """Whether the path is a regular file."""
        body = wire.ExistsRequest(path=path)
        response = self._post(self._url("exists"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.ExistsResponse)
        return parsed is not None and parsed.exists

    def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory, or `[]` when it cannot be listed."""
        body = wire.LsRequest(path=path)
        response = self._post(self._url("ls"), body.model_dump(mode="json"))
        return _to_file_infos(response)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match files by glob, or `[]` when the search cannot be run."""
        body = wire.GlobRequest(pattern=pattern, path=path)
        response = self._post(self._url("glob"), body.model_dump(mode="json"))
        return _to_file_infos(response)

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search file contents, returning matches or an error string."""
        body = wire.GrepRequest(pattern=pattern, path=path, glob=glob, ignore_hidden=ignore_hidden)
        response = self._post(self._url("grep"), body.model_dump(mode="json"))
        parsed = _parse(response, wire.GrepResponse)
        if parsed is None:
            return f"Error: could not search for {pattern!r}"
        if parsed.error is not None:
            return parsed.error
        return [
            GrepMatch(path=m.path, line_number=m.line_number, line=m.line) for m in parsed.matches
        ]

session_id property

Alias for the sandbox id, for parity with DockerSandbox.

__init__(service_url='http://localhost:8080', *, token='', session_id=None, runtime=None, tenant=None, reuse=False, timeout=DEFAULT_TIMEOUT_SECONDS, client=None)

Source code in src/pydantic_ai_backends/remote/client.py

def __init__(
    self,
    service_url: str = "http://localhost:8080",
    *,
    token: str = "",
    session_id: str | None = None,
    runtime: str | None = None,
    tenant: str | None = None,
    reuse: bool = False,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    client: httpx.Client | None = None,
) -> None:
    super().__init__(session_id or f"remote-{uuid.uuid4().hex[:12]}")
    self._service_token = token
    self._session_token = token
    # Built here rather than in `start()` so an id or tenant the service
    # would reject fails at construction, where the caller is looking, and
    # not out of whichever tool call happens to open the session.
    self._create_request = wire.CreateSessionRequest(
        session_id=self._id, runtime=runtime, tenant=tenant, reuse=reuse
    )
    self._timeout = timeout
    # Learned from the service when the session opens. Until then the local
    # default stands, which is the right answer for a service we cannot ask.
    self._server_timeout = timeout
    self._started = False
    # Operations run on a thread pool, so two of them can arrive at an
    # unopened session at once; without this both would POST /sessions and
    # one would lose to its own 409.
    self._open_lock = threading.Lock()
    self.touch()

    self._service_url = service_url.rstrip("/")
    if client is not None:
        self._http = client
        self._owns_client = False
    else:
        self._http = self._new_client()
        self._owns_client = True

start()

Open the remote session, or attach to it when reuse is set.

Idempotent within one instance: repeated calls do nothing once a session is open. With reuse, a new instance naming an existing session id attaches to it rather than failing — which is what makes a sandbox outlive the run that created it.

Usable again after :meth:stop, which opens a fresh session — the same way DockerSandbox starts a new container after being stopped.

Raises:

Type	Description
RuntimeError	If the service refuses to open a session. Unlike the file operations this does raise, because a caller that cannot get a sandbox at all needs to know why.

Source code in src/pydantic_ai_backends/remote/client.py

def start(self) -> None:
    """Open the remote session, or attach to it when `reuse` is set.

    Idempotent within one instance: repeated calls do nothing once a session
    is open. With `reuse`, a *new* instance naming an existing session id
    attaches to it rather than failing — which is what makes a sandbox
    outlive the run that created it.

    Usable again after :meth:`stop`, which opens a fresh session — the same
    way `DockerSandbox` starts a new container after being stopped.

    Raises:
        RuntimeError: If the service refuses to open a session. Unlike the
            file operations this does raise, because a caller that cannot
            get a sandbox at all needs to know why.
    """
    if self._started:
        return

    if self._owns_client and self._http.is_closed:
        # `stop()` closed the client it owned. Rebuilt rather than left
        # broken, so a stopped sandbox behaves like `DockerSandbox` — usable
        # again — instead of silently reporting the service as unreachable
        # for the rest of its life.
        self._http = self._new_client()

    try:
        response = self._http.post(
            "/sessions",
            json=self._create_request.model_dump(mode="json"),
            headers={wire.TOKEN_HEADER: self._service_token},
            timeout=self._timeout,
        )
    except Exception as exc:
        raise RuntimeError(f"Could not reach the sandbox service: {exc}") from exc

    if response.status_code >= 400:
        raise RuntimeError(
            f"Sandbox service refused to open a session "
            f"(HTTP {response.status_code}): {response.text}"
        )

    created = _parse(response, wire.SessionCreated)
    if created is None:
        raise RuntimeError(
            "Sandbox service answered the open with something that is not a "
            "session. Is the URL pointing at sandboxd rather than a proxy?"
        )
    self._id = created.session.session_id
    self._session_token = created.token
    self._started = True
    self._server_timeout = self._fetch_server_timeout()

stop(purge=False)

Delete the remote session and close an owned HTTP client.

Parameters:

Name	Type	Description	Default
purge	bool	Also discard what the session accumulated — its container's filesystem and its host workspace. Leave it off to end a turn while keeping the files for the next one; turn it on when the thing the session belonged to is gone for good.	False

Source code in src/pydantic_ai_backends/remote/client.py

def stop(self, purge: bool = False) -> None:
    """Delete the remote session and close an owned HTTP client.

    Args:
        purge: Also discard what the session accumulated — its container's
            filesystem and its host workspace. Leave it off to end a turn
            while keeping the files for the next one; turn it on when the
            thing the session belonged to is gone for good.
    """
    if self._started:
        # Best effort: the service reaps idle sessions on its own, so a
        # failed teardown costs a timeout, not a leak.
        with contextlib.suppress(Exception):
            self._http.delete(
                f"/sessions/{self._id}",
                params={"purge": "true"} if purge else None,
                headers={wire.TOKEN_HEADER: self._session_token},
                timeout=self._timeout,
            )
        self._started = False

    if self._owns_client:
        with contextlib.suppress(Exception):
            self._http.close()

is_alive()

Whether the remote session exists and its sandbox is running.

Source code in src/pydantic_ai_backends/remote/client.py

def is_alive(self) -> bool:
    """Whether the remote session exists and its sandbox is running."""
    try:
        response = self._http.get(
            f"/sessions/{self._id}",
            headers={wire.TOKEN_HEADER: self._session_token},
            timeout=self._timeout,
        )
    except Exception:
        return False
    info = _parse(response if response.status_code < 400 else None, wire.SessionInfo)
    return info is not None and info.alive

resource_usage()

Sample the remote sandbox's resource usage.

Source code in src/pydantic_ai_backends/remote/client.py

def resource_usage(self) -> SandboxUsage | None:
    """Sample the remote sandbox's resource usage."""
    try:
        response = self._http.get(
            f"/sessions/{self._id}",
            params={"usage": "true"},
            headers={wire.TOKEN_HEADER: self._session_token},
            timeout=self._timeout,
        )
    except Exception:
        return None
    info = _parse(response if response.status_code < 400 else None, wire.SessionInfo)
    usage = info.usage if info is not None else None
    if usage is None:
        return None
    return SandboxUsage(
        memory_bytes=usage.memory_bytes,
        memory_limit_bytes=usage.memory_limit_bytes,
        cpu_percent=usage.cpu_percent,
        pids=usage.pids,
    )

execute(command, timeout=None)

Run a command in the remote sandbox.

Without an explicit timeout the command runs under the service's ceiling, and the transport waits that long plus slack — the two are one contract, and the client giving up first turned a slow-but-successful command into a reported outage.

Source code in src/pydantic_ai_backends/remote/client.py

def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a command in the remote sandbox.

    Without an explicit `timeout` the command runs under the service's
    ceiling, and the transport waits that long plus slack — the two are one
    contract, and the client giving up first turned a slow-but-successful
    command into a reported outage.
    """
    body = wire.ExecRequest(command=command, timeout_seconds=timeout)
    effective = float(timeout) if timeout is not None else self._server_timeout
    transport_timeout = effective + TRANSPORT_SLACK_SECONDS
    response = self._post(self._url("exec"), body.model_dump(mode="json"), transport_timeout)
    parsed = _parse(response, wire.ExecResponse)
    if parsed is None:
        return ExecuteResponse(output="Error: sandbox service unavailable", exit_code=1)
    return ExecuteResponse(
        output=parsed.output,
        exit_code=parsed.exit_code,
        truncated=parsed.truncated,
    )

read(path, offset=0, limit=2000)

Read a slice of a text file.

Source code in src/pydantic_ai_backends/remote/client.py

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read a slice of a text file."""
    body = wire.ReadRequest(path=path, offset=offset, limit=limit)
    response = self._post(self._url("read"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.ReadResponse)
    if parsed is None:
        return f"Error: could not read '{path}'"
    return parsed.content

read_bytes(path)

Read a whole file as bytes, or b"" when unreadable.

Source code in src/pydantic_ai_backends/remote/client.py

def read_bytes(self, path: str) -> bytes:
    """Read a whole file as bytes, or `b""` when unreadable."""
    body = wire.ReadBytesRequest(path=path)
    response = self._post(self._url("read_bytes"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.ReadBytesResponse)
    if parsed is None:
        return b""
    try:
        return base64.b64decode(parsed.content_b64, validate=True)
    except Exception:
        return b""

write(path, content)

Write a file.

Source code in src/pydantic_ai_backends/remote/client.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write a file."""
    raw = content if isinstance(content, bytes) else content.encode("utf-8")
    body = wire.WriteRequest(path=path, content_b64=base64.b64encode(raw).decode("ascii"))
    response = self._post(self._url("write"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.WriteResponse)
    if parsed is None:
        return WriteResult(error=f"Error: could not write '{path}'")
    return WriteResult(path=parsed.path, error=parsed.error)

edit(path, old_string, new_string, replace_all=False)

Replace a string inside a file.

Source code in src/pydantic_ai_backends/remote/client.py

def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Replace a string inside a file."""
    body = wire.EditRequest(
        path=path,
        old_string=old_string,
        new_string=new_string,
        replace_all=replace_all,
    )
    response = self._post(self._url("edit"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.EditResponse)
    if parsed is None:
        return EditResult(error=f"Error: could not edit '{path}'")
    return EditResult(path=parsed.path, error=parsed.error, occurrences=parsed.occurrences)

exists(path)

Whether the path is a regular file.

Source code in src/pydantic_ai_backends/remote/client.py

def exists(self, path: str) -> bool:
    """Whether the path is a regular file."""
    body = wire.ExistsRequest(path=path)
    response = self._post(self._url("exists"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.ExistsResponse)
    return parsed is not None and parsed.exists

ls_info(path)

List one directory, or [] when it cannot be listed.

Source code in src/pydantic_ai_backends/remote/client.py

def ls_info(self, path: str) -> list[FileInfo]:
    """List one directory, or `[]` when it cannot be listed."""
    body = wire.LsRequest(path=path)
    response = self._post(self._url("ls"), body.model_dump(mode="json"))
    return _to_file_infos(response)

glob_info(pattern, path='/')

Match files by glob, or [] when the search cannot be run.

Source code in src/pydantic_ai_backends/remote/client.py

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Match files by glob, or `[]` when the search cannot be run."""
    body = wire.GlobRequest(pattern=pattern, path=path)
    response = self._post(self._url("glob"), body.model_dump(mode="json"))
    return _to_file_infos(response)

grep_raw(pattern, path=None, glob=None, ignore_hidden=True)

Search file contents, returning matches or an error string.

Source code in src/pydantic_ai_backends/remote/client.py

def grep_raw(
    self,
    pattern: str,
    path: str | None = None,
    glob: str | None = None,
    ignore_hidden: bool = True,
) -> list[GrepMatch] | str:
    """Search file contents, returning matches or an error string."""
    body = wire.GrepRequest(pattern=pattern, path=path, glob=glob, ignore_hidden=ignore_hidden)
    response = self._post(self._url("grep"), body.model_dump(mode="json"))
    parsed = _parse(response, wire.GrepResponse)
    if parsed is None:
        return f"Error: could not search for {pattern!r}"
    if parsed.error is not None:
        return parsed.error
    return [
        GrepMatch(path=m.path, line_number=m.line_number, line=m.line) for m in parsed.matches
    ]

WorkspaceArchive

Read-only view of the files sessions left behind, with no sandbox running. Needs the remote extra.

pydantic_ai_backends.remote.client.WorkspaceArchive

Read-only view of the files sessions left behind.

Reads the service's host volume directly, so a session reaped long ago is still browsable and listing a conversation's files costs no container start.

Unlike :class:RemoteSandbox, this raises rather than degrading. Nothing here is in an agent's tool path — the caller is an application answering a user who asked to see some files, and it needs to tell "there are none" apart from "the service is misconfigured".

Requires the service to be running with SandboxdConfig.workspace_root set, and the service token: a reaped session has no token of its own left, and the intended caller is a backend applying its own authorization first.

Parameters:

Name	Type	Description	Default
service_url	str	Base URL of the service. Ignored when client is supplied.	'http://localhost:8080'
token	str	Service token.	''
timeout	float	Request timeout in seconds.	DEFAULT_TIMEOUT_SECONDS
client	Client | None	Pre-built httpx.Client, to share a connection pool or drive an in-process ASGI app.	None

Example

Python

from pydantic_ai_backends.remote import WorkspaceArchive

archive = WorkspaceArchive("http://sandboxd:8080", token="...")
for entry in archive.ls(session_id):
    print(entry["path"], entry["size"])
print(archive.read(session_id, "report.md"))

Source code in src/pydantic_ai_backends/remote/client.py

class WorkspaceArchive:
    """Read-only view of the files sessions left behind.

    Reads the service's host volume directly, so a session reaped long ago is
    still browsable and listing a conversation's files costs no container start.

    Unlike :class:`RemoteSandbox`, this **raises** rather than degrading. Nothing
    here is in an agent's tool path — the caller is an application answering a
    user who asked to see some files, and it needs to tell "there are none" apart
    from "the service is misconfigured".

    Requires the service to be running with `SandboxdConfig.workspace_root` set,
    and the service token: a reaped session has no token of its own left, and the
    intended caller is a backend applying its own authorization first.

    Args:
        service_url: Base URL of the service. Ignored when `client` is supplied.
        token: Service token.
        timeout: Request timeout in seconds.
        client: Pre-built `httpx.Client`, to share a connection pool or drive an
            in-process ASGI app.

    Example:
        ```python
        from pydantic_ai_backends.remote import WorkspaceArchive

        archive = WorkspaceArchive("http://sandboxd:8080", token="...")
        for entry in archive.ls(session_id):
            print(entry["path"], entry["size"])
        print(archive.read(session_id, "report.md"))
        ```
    """

    def __init__(
        self,
        service_url: str = "http://localhost:8080",
        *,
        token: str = "",
        timeout: float = DEFAULT_TIMEOUT_SECONDS,
        client: httpx.Client | None = None,
    ) -> None:
        self._token = token
        self._timeout = timeout
        if client is not None:
            self._http = client
            self._owns_client = False
        else:
            httpx_module = load("httpx", purpose="WorkspaceArchive")
            self._http = httpx_module.Client(
                base_url=service_url.rstrip("/"),
                timeout=httpx_module.Timeout(timeout),
            )
            self._owns_client = True

    def ls(self, session_id: str, path: str = ".") -> list[FileInfo]:
        """List one directory of a stored workspace.

        Args:
            session_id: Session whose files are wanted.
            path: Directory to list. An absolute in-container path works, so a
                path taken from a live session's listing can be handed straight
                back.

        Raises:
            WorkspaceArchiveError: If the workspace or directory is absent, the
                path escapes the workspace, or the service cannot be reached.
        """
        payload = wire.LsRequest(path=path).model_dump(mode="json")
        response = self._post(f"/workspaces/{session_id}/ls", payload)
        entries = [wire.FileEntry.model_validate(row) for row in response.json()]
        return [FileInfo(name=e.name, path=e.path, is_dir=e.is_dir, size=e.size) for e in entries]

    def read(self, session_id: str, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a stored workspace file.

        Decoded exactly as a live session would decode it, so the archive and the
        sandbox never disagree about what a file says.

        Raises:
            WorkspaceArchiveError: If the file is absent, too large, not readable
                as text, outside the workspace, or the service is unreachable.
        """
        payload = wire.ReadRequest(path=path, offset=offset, limit=limit).model_dump(mode="json")
        response = self._post(f"/workspaces/{session_id}/read", payload)
        return wire.ReadResponse.model_validate(response.json()).content

    def close(self) -> None:
        """Close the HTTP client, when this object built it."""
        if self._owns_client:
            with contextlib.suppress(Exception):
                self._http.close()

    def _post(self, url: str, payload: dict[str, Any]) -> Any:
        """POST with the service token, raising on anything but success."""
        try:
            response = self._http.post(
                url,
                json=payload,
                headers={wire.TOKEN_HEADER: self._token},
                timeout=self._timeout,
            )
        except Exception as exc:
            raise WorkspaceArchiveError(f"Could not reach the sandbox service: {exc}") from exc

        if response.status_code >= 400:
            raise WorkspaceArchiveError(response.text, status_code=response.status_code)
        return response

__init__(service_url='http://localhost:8080', *, token='', timeout=DEFAULT_TIMEOUT_SECONDS, client=None)

Source code in src/pydantic_ai_backends/remote/client.py

def __init__(
    self,
    service_url: str = "http://localhost:8080",
    *,
    token: str = "",
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    client: httpx.Client | None = None,
) -> None:
    self._token = token
    self._timeout = timeout
    if client is not None:
        self._http = client
        self._owns_client = False
    else:
        httpx_module = load("httpx", purpose="WorkspaceArchive")
        self._http = httpx_module.Client(
            base_url=service_url.rstrip("/"),
            timeout=httpx_module.Timeout(timeout),
        )
        self._owns_client = True

ls(session_id, path='.')

List one directory of a stored workspace.

Parameters:

Name	Type	Description	Default
session_id	str	Session whose files are wanted.	required
path	str	Directory to list. An absolute in-container path works, so a path taken from a live session's listing can be handed straight back.	'.'

Raises:

Type	Description
WorkspaceArchiveError	If the workspace or directory is absent, the path escapes the workspace, or the service cannot be reached.

Source code in src/pydantic_ai_backends/remote/client.py

def ls(self, session_id: str, path: str = ".") -> list[FileInfo]:
    """List one directory of a stored workspace.

    Args:
        session_id: Session whose files are wanted.
        path: Directory to list. An absolute in-container path works, so a
            path taken from a live session's listing can be handed straight
            back.

    Raises:
        WorkspaceArchiveError: If the workspace or directory is absent, the
            path escapes the workspace, or the service cannot be reached.
    """
    payload = wire.LsRequest(path=path).model_dump(mode="json")
    response = self._post(f"/workspaces/{session_id}/ls", payload)
    entries = [wire.FileEntry.model_validate(row) for row in response.json()]
    return [FileInfo(name=e.name, path=e.path, is_dir=e.is_dir, size=e.size) for e in entries]

read(session_id, path, offset=0, limit=2000)

Read a slice of a stored workspace file.

Decoded exactly as a live session would decode it, so the archive and the sandbox never disagree about what a file says.

Raises:

Type	Description
WorkspaceArchiveError	If the file is absent, too large, not readable as text, outside the workspace, or the service is unreachable.

Source code in src/pydantic_ai_backends/remote/client.py

def read(self, session_id: str, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read a slice of a stored workspace file.

    Decoded exactly as a live session would decode it, so the archive and the
    sandbox never disagree about what a file says.

    Raises:
        WorkspaceArchiveError: If the file is absent, too large, not readable
            as text, outside the workspace, or the service is unreachable.
    """
    payload = wire.ReadRequest(path=path, offset=offset, limit=limit).model_dump(mode="json")
    response = self._post(f"/workspaces/{session_id}/read", payload)
    return wire.ReadResponse.model_validate(response.json()).content

close()

Close the HTTP client, when this object built it.

Source code in src/pydantic_ai_backends/remote/client.py

def close(self) -> None:
    """Close the HTTP client, when this object built it."""
    if self._owns_client:
        with contextlib.suppress(Exception):
            self._http.close()

pydantic_ai_backends.remote.client.WorkspaceArchiveError

Bases: Exception

Raised when a stored workspace cannot be listed or read.

Attributes:

Name	Type	Description
status_code		What the service answered, or None when it could not be reached at all. An application proxying file views to its own users maps this onto its own response.

Source code in src/pydantic_ai_backends/remote/client.py

class WorkspaceArchiveError(Exception):
    """Raised when a stored workspace cannot be listed or read.

    Attributes:
        status_code: What the service answered, or `None` when it could not be
            reached at all. An application proxying file views to its own users
            maps this onto its own response.
    """

    def __init__(self, message: str, status_code: int | None = None) -> None:
        self.status_code = status_code
        super().__init__(message)

SandboxdConfig

Everything the service decides on a client's behalf. Needs the server extra.

pydantic_ai_backends.remote.server.SandboxdConfig dataclass

Everything the service decides on a client's behalf.

Attributes:

Name	Type	Description
token	str	Service token. Required to open, list or inspect any session.
runtimes	Mapping[str, str | SandboxRuntime]	Allowlist mapping a client-visible alias to what may run under it — a bare image string, or a :class:SandboxRuntime carrying its own ceilings and, if it needs them, packages to build in. A request naming anything else is rejected; this is what stops a client from running an image of its choosing.
default_runtime	str	Alias used when a request names none. Empty — the default — takes the first entry in runtimes, so the shipped allowlist defaults to coding while an operator who supplies their own gets whichever they listed first. Naming a specific alias here still wins; what it no longer does is force every custom allowlist to contain a particular key.
max_sessions	int | None	Ceiling on resident sandboxes across the service, or None for no ceiling. Together with the largest per-runtime memory ceiling this is what bounds the worst case an operator has to size the host for, so None is for hosts where something else does the bounding. It does not bound how many sessions may exist — see max_open_sessions.
max_open_sessions	int | None	Ceiling on sessions that exist, resident or hibernated, or None for no ceiling. A hibernated session costs disk and a few kilobytes of bookkeeping rather than memory, so this number is properly much larger than max_sessions — that gap is the difference between the sessions a host can carry and the ones it can run at once. Only meaningful with evict_idle_after, which is what demotes a session to hibernated in the first place.
evict_idle_after	int | None	Seconds of inactivity after which a session at the ceiling may be hibernated to make room for a new one, instead of the new one being refused with 429. This is what turns max_sessions from a hard cap on how many sessions may exist into a working-set size: the hibernated session keeps its token, its event log and its files, and its next request wakes it where it left off. None refuses instead of evicting. Requires workspace_root — hibernating a session whose files live only in its container would discard them silently, which is not a trade a config should make quietly.
max_sessions_per_tenant	int | None	Ceiling on simultaneous sessions carrying one tenant label. Without it, one tenant of the calling application can occupy the whole pool and every other tenant gets 429.
mem_limit	str | None	Default memory ceiling per sandbox, in Docker syntax. A runtime naming its own overrides this, upwards or downwards.
memswap_limit	str | None	Default ceiling on memory and swap combined. None pins it to mem_limit, so a sandbox at its ceiling is killed rather than allowed to swap — which is right when swap is a disk, because one swapping sandbox slows every other one on the host. Raise it only where swap is zram. There the pages stay in RAM compressed at roughly 3:1, so a sandbox briefly over its ceiling costs CPU instead of the whole host's responsiveness, and on a small box that trade is what turns an OOM kill into a slow command.
cpus	float | None	Default hard CPU ceiling per sandbox, in cores. None leaves sandboxes free to use whatever is idle, bounded only by cpu_shares.
cpu_shares	int | None	Default relative CPU weight, applied only under contention. Preferable to a hard cpus on a small host, where capping each sandbox to one core leaves the others unused while a single agent waits.
tmpfs_size	str | None	Size of an in-memory /tmp given to every sandbox, in Docker syntax. Scratch writes then never reach the container's write layer, which is both faster and the difference between a busy sandbox growing on disk and not. None leaves /tmp on the overlay. Sized against mem_limit rather than on top of it, because tmpfs pages are charged to the container's own memory cgroup — so this is memory taken away from the workload, and worth keeping small when many sessions are open at once.
pids_limit	int | None	Default process ceiling per sandbox.
network_mode	str | None	Default Docker network mode. "none" because a service handing sandboxes to untrusted code should not give them the network unless that is a deliberate choice, per runtime or service-wide.
oci_runtime	str | None	Default low-level runtime for every sandbox, overridable per runtime. None takes the daemon's default — which is the right default for us to ship, because naming a runtime the host has not registered turns every session into a 502. Set it to "runsc" once gVisor is installed to put every sandbox behind a userspace syscall layer; crun belongs in the daemon's own config instead, since it is a faster runc rather than a different isolation model.
work_dir	str	Working directory inside every sandbox. Applied to built runtimes as well, overriding whatever their RuntimeConfig says, so that one directory is where the workspace volume is mounted, what the archive endpoints read, and what a client's paths resolve against. Three places disagreeing about it is how files go missing.
workspace_root	str | None	Host directory backing each sandbox's work directory, as {workspace_root}/{session_id}/workspace. Without it a sandbox's files live only in the container's write layer, so idle reaping destroys them — which a session spanning several runs will notice. Session ids are pattern-checked before they reach here, so one cannot traverse out of this root.
prewarm	bool	Pull and build the whole allowlist in the background at startup. Without it the first session on a built runtime pays for the image build — a measured ten seconds and upwards — and pays it in the middle of somebody's request. Only applies to the default Docker builder, since nothing else knows how to warm a custom one.
persist_containers	bool	Give each sandbox a stable container name, so a reaped session keeps its filesystem and the next attach restarts the same container. workspace_root alone preserves only the work directory — packages installed with pip or apt live in the container's write layer, and an agent that reinstalls them after every idle timeout is not working in "the same" machine. The cost is that stopped containers accumulate until a session is closed with purge, so this is off by default.
idle_timeout	int	Seconds of inactivity before a session is reaped.
cleanup_interval	int	Seconds between reaping passes.
workspace_ttl	int | None	Seconds an unused workspace directory is kept before it is swept, timed from when its session was last opened. None — the default — keeps a session's files for ever, which is usually what an agent's user expects: the notes and scripts are the work. Set it only where a retention policy says otherwise. Only meaningful with workspace_root.
container_ttl	int | None	Seconds a stopped persisted container is kept before it is removed, timed from when it stopped. This reclaims what a session installed inside its container — the build, the wheels, the node_modules — while leaving its workspace untouched, so the files survive and only the rebuildable part goes. None keeps them. Only meaningful with persist_containers.
sandbox_uid	int | None	Run every built runtime's sandbox as this unprivileged user instead of root, and give each session's workspace to it. None — the default — keeps the current behaviour, where a sandbox runs as root. Worth turning on. A container escape starts from whoever the container runs as, and every file an agent writes into its bind-mounted workspace is owned by that user on the host — as root, a sandboxd running unprivileged cannot clean up after its own sessions. The image is built around the uid: a real account, a home directory, and a virtualenv it owns first on PATH, without which an agent's pip install fails and uv — having no --user mode — fails with no way forward. Two things it asks of the deployment. The service must be able to chown each session's workspace to this uid, which means either running with the privilege or running as that uid so the directories are created owned by it; a service that can do neither is told so when the session opens rather than handing an agent a workspace it cannot write to. And ready-made runtimes are left as root, since an image nobody built for this has no such user and no virtualenv, so an agent inside one could install nothing.
max_read_bytes	int	Largest single file a client may read out of a sandbox.
execute_timeout	int	Hard ceiling applied to every command, so one client cannot occupy a worker indefinitely.
max_workers	int	Threads dedicated to blocking sandbox calls. Sandbox commands are long, so this pool is kept separate from asyncio's shared default one.
ui_enabled	bool	Serve the bundled dashboard at /ui. Off by default: the page asks a human for the service token, and that token can start containers on the host, so it belongs on localhost or a private network and never on a public listener. The HTTP API is unaffected either way.

Source code in src/pydantic_ai_backends/remote/server.py

@dataclass(slots=True)
class SandboxdConfig:
    """Everything the service decides on a client's behalf.

    Attributes:
        token: Service token. Required to open, list or inspect any session.
        runtimes: Allowlist mapping a client-visible alias to what may run
            under it — a bare image string, or a :class:`SandboxRuntime` carrying
            its own ceilings and, if it needs them, packages to build in. A
            request naming anything else is rejected; this is what stops a client
            from running an image of its choosing.
        default_runtime: Alias used when a request names none. Empty — the
            default — takes the first entry in `runtimes`, so the shipped
            allowlist defaults to `coding` while an operator who supplies their
            own gets whichever they listed first. Naming a specific alias here
            still wins; what it no longer does is force every custom allowlist
            to contain a particular key.
        max_sessions: Ceiling on *resident sandboxes* across the service, or
            `None` for no ceiling. Together with the largest per-runtime memory
            ceiling this is what bounds the worst case an operator has to size
            the host for, so `None` is for hosts where something else does the
            bounding. It does not bound how many sessions may exist — see
            `max_open_sessions`.
        max_open_sessions: Ceiling on sessions that *exist*, resident or
            hibernated, or `None` for no ceiling. A hibernated session costs
            disk and a few kilobytes of bookkeeping rather than memory, so this
            number is properly much larger than `max_sessions` — that gap is the
            difference between the sessions a host can carry and the ones it can
            run at once. Only meaningful with `evict_idle_after`, which is what
            demotes a session to hibernated in the first place.
        evict_idle_after: Seconds of inactivity after which a session at the
            ceiling may be hibernated to make room for a new one, instead of the
            new one being refused with `429`. This is what turns `max_sessions`
            from a hard cap on how many sessions may *exist* into a working-set
            size: the hibernated session keeps its token, its event log and its
            files, and its next request wakes it where it left off. `None`
            refuses instead of evicting. Requires `workspace_root` — hibernating
            a session whose files live only in its container would discard them
            silently, which is not a trade a config should make quietly.
        max_sessions_per_tenant: Ceiling on simultaneous sessions carrying one
            `tenant` label. Without it, one tenant of the calling application can
            occupy the whole pool and every other tenant gets `429`.
        mem_limit: Default memory ceiling per sandbox, in Docker syntax. A
            runtime naming its own overrides this, upwards or downwards.
        memswap_limit: Default ceiling on memory and swap combined. `None` pins
            it to `mem_limit`, so a sandbox at its ceiling is killed rather than
            allowed to swap — which is right when swap is a disk, because one
            swapping sandbox slows every other one on the host.

            Raise it only where swap is `zram`. There the pages stay in RAM
            compressed at roughly 3:1, so a sandbox briefly over its ceiling
            costs CPU instead of the whole host's responsiveness, and on a small
            box that trade is what turns an OOM kill into a slow command.
        cpus: Default hard CPU ceiling per sandbox, in cores. `None` leaves
            sandboxes free to use whatever is idle, bounded only by `cpu_shares`.
        cpu_shares: Default relative CPU weight, applied only under contention.
            Preferable to a hard `cpus` on a small host, where capping each
            sandbox to one core leaves the others unused while a single agent
            waits.
        tmpfs_size: Size of an in-memory `/tmp` given to every sandbox, in Docker
            syntax. Scratch writes then never reach the container's write layer,
            which is both faster and the difference between a busy sandbox
            growing on disk and not. `None` leaves `/tmp` on the overlay.

            Sized against `mem_limit` rather than on top of it, because tmpfs
            pages are charged to the container's own memory cgroup — so this is
            memory taken away from the workload, and worth keeping small when
            many sessions are open at once.
        pids_limit: Default process ceiling per sandbox.
        network_mode: Default Docker network mode. `"none"` because a service
            handing sandboxes to untrusted code should not give them the network
            unless that is a deliberate choice, per runtime or service-wide.
        oci_runtime: Default low-level runtime for every sandbox, overridable per
            runtime. `None` takes the daemon's default — which is the right
            default for us to ship, because naming a runtime the host has not
            registered turns every session into a `502`. Set it to `"runsc"` once
            gVisor is installed to put every sandbox behind a userspace syscall
            layer; `crun` belongs in the daemon's own config instead, since it is
            a faster `runc` rather than a different isolation model.
        work_dir: Working directory inside every sandbox. Applied to built
            runtimes as well, overriding whatever their `RuntimeConfig` says, so
            that one directory is where the workspace volume is mounted, what
            the archive endpoints read, and what a client's paths resolve
            against. Three places disagreeing about it is how files go missing.
        workspace_root: Host directory backing each sandbox's work directory, as
            `{workspace_root}/{session_id}/workspace`. Without it a sandbox's
            files live only in the container's write layer, so idle reaping
            destroys them — which a session spanning several runs will notice.
            Session ids are pattern-checked before they reach here, so one
            cannot traverse out of this root.
        prewarm: Pull and build the whole allowlist in the background at
            startup. Without it the first session on a built runtime pays for the
            image build — a measured ten seconds and upwards — and pays it in the
            middle of somebody's request. Only applies to the default Docker
            builder, since nothing else knows how to warm a custom one.
        persist_containers: Give each sandbox a stable container name, so a
            reaped session keeps its filesystem and the next attach restarts the
            same container. `workspace_root` alone preserves only the work
            directory — packages installed with `pip` or `apt` live in the
            container's write layer, and an agent that reinstalls them after
            every idle timeout is not working in "the same" machine. The cost is
            that stopped containers accumulate until a session is closed with
            `purge`, so this is off by default.
        idle_timeout: Seconds of inactivity before a session is reaped.
        cleanup_interval: Seconds between reaping passes.
        workspace_ttl: Seconds an unused workspace directory is kept before it is
            swept, timed from when its session was last opened. `None` — the
            default — keeps a session's **files** for ever, which is usually what
            an agent's user expects: the notes and scripts are the work. Set it
            only where a retention policy says otherwise. Only meaningful with
            `workspace_root`.
        container_ttl: Seconds a *stopped* persisted container is kept before it
            is removed, timed from when it stopped. This reclaims what a session
            installed inside its container — the build, the wheels, the
            `node_modules` — while leaving its workspace untouched, so the files
            survive and only the rebuildable part goes. `None` keeps them.
            Only meaningful with `persist_containers`.
        sandbox_uid: Run every built runtime's sandbox as this unprivileged
            user instead of root, and give each session's workspace to it.
            `None` — the default — keeps the current behaviour, where a sandbox
            runs as root.

            Worth turning on. A container escape starts from whoever the
            container runs as, and every file an agent writes into its
            bind-mounted workspace is owned by that user on the host — as root,
            a `sandboxd` running unprivileged cannot clean up after its own
            sessions. The image is built around the uid: a real account, a home
            directory, and a virtualenv it owns first on `PATH`, without which
            an agent's `pip install` fails and `uv` — having no `--user` mode —
            fails with no way forward.

            Two things it asks of the deployment. The service must be able to
            `chown` each session's workspace to this uid, which means either
            running with the privilege or running *as* that uid so the
            directories are created owned by it; a service that can do neither
            is told so when the session opens rather than handing an agent a
            workspace it cannot write to. And ready-made runtimes are left as
            root, since an image nobody built for this has no such user and no
            virtualenv, so an agent inside one could install nothing.
        max_read_bytes: Largest single file a client may read out of a sandbox.
        execute_timeout: Hard ceiling applied to every command, so one client
            cannot occupy a worker indefinitely.
        max_workers: Threads dedicated to blocking sandbox calls. Sandbox
            commands are long, so this pool is kept separate from asyncio's
            shared default one.
        ui_enabled: Serve the bundled dashboard at `/ui`. Off by default: the
            page asks a human for the service token, and that token can start
            containers on the host, so it belongs on localhost or a private
            network and never on a public listener. The HTTP API is unaffected
            either way.
    """

    token: str
    runtimes: Mapping[str, str | SandboxRuntime] = field(
        default_factory=lambda: dict(DEFAULT_RUNTIMES)
    )
    default_runtime: str = ""
    max_sessions: int | None = 20
    max_open_sessions: int | None = None
    max_sessions_per_tenant: int | None = None
    evict_idle_after: int | None = None
    mem_limit: str | None = "1g"
    memswap_limit: str | None = None
    cpus: float | None = 2.0
    cpu_shares: int | None = None
    pids_limit: int | None = 512
    tmpfs_size: str | None = "64m"
    network_mode: str | None = "none"
    oci_runtime: str | None = None
    work_dir: str = "/workspace"
    workspace_root: str | None = None
    persist_containers: bool = False
    prewarm: bool = True
    idle_timeout: int = 1800
    cleanup_interval: int = 300
    workspace_ttl: int | None = None
    container_ttl: int | None = None
    sandbox_uid: int | None = None
    max_read_bytes: int = 8 * 1024 * 1024
    execute_timeout: int = 300
    max_workers: int = 32
    ui_enabled: bool = False

    def __post_init__(self) -> None:
        if not self.token:
            raise ValueError("SandboxdConfig.token must not be empty")
        if not self.runtimes:
            raise ValueError("SandboxdConfig.runtimes must allow at least one image")
        if not self.default_runtime:
            self.default_runtime = next(iter(self.runtimes))
        if self.default_runtime not in self.runtimes:
            raise ValueError(
                f"default_runtime {self.default_runtime!r} is not in runtimes "
                f"({', '.join(sorted(self.runtimes))})"
            )
        if self.evict_idle_after is not None and self.workspace_root is None:
            raise ValueError(
                "evict_idle_after needs workspace_root: hibernating a session whose "
                "files live only in its container would discard them silently"
            )
        if (
            self.max_open_sessions is not None
            and self.max_sessions is not None
            and self.max_open_sessions < self.max_sessions
        ):
            raise ValueError(
                f"max_open_sessions ({self.max_open_sessions}) is below max_sessions "
                f"({self.max_sessions}): a session has to exist to be resident"
            )

    def resolve_runtime(self, alias: str | None) -> tuple[str, SandboxRuntime]:
        """Resolve a client-supplied alias to an allowed runtime.

        Args:
            alias: Runtime alias from the request, or `None` for the default.

        Returns:
            The `(alias, runtime)` pair to use.

        Raises:
            KeyError: If the alias is not on the allowlist.
        """
        chosen = alias or self.default_runtime
        return chosen, _as_runtime(self.runtimes[chosen])

    def limits_for(self, runtime: SandboxRuntime) -> dict[str, Any]:
        """The ceilings one runtime actually runs under.

        Its own values where it names them, the service defaults otherwise.
        """
        return {
            "mem_limit": runtime.mem_limit if runtime.mem_limit is not None else self.mem_limit,
            "memswap_limit": (
                runtime.memswap_limit if runtime.memswap_limit is not None else self.memswap_limit
            ),
            "cpus": runtime.cpus if runtime.cpus is not None else self.cpus,
            "cpu_shares": (
                runtime.cpu_shares if runtime.cpu_shares is not None else self.cpu_shares
            ),
            "pids_limit": (
                runtime.pids_limit if runtime.pids_limit is not None else self.pids_limit
            ),
            "network_mode": (
                runtime.network_mode if runtime.network_mode is not None else self.network_mode
            ),
            "oci_runtime": (
                runtime.oci_runtime if runtime.oci_runtime is not None else self.oci_runtime
            ),
        }

resolve_runtime(alias)

Resolve a client-supplied alias to an allowed runtime.

Parameters:

Name	Type	Description	Default
alias	str | None	Runtime alias from the request, or None for the default.	required

Returns:

Type	Description
tuple[str, SandboxRuntime]	The (alias, runtime) pair to use.

Raises:

Type	Description
KeyError	If the alias is not on the allowlist.

Source code in src/pydantic_ai_backends/remote/server.py

def resolve_runtime(self, alias: str | None) -> tuple[str, SandboxRuntime]:
    """Resolve a client-supplied alias to an allowed runtime.

    Args:
        alias: Runtime alias from the request, or `None` for the default.

    Returns:
        The `(alias, runtime)` pair to use.

    Raises:
        KeyError: If the alias is not on the allowlist.
    """
    chosen = alias or self.default_runtime
    return chosen, _as_runtime(self.runtimes[chosen])

limits_for(runtime)

The ceilings one runtime actually runs under.

Its own values where it names them, the service defaults otherwise.

Source code in src/pydantic_ai_backends/remote/server.py

def limits_for(self, runtime: SandboxRuntime) -> dict[str, Any]:
    """The ceilings one runtime actually runs under.

    Its own values where it names them, the service defaults otherwise.
    """
    return {
        "mem_limit": runtime.mem_limit if runtime.mem_limit is not None else self.mem_limit,
        "memswap_limit": (
            runtime.memswap_limit if runtime.memswap_limit is not None else self.memswap_limit
        ),
        "cpus": runtime.cpus if runtime.cpus is not None else self.cpus,
        "cpu_shares": (
            runtime.cpu_shares if runtime.cpu_shares is not None else self.cpu_shares
        ),
        "pids_limit": (
            runtime.pids_limit if runtime.pids_limit is not None else self.pids_limit
        ),
        "network_mode": (
            runtime.network_mode if runtime.network_mode is not None else self.network_mode
        ),
        "oci_runtime": (
            runtime.oci_runtime if runtime.oci_runtime is not None else self.oci_runtime
        ),
    }

SandboxRuntime

One entry in the service's allowlist: what an alias runs, and under which ceilings.

pydantic_ai_backends.remote.server.SandboxRuntime dataclass

One entry in the service's runtime allowlist.

A client names the alias this is registered under and nothing else, so everything here is the operator's decision. Give it either a ready-made image, or a runtime whose packages are built into an image on first use.

The ceilings are per runtime, because they are not one number for a whole service: a notebook-style data runtime needs several gigabytes where a plain shell needs a few hundred megabytes, and forcing one value on both either starves the first or over-commits the host for the second. A ceiling left None takes the service-wide default from :class:SandboxdConfig, which is a default rather than a maximum — a runtime may name more.

Attributes:

Name	Type	Description
image	str | None	Ready-made image, started as-is.
runtime	RuntimeConfig | str | None	RuntimeConfig (or the name of a built-in one) whose packages are installed into an image the first time this runtime is used. Later sessions hit the build cache.
description	str	What this environment is for, shown in the dashboard.
mem_limit	str | None	Memory ceiling in Docker syntax, e.g. "2g".
memswap_limit	str | None	Ceiling on memory and swap combined. None pins it to mem_limit, denying the sandbox swap. Only worth raising on a host whose swap is zram, where the pages stay in compressed RAM and the alternative to a little swapping is an OOM kill.
cpus	float | None	Hard CPU ceiling in cores. A sandbox never exceeds it — and so cannot use cores that are idle.
cpu_shares	int | None	Relative CPU weight, which applies only under contention. On a small host this is usually the better knob: one active sandbox may take the whole machine, and several are still divided fairly.
pids_limit	int | None	Process ceiling.
network_mode	str | None	Docker network mode. Naming anything but "none" here is how one runtime is allowed the network while others are not.
oci_runtime	str | None	Low-level runtime the daemon starts this runtime's containers with — "runsc" for gVisor, "kata" for a microVM per container. Unlike the ceilings above this changes the isolation boundary, and it is per runtime for the same reason they are: the runtime that installs arbitrary packages off the network is the one worth paying gVisor's I/O overhead for, while a plain shell is not. Must be registered with the daemon; None takes its default.

Source code in src/pydantic_ai_backends/remote/server.py

@dataclass(frozen=True, slots=True)
class SandboxRuntime:
    """One entry in the service's runtime allowlist.

    A client names the *alias* this is registered under and nothing else, so
    everything here is the operator's decision. Give it either a ready-made
    `image`, or a `runtime` whose packages are built into an image on first use.

    The ceilings are per runtime, because they are not one number for a whole
    service: a notebook-style data runtime needs several gigabytes where a plain
    shell needs a few hundred megabytes, and forcing one value on both either
    starves the first or over-commits the host for the second. A ceiling left
    `None` takes the service-wide default from :class:`SandboxdConfig`, which is
    a *default* rather than a maximum — a runtime may name more.

    Attributes:
        image: Ready-made image, started as-is.
        runtime: `RuntimeConfig` (or the name of a built-in one) whose packages
            are installed into an image the first time this runtime is used.
            Later sessions hit the build cache.
        description: What this environment is for, shown in the dashboard.
        mem_limit: Memory ceiling in Docker syntax, e.g. `"2g"`.
        memswap_limit: Ceiling on memory and swap combined. `None` pins it to
            `mem_limit`, denying the sandbox swap. Only worth raising on a host
            whose swap is `zram`, where the pages stay in compressed RAM and the
            alternative to a little swapping is an OOM kill.
        cpus: Hard CPU ceiling in cores. A sandbox never exceeds it — and so
            cannot use cores that are idle.
        cpu_shares: Relative CPU weight, which applies only under contention. On
            a small host this is usually the better knob: one active sandbox may
            take the whole machine, and several are still divided fairly.
        pids_limit: Process ceiling.
        network_mode: Docker network mode. Naming anything but `"none"` here is
            how one runtime is allowed the network while others are not.
        oci_runtime: Low-level runtime the daemon starts this runtime's
            containers with — `"runsc"` for gVisor, `"kata"` for a microVM per
            container. Unlike the ceilings above this changes the *isolation
            boundary*, and it is per runtime for the same reason they are: the
            runtime that installs arbitrary packages off the network is the one
            worth paying gVisor's I/O overhead for, while a plain shell is not.
            Must be registered with the daemon; `None` takes its default.
    """

    image: str | None = None
    runtime: RuntimeConfig | str | None = None
    description: str = ""
    mem_limit: str | None = None
    memswap_limit: str | None = None
    cpus: float | None = None
    cpu_shares: int | None = None
    pids_limit: int | None = None
    network_mode: str | None = None
    oci_runtime: str | None = None

    def __post_init__(self) -> None:
        if (self.image is None) == (self.runtime is None):
            raise ValueError("A SandboxRuntime needs exactly one of image or runtime")

    @property
    def builds(self) -> bool:
        """Whether first use builds an image rather than pulling a ready one."""
        return self.runtime is not None

    def resolved_runtime(self) -> RuntimeConfig | None:
        """The `RuntimeConfig` this entry builds, looked up when named."""
        if self.runtime is None:
            return None
        if isinstance(self.runtime, str):
            return get_runtime(self.runtime)
        return self.runtime

    def describes(self) -> str:
        """Human-readable summary of what will run, for the policy view."""
        if self.description:
            return self.description
        built = self.resolved_runtime()
        if built is not None:
            return built.description or f"built from {built.base_image or built.image}"
        return self.image or ""

    def image_label(self) -> str:
        """The image, or what it is built from when there is not one yet."""
        if self.image is not None:
            return self.image
        built = self.resolved_runtime()
        assert built is not None
        return built.image or f"{built.base_image} + {len(built.packages)} package(s)"

builds property

Whether first use builds an image rather than pulling a ready one.

resolved_runtime()

The RuntimeConfig this entry builds, looked up when named.

Source code in src/pydantic_ai_backends/remote/server.py

def resolved_runtime(self) -> RuntimeConfig | None:
    """The `RuntimeConfig` this entry builds, looked up when named."""
    if self.runtime is None:
        return None
    if isinstance(self.runtime, str):
        return get_runtime(self.runtime)
    return self.runtime

describes()

Human-readable summary of what will run, for the policy view.

Source code in src/pydantic_ai_backends/remote/server.py

def describes(self) -> str:
    """Human-readable summary of what will run, for the policy view."""
    if self.description:
        return self.description
    built = self.resolved_runtime()
    if built is not None:
        return built.description or f"built from {built.base_image or built.image}"
    return self.image or ""

image_label()

The image, or what it is built from when there is not one yet.

Source code in src/pydantic_ai_backends/remote/server.py

def image_label(self) -> str:
    """The image, or what it is built from when there is not one yet."""
    if self.image is not None:
        return self.image
    built = self.resolved_runtime()
    assert built is not None
    return built.image or f"{built.base_image} + {len(built.packages)} package(s)"

create_app

pydantic_ai_backends.remote.server.create_app(config, *, sandbox_builder=None)

Build the sandbox service.

Parameters:

Name	Type	Description	Default
config	SandboxdConfig	Service policy. See :class:SandboxdConfig.	required
sandbox_builder	SandboxBuilder | None	Override for how a sandbox is constructed, receiving (session_id, image). Defaults to :class:DockerSandbox configured from config. Supply one to embed a different sandbox type, or to test without a Docker daemon.	None

Returns:

Type	Description
FastAPI	A configured FastAPI application.

Source code in src/pydantic_ai_backends/remote/server.py

def create_app(
    config: SandboxdConfig,
    *,
    sandbox_builder: SandboxBuilder | None = None,
) -> FastAPI:
    """Build the sandbox service.

    Args:
        config: Service policy. See :class:`SandboxdConfig`.
        sandbox_builder: Override for how a sandbox is constructed, receiving
            `(session_id, image)`. Defaults to :class:`DockerSandbox` configured
            from `config`. Supply one to embed a different sandbox type, or to
            test without a Docker daemon.

    Returns:
        A configured `FastAPI` application.
    """
    # Prewarming and the container sweep only make sense for the builder that
    # knows what an image is; an injected builder may have nothing to pull and no
    # daemon to ask.
    service = _Service(
        config,
        sandbox_builder or _default_builder(config),
        prewarm=None if sandbox_builder else _default_prewarm(config),
        docker_client=None if sandbox_builder else _default_docker_client,
    )

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        service.startup()
        try:
            yield
        finally:
            await service.shutdown()

    app = FastAPI(
        title="sandboxd",
        summary="Sandbox execution service for AI agents",
        lifespan=lifespan,
    )
    app.state.service = service
    _register_session_routes(app, service)
    _register_workspace_routes(app, service)
    _register_operation_routes(app, service)
    return app

Wire protocol

The HTTP contract as Pydantic models — one source of truth for both sides.

Wire protocol shared by the remote sandbox client and server.

These models are the single source of truth for the HTTP contract. The operation endpoints (/exec, /read, /write, /ls, /glob) keep the field names that :class:~pydantic_ai_backends.backends.kubernetes.KubernetesPodSandbox already sends in mode="http", so one server can serve both clients; /edit, /grep, /exists and /read_bytes are additions.

Binary-capable payloads travel base64-encoded (content_b64) because JSON cannot carry arbitrary bytes, and a sandbox holds real files.

CreateSessionRequest

Bases: BaseModel

Open a sandbox session.

Deliberately carries no container settings. Image, mounts, network mode and every resource ceiling come from the server's configuration, because a process holding the Docker socket can start a privileged container that mounts the host — a client that could name its own image or volumes would own the machine.

Source code in src/pydantic_ai_backends/remote/wire.py

class CreateSessionRequest(BaseModel):
    """Open a sandbox session.

    Deliberately carries no container settings. Image, mounts, network mode and
    every resource ceiling come from the server's configuration, because a
    process holding the Docker socket can start a privileged container that
    mounts the host — a client that could name its own image or volumes would
    own the machine.
    """

    session_id: str | None = Field(default=None, pattern=SESSION_ID_PATTERN)
    runtime: str | None = None
    """Alias of a server-allowed runtime. `None` selects the server default."""

    tenant: str | None = Field(default=None, pattern=TENANT_PATTERN)
    """Who this session is opened on behalf of, for the per-tenant ceiling.

    Declared by the client rather than derived from `session_id`, so the service
    imposes no naming convention on ids. Only a caller holding the service token
    can open a session at all, so this is a capacity label from a trusted
    application — not an authorization claim, and it grants nothing.
    """

    reuse: bool = False
    """Attach to the session under `session_id` when one is already open.

    Off by default, so an id collision is reported rather than silently sharing
    somebody else's sandbox. Turn it on for a session that spans several runs —
    one conversation, say — where the second run has to reach the files the
    first one wrote. The runtime of an attached session is whatever it was
    opened with; a `runtime` that disagrees is rejected rather than ignored.
    """

runtime = None class-attribute instance-attribute

Alias of a server-allowed runtime. None selects the server default.

tenant = Field(default=None, pattern=TENANT_PATTERN) class-attribute instance-attribute

Who this session is opened on behalf of, for the per-tenant ceiling.

Declared by the client rather than derived from session_id, so the service imposes no naming convention on ids. Only a caller holding the service token can open a session at all, so this is a capacity label from a trusted application — not an authorization claim, and it grants nothing.

reuse = False class-attribute instance-attribute

Attach to the session under session_id when one is already open.

Off by default, so an id collision is reported rather than silently sharing somebody else's sandbox. Turn it on for a session that spans several runs — one conversation, say — where the second run has to reach the files the first one wrote. The runtime of an attached session is whatever it was opened with; a runtime that disagrees is rejected rather than ignored.

SessionCreated

Bases: BaseModel

A freshly opened session and the token scoped to it.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionCreated(BaseModel):
    """A freshly opened session and the token scoped to it."""

    session: SessionInfo
    token: str
    """Grants access to this session only. The service token also works."""

token instance-attribute

Grants access to this session only. The service token also works.

SessionInfo

Bases: BaseModel

Observable state of one session.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionInfo(BaseModel):
    """Observable state of one session."""

    session_id: str
    runtime: str
    tenant: str | None = None
    """Whoever the session was opened for, when the client said."""
    alive: bool
    state: Literal["running", "hibernated"] = "running"
    """Whether the session holds a sandbox, or was stopped to free a slot and is
    waiting for its next request to wake it. A hibernated session is never
    `alive`; it still has its token, its event log and its files."""
    created_at: float
    last_activity: float
    idle_seconds: float
    usage: SessionUsage | None = None

tenant = None class-attribute instance-attribute

Whoever the session was opened for, when the client said.

state = 'running' class-attribute instance-attribute

Whether the session holds a sandbox, or was stopped to free a slot and is waiting for its next request to wake it. A hibernated session is never alive; it still has its token, its event log and its files.

SessionList

Bases: BaseModel

Every open session, for monitoring.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionList(BaseModel):
    """Every open session, for monitoring."""

    sessions: list[SessionInfo] = Field(default_factory=list)
    limit: int | None = None
    """Configured `max_sessions` — the ceiling on *resident* sandboxes."""
    open_limit: int | None = None
    """Configured `max_open_sessions`, or `None` when uncapped."""
    tenant_limit: int | None = None
    """Configured `max_sessions_per_tenant`, or `None` when uncapped."""

limit = None class-attribute instance-attribute

Configured max_sessions — the ceiling on resident sandboxes.

open_limit = None class-attribute instance-attribute

Configured max_open_sessions, or None when uncapped.

tenant_limit = None class-attribute instance-attribute

Configured max_sessions_per_tenant, or None when uncapped.

SessionUsage

Bases: BaseModel

Point-in-time resource usage for one sandbox.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionUsage(BaseModel):
    """Point-in-time resource usage for one sandbox."""

    memory_bytes: int | None = None
    memory_limit_bytes: int | None = None
    cpu_percent: float | None = None
    pids: int | None = None

SessionEvent

Bases: BaseModel

One operation performed against a session.

Records what was asked for and how it went — never file contents or command output, which would turn an audit trail into a data leak and grow without bound.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionEvent(BaseModel):
    """One operation performed against a session.

    Records what was asked for and how it went — never file contents or command
    output, which would turn an audit trail into a data leak and grow without
    bound.
    """

    seq: int
    """Monotonic per-session sequence number, for incremental polling."""
    at: float
    op: str
    """Operation name: `exec`, `read`, `write`, `edit`, `ls`, `glob`, `grep`, `exists`."""
    target: str
    """Command or path the operation addressed, truncated."""
    ok: bool
    detail: str = ""
    """Short outcome summary, e.g. `exit 0` or `14 entries`."""
    duration_ms: float = 0.0

seq instance-attribute

Monotonic per-session sequence number, for incremental polling.

op instance-attribute

Operation name: exec, read, write, edit, ls, glob, grep, exists.

target instance-attribute

Command or path the operation addressed, truncated.

detail = '' class-attribute instance-attribute

Short outcome summary, e.g. exit 0 or 14 entries.

SessionEvents

Bases: BaseModel

A slice of a session's activity log.

Source code in src/pydantic_ai_backends/remote/wire.py

class SessionEvents(BaseModel):
    """A slice of a session's activity log."""

    events: list[SessionEvent] = Field(default_factory=list)
    latest_seq: int = 0
    """Highest sequence number the service holds; pass it back as `after`."""

latest_seq = 0 class-attribute instance-attribute

Highest sequence number the service holds; pass it back as after.

ServiceHealth

Bases: BaseModel

Liveness and capacity summary. Unauthenticated.

Source code in src/pydantic_ai_backends/remote/wire.py

class ServiceHealth(BaseModel):
    """Liveness and capacity summary. Unauthenticated."""

    status: str = "ok"
    sessions: int = 0
    """Sessions holding a sandbox right now."""
    limit: int | None = None
    open_sessions: int = 0
    """Sessions that exist, resident and hibernated together."""
    open_limit: int | None = None
    runtimes: list[str] = Field(default_factory=list)

sessions = 0 class-attribute instance-attribute

Sessions holding a sandbox right now.

open_sessions = 0 class-attribute instance-attribute

Sessions that exist, resident and hibernated together.

ServicePolicy

Bases: BaseModel

The ceilings and allowlist the service imposes on every sandbox.

Authenticated, because it describes the host's configuration — though it holds nothing a caller with the service token is not already subject to. Exists so an operator can see the limits actually in force rather than inferring them from a config file.

Source code in src/pydantic_ai_backends/remote/wire.py

class ServicePolicy(BaseModel):
    """The ceilings and allowlist the service imposes on every sandbox.

    Authenticated, because it describes the host's configuration — though it
    holds nothing a caller with the service token is not already subject to.
    Exists so an operator can see the limits actually in force rather than
    inferring them from a config file.
    """

    runtimes: list[RuntimePolicy] = Field(default_factory=list)
    """Every allowed runtime, with the ceilings it actually runs under."""
    default_runtime: str = ""
    max_sessions: int | None = None
    """Ceiling on resident sandboxes."""
    max_open_sessions: int | None = None
    """Ceiling on sessions that exist, resident or hibernated."""
    max_sessions_per_tenant: int | None = None
    evict_idle_after: int | None = None
    """Seconds after which an idle session may be hibernated to free a slot."""
    mem_limit: str | None = None
    memswap_limit: str | None = None
    """Default memory-plus-swap ceiling, or `None` when swap is pinned to memory."""
    cpus: float | None = None
    cpu_shares: int | None = None
    pids_limit: int | None = None
    network_mode: str | None = None
    oci_runtime: str | None = None
    """Default low-level runtime, or `None` for whatever the daemon defaults to."""

    work_dir: str = ""
    idle_timeout: int = 0
    execute_timeout: int = 0
    max_read_bytes: int = 0
    persist_containers: bool = False
    """Whether a stopped sandbox keeps its filesystem for the next attach."""
    sandbox_uid: int | None = None
    """Uid built runtimes run as, or `None` when sandboxes run as root."""
    workspace_ttl: int | None = None
    """Seconds an unused workspace is kept before it is swept, or `None`."""
    container_ttl: int | None = None
    """Seconds a stopped container is kept before its build is reclaimed."""
    tmpfs_size: str | None = None
    """Size of the in-memory `/tmp` each sandbox gets, or `None` when it has none."""
    prewarm: bool = False
    """Whether the allowlist is pulled and built at startup rather than on demand."""
    buildkit: bool = False
    """Whether image builds use BuildKit, and so keep package caches between them."""

runtimes = Field(default_factory=list) class-attribute instance-attribute

Every allowed runtime, with the ceilings it actually runs under.

max_sessions = None class-attribute instance-attribute

Ceiling on resident sandboxes.

max_open_sessions = None class-attribute instance-attribute

Ceiling on sessions that exist, resident or hibernated.

evict_idle_after = None class-attribute instance-attribute

Seconds after which an idle session may be hibernated to free a slot.

memswap_limit = None class-attribute instance-attribute

Default memory-plus-swap ceiling, or None when swap is pinned to memory.

oci_runtime = None class-attribute instance-attribute

Default low-level runtime, or None for whatever the daemon defaults to.

persist_containers = False class-attribute instance-attribute

Whether a stopped sandbox keeps its filesystem for the next attach.

sandbox_uid = None class-attribute instance-attribute

Uid built runtimes run as, or None when sandboxes run as root.

workspace_ttl = None class-attribute instance-attribute

Seconds an unused workspace is kept before it is swept, or None.

container_ttl = None class-attribute instance-attribute

Seconds a stopped container is kept before its build is reclaimed.

tmpfs_size = None class-attribute instance-attribute

Size of the in-memory /tmp each sandbox gets, or None when it has none.

prewarm = False class-attribute instance-attribute

Whether the allowlist is pulled and built at startup rather than on demand.

buildkit = False class-attribute instance-attribute

Whether image builds use BuildKit, and so keep package caches between them.

ServiceIndex

Bases: BaseModel

What this service is and where to find its endpoints.

Served at / so that opening the base URL says something useful instead of a bare 404. Unauthenticated, and deliberately free of any session detail.

Source code in src/pydantic_ai_backends/remote/wire.py

class ServiceIndex(BaseModel):
    """What this service is and where to find its endpoints.

    Served at `/` so that opening the base URL says something useful instead of
    a bare 404. Unauthenticated, and deliberately free of any session detail.
    """

    service: str = "sandboxd"
    health: ServiceHealth
    docs_url: str = "/docs"
    openapi_url: str = "/openapi.json"
    ui_url: str | None = None
    """Where the dashboard lives, or `None` when it is not enabled."""
    endpoints: list[str] = Field(default_factory=list)
    """Routed paths, derived from the app so the list cannot go stale."""

ui_url = None class-attribute instance-attribute

Where the dashboard lives, or None when it is not enabled.

endpoints = Field(default_factory=list) class-attribute instance-attribute

Routed paths, derived from the app so the list cannot go stale.