agentscope.app.workspace_managerE2BWorkspaceManager
class E2BWorkspaceManager(WorkspacePrewarmMixin[E2BWorkspace], WorkspaceManagerBase)
agentscope/app/workspace_manager/_e2b_workspace_manager.py:55–479E2B 工作区管理器,承载智能体的部署、调度与服务化运行。位于 agentscope.app.workspace_manager 包,继承自 WorkspacePrewarmMixin[E2BWorkspace], WorkspaceManagerBase。核心成员包括 get_workspace、create_workspace、close、close_all。
Use the manager as an async with context manager: entering it
starts the TTL sweeper task, exiting it stops the sweeper and then
closes every cached workspace via :meth:close_all.
构造器摘要Constructor Summary
| 构造器Constructor | 说明Description |
|---|---|
def __init__(*, isolation: IsolationPolicy = IsolationPolicy.PER_AGENT, template: str = DEFAULT_TEMPLATE, api_key: str… | 初始化 `E2BWorkspaceManager` 实例。 Initialize the E2B workspace manager. |
方法摘要Method Summary
| 方法Method | 说明Description |
|---|---|
async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str | None = None) -> E2BWorkspaceasync | 获取工作区。 Return an initialised workspace, reattaching on cache miss. |
async def create_workspace(user_id: str, agent_id: str, session_id: str) -> E2BWorkspaceasyncdeprecateddeprecated | 创建工作区。 Build a brand-new workspace and track it. |
async def close(workspace_id: str) -> Noneasync | 关闭连接并释放相关资源。 Close (= pause the sandbox) and evict a single workspace. |
async def close_all() -> Noneasync | 关闭所有。 Close every cached workspace in parallel. |
async def __aenter__() -> Selfasync | 进入异步上下文并返回自身。 Start the TTL sweeper task and fill the pre-warm buffer. |
async def __aexit__(*exc: object) -> Noneasync | 退出异步上下文并清理资源。 Stop the sweeper and buffer, then close every workspace. |
构造器详情Constructor Detail
E2BWorkspaceManager
def __init__(*, isolation: IsolationPolicy = IsolationPolicy.PER_AGENT, template: str = DEFAULT_TEMPLATE, api_key: str = '', domain: str = '', timeout_seconds: int = DEFAULT_TIMEOUT, gateway_port: int = DEFAULT_GATEWAY_PORT, env: dict[str, str] | None = None, sandbox_metadata: dict[str, str] | None = None, extra_pip: list[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, prewarm: PrewarmConfig | None = None) -> None
初始化 E2BWorkspaceManager 实例。
Initialize the E2B workspace manager.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
isolationkw默认default IsolationPolicy.PER_AGENT | IsolationPolicy | 隔离(类型 `IsolationPolicy`)。Isolation grain for :meth:`assign_workspace_id`. See :class:`DockerWorkspaceManager` for semantics. |
templatekw默认default DEFAULT_TEMPLATE | str | 模板(类型 `str`)。E2B template id passed to every workspace this manager produces. Defaults to `"base"`. |
api_keykw默认default '' | str | API 密钥(类型 `str`)。E2B API key. `""` falls back to the `E2B_API_KEY` env var on the SDK side. |
domainkw默认default '' | str | domain(类型 `str`)。Optional custom E2B domain (self-hosted etc.). |
timeout_secondskw默认default DEFAULT_TIMEOUT | int | 超时秒数(类型 `int`)。Sandbox keep-alive timeout passed to `AsyncSandbox.create` / `AsyncSandbox.connect`. |
gateway_portkw默认default DEFAULT_GATEWAY_PORT | int | 网关 port(类型 `int`)。TCP port the in-sandbox gateway listens on. |
envkw默认default None | dict[str, str] | None | env(类型 `dict[str, str] | None`)。Environment variables baked into the sandbox at create time. |
sandbox_metadatakw默认default None | dict[str, str] | None | 沙箱元数据(类型 `dict[str, str] | None`)。Extra metadata merged with the per-workspace `agentscope.workspace.id` / `agentscope.user.id` / `agentscope.agent.id` keys. Useful for downstream E2B dashboard filtering. |
extra_pipkw默认default None | list[str] | None | extra pip(类型 `list[str] | None`)。Extra Python packages to install into the gateway venv during bootstrap. |
default_mcpskw默认default None | list[MCPClient] | None | 默认 mcps(类型 `list[MCPClient] | None`)。MCP clients seeded into brand-new workspaces. Ignored on subsequent reattachments — the sandbox's persisted `.mcp` file wins. |
skill_pathskw默认default None | list[str] | None | 技能 paths(类型 `list[str] | None`)。Skill directories seeded into brand-new workspaces. |
ttlkw默认default 3600.0 | float | 存活时间(类型 `float`)。Seconds before an idle cached workspace is evicted and its sandbox paused. |
sweep_intervalkw默认default DEFAULT_SWEEP_INTERVAL | float | sweep 间隔(类型 `float`)。How often (seconds) the background sweeper wakes up to look for idle workspaces. Defaults to 5 minutes. |
prewarmkw默认default None | PrewarmConfig | None | 预热(类型 `PrewarmConfig | None`)。Keep this many sandboxes created and idle, ready to be handed to the next session that needs one. `None` disables pre-warming. Idle sandboxes stay running, so they bill while they wait — keep the buffer small. |
agentscope/app/workspace_manager/_e2b_workspace_manager.py:66–150方法详情Method Detail
get_workspace
async def get_workspace(user_id: str, agent_id: str, session_id: str, workspace_id: str | None = None) -> E2BWorkspace
获取工作区。
On miss the manager calls E2BWorkspace(workspace_id=…) and relies on its initialize to find any existing sandbox via AsyncSandbox.list(query=SandboxQuery(metadata=...)) and connect to it (auto-resuming if paused) — or to create a fresh sandbox otherwise.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
user_id | str | 用户标识(类型 `str`)。Owning user identifier (forwarded as sandbox metadata only — not part of the cache key). |
agent_id | str | 智能体标识(类型 `str`)。Agent identifier (forwarded as sandbox metadata only — not part of the cache key). |
session_id | str | 会话标识(类型 `str`)。Session identifier (unused; sandboxes are per-workspace, sessions partition under `sessions/ /`). |
workspace_id默认default None | str | None | 工作区标识(类型 `str | None`)。Stable workspace identifier — the cache key and the value stored in the sandbox's `agentscope.workspace.id` metadata. When unset the manager falls back to :meth:`assign_workspace_id`. |
agentscope/app/workspace_manager/_e2b_workspace_manager.py:263–335create_workspace
async def create_workspace(user_id: str, agent_id: str, session_id: str) -> E2BWorkspace
创建工作区。
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.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
user_id | str | 用户标识(类型 `str`)。Owning user identifier (forwarded as sandbox metadata). |
agent_id | str | 智能体标识(类型 `str`)。Agent identifier (forwarded as sandbox metadata). |
session_id | str | 会话标识(类型 `str`)。Session identifier (accepted for parity; not used here). |
agentscope/app/workspace_manager/_e2b_workspace_manager.py:342–377close
async def close(workspace_id: str) -> None
关闭连接并释放相关资源。
No-op when the workspace_id is not tracked.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
workspace_id | str | 工作区到关闭。The workspace to close. |
agentscope/app/workspace_manager/_e2b_workspace_manager.py:379–393close_all
async def close_all() -> None
关闭所有。
sandbox.pause() is a remote round-trip per sandbox; doing it sequentially on app shutdown produces a noticeable stall, so we fan the calls out with :func:asyncio.gather.
agentscope/app/workspace_manager/_e2b_workspace_manager.py:395–410__aenter__
async def __aenter__() -> Self
进入异步上下文并返回自身。
Start the TTL sweeper task and fill the pre-warm buffer.
agentscope/app/workspace_manager/_e2b_workspace_manager.py:414–419__aexit__
async def __aexit__(*exc: object) -> None
退出异步上下文并清理资源。
Stop the sweeper and buffer, then close every workspace.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
exc | object | exc(类型 `object`)。The exc. |
agentscope/app/workspace_manager/_e2b_workspace_manager.py:421–431