> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.

# 工作区管理

> 在服务中分配、管理智能体的运行环境（Workspace）

**工作区管理器**（Workspace Manager）持有隔离策略，并管理智能体服务中每个智能体的 [`Workspace`](/versions/2.0.5dev/zh/building-blocks/workspace) 实例的生命周期。它是 `agentscope.workspace` 模块在服务侧的对应实现 —— 每一种 `WorkspaceBase` 子类都有一个配套的 `WorkspaceManagerBase` 的子类来负责创建、缓存与删除。

它的特点：

* **可配置的隔离粒度** —— `PER_AGENT`（默认）、`PER_SESSION` 或 `PER_USER` —— 决定 workspace 在 `(user_id, agent_id, session_id)` 三个维度上如何共享或隔离，让同一套部署可以干净地服务多个租户。
* **可插拔的沙箱后端** —— 同一份生命周期契约既可以包裹本地目录、Docker 容器、E2B 云沙箱，也可以对应 Kubernetes Pod。切换后端只需在 `create_app` 中改一行。

## 可选的实现

| 类                        | 说明                                                                             |
| ------------------------ | ------------------------------------------------------------------------------ |
| `LocalWorkspaceManager`  | 直接落在宿主目录下的裸机式 workspace。零基础设施、零沙箱化 —— 智能体直接跑在开发者的宿主文件系统上。                      |
| `DockerWorkspaceManager` | 每个 workspace 对应一个 Docker 容器，宿主目录以 bind mount 的方式挂进去做持久化。需要一个可达的 Docker daemon。 |
| `E2BWorkspaceManager`    | 通过 [E2B](https://e2b.dev) 托管的云沙箱。空闲时自动挂起，下一次访问时恢复。                             |
| `K8sWorkspaceManager`    | 在 Kubernetes 集群上为每个 workspace 起一个 Pod + PVC。适合已经在跑其他负载的生产集群。                   |

<Tip>
  想深入了解 workspace *究竟是什么*（文件系统布局、gateway、MCP 接线、内置工具），请看 [Workspace](/versions/2.0.5dev/zh/building-blocks/workspace) 章节。
</Tip>

## 接入智能体服务

在 `create_app` 入口处设置对应的 `WorkspaceManager` 实例即可。下面是四种内置实现的示例：

<CodeGroup>
  ```python Local theme={null}
  from agentscope.app import create_app
  from agentscope.app.workspace_manager import (
      IsolationPolicy,
      LocalWorkspaceManager,
  )

  workspace_manager = LocalWorkspaceManager(
      # 宿主根目录；每个 agent 的工作目录落在 `<basedir>/<user_id>/<agent_id>` 下。
      basedir="/data/workspaces",
      # `PER_AGENT`（默认）：同一 agent 的所有 session 共享一个 workspace。
      # `PER_SESSION` / `PER_USER` 改变粒度，详见下文。
      isolation=IsolationPolicy.PER_AGENT,
  )

  app = create_app(
      # ...existing code...
      workspace_manager=workspace_manager,
  )
  ```

  ```python Docker theme={null}
  from agentscope.app import create_app
  from agentscope.app.workspace_manager import (
      DockerWorkspaceManager,
      IsolationPolicy,
  )

  workspace_manager = DockerWorkspaceManager(
      # 宿主根目录，会 bind mount 到每个容器内的 `/workspace`。
      basedir="/data/workspaces",
      isolation=IsolationPolicy.PER_AGENT,
      # 基础镜像，必须自带 `python3`。镜像按内容哈希，
      # 只有 Dockerfile 输入变化时才会重新构建。
      base_image="python:3.11-slim",
      # 镜像里内置的 Node.js 主版本号（基于 npx 的 MCP 会用到）。
      node_version="20",
      # 空闲容器在被 sweeper 清掉前的缓存时间（秒）。
      ttl=3600.0,
  )

  app = create_app(
      # ...existing code...
      workspace_manager=workspace_manager,
  )
  ```

  ```python E2B theme={null}
  from agentscope.app import create_app
  from agentscope.app.workspace_manager import (
      E2BWorkspaceManager,
      IsolationPolicy,
  )

  workspace_manager = E2BWorkspaceManager(
      isolation=IsolationPolicy.PER_AGENT,
      # E2B template id，模板里要自带 agent 所需的运行时。
      template="base",
      # 传 `""` 时会回退去读环境变量 `E2B_API_KEY`。
      api_key="",
      # E2B 侧的沙箱 keep-alive 超时时间。
      timeout_seconds=300,
      ttl=3600.0,
  )

  app = create_app(
      # ...existing code...
      workspace_manager=workspace_manager,
  )
  ```

  ```python Kubernetes theme={null}
  from agentscope.app import create_app
  from agentscope.app.workspace_manager import (
      IsolationPolicy,
      K8sWorkspaceManager,
  )

  workspace_manager = K8sWorkspaceManager(
      isolation=IsolationPolicy.PER_AGENT,
      # 存放 workspace Pod 与 PVC 的 K8s namespace。
      namespace="agentscope",
      # 传 `None` 用集群内配置；集群外使用时传一个 kubeconfig 路径。
      kubeconfig=None,
      # 每个 Pod 的容器镜像。
      image="python:3.11-slim",
      # 为 workspace 文件系统兜底的 PVC 大小。
      storage_size="1Gi",
      ttl=3600.0,
  )

  app = create_app(
      # ...existing code...
      workspace_manager=workspace_manager,
  )
  ```
</CodeGroup>

## 隔离粒度

`isolation` 决定 workspace 在 `(user_id, agent_id, session_id)` 三元组上如何共享或隔离。Manager 会在 session 创建时按所选策略生成一个 `workspace_id`；后续所有带着同一个 `workspace_id` 的请求都会落到同一个底层 workspace 上。

| 取值              | 共享规则                                                  | 典型场景                                                              |
| --------------- | ----------------------------------------------------- | ----------------------------------------------------------------- |
| `PER_AGENT`（默认） | 同一 `(user_id, agent_id)` 下的所有 session 共享一个 workspace  | 让每个 agent 拥有一份持久化的、按用户区分的工作目录 —— 文件、skill、MCP 注册跨 session 也能保留下来。 |
| `PER_SESSION`   | 每个 session 一个独立的 workspace                            | 需要保证 session 之间互不泄露状态的场景 —— 例如一次性沙箱评测、短生命周期的自动化任务。                |
| `PER_USER`      | 同一 `user_id` 下的所有 session 共享一个 workspace，不区分是哪个 agent | 同一用户的多个 agent 在同一份文件系统上协作（少见，谨慎使用）。                               |

<Note>
  在 session 创建请求中显式指定 `workspace_id` 永远会覆盖策略。这就是内置团队工具（`AgentCreate` / `AgentInvite`）用来让子 agent 与团队 leader 共用同一个 workspace 的机制。
</Note>

## 在智能体服务中的工作原理

工作区管理器是一个**应用级单例资源**：一个智能体服务进程只有一份实例，所有请求共享。它的职责有两个：（a）在配置的 `isolation` 策略下*判断*一次请求属于哪个 workspace；（b）把该 workspace 作为一个廉价依赖交给需要它的路由和服务。

### 接线方式

开发者传给 `create_app(workspace_manager=...)` 的实例会被挂到 `app.state` 上，并通过一个 FastAPI 依赖对外暴露：

```python theme={null}
# agentscope/app/_app.py
app.state.workspace_manager = workspace_manager

# agentscope/app/deps.py
async def get_workspace_manager(request: Request) -> WorkspaceManagerBase:
    return request.app.state.workspace_manager
```

所有会碰到 workspace 的 router、service 与内置工具都通过这个依赖来消费它（服务类则通过构造函数注入）：

* **Router** —— `/session`、`/workspace/*`（MCP、skill）。它们调用 `Depends(get_workspace_manager)`，然后用 `manager.get_workspace(...)` 解析出调用方对应的 workspace。
* **`ChatService`** —— 在 lifespan 里通过构造函数拿到 manager，每次 chat 运行时调用 `get_workspace` 拿到 agent 将要作用其中的 workspace。
* **内置团队工具**（`AgentCreate` / `AgentInvite`）—— 拿到 manager 后在派生子 agent session 时调用 `assign_workspace_id(...)`，让子 agent 落到与 leader 相同的 workspace 里。

### 生命周期

Manager 是一个**异步上下文管理器**。应用的 `lifespan` 会在启动时进入它，在关闭时通过一个 `AsyncExitStack` 退出它，所以它所拥有的任何后台机制（例如 `DockerWorkspaceManager` / `E2BWorkspaceManager` / `K8sWorkspaceManager` 的空闲 TTL sweeper）都会与服务的其他部分同步启停：

```python theme={null}
# agentscope/app/_lifespan.py（节选）
workspace_manager = app.state.workspace_manager
async with AsyncExitStack() as stack:
    await stack.enter_async_context(storage)
    await stack.enter_async_context(message_bus)
    await stack.enter_async_context(workspace_manager)  # ← 就是这里
    ...
    yield
# ← 退出时，`__aexit__` 会调用 `close_all()` 拆掉每一个缓存中的 workspace。
```

### 端到端请求流程

服务跑起来之后，一次典型的交互长这样：

1. **创建 session** —— `/session` router 只计算一次 workspace id 并把它写到 session 记录上：

   ```python theme={null}
   # agentscope/app/_router/_session.py（节选）
   resolved_workspace_id = body.workspace_id or (
       workspace_manager.assign_workspace_id(
           user_id=user_id,
           agent_id=agent_id,
           session_id=session_id,
       )
   )
   ```

   显式传入的 `body.workspace_id` 永远优先（团队工具就是靠它让子 agent 共享 leader 的 workspace）；否则由 manager 按配置的 `isolation` 策略生成一个。`assign_workspace_id` 是一个纯函数 —— 不做 I/O，不查缓存 —— 所以在热路径上调用非常便宜。

2. **后续每个请求**带着这个 session —— 无论是 chat 运行、MCP 注册还是 skill 上传 —— 都会从 session 记录里读出保存好的 `workspace_id`，然后请求 manager 把它落到具体的 workspace 上：

   ```python theme={null}
   workspace = await workspace_manager.get_workspace(
       user_id, agent_id, session_id, session_record.config.workspace_id,
   )
   ```

   `get_workspace` 是缓存命中的路径。它在 manager 内部的 `dict` 里查 `workspace_id`，刷新一下最后访问时间，就在 O(1) 内返回已初始化好的 `Workspace`。只有 miss 的时候才会抢一把 `asyncio.Lock`，把后端准备好（构建 Docker 镜像、启动 E2B 沙箱、创建 K8s Pod……），再把结果缓存起来 —— 这把锁能防止多个请求同时冲进来给同一个 id 起两份后端。

3. **空闲驱逐** —— 沙箱类的 manager 会跑一个后台 sweeper（在 `__aenter__` 里启动），周期性遍历缓存，把最后访问时间超过 `ttl` 的 workspace 关闭并踢出。下一次访问该 id 时会透明地重新准备一份。

4. **关闭** —— lifespan 退出 manager，manager 调用 `close_all()` 拆掉所有缓存中的 workspace 及其后台 sweeper。

<Note>
  由于 `workspace_id` **只在 session 创建时生成一次**并持久化，运行中的部署改动 `isolation` 策略并**不会**对现存 session 重新分组 —— 它们会继续保留当初分配的 id。只有新 session 才会走新策略。
</Note>

## 核心 API

每个 manager 都要实现来自 `WorkspaceManagerBase` 的以下契约。自定义子类只需要补齐这些抽象方法即可；`assign_workspace_id` 里的隔离逻辑由基类提供，并由 `isolation` 构造参数驱动。

| 方法                                                                                 | 用途                                                                                                                             |
| ---------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ |
| `assign_workspace_id(*, user_id, agent_id, session_id) -> str`                     | 在配置的 `isolation` 策略下为一个新 session 生成 workspace id。纯函数，无 I/O。当客户端在 session 创建请求中省略了 `workspace_id` 时会被调用。                        |
| `get_workspace(user_id, agent_id, session_id, workspace_id=None) -> WorkspaceBase` | 返回一个绑定到 `workspace_id` 的已初始化 workspace。热路径上的缓存命中路径；miss 时 manager 会准备底层后端并缓存结果。`workspace_id=None` 会回退到 `assign_workspace_id`。 |
| `create_workspace(user_id, agent_id, session_id) -> WorkspaceBase`                 | 全新准备一个 workspace 并纳入管理。调用方还没有可复用的 id 时使用。                                                                                      |
| `close(workspace_id)`                                                              | 把某一个 workspace 从缓存中踢掉，并拆掉它的后端。                                                                                                 |
| `close_all()`                                                                      | 踢掉所有缓存的 workspace —— 服务关闭时调用。                                                                                                  |
| `async with manager: ...`                                                          | 进入 / 退出 manager 的生命周期。进入时启动后台机制（例如 TTL sweeper）；退出时调用 `close_all`。                                                             |

## 自定义 Manager

任何基于 [`WorkspaceBase`](/versions/2.0.5dev/zh/building-blocks/workspace) 构建的 workspace 类，都可以通过配一个 `WorkspaceManagerBase` 子类的方式接到智能体服务里。大多数情况下只需要写一小段准备 + 缓存的样板代码 —— 隔离策略由基类自动继承。

```python 自定义 Manager theme={null}
import asyncio
import time
from typing import Self

from agentscope.app.workspace_manager import (
    IsolationPolicy,
    WorkspaceManagerBase,
)
from agentscope.workspace import WorkspaceBase


class MyWorkspaceManager(WorkspaceManagerBase):
    """封装自定义 `MyWorkspace` 后端的轻量 manager。"""

    def __init__(
        self,
        *,
        isolation: IsolationPolicy = IsolationPolicy.PER_AGENT,
        ttl: float = 3600.0,
    ) -> None:
        super().__init__(isolation=isolation)
        self._ttl = ttl
        # workspace_id -> (workspace, last_access_monotonic)
        self._cache: dict[str, tuple[WorkspaceBase, float]] = {}
        self._lock = asyncio.Lock()

    async def get_workspace(
        self,
        user_id: str,
        agent_id: str,
        session_id: str,
        workspace_id: str | None = None,
    ) -> WorkspaceBase:
        # 当此 session 没有持久化的显式绑定时，
        # 回退到 manager 的 isolation 策略。
        if workspace_id is None:
            workspace_id = self.assign_workspace_id(
                user_id=user_id,
                agent_id=agent_id,
                session_id=session_id,
            )
        async with self._lock:
            hit = self._cache.get(workspace_id)
            if hit is not None:
                ws, _ = hit
                self._cache[workspace_id] = (ws, time.monotonic())
                return ws
            ws = MyWorkspace(workspace_id=workspace_id)  # 你的后端
            await ws.initialize()
            self._cache[workspace_id] = (ws, time.monotonic())
            return ws

    async def create_workspace(
        self,
        user_id: str,
        agent_id: str,
        session_id: str,
    ) -> WorkspaceBase:
        ws = MyWorkspace()  # 让 workspace 自己生成 id
        await ws.initialize()
        async with self._lock:
            self._cache[ws.workspace_id] = (ws, time.monotonic())
        return ws

    async def close(self, workspace_id: str) -> None:
        async with self._lock:
            entry = self._cache.pop(workspace_id, None)
        if entry is not None:
            await entry[0].close()

    async def close_all(self) -> None:
        async with self._lock:
            entries = list(self._cache.values())
            self._cache.clear()
        await asyncio.gather(
            *(ws.close() for ws, _ in entries),
            return_exceptions=True,
        )
```

把一个 `MyWorkspaceManager` 实例传给 `create_app(workspace_manager=...)`，服务就会把它当作内置 manager 一样使用。

<Tip>
  如果后端需要一个后台 sweeper 来做空闲驱逐，或者需要跑长时间的准备逻辑，重写 `__aenter__` / `__aexit__` 来启停这些机制 —— `DockerWorkspaceManager` 的源码就是一个完整示例。
</Tip>

## 分布式部署

`LocalWorkspaceManager` 与 `DockerWorkspaceManager` 都是**单节点**的 —— workspace 状态活在跑服务进程的那台机器上。如果做水平扩容（负载均衡后挂多个智能体服务 worker 节点），同一个 `workspace_id` 的请求可能落到任意一个节点，而这两个 manager 都够不到其他节点上准备好的 workspace。

分布式部署要用**云托管的沙箱后端**：

* **`E2BWorkspaceManager`** —— 沙箱可以通过 metadata 跨节点寻址；缓存 miss 时服务直接重连已存在的沙箱即可。
* **`K8sWorkspaceManager`** —— Pod 与 PVC 都是集群级资源；集群内任意服务副本都可以按 workspace-id 派生出的名字重连到同一个 Pod。

<Tip>
  本地开发用 `LocalWorkspaceManager`，单机生产用 `DockerWorkspaceManager`，一旦要横向扩容就切到 E2B 或 K8s。
</Tip>