Workspaces API

Pydantic AI workspace capabilities backed by this library's sandboxes. See Workspaces for the concepts. Needs the workspaces extra, plus the provider's own (docker, kubernetes, daytona).

DockerWorkspace

pydantic_ai_backends.workspaces.DockerWorkspace dataclass

Bases: AbstractCapability[object]

Supply a Docker container on this host as the run's workspace.

A run with no ref creates a container; one carrying a "docker" ref attaches to that container, starting it if it was stopped. Containers are kept after the run — the ref is how to come back to them, and :meth:destroy is how to remove one.

This supplies the environment only. Compose it with something that uses the workspace: Coder, Shell or FileSystem from the Pydantic AI harness, or this library's ConsoleCapability.

A container is only as isolated as its runtime: Docker's default runc shares the host kernel. See oci_runtime.

Example

Python

from pydantic_ai import Agent
from pydantic_ai_harness.coder import Coder

from pydantic_ai_backends.workspaces import DockerWorkspace

agent = Agent("anthropic:claude-opus-5-5", capabilities=[DockerWorkspace(), Coder()])

Source code in src/pydantic_ai_backends/workspaces/_docker.py

@dataclass(kw_only=True)
class DockerWorkspace(AbstractCapability[object]):
    """Supply a Docker container on this host as the run's workspace.

    A run with no ref creates a container; one carrying a `"docker"` ref attaches
    to that container, starting it if it was stopped. Containers are kept after
    the run — the ref is how to come back to them, and :meth:`destroy` is how to
    remove one.

    This supplies the environment only. Compose it with something that uses the
    workspace: `Coder`, `Shell` or `FileSystem` from the Pydantic AI harness, or
    this library's `ConsoleCapability`.

    A container is only as isolated as its runtime: Docker's default `runc`
    shares the host kernel. See `oci_runtime`.

    Example:
        ```python
        from pydantic_ai import Agent
        from pydantic_ai_harness.coder import Coder

        from pydantic_ai_backends.workspaces import DockerWorkspace

        agent = Agent("anthropic:claude-opus-5-5", capabilities=[DockerWorkspace(), Coder()])
        ```
    """

    image: str = "python:3.12-slim"
    """Image for a new container. Ignored when `runtime` is given."""

    runtime: RuntimeConfig | str | None = None
    """A `RuntimeConfig` or the name of a built-in runtime, which also sets the work directory."""

    work_dir: str = "/workspace"
    """Directory commands start in. Ignored when `runtime` is given."""

    network_mode: str | None = None
    """Docker network mode; `"none"` keeps the container off the network."""

    mem_limit: str | None = None
    """Memory ceiling in Docker syntax, such as `"512m"`."""

    cpus: float | None = None
    """Hard CPU ceiling in cores."""

    oci_runtime: str | None = None
    """Low-level runtime, Docker's `--runtime`: `"runsc"` (gVisor) for a stronger boundary."""

    env: Mapping[str, str] | None = field(default=None, repr=False)
    """Variables every command gets. Nothing is read from the host's environment."""

    volumes: Mapping[str, str] | None = None
    """Host directories mounted into the container, as `{"/host/path": "/container/path"}`.

    Mounting a project at `work_dir` lets the agent work on its files in place;
    the container then reaches exactly those host files.
    """

    container_name: str | None = None
    """One container for every run, named by you: created on first use, attached after.

    Without it each run without a ref gets a new container and only the ref
    leads back to it. With it a later process finds the same container -
    installed packages included - by name. A ref naming another container is
    left to another capability.
    """

    def __post_init__(self) -> None:
        if self.defer_loading:
            raise UserError(
                "`DockerWorkspace` does not support `defer_loading=True`: "
                "the workspace is selected before deferred capabilities load."
            )

    def _sandbox(self, name: str) -> DockerSandbox:
        return DockerSandbox(
            image=self.image,
            runtime=self.runtime,
            work_dir=self.work_dir,
            container_name=name,
            network_mode=self.network_mode,
            mem_limit=self.mem_limit,
            cpus=self.cpus,
            oci_runtime=self.oci_runtime,
            volumes=dict(self.volumes) if self.volumes else None,
        )

    def backend(self, ref: WorkspaceRef | None = None) -> DockerWorkspaceBackend:
        """A backend for `ref`, or for a new container; no I/O until its first operation."""
        return DockerWorkspaceBackend(
            sandbox_factory=self._sandbox,
            ref=ref,
            env=self.env,
            container_name=self.container_name,
        )

    def get_workspace(
        self, ctx: RunContext[object], *, ref: WorkspaceRef | None
    ) -> WorkspaceBackend | None:
        """This run's backend, or `None` for a ref another provider owns."""
        del ctx
        if ref is not None and ref.provider != DOCKER_PROVIDER:
            return None
        if ref is not None and self.container_name is not None and ref.id != self.container_name:
            return None
        return self.backend(ref)

    async def destroy(self, ref: WorkspaceRef) -> None:
        """Remove the container `ref` names, files and all. Already gone is fine.

        Raises:
            ValueError: `ref` belongs to another provider, or names a container
                no `DockerWorkspace` created.
        """
        if ref.provider != DOCKER_PROVIDER:
            raise ValueError(f"expected a {DOCKER_PROVIDER!r} workspace ref, got {ref.provider!r}")
        if ref.id != self.container_name and not _created_here(ref.id):
            raise ValueError(f"container {ref.id!r} was not created by a DockerWorkspace")
        await anyio.to_thread.run_sync(_remove_container, ref.id)

backend(ref=None)

A backend for ref, or for a new container; no I/O until its first operation.

Source code in src/pydantic_ai_backends/workspaces/_docker.py

def backend(self, ref: WorkspaceRef | None = None) -> DockerWorkspaceBackend:
    """A backend for `ref`, or for a new container; no I/O until its first operation."""
    return DockerWorkspaceBackend(
        sandbox_factory=self._sandbox,
        ref=ref,
        env=self.env,
        container_name=self.container_name,
    )

get_workspace(ctx, *, ref)

This run's backend, or None for a ref another provider owns.

Source code in src/pydantic_ai_backends/workspaces/_docker.py

def get_workspace(
    self, ctx: RunContext[object], *, ref: WorkspaceRef | None
) -> WorkspaceBackend | None:
    """This run's backend, or `None` for a ref another provider owns."""
    del ctx
    if ref is not None and ref.provider != DOCKER_PROVIDER:
        return None
    if ref is not None and self.container_name is not None and ref.id != self.container_name:
        return None
    return self.backend(ref)

destroy(ref) async

Remove the container ref names, files and all. Already gone is fine.

Raises:

Type	Description
ValueError	ref belongs to another provider, or names a container no DockerWorkspace created.

Source code in src/pydantic_ai_backends/workspaces/_docker.py

async def destroy(self, ref: WorkspaceRef) -> None:
    """Remove the container `ref` names, files and all. Already gone is fine.

    Raises:
        ValueError: `ref` belongs to another provider, or names a container
            no `DockerWorkspace` created.
    """
    if ref.provider != DOCKER_PROVIDER:
        raise ValueError(f"expected a {DOCKER_PROVIDER!r} workspace ref, got {ref.provider!r}")
    if ref.id != self.container_name and not _created_here(ref.id):
        raise ValueError(f"container {ref.id!r} was not created by a DockerWorkspace")
    await anyio.to_thread.run_sync(_remove_container, ref.id)

pydantic_ai_backends.workspaces.DockerWorkspaceBackend

Bases: ContainerWorkspaceBackend

One Docker container, as the environment an agent run works in.

The container is named after the ref and never auto-removed, so a later run can attach to it and find its files, installed packages included; a stopped one is started again. It lives until :meth:DockerWorkspace.destroy. A ref attaches only to a container named the way this class names the ones it creates, never to any other container on the host.

Parameters:

Name	Type	Description	Default
sandbox_factory	SandboxFactory	Builds the DockerSandbox for a container name. Holds the image, runtime and limits, and must not start the container.	required
ref	WorkspaceRef | None	The workspace to attach to; None creates one on first use.	None
env	Mapping[str, str] | None	Variables every command gets, under any a call passes.	None
container_name	str | None	A container chosen by whoever configures the workspace rather than by this class: created under that name on first use when it does not exist, attached when it does. A ref naming it must still find it there, and no ref reaches any other container.	None

Source code in src/pydantic_ai_backends/workspaces/_docker.py

class DockerWorkspaceBackend(ContainerWorkspaceBackend):
    """One Docker container, as the environment an agent run works in.

    The container is named after the ref and never auto-removed, so a later run
    can attach to it and find its files, installed packages included; a stopped
    one is started again. It lives until :meth:`DockerWorkspace.destroy`. A ref
    attaches only to a container named the way this class names the ones it
    creates, never to any other container on the host.

    Args:
        sandbox_factory: Builds the `DockerSandbox` for a container name. Holds
            the image, runtime and limits, and must not start the container.
        ref: The workspace to attach to; `None` creates one on first use.
        env: Variables every command gets, under any a call passes.
        container_name: A container chosen by whoever configures the workspace
            rather than by this class: created under that name on first use
            when it does not exist, attached when it does. A ref naming it must
            still find it there, and no ref reaches any other container.
    """

    def __init__(
        self,
        *,
        sandbox_factory: SandboxFactory,
        ref: WorkspaceRef | None = None,
        env: Mapping[str, str] | None = None,
        container_name: str | None = None,
    ) -> None:
        async def open_container(name: str | None) -> tuple[str, RunnerSandbox]:
            if container_name is not None and name is None:
                # First use of a configured container: whatever state it is in,
                # it is the one asked for, and starting the sandbox creates it.
                name = container_name
            elif name is None:
                name = f"{CONTAINER_PREFIX}{uuid.uuid4().hex[:16]}"
            elif name != container_name and not _created_here(name):
                raise SandboxUnavailableError(
                    f"container {name!r} was not created by a DockerWorkspace"
                )
            else:
                status = await anyio.to_thread.run_sync(_container_status, name)
                if status != "running" and status not in REATTACHABLE_STATUSES:
                    raise SandboxUnavailableError(
                        f"container {name!r} no longer exists"
                        if status is None
                        else f"container {name!r} is {status}"
                    )
            sandbox = sandbox_factory(name)
            await anyio.to_thread.run_sync(sandbox.start)
            return name, sandbox

        super().__init__(provider=DOCKER_PROVIDER, opener=open_container, ref=ref, env=env)

SandboxdWorkspace

pydantic_ai_backends.workspaces.SandboxdWorkspace dataclass

Bases: AbstractCapability[object]

Supply a sandboxd session as the run's workspace.

A run with no ref opens a session; one carrying this capability's provider attaches to that session, or to the workspace the service kept after it was reaped. Sessions are kept after the run — the ref is how to come back, and :meth:destroy is how to remove one with its files.

This supplies the environment only. Compose it with something that uses the workspace: Coder, Shell or FileSystem from the Pydantic AI harness, or this library's ConsoleCapability.

Example

Python

from pydantic_ai import Agent
from pydantic_ai_harness.coder import Coder

from pydantic_ai_backends.workspaces import SandboxdWorkspace

sandbox = SandboxdWorkspace(service_url="http://sandboxd:8080", token="...")
agent = Agent("anthropic:claude-opus-5-5", capabilities=[sandbox, Coder()])

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

@dataclass(kw_only=True)
class SandboxdWorkspace(AbstractCapability[object]):
    """Supply a `sandboxd` session as the run's workspace.

    A run with no ref opens a session; one carrying this capability's provider
    attaches to that session, or to the workspace the service kept after it was
    reaped. Sessions are kept after the run — the ref is how to come back, and
    :meth:`destroy` is how to remove one with its files.

    This supplies the environment only. Compose it with something that uses the
    workspace: `Coder`, `Shell` or `FileSystem` from the Pydantic AI harness, or
    this library's `ConsoleCapability`.

    Example:
        ```python
        from pydantic_ai import Agent
        from pydantic_ai_harness.coder import Coder

        from pydantic_ai_backends.workspaces import SandboxdWorkspace

        sandbox = SandboxdWorkspace(service_url="http://sandboxd:8080", token="...")
        agent = Agent("anthropic:claude-opus-5-5", capabilities=[sandbox, Coder()])
        ```
    """

    service_url: str
    """Base URL of the service."""

    token: str = field(repr=False)
    """The service token: it can open sessions on the host, so keep it out of logs."""

    provider: str = SANDBOXD_PROVIDER
    """Provider name in refs; distinct per service when an agent can reach several."""

    runtime: str | None = None
    """Runtime alias for new sessions; the service default when `None`."""

    tenant: str | None = None
    """Who sessions are opened for, against the service's per-tenant ceiling."""

    env: Mapping[str, str] | None = field(default=None, repr=False)
    """Variables every command gets. Nothing is read from the host's environment."""

    client: httpx.AsyncClient | None = field(default=None, repr=False, compare=False)
    """An `httpx.AsyncClient` to share across runs, owned and closed by the caller."""

    session_name: str | None = None
    """One session for every run, named by you: opened on first use, attached after.

    Without it each run without a ref gets a new session and only the ref leads
    back to it. With it a later process finds the same session - or the files the
    service kept after reaping it - by name. A ref naming another session is left
    to another capability.
    """

    def __post_init__(self) -> None:
        if self.defer_loading:
            raise UserError(
                "`SandboxdWorkspace` does not support `defer_loading=True`: "
                "the workspace is selected before deferred capabilities load."
            )

    def backend(self, ref: WorkspaceRef | None = None) -> SandboxdWorkspaceBackend:
        """A backend for `ref`, or for a new session; no I/O until its first operation."""
        return SandboxdWorkspaceBackend(
            self.service_url,
            token=self.token,
            ref=ref,
            provider=self.provider,
            runtime=self.runtime,
            tenant=self.tenant,
            env=self.env,
            client=self.client,
            session_name=self.session_name,
        )

    def get_workspace(
        self, ctx: RunContext[object], *, ref: WorkspaceRef | None
    ) -> WorkspaceBackend | None:
        """This run's backend, or `None` for a ref another provider owns."""
        del ctx
        if ref is not None and ref.provider != self.provider:
            return None
        if ref is not None and self.session_name is not None and ref.id != self.session_name:
            return None
        return self.backend(ref)

    async def destroy(self, ref: WorkspaceRef) -> None:
        """Close the session `ref` names and delete its files. Already gone is fine.

        Raises:
            ValueError: `ref` belongs to another provider.
            httpx.HTTPError: The service could not be reached or refused.
        """
        if ref.provider != self.provider:
            raise ValueError(f"expected a {self.provider!r} workspace ref, got {ref.provider!r}")
        await self.backend(ref).purge()

backend(ref=None)

A backend for ref, or for a new session; no I/O until its first operation.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

def backend(self, ref: WorkspaceRef | None = None) -> SandboxdWorkspaceBackend:
    """A backend for `ref`, or for a new session; no I/O until its first operation."""
    return SandboxdWorkspaceBackend(
        self.service_url,
        token=self.token,
        ref=ref,
        provider=self.provider,
        runtime=self.runtime,
        tenant=self.tenant,
        env=self.env,
        client=self.client,
        session_name=self.session_name,
    )

get_workspace(ctx, *, ref)

This run's backend, or None for a ref another provider owns.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

def get_workspace(
    self, ctx: RunContext[object], *, ref: WorkspaceRef | None
) -> WorkspaceBackend | None:
    """This run's backend, or `None` for a ref another provider owns."""
    del ctx
    if ref is not None and ref.provider != self.provider:
        return None
    if ref is not None and self.session_name is not None and ref.id != self.session_name:
        return None
    return self.backend(ref)

destroy(ref) async

Close the session ref names and delete its files. Already gone is fine.

Raises:

Type	Description
ValueError	ref belongs to another provider.
HTTPError	The service could not be reached or refused.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

async def destroy(self, ref: WorkspaceRef) -> None:
    """Close the session `ref` names and delete its files. Already gone is fine.

    Raises:
        ValueError: `ref` belongs to another provider.
        httpx.HTTPError: The service could not be reached or refused.
    """
    if ref.provider != self.provider:
        raise ValueError(f"expected a {self.provider!r} workspace ref, got {ref.provider!r}")
    await self.backend(ref).purge()

pydantic_ai_backends.workspaces.SandboxdWorkspaceBackend

Bases: WorkspaceBackend, SupportsCommands

One sandboxd session, as the environment an agent run works in.

Commands go through the service's /run, which keeps stdout and stderr apart, reports a vanished sandbox as one and can be stopped; Workspace derives file operations through them. The session's files live on the service host, so a ref still attaches after the container was reaped, for as long as the service keeps the workspace (its workspace_ttl).

Every command is bounded by the service's execute_timeout, including one asking for no timeout: the service enforces its ceiling on every caller.

Parameters:

Name	Type	Description	Default
service_url	str	Base URL of the service.	required
token	str	The service token. It can open a session on the service's host, so treat it as the Docker socket it sits in front of.	required
ref	WorkspaceRef | None	The workspace to attach to; None opens a new session on first use.	None
provider	str	Provider name in refs. Give each service its own when an agent can reach more than one, or a ref from one would attach on another.	SANDBOXD_PROVIDER
runtime	str | None	Runtime alias for a new session; the service default when None.	None
tenant	str | None	Who the session is opened for, counted against the service's per-tenant ceiling.	None
env	Mapping[str, str] | None	Variables every command gets, under any a call passes.	None
client	AsyncClient | None	An httpx.AsyncClient to share. Owned by the caller, who closes it; without one each request uses a client of its own.	None
request_timeout	float	Seconds for a request that runs no command.	DEFAULT_REQUEST_TIMEOUT
session_name	str | None	A session id chosen by whoever configures the workspace rather than by the service: with no ref, the first operation opens the session under it, or attaches when it is open or its files are kept. A ref is still attach-only, so a caller that knows the session existed learns that it is gone instead of starting over in an empty one.	None

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

class SandboxdWorkspaceBackend(WorkspaceBackend, SupportsCommands):
    """One `sandboxd` session, as the environment an agent run works in.

    Commands go through the service's `/run`, which keeps stdout and stderr
    apart, reports a vanished sandbox as one and can be stopped; `Workspace`
    derives file operations through them. The session's files live on the
    service host, so a ref still attaches after the container was reaped, for as
    long as the service keeps the workspace (its `workspace_ttl`).

    Every command is bounded by the service's `execute_timeout`, including one
    asking for no timeout: the service enforces its ceiling on every caller.

    Args:
        service_url: Base URL of the service.
        token: The service token. It can open a session on the service's host,
            so treat it as the Docker socket it sits in front of.
        ref: The workspace to attach to; `None` opens a new session on first use.
        provider: Provider name in refs. Give each service its own when an agent
            can reach more than one, or a ref from one would attach on another.
        runtime: Runtime alias for a new session; the service default when `None`.
        tenant: Who the session is opened for, counted against the service's
            per-tenant ceiling.
        env: Variables every command gets, under any a call passes.
        client: An `httpx.AsyncClient` to share. Owned by the caller, who
            closes it; without one each request uses a client of its own.
        request_timeout: Seconds for a request that runs no command.
        session_name: A session id chosen by whoever configures the workspace
            rather than by the service: with no ref, the first operation opens
            the session under it, or attaches when it is open or its files are
            kept. A ref is still attach-only, so a caller that knows the
            session existed learns that it is gone instead of starting over in
            an empty one.
    """

    def __init__(
        self,
        service_url: str,
        *,
        token: str,
        ref: WorkspaceRef | None = None,
        provider: str = SANDBOXD_PROVIDER,
        runtime: str | None = None,
        tenant: str | None = None,
        env: Mapping[str, str] | None = None,
        client: httpx.AsyncClient | None = None,
        request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
        session_name: str | None = None,
    ) -> None:
        if ref is not None and ref.provider != provider:
            raise ValueError(f"expected a {provider!r} workspace ref, got {ref.provider!r}")
        if session_name is not None and not re.fullmatch(wire.SESSION_ID_PATTERN, session_name):
            raise ValueError(
                f"session_name {session_name!r} is not a session id the service accepts"
            )
        self._service_url = service_url.rstrip("/")
        self._token = token
        self._ref = ref
        self._provider = provider
        self._runtime = runtime
        self._tenant = tenant
        self._env = dict(env) if env else None
        self._client = client
        self._request_timeout = request_timeout
        self._session_name = session_name
        self._session: _Session | None = None
        self._working_dir: str | None = None
        self._lock = anyio.Lock()

    @property
    def ref(self) -> WorkspaceRef | None:
        """The session id once it exists; `None` before the first operation."""
        return self._ref

    @contextlib.asynccontextmanager
    async def _http(self) -> AsyncIterator[httpx.AsyncClient]:
        if self._client is not None:
            yield self._client
            return
        async with httpx.AsyncClient(base_url=self._service_url) as client:
            yield client

    def _url(self, path: str) -> str:
        # A shared client may carry its own base URL or none, so every request
        # names the service in full.
        return f"{self._service_url}{path}"

    async def _post(
        self, path: str, body: BaseModel | None, *, token: str, timeout: float
    ) -> httpx.Response:
        async with self._http() as client:
            return await client.post(
                self._url(path),
                json=None if body is None else body.model_dump(mode="json"),
                headers={wire.TOKEN_HEADER: token},
                timeout=timeout,
            )

    async def _open(self) -> _Session:
        """Open or attach to the session, then learn the service's command ceiling."""
        named = self._ref is None and self._session_name is not None
        request = wire.CreateSessionRequest(
            session_id=self._session_name if self._ref is None else self._ref.id,
            runtime=self._runtime,
            tenant=self._tenant,
            reuse=named,
            attach=self._ref is not None,
        )
        response = await self._post(
            "/sessions", request, token=self._token, timeout=self._request_timeout
        )
        # With `reuse` a 409 means another client is opening the same name right
        # now - two first runs of one conversation, say. It is theirs a moment
        # later, so ask again rather than fail one of them.
        give_up = anyio.current_time() + self._request_timeout
        while named and response.status_code == 409 and anyio.current_time() < give_up:
            await anyio.sleep(OPENING_RETRY_SECONDS)
            response = await self._post(
                "/sessions", request, token=self._token, timeout=self._request_timeout
            )
        if self._ref is not None and response.status_code == 404:
            raise WorkspaceUnavailableError(f"sandboxd session {self._ref.id!r} no longer exists")
        response.raise_for_status()
        created = wire.SessionCreated.model_validate_json(response.content)
        if self._ref is None:
            # Recorded before anything else can fail, so a caller holding this
            # backend can always remove what it opened.
            self._ref = WorkspaceRef(provider=self._provider, id=created.session.session_id)

        async with self._http() as client:
            policy_response = await client.get(
                self._url("/policy"),
                headers={wire.TOKEN_HEADER: self._token},
                timeout=self._request_timeout,
            )
        policy_response.raise_for_status()
        policy = wire.ServicePolicy.model_validate_json(policy_response.content)
        return _Session(
            session_id=created.session.session_id,
            token=created.token,
            execute_timeout=float(policy.execute_timeout),
        )

    async def _connect(self) -> _Session:
        async with self._lock:
            if self._session is not None:
                return self._session
            # Shielded so a caller cancelled mid-open still records the ref of a
            # session the service has already opened. Bounded by the request
            # timeout of the two requests inside.
            with anyio.CancelScope(shield=True):
                opened = await self._open()
                self._session = opened
            return opened

    async def working_dir(self) -> str:
        """The directory commands start in, as the sandbox resolves it."""
        if self._working_dir is None:
            result = await self.run(["sh", "-c", "pwd -P"])
            if result.exit_code != 0:
                raise WorkspaceUnavailableError(
                    f"could not resolve the sandbox's working directory: {result.stderr.strip()}"
                )
            self._working_dir = result.stdout.rstrip("\n")
        return self._working_dir

    async def run(
        self,
        command: WorkspaceCommand,
        *,
        shell: bool = False,
        env: Mapping[str, str] | None = None,
        timeout: float | None = None,
    ) -> CommandResult:
        """Run a command in the session's sandbox; see `SupportsCommands.run`.

        Raises:
            WorkspaceUnavailableError: The session or its sandbox is gone.
            WorkspaceTimeoutError: The command reached `timeout`, or the
                service's own ceiling first.
            WorkspaceOutputLimitError: Its combined output passed the limit.
            httpx.HTTPError: The service could not be reached or answered with
                an unexpected status — transient, and left for a caller to retry.
        """
        argv = command_argv(command, shell)
        check_timeout(timeout)
        session = await self._connect()
        request = wire.RunRequest(
            argv=argv,
            env=layered_env(self._env, env) or {},
            timeout_seconds=timeout,
            run_id=uuid.uuid4().hex,
        )
        deadline = (
            session.execute_timeout if timeout is None else min(timeout, session.execute_timeout)
        )
        path = f"/sessions/{session.session_id}"
        answered = False
        try:
            response = await self._post(
                f"{path}/run",
                request,
                token=session.token,
                timeout=deadline + TRANSPORT_SLACK_SECONDS,
            )
            answered = True
        finally:
            # A caller cancelled mid-command, or a request that failed in
            # transit, may have left the command running on the service.
            if not answered:
                await self._stop(f"{path}/runs/{request.run_id}/stop", session.token)
        if response.status_code in _GONE:
            raise WorkspaceUnavailableError(
                f"sandboxd session {session.session_id!r} is gone: {response.text}"
            )
        response.raise_for_status()
        try:
            ran = wire.RunResponse.model_validate_json(response.content)
        except ValidationError as error:
            raise ValueError(f"sandboxd answered /run with something else: {error}") from error
        outcome = CommandOutcome(
            stdout=ran.stdout,
            stderr=ran.stderr,
            exit_code=ran.exit_code,
            timed_out=ran.timed_out,
            output_limited=ran.output_limited,
        )
        # A timeout past the service's ceiling never applied: the ceiling did.
        applied = timeout if timeout is not None and timeout <= session.execute_timeout else None
        return command_result(
            outcome, timeout=applied, limit=ran.output_limit or MAX_RUN_OUTPUT_BYTES
        )

    async def purge(self) -> None:
        """Close this backend's session and delete its files. Already gone is fine.

        Attaches first when the session is closed but its workspace is kept, since
        the service deletes a workspace only through its session.

        Raises:
            httpx.HTTPError: The service could not be reached or refused.
        """
        try:
            session = await self._connect()
        except WorkspaceUnavailableError:
            return
        async with self._http() as client:
            response = await client.delete(
                self._url(f"/sessions/{session.session_id}"),
                params={"purge": "true"},
                headers={wire.TOKEN_HEADER: self._token},
                timeout=self._request_timeout,
            )
        if response.status_code not in _GONE:
            response.raise_for_status()

    async def _stop(self, path: str, token: str) -> None:
        """Ask the service to stop a run whose caller stopped waiting, if it can.

        Best effort: the caller is leaving with its own error or cancellation,
        which a failed stop must not replace.
        """
        with (
            anyio.CancelScope(shield=True),
            anyio.move_on_after(STOP_GRACE_SECONDS),
            contextlib.suppress(httpx.HTTPError),
        ):
            await self._post(path, None, token=token, timeout=STOP_GRACE_SECONDS)

ref property

The session id once it exists; None before the first operation.

working_dir() async

The directory commands start in, as the sandbox resolves it.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

async def working_dir(self) -> str:
    """The directory commands start in, as the sandbox resolves it."""
    if self._working_dir is None:
        result = await self.run(["sh", "-c", "pwd -P"])
        if result.exit_code != 0:
            raise WorkspaceUnavailableError(
                f"could not resolve the sandbox's working directory: {result.stderr.strip()}"
            )
        self._working_dir = result.stdout.rstrip("\n")
    return self._working_dir

run(command, *, shell=False, env=None, timeout=None) async

Run a command in the session's sandbox; see SupportsCommands.run.

Raises:

Type	Description
WorkspaceUnavailableError	The session or its sandbox is gone.
WorkspaceTimeoutError	The command reached timeout, or the service's own ceiling first.
WorkspaceOutputLimitError	Its combined output passed the limit.
HTTPError	The service could not be reached or answered with an unexpected status — transient, and left for a caller to retry.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

async def run(
    self,
    command: WorkspaceCommand,
    *,
    shell: bool = False,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> CommandResult:
    """Run a command in the session's sandbox; see `SupportsCommands.run`.

    Raises:
        WorkspaceUnavailableError: The session or its sandbox is gone.
        WorkspaceTimeoutError: The command reached `timeout`, or the
            service's own ceiling first.
        WorkspaceOutputLimitError: Its combined output passed the limit.
        httpx.HTTPError: The service could not be reached or answered with
            an unexpected status — transient, and left for a caller to retry.
    """
    argv = command_argv(command, shell)
    check_timeout(timeout)
    session = await self._connect()
    request = wire.RunRequest(
        argv=argv,
        env=layered_env(self._env, env) or {},
        timeout_seconds=timeout,
        run_id=uuid.uuid4().hex,
    )
    deadline = (
        session.execute_timeout if timeout is None else min(timeout, session.execute_timeout)
    )
    path = f"/sessions/{session.session_id}"
    answered = False
    try:
        response = await self._post(
            f"{path}/run",
            request,
            token=session.token,
            timeout=deadline + TRANSPORT_SLACK_SECONDS,
        )
        answered = True
    finally:
        # A caller cancelled mid-command, or a request that failed in
        # transit, may have left the command running on the service.
        if not answered:
            await self._stop(f"{path}/runs/{request.run_id}/stop", session.token)
    if response.status_code in _GONE:
        raise WorkspaceUnavailableError(
            f"sandboxd session {session.session_id!r} is gone: {response.text}"
        )
    response.raise_for_status()
    try:
        ran = wire.RunResponse.model_validate_json(response.content)
    except ValidationError as error:
        raise ValueError(f"sandboxd answered /run with something else: {error}") from error
    outcome = CommandOutcome(
        stdout=ran.stdout,
        stderr=ran.stderr,
        exit_code=ran.exit_code,
        timed_out=ran.timed_out,
        output_limited=ran.output_limited,
    )
    # A timeout past the service's ceiling never applied: the ceiling did.
    applied = timeout if timeout is not None and timeout <= session.execute_timeout else None
    return command_result(
        outcome, timeout=applied, limit=ran.output_limit or MAX_RUN_OUTPUT_BYTES
    )

purge() async

Close this backend's session and delete its files. Already gone is fine.

Attaches first when the session is closed but its workspace is kept, since the service deletes a workspace only through its session.

Raises:

Type	Description
HTTPError	The service could not be reached or refused.

Source code in src/pydantic_ai_backends/workspaces/_sandboxd.py

async def purge(self) -> None:
    """Close this backend's session and delete its files. Already gone is fine.

    Attaches first when the session is closed but its workspace is kept, since
    the service deletes a workspace only through its session.

    Raises:
        httpx.HTTPError: The service could not be reached or refused.
    """
    try:
        session = await self._connect()
    except WorkspaceUnavailableError:
        return
    async with self._http() as client:
        response = await client.delete(
            self._url(f"/sessions/{session.session_id}"),
            params={"purge": "true"},
            headers={wire.TOKEN_HEADER: self._token},
            timeout=self._request_timeout,
        )
    if response.status_code not in _GONE:
        response.raise_for_status()

KubernetesWorkspace

pydantic_ai_backends.workspaces.KubernetesWorkspace dataclass

Bases: AbstractCapability[object]

Supply a Kubernetes pod as the run's workspace.

A run with no ref creates a pod; one carrying this capability's provider attaches to that pod. Pods are kept after the run — the ref is how to come back, and :meth:destroy deletes one. Commands go through pods/exec, so the caller needs that RBAC and the image needs /bin/sh.

Not checked against a live cluster in this repository's CI.

Example

Python

from pydantic_ai import Agent

from pydantic_ai_backends import ConsoleCapability
from pydantic_ai_backends.workspaces import KubernetesWorkspace

pods = KubernetesWorkspace(image="python:3.12-slim", namespace="agents")
agent = Agent("anthropic:claude-opus-5-5", capabilities=[pods, ConsoleCapability()])

Source code in src/pydantic_ai_backends/workspaces/_kubernetes.py

@dataclass(kw_only=True)
class KubernetesWorkspace(AbstractCapability[object]):
    """Supply a Kubernetes pod as the run's workspace.

    A run with no ref creates a pod; one carrying this capability's provider
    attaches to that pod. Pods are kept after the run — the ref is how to come
    back, and :meth:`destroy` deletes one. Commands go through `pods/exec`, so
    the caller needs that RBAC and the image needs `/bin/sh`.

    Not checked against a live cluster in this repository's CI.

    Example:
        ```python
        from pydantic_ai import Agent

        from pydantic_ai_backends import ConsoleCapability
        from pydantic_ai_backends.workspaces import KubernetesWorkspace

        pods = KubernetesWorkspace(image="python:3.12-slim", namespace="agents")
        agent = Agent("anthropic:claude-opus-5-5", capabilities=[pods, ConsoleCapability()])
        ```
    """

    image: str
    """Container image; needs `/bin/sh`."""

    namespace: str = "default"
    """Namespace the pods live in."""

    work_dir: str = DEFAULT_WORK_DIR
    """Directory commands start in."""

    pod_template: dict[str, Any] | None = None
    """A full pod spec instead of the hardened default; its first container must stay up."""

    kube_config_path: str | None = None
    """A kubeconfig; in-cluster config, then `~/.kube/config` when omitted."""

    service_account_name: str = "default"
    """The pods' service account; a dedicated one with no permissions is the right choice."""

    provider: str = KUBERNETES_PROVIDER
    """Provider name in refs; distinct per cluster when an agent can reach several."""

    env: Mapping[str, str] | None = field(default=None, repr=False)
    """Variables every command gets. Nothing is read from the host's environment."""

    def __post_init__(self) -> None:
        if self.defer_loading:
            raise UserError(
                "`KubernetesWorkspace` does not support `defer_loading=True`: "
                "the workspace is selected before deferred capabilities load."
            )

    def _pod(self, sandbox_id: str) -> KubernetesPodSandbox:
        return KubernetesPodSandbox(
            self.image,
            namespace=self.namespace,
            sandbox_id=sandbox_id,
            work_dir=self.work_dir,
            pod_template=self.pod_template,
            kube_config_path=self.kube_config_path,
            service_account_name=self.service_account_name,
        )

    def backend(self, ref: WorkspaceRef | None = None) -> KubernetesWorkspaceBackend:
        """A backend for `ref`, or for a new pod; no API call until its first operation."""
        return KubernetesWorkspaceBackend(
            pod_factory=self._pod, ref=ref, env=self.env, provider=self.provider
        )

    def get_workspace(
        self, ctx: RunContext[object], *, ref: WorkspaceRef | None
    ) -> WorkspaceBackend | None:
        """This run's backend, or `None` for a ref another provider owns."""
        del ctx
        if ref is not None and ref.provider != self.provider:
            return None
        return self.backend(ref)

    async def destroy(self, ref: WorkspaceRef) -> None:
        """Delete the pod `ref` names. Already gone is fine.

        Raises:
            ValueError: `ref` belongs to another provider.
        """
        if ref.provider != self.provider:
            raise ValueError(f"expected a {self.provider!r} workspace ref, got {ref.provider!r}")
        pod = await anyio.to_thread.run_sync(self._pod, ref.id)
        await anyio.to_thread.run_sync(pod.stop)

backend(ref=None)

A backend for ref, or for a new pod; no API call until its first operation.

Source code in src/pydantic_ai_backends/workspaces/_kubernetes.py

def backend(self, ref: WorkspaceRef | None = None) -> KubernetesWorkspaceBackend:
    """A backend for `ref`, or for a new pod; no API call until its first operation."""
    return KubernetesWorkspaceBackend(
        pod_factory=self._pod, ref=ref, env=self.env, provider=self.provider
    )

get_workspace(ctx, *, ref)

This run's backend, or None for a ref another provider owns.

Source code in src/pydantic_ai_backends/workspaces/_kubernetes.py

def get_workspace(
    self, ctx: RunContext[object], *, ref: WorkspaceRef | None
) -> WorkspaceBackend | None:
    """This run's backend, or `None` for a ref another provider owns."""
    del ctx
    if ref is not None and ref.provider != self.provider:
        return None
    return self.backend(ref)

destroy(ref) async

Delete the pod ref names. Already gone is fine.

Raises:

Type	Description
ValueError	ref belongs to another provider.

Source code in src/pydantic_ai_backends/workspaces/_kubernetes.py

async def destroy(self, ref: WorkspaceRef) -> None:
    """Delete the pod `ref` names. Already gone is fine.

    Raises:
        ValueError: `ref` belongs to another provider.
    """
    if ref.provider != self.provider:
        raise ValueError(f"expected a {self.provider!r} workspace ref, got {ref.provider!r}")
    pod = await anyio.to_thread.run_sync(self._pod, ref.id)
    await anyio.to_thread.run_sync(pod.stop)

pydantic_ai_backends.workspaces.KubernetesWorkspaceBackend

Bases: ContainerWorkspaceBackend

One pod, as the environment an agent run works in.

The first operation creates the pod and waits for it to be Ready; a ref attaches to a running pod and fails with WorkspaceUnavailableError when it is gone or has finished. The pod lives until :meth:KubernetesWorkspace.destroy.

Parameters:

Name	Type	Description	Default
pod_factory	PodFactory	Builds the KubernetesPodSandbox for a sandbox id. Holds the image, namespace and pod template, and must not create the pod.	required
ref	WorkspaceRef | None	The workspace to attach to; None creates one on first use.	None
env	Mapping[str, str] | None	Variables every command gets, under any a call passes.	None
provider	str	Provider name in refs; distinct per cluster when an agent can reach several.	KUBERNETES_PROVIDER

Source code in src/pydantic_ai_backends/workspaces/_kubernetes.py

class KubernetesWorkspaceBackend(ContainerWorkspaceBackend):
    """One pod, as the environment an agent run works in.

    The first operation creates the pod and waits for it to be Ready; a ref
    attaches to a running pod and fails with `WorkspaceUnavailableError` when it
    is gone or has finished. The pod lives until :meth:`KubernetesWorkspace.destroy`.

    Args:
        pod_factory: Builds the `KubernetesPodSandbox` for a sandbox id. Holds
            the image, namespace and pod template, and must not create the pod.
        ref: The workspace to attach to; `None` creates one on first use.
        env: Variables every command gets, under any a call passes.
        provider: Provider name in refs; distinct per cluster when an agent
            can reach several.
    """

    def __init__(
        self,
        *,
        pod_factory: PodFactory,
        ref: WorkspaceRef | None = None,
        env: Mapping[str, str] | None = None,
        provider: str = KUBERNETES_PROVIDER,
    ) -> None:
        async def open_pod(sandbox_id: str | None) -> tuple[str, RunnerSandbox]:
            # In a thread: building one loads the kubeconfig from disk.
            sandbox = await anyio.to_thread.run_sync(pod_factory, sandbox_id or uuid.uuid4().hex)
            if sandbox_id is None:
                await anyio.to_thread.run_sync(sandbox.start)
            else:
                await anyio.to_thread.run_sync(sandbox.attach)
            return sandbox.id, sandbox

        super().__init__(provider=provider, opener=open_pod, ref=ref, env=env)

DaytonaWorkspace

pydantic_ai_backends.workspaces.DaytonaWorkspace dataclass

Bases: AbstractCapability[object]

Supply a Daytona sandbox as the run's workspace.

A run with no ref creates a sandbox; one carrying a "daytona" ref attaches to it, starting it when Daytona stopped or archived it. Sandboxes are kept after the run, subject to Daytona's own auto-stop and auto-delete; the ref is how to come back, and :meth:destroy deletes one.

Not checked against a live Daytona account in this repository's CI.

Example

Python

from daytona import DaytonaConfig
from pydantic_ai import Agent

from pydantic_ai_backends import ConsoleCapability
from pydantic_ai_backends.workspaces import DaytonaWorkspace

sandboxes = DaytonaWorkspace(config=DaytonaConfig(api_key="dtn_..."))
agent = Agent("anthropic:claude-opus-5-5", capabilities=[sandboxes, ConsoleCapability()])

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

@dataclass(kw_only=True)
class DaytonaWorkspace(AbstractCapability[object]):
    """Supply a Daytona sandbox as the run's workspace.

    A run with no ref creates a sandbox; one carrying a `"daytona"` ref attaches
    to it, starting it when Daytona stopped or archived it. Sandboxes are kept
    after the run, subject to Daytona's own auto-stop and auto-delete; the ref is
    how to come back, and :meth:`destroy` deletes one.

    Not checked against a live Daytona account in this repository's CI.

    Example:
        ```python
        from daytona import DaytonaConfig
        from pydantic_ai import Agent

        from pydantic_ai_backends import ConsoleCapability
        from pydantic_ai_backends.workspaces import DaytonaWorkspace

        sandboxes = DaytonaWorkspace(config=DaytonaConfig(api_key="dtn_..."))
        agent = Agent("anthropic:claude-opus-5-5", capabilities=[sandboxes, ConsoleCapability()])
        ```
    """

    config: DaytonaConfig | None = field(default=None, repr=False)
    """A `daytona.DaytonaConfig`; the environment's `DAYTONA_*` variables when `None`."""

    create_params: CreateParams | None = None
    """Parameters for a new sandbox, such as `CreateSandboxFromSnapshotParams`."""

    env: Mapping[str, str] | None = field(default=None, repr=False)
    """Variables every command gets. Nothing is read from the host's environment."""

    client: AsyncDaytona | None = field(default=None, repr=False, compare=False)
    """An `AsyncDaytona` to share across runs, owned and closed by the caller."""

    sandbox_name: str | None = None
    """One sandbox for every run, named by you: created on first use, attached after.

    Without it each run without a ref gets a new sandbox and only the ref leads
    back to it. With it a later process finds the same sandbox by name. A ref
    naming another sandbox is left to another capability.
    """

    def __post_init__(self) -> None:
        if self.defer_loading:
            raise UserError(
                "`DaytonaWorkspace` does not support `defer_loading=True`: "
                "the workspace is selected before deferred capabilities load."
            )

    def backend(self, ref: WorkspaceRef | None = None) -> DaytonaWorkspaceBackend:
        """A backend for `ref`, or for a new sandbox; no API call until its first operation."""
        return DaytonaWorkspaceBackend(
            config=self.config,
            create_params=self.create_params,
            ref=ref,
            env=self.env,
            client=self.client,
            sandbox_name=self.sandbox_name,
        )

    def get_workspace(
        self, ctx: RunContext[object], *, ref: WorkspaceRef | None
    ) -> WorkspaceBackend | None:
        """This run's backend, or `None` for a ref another provider owns."""
        del ctx
        if ref is not None and ref.provider != DAYTONA_PROVIDER:
            return None
        if ref is not None and self.sandbox_name is not None and ref.id != self.sandbox_name:
            return None
        return self.backend(ref)

    async def destroy(self, ref: WorkspaceRef) -> None:
        """Delete the sandbox `ref` names. Already gone is fine.

        Raises:
            ValueError: `ref` belongs to another provider.
        """
        if ref.provider != DAYTONA_PROVIDER:
            raise ValueError(f"expected a {DAYTONA_PROVIDER!r} workspace ref, got {ref.provider!r}")
        await self.backend(ref).purge()

backend(ref=None)

A backend for ref, or for a new sandbox; no API call until its first operation.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

def backend(self, ref: WorkspaceRef | None = None) -> DaytonaWorkspaceBackend:
    """A backend for `ref`, or for a new sandbox; no API call until its first operation."""
    return DaytonaWorkspaceBackend(
        config=self.config,
        create_params=self.create_params,
        ref=ref,
        env=self.env,
        client=self.client,
        sandbox_name=self.sandbox_name,
    )

get_workspace(ctx, *, ref)

This run's backend, or None for a ref another provider owns.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

def get_workspace(
    self, ctx: RunContext[object], *, ref: WorkspaceRef | None
) -> WorkspaceBackend | None:
    """This run's backend, or `None` for a ref another provider owns."""
    del ctx
    if ref is not None and ref.provider != DAYTONA_PROVIDER:
        return None
    if ref is not None and self.sandbox_name is not None and ref.id != self.sandbox_name:
        return None
    return self.backend(ref)

destroy(ref) async

Delete the sandbox ref names. Already gone is fine.

Raises:

Type	Description
ValueError	ref belongs to another provider.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

async def destroy(self, ref: WorkspaceRef) -> None:
    """Delete the sandbox `ref` names. Already gone is fine.

    Raises:
        ValueError: `ref` belongs to another provider.
    """
    if ref.provider != DAYTONA_PROVIDER:
        raise ValueError(f"expected a {DAYTONA_PROVIDER!r} workspace ref, got {ref.provider!r}")
    await self.backend(ref).purge()

pydantic_ai_backends.workspaces.DaytonaWorkspaceBackend

Bases: WorkspaceBackend, SupportsCommands

One Daytona sandbox, as the environment an agent run works in.

Commands only: Workspace derives the file operations through the shell. The first operation creates the sandbox, or attaches to the one the ref names, starting it when it was stopped or archived; a sandbox that is gone fails with WorkspaceUnavailableError. It lives until :meth:DaytonaWorkspace.destroy or Daytona's own auto-delete.

Parameters:

Name	Type	Description	Default
config	DaytonaConfig | None	A daytona.DaytonaConfig; the environment's DAYTONA_* variables when None.	None
create_params	CreateParams | None	Parameters for a new sandbox, such as CreateSandboxFromSnapshotParams; Daytona's default when None.	None
ref	WorkspaceRef | None	The workspace to attach to; None creates one on first use.	None
env	Mapping[str, str] | None	Variables every command gets, under any a call passes.	None
client	AsyncDaytona | None	An AsyncDaytona to share, owned and closed by the caller; without one each operation opens and closes its own.	None
sandbox_name	str | None	A sandbox name chosen by whoever configures the workspace rather than by Daytona: with no ref, the first operation creates the sandbox under it, or attaches to the one that already has it. A ref is still attach-only, so a caller that knows the sandbox existed learns that it is gone instead of starting over in a new one.	None

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

class DaytonaWorkspaceBackend(WorkspaceBackend, SupportsCommands):
    """One Daytona sandbox, as the environment an agent run works in.

    Commands only: `Workspace` derives the file operations through the shell.
    The first operation creates the sandbox, or attaches to the one the ref
    names, starting it when it was stopped or archived; a sandbox that is gone
    fails with `WorkspaceUnavailableError`. It lives until
    :meth:`DaytonaWorkspace.destroy` or Daytona's own auto-delete.

    Args:
        config: A `daytona.DaytonaConfig`; the environment's `DAYTONA_*`
            variables when `None`.
        create_params: Parameters for a new sandbox, such as
            `CreateSandboxFromSnapshotParams`; Daytona's default when `None`.
        ref: The workspace to attach to; `None` creates one on first use.
        env: Variables every command gets, under any a call passes.
        client: An `AsyncDaytona` to share, owned and closed by the caller;
            without one each operation opens and closes its own.
        sandbox_name: A sandbox name chosen by whoever configures the workspace
            rather than by Daytona: with no ref, the first operation creates the
            sandbox under it, or attaches to the one that already has it. A ref
            is still attach-only, so a caller that knows the sandbox existed
            learns that it is gone instead of starting over in a new one.
    """

    def __init__(
        self,
        *,
        config: DaytonaConfig | None = None,
        create_params: CreateParams | None = None,
        ref: WorkspaceRef | None = None,
        env: Mapping[str, str] | None = None,
        client: AsyncDaytona | None = None,
        sandbox_name: str | None = None,
    ) -> None:
        if ref is not None and ref.provider != DAYTONA_PROVIDER:
            raise ValueError(f"expected a {DAYTONA_PROVIDER!r} workspace ref, got {ref.provider!r}")
        self._sandbox_name = sandbox_name
        self._config = config
        self._create_params = create_params
        self._ref = ref
        self._env = dict(env) if env else None
        self._client = client
        self._working_dir: str | None = None
        self._session_id = f"pab-{uuid.uuid4().hex}"
        self._session_open = False
        self._lock = anyio.Lock()

    @property
    def ref(self) -> WorkspaceRef | None:
        """The sandbox's id once it exists; `None` before the first operation."""
        return self._ref

    @contextlib.asynccontextmanager
    async def _daytona(self) -> AsyncIterator[AsyncDaytona]:
        if self._client is not None:
            yield self._client
            return
        daytona = load("daytona", purpose="DaytonaWorkspace")
        async with daytona.AsyncDaytona(self._config) as client:
            yield client

    async def _sandbox(self, client: AsyncDaytona) -> AsyncSandbox:
        """The sandbox, created or attached under the lock, with a session open in it."""
        async with self._lock:
            if self._ref is not None:
                sandbox = await _attach(client, self._ref.id)
            elif self._sandbox_name is not None:
                sandbox = await self._open_named(client, self._sandbox_name)
            else:
                # Shielded so a caller cancelled mid-create still leaves the ref
                # of a sandbox Daytona has already made.
                with anyio.CancelScope(shield=True):
                    sandbox = await client.create(self._create_params)
                    self._ref = WorkspaceRef(provider=DAYTONA_PROVIDER, id=sandbox.id)
            if self._working_dir is None:
                self._working_dir = await sandbox.get_work_dir()
            if not self._session_open:
                await sandbox.process.create_session(self._session_id)
                self._session_open = True
            return sandbox

    async def _open_named(self, client: AsyncDaytona, name: str) -> AsyncSandbox:
        """The sandbox called `name`: the existing one, or a new one created under it."""
        daytona = load("daytona", purpose="DaytonaWorkspace")
        with anyio.CancelScope(shield=True):
            try:
                sandbox = await _attach(client, name)
            except WorkspaceUnavailableError as gone:
                if not isinstance(gone.__cause__, daytona.DaytonaNotFoundError):
                    raise
                params = self._create_params or daytona.CreateSandboxFromSnapshotParams()
                try:
                    sandbox = await client.create(params.model_copy(update={"name": name}))
                except daytona.DaytonaConflictError:
                    # Another client created it between our lookup and our create.
                    sandbox = await _attach(client, name)
            # The name, not the id: it is what a later client knows to look for,
            # and `get` takes either.
            self._ref = WorkspaceRef(provider=DAYTONA_PROVIDER, id=name)
        return sandbox

    async def working_dir(self) -> str:
        """The sandbox's working directory, as Daytona reports it."""
        async with self._daytona() as client:
            await self._sandbox(client)
        assert self._working_dir is not None
        return self._working_dir

    async def run(
        self,
        command: WorkspaceCommand,
        *,
        shell: bool = False,
        env: Mapping[str, str] | None = None,
        timeout: float | None = None,
    ) -> CommandResult:
        """Run a command in the sandbox, stdin at EOF; see `SupportsCommands.run`.

        Raises:
            WorkspaceUnavailableError: The sandbox is gone.
            WorkspaceTimeoutError: The command reached `timeout`. Daytona answers
                a session command only once it ends, so no partial output comes
                with it.
            WorkspaceOutputLimitError: Its combined output passed 10 MiB.
        """
        argv = command_argv(command, shell)
        check_timeout(timeout)
        daytona = load("daytona", purpose="DaytonaWorkspace")
        run_id = uuid.uuid4().hex
        assignments = [f"{k}={v}" for k, v in (layered_env(self._env, env) or {}).items()]
        async with self._daytona() as client:
            sandbox = await self._sandbox(client)
            assert self._working_dir is not None
            line = shlex.join(
                [
                    "sh",
                    "-c",
                    _CD_THEN_EXEC,
                    "sh",
                    self._working_dir,
                    "env",
                    *assignments,
                    *wrapped_argv(argv, run_id),
                ]
            )
            request = daytona.SessionExecuteRequest(
                command=f"{line} </dev/null", run_async=False, suppress_input_echo=True
            )
            answered = False
            try:
                with anyio.move_on_after(timeout) as deadline:
                    response = await sandbox.process.execute_session_command(
                        self._session_id, request
                    )
                    answered = True
            except daytona.DaytonaNotFoundError as error:
                raise WorkspaceUnavailableError(
                    f"Daytona sandbox {sandbox.id!r} went away during the command"
                ) from error
            finally:
                if not answered:
                    await self._stop(sandbox, run_id)
        if deadline.cancelled_caught:
            return command_result(
                CommandOutcome(stdout="", stderr="", timed_out=True),
                timeout=timeout,
                limit=MAX_RUN_OUTPUT_BYTES,
            )
        stdout, stderr = response.stdout or "", response.stderr or ""
        limited = len(stdout.encode()) + len(stderr.encode()) > MAX_RUN_OUTPUT_BYTES
        outcome = CommandOutcome(
            stdout=stdout[:PARTIAL_OUTPUT_BYTES] if limited else stdout,
            stderr=stderr[:PARTIAL_OUTPUT_BYTES] if limited else stderr,
            exit_code=None if limited else response.exit_code,
            output_limited=limited,
        )
        return command_result(outcome, timeout=timeout, limit=MAX_RUN_OUTPUT_BYTES)

    async def _stop(self, sandbox: AsyncSandbox, run_id: str) -> None:
        """Stop a command whose caller stopped waiting, and everything it started."""
        stopper = shlex.join(["sh", "-c", STOPPER, "sh", pid_file(run_id)])
        with (
            anyio.CancelScope(shield=True),
            anyio.move_on_after(STOP_GRACE_SECONDS),
            contextlib.suppress(Exception),
        ):
            await sandbox.process.exec(stopper, timeout=int(STOP_GRACE_SECONDS))

    async def purge(self) -> None:
        """Delete this backend's sandbox. Already gone is fine."""
        if self._ref is None:
            return
        daytona = load("daytona", purpose="DaytonaWorkspace")
        async with self._daytona() as client:
            try:
                sandbox = await client.get(self._ref.id)
            except daytona.DaytonaNotFoundError:
                return
            await client.delete(sandbox)

ref property

The sandbox's id once it exists; None before the first operation.

working_dir() async

The sandbox's working directory, as Daytona reports it.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

async def working_dir(self) -> str:
    """The sandbox's working directory, as Daytona reports it."""
    async with self._daytona() as client:
        await self._sandbox(client)
    assert self._working_dir is not None
    return self._working_dir

run(command, *, shell=False, env=None, timeout=None) async

Run a command in the sandbox, stdin at EOF; see SupportsCommands.run.

Raises:

Type	Description
WorkspaceUnavailableError	The sandbox is gone.
WorkspaceTimeoutError	The command reached timeout. Daytona answers a session command only once it ends, so no partial output comes with it.
WorkspaceOutputLimitError	Its combined output passed 10 MiB.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

async def run(
    self,
    command: WorkspaceCommand,
    *,
    shell: bool = False,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> CommandResult:
    """Run a command in the sandbox, stdin at EOF; see `SupportsCommands.run`.

    Raises:
        WorkspaceUnavailableError: The sandbox is gone.
        WorkspaceTimeoutError: The command reached `timeout`. Daytona answers
            a session command only once it ends, so no partial output comes
            with it.
        WorkspaceOutputLimitError: Its combined output passed 10 MiB.
    """
    argv = command_argv(command, shell)
    check_timeout(timeout)
    daytona = load("daytona", purpose="DaytonaWorkspace")
    run_id = uuid.uuid4().hex
    assignments = [f"{k}={v}" for k, v in (layered_env(self._env, env) or {}).items()]
    async with self._daytona() as client:
        sandbox = await self._sandbox(client)
        assert self._working_dir is not None
        line = shlex.join(
            [
                "sh",
                "-c",
                _CD_THEN_EXEC,
                "sh",
                self._working_dir,
                "env",
                *assignments,
                *wrapped_argv(argv, run_id),
            ]
        )
        request = daytona.SessionExecuteRequest(
            command=f"{line} </dev/null", run_async=False, suppress_input_echo=True
        )
        answered = False
        try:
            with anyio.move_on_after(timeout) as deadline:
                response = await sandbox.process.execute_session_command(
                    self._session_id, request
                )
                answered = True
        except daytona.DaytonaNotFoundError as error:
            raise WorkspaceUnavailableError(
                f"Daytona sandbox {sandbox.id!r} went away during the command"
            ) from error
        finally:
            if not answered:
                await self._stop(sandbox, run_id)
    if deadline.cancelled_caught:
        return command_result(
            CommandOutcome(stdout="", stderr="", timed_out=True),
            timeout=timeout,
            limit=MAX_RUN_OUTPUT_BYTES,
        )
    stdout, stderr = response.stdout or "", response.stderr or ""
    limited = len(stdout.encode()) + len(stderr.encode()) > MAX_RUN_OUTPUT_BYTES
    outcome = CommandOutcome(
        stdout=stdout[:PARTIAL_OUTPUT_BYTES] if limited else stdout,
        stderr=stderr[:PARTIAL_OUTPUT_BYTES] if limited else stderr,
        exit_code=None if limited else response.exit_code,
        output_limited=limited,
    )
    return command_result(outcome, timeout=timeout, limit=MAX_RUN_OUTPUT_BYTES)

purge() async

Delete this backend's sandbox. Already gone is fine.

Source code in src/pydantic_ai_backends/workspaces/_daytona.py

async def purge(self) -> None:
    """Delete this backend's sandbox. Already gone is fine."""
    if self._ref is None:
        return
    daytona = load("daytona", purpose="DaytonaWorkspace")
    async with self._daytona() as client:
        try:
            sandbox = await client.get(self._ref.id)
        except daytona.DaytonaNotFoundError:
            return
        await client.delete(sandbox)

StateWorkspace

pydantic_ai_backends.workspaces.StateWorkspace dataclass

Bases: AbstractCapability[object]

Supply a StateBackend document as the run's workspace.

A run with no ref adds a document to store; one carrying a "state" ref works in the document under that id, and one whose document is gone fails with WorkspaceUnavailableError. Commands are not available: compose it with file tools such as the harness's FileSystem, or ConsoleCapability(include_execute=False).

The default store lives as long as this capability. An application that keeps documents elsewhere fills the mapping before a run and saves the document afterwards — files and sorted(directories) are JSON.

Example

Python

from pydantic_ai import Agent
from pydantic_ai_harness.filesystem import FileSystem

from pydantic_ai_backends.workspaces import StateWorkspace

agent = Agent("anthropic:claude-opus-5-5", capabilities=[StateWorkspace(), FileSystem()])

Source code in src/pydantic_ai_backends/workspaces/_state.py

@dataclass(kw_only=True)
class StateWorkspace(AbstractCapability[object]):
    """Supply a `StateBackend` document as the run's workspace.

    A run with no ref adds a document to `store`; one carrying a `"state"` ref
    works in the document under that id, and one whose document is gone fails
    with `WorkspaceUnavailableError`. Commands are not available: compose it
    with file tools such as the harness's `FileSystem`, or
    `ConsoleCapability(include_execute=False)`.

    The default `store` lives as long as this capability. An application that
    keeps documents elsewhere fills the mapping before a run and saves the
    document afterwards — `files` and `sorted(directories)` are JSON.

    Example:
        ```python
        from pydantic_ai import Agent
        from pydantic_ai_harness.filesystem import FileSystem

        from pydantic_ai_backends.workspaces import StateWorkspace

        agent = Agent("anthropic:claude-opus-5-5", capabilities=[StateWorkspace(), FileSystem()])
        ```
    """

    store: MutableMapping[str, StateBackend] = field(default_factory=dict, repr=False)
    """Documents by id; the default is an in-process dict."""

    def __post_init__(self) -> None:
        if self.defer_loading:
            raise UserError(
                "`StateWorkspace` does not support `defer_loading=True`: "
                "the workspace is selected before deferred capabilities load."
            )

    def backend(self, ref: WorkspaceRef | None = None) -> StateWorkspaceBackend:
        """A backend for `ref`, or for a new document created on first use."""
        return StateWorkspaceBackend(self.store, ref=ref)

    def get_workspace(
        self, ctx: RunContext[object], *, ref: WorkspaceRef | None
    ) -> WorkspaceBackend | None:
        """This run's backend, or `None` for a ref another provider owns."""
        del ctx
        if ref is not None and ref.provider != STATE_PROVIDER:
            return None
        return self.backend(ref)

    async def destroy(self, ref: WorkspaceRef) -> None:
        """Drop the document `ref` names. Already gone is fine.

        Raises:
            ValueError: `ref` belongs to another provider.
        """
        if ref.provider != STATE_PROVIDER:
            raise ValueError(f"expected a {STATE_PROVIDER!r} workspace ref, got {ref.provider!r}")
        self.store.pop(ref.id, None)

backend(ref=None)

A backend for ref, or for a new document created on first use.

Source code in src/pydantic_ai_backends/workspaces/_state.py

def backend(self, ref: WorkspaceRef | None = None) -> StateWorkspaceBackend:
    """A backend for `ref`, or for a new document created on first use."""
    return StateWorkspaceBackend(self.store, ref=ref)

get_workspace(ctx, *, ref)

This run's backend, or None for a ref another provider owns.

Source code in src/pydantic_ai_backends/workspaces/_state.py

def get_workspace(
    self, ctx: RunContext[object], *, ref: WorkspaceRef | None
) -> WorkspaceBackend | None:
    """This run's backend, or `None` for a ref another provider owns."""
    del ctx
    if ref is not None and ref.provider != STATE_PROVIDER:
        return None
    return self.backend(ref)

destroy(ref) async

Drop the document ref names. Already gone is fine.

Raises:

Type	Description
ValueError	ref belongs to another provider.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def destroy(self, ref: WorkspaceRef) -> None:
    """Drop the document `ref` names. Already gone is fine.

    Raises:
        ValueError: `ref` belongs to another provider.
    """
    if ref.provider != STATE_PROVIDER:
        raise ValueError(f"expected a {STATE_PROVIDER!r} workspace ref, got {ref.provider!r}")
    self.store.pop(ref.id, None)

pydantic_ai_backends.workspaces.StateWorkspaceBackend

Bases: WorkspaceBackend, SupportsFilesystem

A StateBackend document, as the environment an agent run works in.

Files only: there is nothing here to run a command in, so ctx.workspace.run refuses and file tools work. The document lives in store under the ref's id; a run with no ref adds a new one on its first operation.

Parameters:

Name	Type	Description	Default
store	MutableMapping[str, StateBackend]	Documents by id. The application owns it — an in-process dict, or a mapping it filled from its own storage before the run.	required
ref	WorkspaceRef | None	The document to work in; None creates one on first use.	None

Source code in src/pydantic_ai_backends/workspaces/_state.py

class StateWorkspaceBackend(WorkspaceBackend, SupportsFilesystem):
    """A `StateBackend` document, as the environment an agent run works in.

    Files only: there is nothing here to run a command in, so `ctx.workspace.run`
    refuses and file tools work. The document lives in `store` under the ref's
    id; a run with no ref adds a new one on its first operation.

    Args:
        store: Documents by id. The application owns it — an in-process dict, or
            a mapping it filled from its own storage before the run.
        ref: The document to work in; `None` creates one on first use.
    """

    def __init__(
        self, store: MutableMapping[str, StateBackend], *, ref: WorkspaceRef | None = None
    ) -> None:
        if ref is not None and ref.provider != STATE_PROVIDER:
            raise ValueError(f"expected a {STATE_PROVIDER!r} workspace ref, got {ref.provider!r}")
        self._store = store
        self._ref = ref

    @property
    def ref(self) -> WorkspaceRef | None:
        """The document's id once it exists; `None` before the first operation."""
        return self._ref

    def _state(self) -> StateBackend:
        """The document, created on the first operation when there is no ref yet."""
        if self._ref is None:
            document_id = uuid.uuid4().hex
            self._store[document_id] = StateBackend()
            self._ref = WorkspaceRef(provider=STATE_PROVIDER, id=document_id)
        state = self._store.get(self._ref.id)
        if state is None:
            raise WorkspaceUnavailableError(f"state document {self._ref.id!r} no longer exists")
        return state

    async def working_dir(self) -> str:
        """The document's root, which relative paths resolve against."""
        self._state()
        return STATE_ROOT

    async def read_bytes(self, path: str) -> bytes:
        """A file's exact bytes."""
        return self._state().read_bytes(path)

    async def write_bytes(self, path: str, data: bytes) -> None:
        """Store a file, creating missing parents."""
        self._state().write_bytes(path, data)

    async def stat(self, path: str) -> FileEntry:
        """Metadata for a file or directory."""
        state = self._state()
        normal = posixpath.normpath(path)
        if state.is_dir(normal):
            return FileEntry(name=posixpath.basename(normal), path=normal, is_dir=True, size=None)
        return FileEntry(
            name=posixpath.basename(normal), path=normal, is_dir=False, size=state.size(normal)
        )

    async def list_dir(self, path: str) -> Sequence[FileEntry]:
        """The entries directly in a directory."""
        state = self._state()
        normal = posixpath.normpath(path)
        entries: list[FileEntry] = []
        for name, is_dir in state.list_dir(normal):
            child = posixpath.join(normal, name)
            size = None if is_dir else state.size(child)
            entries.append(FileEntry(name=name, path=child, is_dir=is_dir, size=size))
        return entries

    async def make_dir(self, path: str) -> None:
        """Create a directory and any missing parents."""
        self._state().make_dir(path)

    async def remove(self, path: str) -> None:
        """Remove a file, or a directory and everything under it.

        Raises:
            ValueError: `path` is the root, which holds the whole workspace.
        """
        state = self._state()
        if posixpath.normpath(path) == STATE_ROOT:
            raise ValueError("refusing to remove the workspace's working directory")
        state.remove(path)

    async def exists(self, path: str) -> bool:
        """Whether a file or directory is at `path`."""
        return self._state().exists(path)

ref property

The document's id once it exists; None before the first operation.

working_dir() async

The document's root, which relative paths resolve against.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def working_dir(self) -> str:
    """The document's root, which relative paths resolve against."""
    self._state()
    return STATE_ROOT

read_bytes(path) async

A file's exact bytes.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def read_bytes(self, path: str) -> bytes:
    """A file's exact bytes."""
    return self._state().read_bytes(path)

write_bytes(path, data) async

Store a file, creating missing parents.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def write_bytes(self, path: str, data: bytes) -> None:
    """Store a file, creating missing parents."""
    self._state().write_bytes(path, data)

stat(path) async

Metadata for a file or directory.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def stat(self, path: str) -> FileEntry:
    """Metadata for a file or directory."""
    state = self._state()
    normal = posixpath.normpath(path)
    if state.is_dir(normal):
        return FileEntry(name=posixpath.basename(normal), path=normal, is_dir=True, size=None)
    return FileEntry(
        name=posixpath.basename(normal), path=normal, is_dir=False, size=state.size(normal)
    )

list_dir(path) async

The entries directly in a directory.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def list_dir(self, path: str) -> Sequence[FileEntry]:
    """The entries directly in a directory."""
    state = self._state()
    normal = posixpath.normpath(path)
    entries: list[FileEntry] = []
    for name, is_dir in state.list_dir(normal):
        child = posixpath.join(normal, name)
        size = None if is_dir else state.size(child)
        entries.append(FileEntry(name=name, path=child, is_dir=is_dir, size=size))
    return entries

make_dir(path) async

Create a directory and any missing parents.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def make_dir(self, path: str) -> None:
    """Create a directory and any missing parents."""
    self._state().make_dir(path)

remove(path) async

Remove a file, or a directory and everything under it.

Raises:

Type	Description
ValueError	path is the root, which holds the whole workspace.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def remove(self, path: str) -> None:
    """Remove a file, or a directory and everything under it.

    Raises:
        ValueError: `path` is the root, which holds the whole workspace.
    """
    state = self._state()
    if posixpath.normpath(path) == STATE_ROOT:
        raise ValueError("refusing to remove the workspace's working directory")
    state.remove(path)

exists(path) async

Whether a file or directory is at path.

Source code in src/pydantic_ai_backends/workspaces/_state.py

async def exists(self, path: str) -> bool:
    """Whether a file or directory is at `path`."""
    return self._state().exists(path)

Commands under the workspace contract

pydantic_ai_backends.types.CommandOutcome dataclass

How one CommandRunner.run_command call ended.

Exactly one of three shapes: finished (exit_code set, non-zero included), timed_out, or output_limited. The last two carry only the beginning of each stream and no exit code, because the command was stopped rather than allowed to finish. Undecodable bytes are replaced, never dropped.

Source code in src/pydantic_ai_backends/types.py

@dataclass(frozen=True)
class CommandOutcome:
    """How one `CommandRunner.run_command` call ended.

    Exactly one of three shapes: finished (`exit_code` set, non-zero included),
    `timed_out`, or `output_limited`. The last two carry only the beginning of
    each stream and no exit code, because the command was stopped rather than
    allowed to finish. Undecodable bytes are replaced, never dropped.
    """

    stdout: str
    stderr: str
    exit_code: int | None = None
    timed_out: bool = False
    output_limited: bool = False