MCP Server for Apache OpenDAL™

Instructions

List files in OpenDAL service

Args:
    uri: resource URI, e.g. mys3://path/to/dir

Returns:
    String containing directory content

Implementation Reference

• src/mcp_server_opendal/server.py:84-108 (handler)

  Main handler function for the 'list' tool. Decorated with @mcp.tool() which registers it as an MCP tool. Parses the URI, ensures path ends with slash, calls resource.list(), and returns string representation of entries.

  @mcp.tool()
  async def list(uri: str) -> str:
      """
      List files in OpenDAL service
  
      Args:
          uri: resource URI, e.g. mys3://path/to/dir
  
      Returns:
          String containing directory content
      """
      logger.debug(f"Listing directory content: {uri}")
      try:
          resource, path = parse_uri(uri)
  
          # Ensure directory path ends with a slash
          if path and not path.endswith("/"):
              path = path + "/"
  
          entries = await resource.list(path)
  
          return str(entries)
      except Exception as e:
          logger.error(f"Failed to list directory content: {e!s}")
          return f"Error: {e!s}"

• src/mcp_server_opendal/resource.py:81-100 (helper)

  Core listing implementation in OpendalResource class. Uses opendal operator to list entries with given prefix, limits to max_keys, returns list of Entry objects.

  async def list(
      self, prefix: Union[str, os.PathLike], max_keys: int = 1000
  ) -> List[Entry]:
      """List entries with the given prefix"""
      logger.debug(f"Listing entries with prefix: {prefix}")
  
      if max_keys <= 0:
          return []
  
      entries = []
  
      it = await self.op.list(prefix)
  
      async for entry in it:
          logger.debug(f"Listing entry: {entry}")
          entries.append(entry)
          if len(entries) >= max_keys:
              break
  
      return entries

• src/mcp_server_opendal/resource.py:137-147 (helper)

  Utility function to parse the tool input URI into scheme resource and path, used by the list tool handler.

  def parse_uri(uri: str) -> Tuple[OpendalResource, str]:
      """Parse a URI into a resource and path"""
      from urllib.parse import unquote, urlparse
  
      logger.debug(f"Parsing URI: {uri}")
      parsed = urlparse(uri)
  
      scheme = parsed.scheme
      path = parsed.netloc + parsed.path
      path = unquote(path)  # Decode URL-encoded characters
      return (OpendalResource(scheme), path)