Workspace Manager

• Configurable isolation grain — PER_AGENT (default), PER_SESSION, or PER_USER — decides how workspaces are shared or isolated across the (user_id, agent_id, session_id) dimensions, so one deployment can serve many tenants cleanly.
• Pluggable sandbox backends — the same lifecycle contract wraps a local directory, a Docker container, an E2B cloud sandbox, or a Kubernetes Pod. Swapping backend is a one-line change in create_app.

​

Available managers

​

Integrate with Agent Service

from agentscope.app import create_app
from agentscope.app.workspace_manager import (
    IsolationPolicy,
    LocalWorkspaceManager,
)

workspace_manager = LocalWorkspaceManager(
    # Host root; per-agent workdirs live at `<basedir>/<user_id>/<agent_id>`.
    basedir="/data/workspaces",
    # `PER_AGENT` (default): sessions of the same agent share one workspace.
    # `PER_SESSION` / `PER_USER` change the grain; see below.
    isolation=IsolationPolicy.PER_AGENT,
)

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

​

Isolation grain

​

How it works in the Agent Service

​

Wiring

# 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

• Routers — /session, /workspace/* (MCP, skills). They call Depends(get_workspace_manager) and then manager.get_workspace(...) to resolve the caller’s workspace.
• ChatService — receives the manager in its constructor from the lifespan and calls get_workspace on every chat run to obtain the workspace the agent will act inside.
• Built-in team tools (AgentCreate / AgentInvite) — receive the manager to call assign_workspace_id(...) when spinning up sub-agent sessions, so the sub-agent lands in the same workspace as the leader.

​

Lifecycle

# agentscope/app/_lifespan.py (excerpt)
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)  # ← here
    ...
    yield
# ← on exit, `__aexit__` calls `close_all()` and tears down every cached workspace.

​

End-to-end request flow

1. Session creation — the /session router computes the workspace id once and persists it on the session record:
   # agentscope/app/_router/_session.py (excerpt)
   resolved_workspace_id = body.workspace_id or (
       workspace_manager.assign_workspace_id(
           user_id=user_id,
           agent_id=agent_id,
           session_id=session_id,
       )
   )
   
   An explicit body.workspace_id always wins (that’s how team tools make a sub-agent share the leader’s workspace); otherwise the manager mints one under the configured isolation policy. assign_workspace_id is a pure function — no I/O, no cache lookup — so it is cheap to call on the hot path.
2. Every subsequent request carrying that session — chat runs, MCP registration, skill upload, … — pulls the stored workspace_id off the session record and asks the manager to materialise it:
   workspace = await workspace_manager.get_workspace(
       user_id, agent_id, session_id, session_record.config.workspace_id,
   )
   
   get_workspace is the cache-hit path. It looks up workspace_id in the manager’s internal dict, refreshes the last-access timestamp, and returns the initialised Workspace in O(1). Only on a miss does it acquire an asyncio.Lock, provision the backend (build the Docker image, start the E2B sandbox, create the K8s Pod, …), and cache the result — the lock prevents concurrent requests for the same id from racing to spin up two backends.
3. Idle eviction — the sandbox-backed managers run a background sweeper (started in __aenter__) that periodically walks the cache, closes any workspace whose last-access time exceeds ttl, and drops it. The next request for that id re-provisions transparently.
4. Shutdown — the lifespan exits the manager, which calls close_all() and tears down every cached workspace and its background sweeper.

​

Core APIs

​

Custom manager

import asyncio
import time
from typing import Self

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


class MyWorkspaceManager(WorkspaceManagerBase):
    """Thin manager over a custom `MyWorkspace` backend."""

    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:
        # Fall back to the manager's isolation policy when no explicit
        # binding was persisted for this session.
        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)  # your backend
            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()  # let the workspace mint its own 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,
        )

​

Distributed deployment

• E2BWorkspaceManager — sandboxes are addressable by metadata across nodes; the service simply reattaches to the existing sandbox on cache miss.
• K8sWorkspaceManager — Pods and PVCs are cluster-scoped resources; any service replica in the cluster reattaches to the same Pod by workspace-id-derived name.