get_workspace
Retrieve the folder and file structure of a graph's workspace to understand document organization.
Instructions
Returns the folder and file structure of a graph's workspace. Use this to understand the organization of documents in a graph.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| graph_id | Yes |
Implementation Reference
- src/neem/mcp/tools/hocuspocus.py:321-358 (handler)The core asynchronous handler function `get_workspace_tool` for the "get_workspace" tool. It authenticates the request, ensures connection to the workspace channel via `hp_client.connect_workspace(graph_id)`, fetches the workspace snapshot using `hp_client.get_workspace_snapshot(graph_id)`, and returns a dictionary containing the graph_id and workspace data.@server.tool( name="get_workspace", title="Get Workspace Structure", description=( "Returns the folder and file structure of a graph's workspace. " "Use this to understand the organization of documents in a graph." ), ) async def get_workspace_tool( graph_id: str, context: Context | None = None, ) -> dict: """Get workspace folder structure.""" auth = MCPAuthContext.from_context(context) auth.require_auth() try: # Connect to the workspace channel await hp_client.connect_workspace(graph_id) # Get workspace snapshot snapshot = hp_client.get_workspace_snapshot(graph_id) result = { "graph_id": graph_id, "workspace": snapshot, } return result except Exception as e: logger.error( "Failed to get workspace", extra_context={ "graph_id": graph_id, "error": str(e), }, ) raise RuntimeError(f"Failed to get workspace: {e}")
- src/neem/mcp/server/standalone_server.py:317-317 (registration)Explicit registration of the hocuspocus tools (including "get_workspace") on the MCP server instance by calling `register_hocuspocus_tools(mcp_server)`.register_hocuspocus_tools(mcp_server)
- Supporting method `get_workspace_snapshot` in HocuspocusClient class that retrieves the current state of the workspace Y.js document for a given graph_id, extracting and serializing the folders, artifacts, documents, and ui maps into a plain dictionary.def get_workspace_snapshot(self, graph_id: str) -> Dict[str, Any]: """Get a snapshot of the workspace state for a graph.""" channel = self._workspace_channels.get(graph_id) if channel is None: return {} doc = channel.doc def ymap_to_dict(ymap: pycrdt.Map) -> Dict[str, Any]: result = {} for key in ymap.keys(): value = ymap.get(key) if isinstance(value, pycrdt.Map): result[key] = ymap_to_dict(value) elif isinstance(value, pycrdt.Array): result[key] = list(value) else: result[key] = value return result # Workspace uses four separate YMaps: folders, artifacts, documents, ui folders_map: pycrdt.Map = doc.get("folders", type=pycrdt.Map) artifacts_map: pycrdt.Map = doc.get("artifacts", type=pycrdt.Map) documents_map: pycrdt.Map = doc.get("documents", type=pycrdt.Map) ui_map: pycrdt.Map = doc.get("ui", type=pycrdt.Map) return { "folders": ymap_to_dict(folders_map), "artifacts": ymap_to_dict(artifacts_map), "documents": ymap_to_dict(documents_map), "ui": ymap_to_dict(ui_map), }
- Supporting method `connect_workspace` that establishes the WebSocket connection to the Hocuspocus workspace channel for the given graph_id if not already connected and synced.async def connect_workspace(self, graph_id: str) -> None: """Connect to a workspace channel for the given graph. Args: graph_id: The graph ID """ if graph_id in self._workspace_channels: channel = self._workspace_channels[graph_id] if channel.ws and not channel.ws.closed and channel.synced.is_set(): return # Already connected and synced channel = ChannelState() self._workspace_channels[graph_id] = channel async with channel.lock: await self._connect_channel( channel, f"/hocuspocus/workspace/{graph_id}", f"workspace:{graph_id}", )
- src/neem/mcp/tools/hocuspocus.py:25-25 (registration)The `register_hocuspocus_tools` function definition that sets up the HocuspocusClient on the server and registers all hocuspocus tools including "get_workspace" using the `@server.tool` decorator.def register_hocuspocus_tools(server: FastMCP) -> None: