list

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)

  The MCP tool handler function for 'list'. Decorated with @mcp.tool(), it parses the URI, ensures the path ends with a slash, calls resource.list(), and returns the entries as a string.

  @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/server.py:83-84 (registration)

  Registration of the 'list' tool via the @mcp.tool() decorator on the async function.

  # Modify the list tool to ensure the path ends with a slash
  @mcp.tool()

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

  The OpendalResource.list() method that performs the actual OpenDAL listing operation. Calls self.op.list(prefix) and iterates over entries up to max_keys limit.

  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)

  The parse_uri() helper function used by the list tool handler to parse a URI string into an OpendalResource and a path.

  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)

• tests/test_integration_fs.py:90-100 (registration)

  Integration test for the 'list' tool, verifying directory listing functionality (imports list from server and tests both root and subdirectory listing).

  async def test_list_directory_contents(setup_env, test_files):
      """Test listing directory contents"""
      result = await list("fs://")
  
      assert "test_text.txt" in result
      assert "binary_file.bin" in result
      assert "config.json" in result
      assert "subdir" in result
  
      subdir_result = await list("fs://subdir/")
      assert "nested_file.log" in subdir_result