Mnemosyne MCP

Instructions

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)

• src/neem/hocuspocus/client.py:534-565 (helper)

  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),
      }

• src/neem/hocuspocus/client.py:513-532 (helper)

  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: