类class agentscope.app.workspace_manager

K8sWorkspaceManager

class K8sWorkspaceManager(WorkspaceManagerBase)

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:36–354

K 8 s 工作区管理器，承载智能体的部署、调度与服务化运行。位于 agentscope.app.workspace_manager 包，继承自 WorkspaceManagerBase。核心成员包括 get_workspace、create_workspace、close、close_all。

Use as an async with context manager: entering starts the TTL sweeper, exiting stops it and closes every cached workspace.

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(*, isolation:` `IsolationPolicy = IsolationPolicy.PER_AGENT, namespace: str = 'agentscope', kubeconfig: str…`	初始化 `K8sWorkspaceManager` 实例。 Initialize the K8s workspace manager.

方法摘要Method Summary

方法Method	说明Description
`async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str \| None = None) ->` `K8sWorkspace`async	获取工作区。 Return an initialised workspace, reattaching on cache miss.
`async def create_workspace(user_id: str, agent_id: str, session_id: str) ->` `K8sWorkspace`asyncdeprecateddeprecated	创建工作区。 Build a brand-new workspace and track it.
`async def close(workspace_id: str) -> None`async	关闭连接并释放相关资源。 Close and evict a single workspace.
`async def close_all() -> None`async	关闭所有。 Close every cached workspace in parallel.
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Start the TTL sweeper task.
`async def __aexit__(*exc: object) -> None`async	退出异步上下文并清理资源。 Stop the TTL sweeper, then close every cached workspace.

构造器详情Constructor Detail

K8sWorkspaceManager

def __init__(*, isolation: IsolationPolicy = IsolationPolicy.PER_AGENT, namespace: str = 'agentscope', kubeconfig: str | None = None, image: str = DEFAULT_IMAGE, image_pull_policy: str = 'IfNotPresent', image_pull_secrets: list[str] | None = None, resources: dict[str, Any] | None = None, node_selector: dict[str, str] | None = None, tolerations: list[dict[str, Any]] | None = None, service_account: str | None = None, gateway_port: int = DEFAULT_GATEWAY_PORT, extra_pip: list[str] | None = None, storage_class: str | None = None, storage_size: str = '1Gi', env: dict[str, str] | None = None, default_mcps: list[MCPClient] | None = None, skill_paths: list[str] | None = None, ttl: float = 3600.0, sweep_interval: float = DEFAULT_SWEEP_INTERVAL, delete_pvc_on_close: bool = False) -> None

初始化 K8sWorkspaceManager 实例。

Initialize the K8s workspace manager.

参数Parameters

名称Name	类型Type	说明Description
`isolation`kw默认default `IsolationPolicy.PER_AGENT`	IsolationPolicy	隔离（类型 `IsolationPolicy`）。Isolation grain for :meth:`assign_workspace_id`. See :class:`DockerWorkspaceManager` for semantics.
`namespace`kw默认default `'agentscope'`	str	命名空间（类型 `str`）。K8s namespace for workspace Pods and PVCs.
`kubeconfig`kw默认default `None`	str \| None	kubeconfig（类型 `str \| None`）。Path to kubeconfig file. `None` uses in-cluster config.
`image`kw默认default `DEFAULT_IMAGE`	str	图像（类型 `str`）。Container image for workspace Pods.
`image_pull_policy`kw默认default `'IfNotPresent'`	str	图像 pull 策略（类型 `str`）。K8s imagePullPolicy.
`image_pull_secrets`kw默认default `None`	list[str] \| None	图像 pull secrets（类型 `list[str] \| None`）。Names of K8s image pull secrets.
`resources`kw默认default `None`	dict[str, Any] \| None	resources（类型 `dict[str, Any] \| None`）。K8s ResourceRequirements dict.
`node_selector`kw默认default `None`	dict[str, str] \| None	节点 selector（类型 `dict[str, str] \| None`）。K8s nodeSelector.
`tolerations`kw默认default `None`	list[dict[str, Any]] \| None	tolerations（类型 `list[dict[str, Any]] \| None`）。K8s tolerations list.
`service_account`kw默认default `None`	str \| None	服务 account（类型 `str \| None`）。K8s serviceAccountName.
`gateway_port`kw默认default `DEFAULT_GATEWAY_PORT`	int	网关 port（类型 `int`）。Port the gateway listens on inside the Pod.
`extra_pip`kw默认default `None`	list[str] \| None	extra pip（类型 `list[str] \| None`）。Extra packages for the gateway venv.
`storage_class`kw默认default `None`	str \| None	存储类（类型 `str \| None`）。K8s StorageClass name.
`storage_size`kw默认default `'1Gi'`	str	存储大小（类型 `str`）。PVC size.
`env`kw默认default `None`	dict[str, str] \| None	env（类型 `dict[str, str] \| None`）。Environment variables for workspace containers.
`default_mcps`kw默认default `None`	list[MCPClient] \| None	默认 mcps（类型 `list[MCPClient] \| None`）。MCPs seeded into new workspaces.
`skill_paths`kw默认default `None`	list[str] \| None	技能 paths（类型 `list[str] \| None`）。Skill directories seeded into new workspaces.
`ttl`kw默认default `3600.0`	float	存活时间（类型 `float`）。Seconds before an idle workspace is evicted.
`sweep_interval`kw默认default `DEFAULT_SWEEP_INTERVAL`	float	sweep 间隔（类型 `float`）。How often the sweeper runs.
`delete_pvc_on_close`kw默认default `False`	bool	删除 pvc on 关闭（类型 `bool`）。Default `delete_pvc_on_close` value for newly created workspaces.

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:43–138

方法详情Method Detail

get_workspace

async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str | None = None) -> K8sWorkspace

async

获取工作区。

Return an initialised workspace, reattaching on cache miss.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。Owning user identifier.
`agent_id`	str	智能体标识（类型 `str`）。Agent identifier.
`session_id`	str	会话标识（类型 `str`）。Session identifier (unused; Pods are per-workspace).
`workspace_id`默认default `None`	str \| None	工作区标识（类型 `str \| None`）。Stable workspace identifier — the cache key. When unset the manager falls back to :meth:`assign_workspace_id`.

返回Returns K8sWorkspace

返回K8sWorkspace。`K8sWorkspace`:

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:173–225

create_workspace

async def create_workspace(user_id: str, agent_id: str, session_id: str) -> K8sWorkspace

asyncdeprecateddeprecated

创建工作区。

Build a brand-new workspace and track it.

说明Notes

注意：该行为受运行时配置或底层实现影响，建议结合对应源码与版本说明确认。Deprecated: Use :meth:`get_workspace` with `workspace_id=None` — it falls back to :meth:`assign_workspace_id` under the manager's isolation policy and reuses the cache path.

已弃用。Deprecated. 已弃用，请改用受支持的替代 API。Use :meth:`get_workspace` with `workspace_id=None` — it falls back to :meth:`assign_workspace_id` under the manager's isolation policy and reuses the cache path.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。Owning user identifier.
`agent_id`	str	智能体标识（类型 `str`）。Agent identifier.
`session_id`	str	会话标识（类型 `str`）。Session identifier (unused).

返回Returns K8sWorkspace

返回K8sWorkspace。`K8sWorkspace`:

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:232–262

close

async def close(workspace_id: str) -> None

async

关闭连接并释放相关资源。

Close and evict a single workspace.

参数Parameters

名称Name	类型Type	说明Description
`workspace_id`	str	工作区到关闭。The workspace to close.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:264–276

close_all

async def close_all() -> None

async

关闭所有。

Close every cached workspace in parallel.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:278–288

aenter

async def __aenter__() -> Self

async

进入异步上下文并返回自身。

Start the TTL sweeper task.

返回Returns Self

返回自身。Returns self.

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:292–296

aexit

async def __aexit__(*exc: object) -> None

async

退出异步上下文并清理资源。

Stop the TTL sweeper, then close every cached workspace.

参数Parameters

名称Name	类型Type	说明Description
`exc`	object	exc（类型 `object`）。The exc.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/workspace_manager/_k8s_workspace_manager.py:298–307

继承成员Inherited Members

继承自Inherited from WorkspaceManagerBase

bind_storage, assign_workspace_id