类class agentscope.workspace

WorkspaceBase

class WorkspaceBase

源码Source agentscope/workspace/_base.py:225–1615

工作区基类，在隔离环境中安全执行代码和工具。位于 agentscope.workspace 包。核心成员包括 initialize、close、reset、get_backend、get_instructions、list_tools。

Subclasses provide concrete behaviour for one execution backend (local filesystem, Docker container, E2B sandbox). The base class owns:

lifecycle scaffolding (async with protocol, is_alive);
the canonical workspace layout derived from workdir (data/, skills/, sessions/, .mcp);
shared backend-driven implementations of offload, MCP persistence and a basic skill manager that subclasses can override (LocalWorkspace does, with a hash-indexed variant).

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__(*, workspace_id: str \| None = None, default_mcps: list[MCPClient] \| None = None, skill_paths: list[str] \|…`	初始化 `WorkspaceBase` 实例。 Initialise the shared workspace state.

属性摘要Property Summary

属性Property	说明Description
`is_persistent`	读取 `is_persistent` 属性。 Whether the workspace storage survives :meth:`close`.

属性摘要Attribute Summary

属性Attribute	类型Type	说明Description	值Value
`workspace_id`	str	工作区标识字段。	—
`workdir`	str	`workdir` 字段。	—
`is_alive`	bool	`is_alive` 字段。	—
`default_mcps`	list[MCPClient]	`default_mcps` 字段。	—
`skill_paths`	list[str]	`skill_paths` 字段。	—
`max_live_stateful_mcps`	int	`max_live_stateful_mcps` 字段。	—

方法摘要Method Summary

方法Method	说明Description
`async def initialize() -> None`asyncabstractabstractmethod	初始化 `WorkspaceBase` 实例。 Provision resources, connect MCP servers, copy skills.
`async def close() -> None`asyncabstractabstractmethod	关闭连接并释放相关资源。 Release all resources and connections.
`async def reset() -> None`async	清空或重置内部状态。 Reset the workspace to a clean state.
`def get_backend() ->` `BackendBase`	获取 `get_backend`。 Return the workspace's active filesystem/execution backend.
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Context manager support for ``async with``. Calls ``initialize()`` and returns the workspace instance.
`async def __aexit__(*exc: object) -> None`async	退出异步上下文并清理资源。 Context manager support for ``async with``. Calls ``close()`` and returns the workspace instance.
`async def get_instructions() -> str`asyncabstractabstractmethod	获取指令。 Workspace-specific system prompt fragment.
`async def list_tools() -> list[ToolBase]`async	列出当前连接可用的工具。 Built-in tools scoped to this workspace.
`async def list_mcps(*, agent_id: str \| None = None, session_id: str \| None = None) -> list[MCPClient]`async	列出 `list_mcps`。 Return the MCP clients for one agent/session.
`async def add_mcp(mcp_client:` `MCPClient, *, agent_id: str \| None = None, session_id: str \| None = None) -> MCPClient`asyncabstractabstractmethod	添加MCP。 Register a new MCP server for one agent/session and persist.
`async def remove_mcp(name: str, *, agent_id: str \| None = None, session_id: str \| None = None) -> None`asyncabstractabstractmethod	移除MCP。 Deregister an MCP by name within one agent/session.
`async def purge_session(*, agent_id: str, session_id: str) -> None`async	执行 `purge_session` 方法。 Drop everything this workspace holds for one session.
`async def purge_agent(*, agent_id: str) -> None`async	设置该参数。 Drop everything this workspace holds for one agent.
`async def offload_context(session_id: str, msgs: list[Msg]) -> str`async	执行卸载上下文相关操作。 Persist compressed context for agentic retrieval.
`async def offload_tool_result(session_id: str, tool_result:` `ToolResultBlock) -> str`async	执行卸载工具结果相关操作。 Persist a single tool result as a flat text file.
`async def offload_data_block(block:` `DataBlock) -> DataBlock`async	设置卸载数据块。 Persist a base64 :class:`DataBlock` under ``data/``.
`async def list_skills(*, agent_id: str \| None = None) -> list[Skill]`async	列出技能。 Enumerate the skills one agent can use.
`async def add_skill(skill_path: str, *, agent_id: str \| None = None) -> None`async	添加技能。 Copy a local skill directory into an agent's partition.
`async def add_skill_archive(stream: AsyncIterator[bytes], fmt: Literal['zip', 'tar', 'tar.gz'], dir_name: str, max_extr…`async	添加技能归档。 Install a skill from an archive stream into a partition.
`async def remove_skill(name: str, *, agent_id: str \| None = None) -> None`async	移除技能。 Remove a skill by its agent-facing ``name`` (front matter).

构造器详情Constructor Detail

WorkspaceBase

def __init__(*, workspace_id: str | None = None, default_mcps: list[MCPClient] | None = None, skill_paths: list[str] | None = None, max_live_stateful_mcps: int | None = None) -> None

初始化 WorkspaceBase 实例。

Subclasses must call super().__init__ and then set :attr:workdir themselves before any base-class method is invoked.

参数Parameters

名称Name	类型Type	说明Description
`workspace_id`kw默认default `None`	str \| None	工作区标识（类型 `str \| None`）。Existing identifier to adopt; `None` mints a fresh UUID.
`default_mcps`kw默认default `None`	list[MCPClient] \| None	默认 mcps（类型 `list[MCPClient] \| None`）。MCP clients seeded into any agent/session that has never added or removed an MCP of its own.
`skill_paths`kw默认default `None`	list[str] \| None	技能 paths（类型 `list[str] \| None`）。Local skill directories to copy into `skills/` on first start.
`max_live_stateful_mcps`kw默认default `None`	int \| None	最大 live stateful mcps（类型 `int \| None`）。Cap on concurrently live stateful MCP instances. `None` derives `max(40, 2 * )` so one session can always start its own seeded set.

源码Source agentscope/workspace/_base.py:297–346

属性详情Attribute Detail

workspace_id

str

工作区标识字段。声明为 str。

The workspace id field.

源码Source agentscope/workspace/_base.py:240

workdir

str

workdir 字段。声明为 str。

The workdir field.

源码Source agentscope/workspace/_base.py:243

is_alive

bool

is_alive 字段。声明为 bool。

The is alive field.

源码Source agentscope/workspace/_base.py:246

default_mcps

list[MCPClient]

default_mcps 字段。声明为 list[MCPClient]。

The default mcps field.

源码Source agentscope/workspace/_base.py:252

skill_paths

list[str]

skill_paths 字段。声明为 list[str]。

The skill paths field.

源码Source agentscope/workspace/_base.py:256

max_live_stateful_mcps

int

max_live_stateful_mcps 字段。声明为 int。

The max live stateful mcps field.

源码Source agentscope/workspace/_base.py:259

属性详情Property Detail

is_persistent

def is_persistent() -> bool

读取 is_persistent 属性。

Defaults to True.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/workspace/_base.py:467–475

方法详情Method Detail

initialize

async def initialize() -> None

asyncabstractabstractmethod

初始化 WorkspaceBase 实例。

Provision resources, connect MCP servers, copy skills.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:494–495

close

async def close() -> None

asyncabstractabstractmethod

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

Release all resources and connections.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:498–499

reset

async def reset() -> None

async

清空或重置内部状态。

Closes and removes all registered MCPs, deletes all skills, and wipes per-session state (offloaded context / tool results and any data files).

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:501–512

get_backend

def get_backend() -> BackendBase

获取 get_backend。

Docker and E2B workspaces may replace their backend when reconnecting, so callers should resolve it from the workspace when beginning an operation rather than retaining a stale private _backend value.

返回Returns BackendBase

返回BackendBase。Returns backend base.

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If the workspace has not been initialized or has no active backend.

源码Source agentscope/workspace/_base.py:514–531

aenter

async def __aenter__() -> Self

async

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

Context manager support for async with.

返回Returns Self

返回自身。Returns self.

源码Source agentscope/workspace/_base.py:533–539

aexit

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

async

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

Context manager support for async with.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:541–546

get_instructions

async def get_instructions() -> str

asyncabstractabstractmethod

获取指令。

Workspace-specific system prompt fragment.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/workspace/_base.py:551–552

list_tools

async def list_tools() -> list[ToolBase]

async

列出当前连接可用的工具。

Returns the six builtin tools (:class:Bash, :class:Edit, :class:Glob, :class:Grep, :class:Read, :class:Write), each bound to the workspace's active backend so that all filesystem and process I/O happens inside the workspace's execution environment.

返回Returns list[ToolBase]

返回ToolBase列表。Returns a list of tool base.

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If the workspace has not been initialised yet.

源码Source agentscope/workspace/_base.py:556–584

list_mcps

async def list_mcps(*, agent_id: str | None = None, session_id: str | None = None) -> list[MCPClient]

async

列出 list_mcps。

Instances are built on first use and are private to that agent/session, so stateful sessions (browser cookies, login state) never leak across agents or sessions.

参数Parameters

名称Name	类型Type	说明Description
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The owning agent. `None` means the legacy `""`. Keyword-only: both ids are plain strings, so passing them positionally could silently hit another session.
`session_id`kw默认default `None`	str \| None	会话标识（类型 `str \| None`）。The owning session. `None` means the legacy `""`.

返回Returns list[MCPClient]

返回MCPClient列表。Returns a list of MCP client.

源码Source agentscope/workspace/_base.py:605–659

add_mcp

async def add_mcp(mcp_client: MCPClient, *, agent_id: str | None = None, session_id: str | None = None) -> MCPClient

asyncabstractabstractmethod

添加MCP。

Implementations record the config in :attr:_mcp_specs, start one live instance under :meth:_enforce_mcp_capacity, and call :meth:_save_mcp_file.

参数Parameters

名称Name	类型Type	说明Description
`mcp_client`	MCPClient	MCP 到注册。The MCP to register.
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The owning agent. `None` means the legacy `""`. Keyword-only: both ids are plain strings, so passing them positionally could silently hit another session.
`session_id`kw默认default `None`	str \| None	会话标识（类型 `str \| None`）。The owning session. `None` means the legacy `""`.

返回Returns MCPClient

返回MCPClient。`MCPClient`:

异常Raises

可能抛出 `Exception`。`ValueError`:
可能抛出 `Exception`。If the name already exists for this agent/session.

源码Source agentscope/workspace/_base.py:664–696

remove_mcp

async def remove_mcp(name: str, *, agent_id: str | None = None, session_id: str | None = None) -> None

asyncabstractabstractmethod

移除MCP。

Other agents and sessions keep their own declarations and instances, so there is nothing to reconcile across them.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。MCP name to remove. Unknown names log a warning and return silently.
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The owning agent. `None` means the legacy `""`. Keyword-only: both ids are plain strings, so passing them positionally could silently hit another session.
`session_id`kw默认default `None`	str \| None	会话标识（类型 `str \| None`）。The owning session. `None` means the legacy `""`.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:699–721

purge_session

async def purge_session(*, agent_id: str, session_id: str) -> None

async

执行 purge_session 方法。

Closes the session's MCP instances, forgets its .mcp declaration and deletes its offload directory.

参数Parameters

名称Name	类型Type	说明Description
`agent_id`kw	str	智能体标识（类型 `str`）。The agent that owned the session.
`session_id`kw	str	会话标识（类型 `str`）。The session being deleted.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:725–768

purge_agent

async def purge_agent(*, agent_id: str) -> None

async

设置该参数。

Deletes the agent's skill partition.

参数Parameters

名称Name	类型Type	说明Description
`agent_id`kw	str	智能体标识（类型 `str`）。The agent being deleted.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/workspace/_base.py:770–795

offload_context

async def offload_context(session_id: str, msgs: list[Msg]) -> str

async

执行卸载上下文相关操作。

Appends every message in msgs to ${workdir}/sessions/ /context.jsonl (one message per JSONL line).

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。Session-scope key used to partition offloaded data (one subdirectory per session).
`msgs`	list[Msg]	消息列表（类型 `list[Msg]`）。Conversation messages to offload. Not mutated — a deep copy is used internally.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/workspace/_base.py:1013–1092

offload_tool_result

async def offload_tool_result(session_id: str, tool_result: ToolResultBlock) -> str

async

执行卸载工具结果相关操作。

Writes ${workdir}/sessions/ /tool_result- .txt.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。Session-scope key used to partition offloaded data.
`tool_result`	ToolResultBlock	工具结果（类型 `ToolResultBlock`）。The tool result block to persist.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/workspace/_base.py:1094–1151

offload_data_block

async def offload_data_block(block: DataBlock) -> DataBlock

async

设置卸载数据块。

The decoded payload is stored at `${workdir}/data/ .

参数Parameters

名称Name	类型Type	说明Description
`block`	DataBlock	内容块（类型 `DataBlock`）。A data block. Blocks already backed by a :class:`URLSource` are returned unchanged.

返回Returns DataBlock

返回数据块。`DataBlock`:

源码Source agentscope/workspace/_base.py:1153–1200

list_skills

async def list_skills(*, agent_id: str | None = None) -> list[Skill]

async

列出技能。

Reads the agent's own partition — equipping it from the seed template if this is the agent's first appearance — parses every skills/ / /SKILL.md's YAML front matter, and yields one :class:Skill per file that has both name and description.

参数Parameters

名称Name	类型Type	说明Description
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The agent asking. `None` reads the default partition, which is where an SDK caller driving the workspace without agents puts everything.

返回Returns list[Skill]

返回技能列表。`list[Skill]`:

源码Source agentscope/workspace/_base.py:1204–1266

add_skill

async def add_skill(skill_path: str, *, agent_id: str | None = None) -> None

async

添加技能。

Tars the directory on the host, writes the archive to the backend's tmp area, and extracts it via python3 -c inside the sandbox — two round trips regardless of skill size, and portable across any backend whose image ships python3 (same contract as the gateway shim).

参数Parameters

名称Name	类型Type	说明Description
`skill_path`	str	技能路径（类型 `str`）。Path to a skill directory on the local filesystem.
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The agent taking ownership. `None` installs into the default partition.

返回Returns None

无返回值。No value is returned.

异常Raises

ValueError 可能抛出 `ValueError`。If `SKILL.md` is missing or a directory with the same basename already exists in the partition.
RuntimeError 可能抛出 `RuntimeError`。If extraction inside the sandbox fails.

源码Source agentscope/workspace/_base.py:1268–1344

add_skill_archive

async def add_skill_archive(stream: AsyncIterator[bytes], fmt: Literal['zip', 'tar', 'tar.gz'], dir_name: str, max_extracted_bytes: int = DEFAULT_MAX_EXTRACTED_BYTES, *, agent_id: str | None = None) -> None

async

添加技能归档。

The archive is piped straight into the sandbox and expanded there, so the server never holds it whole.

参数Parameters

名称Name	类型Type	说明Description
`stream`	AsyncIterator[bytes]	归档字节,in 顺序。The archive bytes, in order.
`fmt`	Literal['zip', 'tar', 'tar.gz']	归档格式。The archive format.
`dir_name`	str	目录名称（类型 `str`）。Directory name to install as. A numeric suffix is appended when it is taken.
`max_extracted_bytes`默认default `DEFAULT_MAX_EXTRACTED_BYTES`	int	最大 extracted 字节（类型 `int`）。Ceiling on the archive's expanded size.
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The agent taking ownership. `None` installs into the default partition.

返回Returns None

无返回值。No value is returned.

异常Raises

ValueError 可能抛出 `ValueError`。If `dir_name` is not a plain name, or the archive holds no `SKILL.md`.
RuntimeError 可能抛出 `RuntimeError`。If expanding the archive inside the sandbox fails.

源码Source agentscope/workspace/_base.py:1346–1441

remove_skill

async def remove_skill(name: str, *, agent_id: str | None = None) -> None

async

移除技能。

Looks up the skill via :meth:list_skills and rm -rf-style deletes its directory through the backend.

参数Parameters

名称Name	类型Type	说明Description
`name`	str	名称（类型 `str`）。The agent-facing name of the skill to remove.
`agent_id`kw默认default `None`	str \| None	智能体标识（类型 `str \| None`）。The agent asking. `None` searches every partition.

返回Returns None

无返回值。No value is returned.

异常Raises

KeyError 当If 技能是不找到 in 工作区时抛出 `KeyError`。If the skill is not found in the workspace.

源码Source agentscope/workspace/_base.py:1496–1532