diff --git a/bindings/python/pyproject.toml b/bindings/python/pyproject.toml
index d096ef35ba4..4529a800aee 100644
--- a/bindings/python/pyproject.toml
+++ b/bindings/python/pyproject.toml
@@ -41,7 +41,7 @@ benchmark = [
   "boto3-stubs[essential]",
 ]
 docs = ["pdoc"]
-test = ["pytest", "python-dotenv"]
+test = ["pytest", "python-dotenv", "pytest-asyncio"]
 
 [project.urls]
 Documentation = "https://opendal.apache.org/docs/python/opendal.html"
diff --git a/bindings/python/tests/conftest.py b/bindings/python/tests/conftest.py
index 8bfbfff79bd..9db7dbe92b8 100644
--- a/bindings/python/tests/conftest.py
+++ b/bindings/python/tests/conftest.py
@@ -16,6 +16,8 @@
 # under the License.
 
 from dotenv import load_dotenv
+import pytest
 
 
 load_dotenv()
+pytest_plugins = ("pytest_asyncio",)
diff --git a/bindings/python/tests/test_services.py b/bindings/python/tests/test_services.py
index c8c6356f3ac..836988e7d10 100644
--- a/bindings/python/tests/test_services.py
+++ b/bindings/python/tests/test_services.py
@@ -41,6 +41,7 @@ def setup_method(self):
             raise ValueError(f"Service {self.service_name} test is not enabled.")
 
         self.operator = opendal.Operator(self.service_name, **self.config)
+        self.async_operator = opendal.AsyncOperator(self.service_name, **self.config)
 
     def test_sync_read(self):
         size = randint(1, 1024)
@@ -52,6 +53,21 @@ def test_sync_read(self):
         assert read_content is not None
         assert read_content == content
 
+        self.operator.delete(filename)
+
+    @pytest.mark.asyncio
+    async def test_async_read(self):
+        size = randint(1, 1024)
+        filename = f"random_file_{str(uuid4())}"
+        content = os.urandom(size)
+        await self.async_operator.write(filename, content)
+
+        read_content = await self.async_operator.read(filename)
+        assert read_content is not None
+        assert read_content == content
+
+        await self.async_operator.delete(filename)
+
     def test_sync_read_stat(self):
         size = randint(1, 1024)
         filename = f"random_file_{str(uuid4())}"
@@ -63,10 +79,33 @@ def test_sync_read_stat(self):
         assert metadata.content_length == len(content)
         assert metadata.mode.is_file()
 
+        self.operator.delete(filename)
+
+    @pytest.mark.asyncio
+    async def test_async_read_stat(self):
+        size = randint(1, 1024)
+        filename = f"random_file_{str(uuid4())}"
+        content = os.urandom(size)
+        await self.async_operator.write(filename, content)
+
+        metadata = await self.async_operator.stat(filename)
+        assert metadata is not None
+        assert metadata.content_length == len(content)
+        assert metadata.mode.is_file()
+
+        await self.async_operator.delete(filename)
+
+        self.operator.delete(filename)
+
     def test_sync_read_not_exists(self):
         with pytest.raises(FileNotFoundError):
             self.operator.read(str(uuid4()))
 
+    @pytest.mark.asyncio
+    async def test_async_read_not_exists(self):
+        with pytest.raises(FileNotFoundError):
+            await self.async_operator.read(str(uuid4()))
+
     def test_sync_write(self):
         size = randint(1, 1024)
         filename = f"test_file_{str(uuid4())}.txt"
@@ -80,6 +119,20 @@ def test_sync_write(self):
 
         self.operator.delete(filename)
 
+    @pytest.mark.asyncio
+    async def test_async_write(self):
+        size = randint(1, 1024)
+        filename = f"test_file_{str(uuid4())}.txt"
+        content = os.urandom(size)
+        size = len(content)
+        await self.async_operator.write(filename, content)
+        metadata = await self.async_operator.stat(filename)
+        assert metadata is not None
+        assert metadata.mode.is_file()
+        assert metadata.content_length == size
+
+        await self.async_operator.delete(filename)
+
     def test_sync_write_with_non_ascii_name(self):
         size = randint(1, 1024)
         filename = f"❌😱中文_{str(uuid4())}.test"
@@ -93,6 +146,20 @@ def test_sync_write_with_non_ascii_name(self):
 
         self.operator.delete(filename)
 
+    @pytest.mark.asyncio
+    async def test_async_write_with_non_ascii_name(self):
+        size = randint(1, 1024)
+        filename = f"❌😱中文_{str(uuid4())}.test"
+        content = os.urandom(size)
+        size = len(content)
+        await self.async_operator.write(filename, content)
+        metadata = await self.async_operator.stat(filename)
+        assert metadata is not None
+        assert metadata.mode.is_file()
+        assert metadata.content_length == size
+
+        await self.async_operator.delete(filename)
+
     def test_sync_create_dir(self):
         path = f"test_dir_{str(uuid4())}/"
         self.operator.create_dir(path)
@@ -102,6 +169,16 @@ def test_sync_create_dir(self):
 
         self.operator.delete(path)
 
+    @pytest.mark.asyncio
+    async def test_async_create_dir(self):
+        path = f"test_dir_{str(uuid4())}/"
+        await self.async_operator.create_dir(path)
+        metadata = await self.async_operator.stat(path)
+        assert metadata is not None
+        assert metadata.mode.is_dir()
+
+        await self.async_operator.delete(path)
+
     def test_sync_delete(self):
         size = randint(1, 1024)
         filename = f"test_file_{str(uuid4())}.txt"
@@ -112,6 +189,17 @@ def test_sync_delete(self):
         with pytest.raises(FileNotFoundError):
             self.operator.stat(filename)
 
+    @pytest.mark.asyncio
+    async def test_async_delete(self):
+        size = randint(1, 1024)
+        filename = f"test_file_{str(uuid4())}.txt"
+        content = os.urandom(size)
+        size = len(content)
+        await self.async_operator.write(filename, content)
+        await self.async_operator.delete(filename)
+        with pytest.raises(FileNotFoundError):
+            await self.operator.stat(filename)
+
 
 class TestS3(AbstractTestSuite):
     service_name = "s3"