类class agentscope.workspace

K8sBackend

class K8sBackend(BackendBase)

源码Source agentscope/workspace/_k8s/_k8s_backend.py:29–410

K 8 s Backend，在隔离环境中安全执行代码和工具。位于 agentscope.workspace 包，继承自 BackendBase。核心成员包括 exec_shell、read_file、write_file、write_stream。

Only the three abstract primitives (exec_shell, read_file, write_file) are implemented here; the derived filesystem helpers are inherited from :class:BackendBase.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(api_client: Any, namespace: str, pod_name: str, container_name: str, workdir: str) -> None`	初始化 `K8sBackend` 实例。 Initialize the K8s backend.

方法摘要Method Summary

方法Method	说明Description
`async def exec_shell(command: list[str], *, cwd: str \| None = None, timeout: float \| None = None) ->` `ExecResult`async	执行 `exec_shell` 方法。 Run a program inside the Pod via WebSocket ``exec``.
`async def read_file(path: str) -> bytes`async	设置读取文件。 Read a file from the Pod via tar stream extraction.
`async def write_file(path: str, data: bytes) -> None`async	执行写入文件相关操作。 Write raw bytes to a file inside the Pod via tar stream.
`async def write_stream(path: str, stream: AsyncIterator[bytes]) -> None`async	执行写入流相关操作。 Stream bytes into ``path`` inside the Pod.

构造器详情Constructor Detail

K8sBackend

def __init__(api_client: Any, namespace: str, pod_name: str, container_name: str, workdir: str) -> None

初始化 K8sBackend 实例。

Initialize the K8s backend.

参数Parameters

名称Name	类型Type	说明Description
`api_client`	Any	一种`kubernetes_asyncio.client.ApiClient`实例。A `kubernetes_asyncio.client.ApiClient` instance.
`namespace`	str	命名空间（类型 `str`）。Kubernetes namespace the Pod lives in.
`pod_name`	str	pod 名称（类型 `str`）。Name of the target Pod.
`container_name`	str	容器名称（类型 `str`）。Container name within the Pod.
`workdir`	str	workdir（类型 `str`）。Default working directory for `exec_shell` calls.

源码Source agentscope/workspace/_k8s/_k8s_backend.py:50–76

方法详情Method Detail

exec_shell

async def exec_shell(command: list[str], *, cwd: str | None = None, timeout: float | None = None) -> ExecResult

async

执行 exec_shell 方法。

Opens a multiplexed WebSocket channel with _preload_content=False and reads channel-tagged frames to separate stdout (channel 1), stderr (channel 2) and the exit status (channel 3).

参数Parameters

名称Name	类型Type	说明Description
`command`	list[str]	命令（类型 `list[str]`）。Executable path/name followed by its arguments.
`cwd`kw默认default `None`	str \| None	cwd（类型 `str \| None`）。Working directory inside the Pod. When `None` the backend's default `workdir` is used.
`timeout`kw默认default `None`	float \| None	超时时间（类型 `float \| None`）。Maximum number of seconds to wait. `None` waits indefinitely.

返回Returns ExecResult

返回ExecResult。`ExecResult`:

源码Source agentscope/workspace/_k8s/_k8s_backend.py:80–201

read_file

async def read_file(path: str) -> bytes

async

设置读取文件。

Runs tar cf - -C inside the Pod, collects the tar bytes from stdout, and extracts the file content.

参数Parameters

名称Name	类型Type	说明Description
`path`	str	路径（类型 `str`）。Path to the file inside the Pod.

返回Returns bytes

返回字节。`bytes`:

异常Raises

可能抛出 `Exception`。`FileNotFoundError`:
可能抛出 `Exception`。If the path does not exist inside the Pod.

源码Source agentscope/workspace/_k8s/_k8s_backend.py:205–258

write_file

async def write_file(path: str, data: bytes) -> None

async

执行写入文件相关操作。

Creates the parent directory first, then constructs an in-memory tar archive containing the file and pipes it to tar xf - -C inside the Pod.

参数Parameters

名称Name	类型Type	说明Description
`path`	str	路径（类型 `str`）。Destination path inside the Pod.
`data`	bytes	原始字节到写入。The raw bytes to write.

返回Returns None

无返回值。No value is returned.

异常Raises

可能抛出 `Exception`。`RuntimeError`:
可能抛出 `Exception`。If `tar xf` exits non-zero inside the Pod.

源码Source agentscope/workspace/_k8s/_k8s_backend.py:260–298

write_stream

async def write_stream(path: str, stream: AsyncIterator[bytes]) -> None

async

执行写入流相关操作。

Pipes to cat rather than tar because a tar header needs the member size up front, which a stream cannot give.

参数Parameters

名称Name	类型Type	说明Description
`path`	str	路径（类型 `str`）。Destination path inside the Pod.
`stream`	AsyncIterator[bytes]	流（类型 `AsyncIterator[bytes]`）。The chunks to write, in order.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_k8s/_k8s_backend.py:300–322

继承成员Inherited Members

继承自Inherited from BackendBase

join_path, dirname, basename, isabs, normpath, abspath, read_stream, getcwd, expanduser, file_exists, is_dir, list_dir, scandir, stat, stat_mtime, delete_path, os_name