diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 91d2b2c5..81caa735 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -98,7 +98,7 @@ jobs:
       - name: Run formatters and linters
         run: |
           pip3 install black isort flake8-pyi flake8-noqa flake8-bugbear
-          black --check .
+          black --check --extend-exclude '(_pb2_grpc|_pb2).pyi?$' .
           isort --check . --diff
           flake8 .
       - name: run shellcheck
diff --git a/mypy_protobuf/main.py b/mypy_protobuf/main.py
index 23514a5a..02f1bda2 100644
--- a/mypy_protobuf/main.py
+++ b/mypy_protobuf/main.py
@@ -663,30 +663,77 @@ def _map_key_value_types(
 
         return ktype, vtype
 
-    def _callable_type(self, method: d.MethodDescriptorProto) -> str:
+    def _callable_type(self, method: d.MethodDescriptorProto, is_async: bool = False) -> str:
+        module = "grpc.aio" if is_async else "grpc"
         if method.client_streaming:
             if method.server_streaming:
-                return self._import("grpc", "StreamStreamMultiCallable")
+                return self._import(module, "StreamStreamMultiCallable")
             else:
-                return self._import("grpc", "StreamUnaryMultiCallable")
+                return self._import(module, "StreamUnaryMultiCallable")
         else:
             if method.server_streaming:
-                return self._import("grpc", "UnaryStreamMultiCallable")
+                return self._import(module, "UnaryStreamMultiCallable")
             else:
-                return self._import("grpc", "UnaryUnaryMultiCallable")
+                return self._import(module, "UnaryUnaryMultiCallable")
 
-    def _input_type(self, method: d.MethodDescriptorProto, use_stream_iterator: bool = True) -> str:
+    def _input_type(self, method: d.MethodDescriptorProto) -> str:
         result = self._import_message(method.input_type)
-        if use_stream_iterator and method.client_streaming:
-            result = f"{self._import('collections.abc', 'Iterator')}[{result}]"
         return result
 
-    def _output_type(self, method: d.MethodDescriptorProto, use_stream_iterator: bool = True) -> str:
+    def _servicer_input_type(self, method: d.MethodDescriptorProto) -> str:
+        result = self._import_message(method.input_type)
+        if method.client_streaming:
+            # See write_grpc_async_hacks().
+            result = f"_MaybeAsyncIterator[{result}]"
+        return result
+
+    def _output_type(self, method: d.MethodDescriptorProto) -> str:
         result = self._import_message(method.output_type)
-        if use_stream_iterator and method.server_streaming:
-            result = f"{self._import('collections.abc', 'Iterator')}[{result}]"
         return result
 
+    def _servicer_output_type(self, method: d.MethodDescriptorProto) -> str:
+        result = self._import_message(method.output_type)
+        if method.server_streaming:
+            # Union[Iterator[Resp], AsyncIterator[Resp]] is subtyped by Iterator[Resp] and AsyncIterator[Resp].
+            # So both can be used in the covariant function return position.
+            iterator = f"{self._import('collections.abc', 'Iterator')}[{result}]"
+            aiterator = f"{self._import('collections.abc', 'AsyncIterator')}[{result}]"
+            result = f"{self._import('typing', 'Union')}[{iterator}, {aiterator}]"
+        else:
+            # Union[Resp, Awaitable[Resp]] is subtyped by Resp and Awaitable[Resp].
+            # So both can be used in the covariant function return position.
+            # Awaitable[Resp] is equivalent to async def.
+            awaitable = f"{self._import('collections.abc', 'Awaitable')}[{result}]"
+            result = f"{self._import('typing', 'Union')}[{result}, {awaitable}]"
+        return result
+
+    def write_grpc_async_hacks(self) -> None:
+        wl = self._write_line
+        # _MaybeAsyncIterator[Req] is supertyped by Iterator[Req] and AsyncIterator[Req].
+        # So both can be used in the contravariant function parameter position.
+        wl("_T = {}('_T')", self._import("typing", "TypeVar"))
+        wl("")
+        wl(
+            "class _MaybeAsyncIterator({}[_T], {}[_T], metaclass={}):",
+            self._import("collections.abc", "AsyncIterator"),
+            self._import("collections.abc", "Iterator"),
+            self._import("abc", "ABCMeta"),
+        )
+        with self._indent():
+            wl("...")
+        wl("")
+
+        # _ServicerContext is supertyped by grpc.ServicerContext and grpc.aio.ServicerContext
+        # So both can be used in the contravariant function parameter position.
+        wl(
+            "class _ServicerContext({}, {}):  # type: ignore",
+            self._import("grpc", "ServicerContext"),
+            self._import("grpc.aio", "ServicerContext"),
+        )
+        with self._indent():
+            wl("...")
+        wl("")
+
     def write_grpc_methods(self, service: d.ServiceDescriptorProto, scl_prefix: SourceCodeLocation) -> None:
         wl = self._write_line
         methods = [(i, m) for i, m in enumerate(service.method) if m.name not in PYTHON_RESERVED]
@@ -701,12 +748,12 @@ def write_grpc_methods(self, service: d.ServiceDescriptorProto, scl_prefix: Sour
             with self._indent():
                 wl("self,")
                 input_name = "request_iterator" if method.client_streaming else "request"
-                input_type = self._input_type(method)
+                input_type = self._servicer_input_type(method)
                 wl(f"{input_name}: {input_type},")
-                wl("context: {},", self._import("grpc", "ServicerContext"))
+                wl("context: _ServicerContext,")
             wl(
                 ") -> {}:{}",
-                self._output_type(method),
+                self._servicer_output_type(method),
                 " ..." if not self._has_comments(scl) else "",
             )
             if self._has_comments(scl):
@@ -714,7 +761,7 @@ def write_grpc_methods(self, service: d.ServiceDescriptorProto, scl_prefix: Sour
                     if not self._write_comments(scl):
                         wl("...")
 
-    def write_grpc_stub_methods(self, service: d.ServiceDescriptorProto, scl_prefix: SourceCodeLocation) -> None:
+    def write_grpc_stub_methods(self, service: d.ServiceDescriptorProto, scl_prefix: SourceCodeLocation, is_async: bool = False) -> None:
         wl = self._write_line
         methods = [(i, m) for i, m in enumerate(service.method) if m.name not in PYTHON_RESERVED]
         if not methods:
@@ -723,10 +770,10 @@ def write_grpc_stub_methods(self, service: d.ServiceDescriptorProto, scl_prefix:
         for i, method in methods:
             scl = scl_prefix + [d.ServiceDescriptorProto.METHOD_FIELD_NUMBER, i]
 
-            wl("{}: {}[", method.name, self._callable_type(method))
+            wl("{}: {}[", method.name, self._callable_type(method, is_async=is_async))
             with self._indent():
-                wl("{},", self._input_type(method, False))
-                wl("{},", self._output_type(method, False))
+                wl("{},", self._input_type(method))
+                wl("{},", self._output_type(method))
             wl("]")
             self._write_comments(scl)
 
@@ -743,17 +790,31 @@ def write_grpc_services(
             scl = scl_prefix + [i]
 
             # The stub client
-            wl(f"class {service.name}Stub:")
+            wl(
+                "class {}Stub:",
+                service.name,
+            )
             with self._indent():
                 if self._write_comments(scl):
                     wl("")
-                wl(
-                    "def __init__(self, channel: {}) -> None: ...",
-                    self._import("grpc", "Channel"),
-                )
+                # To support casting into FooAsyncStub, allow both Channel and aio.Channel here.
+                channel = f"{self._import('typing', 'Union')}[{self._import('grpc', 'Channel')}, {self._import('grpc.aio', 'Channel')}]"
+                wl("def __init__(self, channel: {}) -> None: ...", channel)
                 self.write_grpc_stub_methods(service, scl)
             wl("")
 
+            # The (fake) async stub client
+            wl(
+                "class {}AsyncStub:",
+                service.name,
+            )
+            with self._indent():
+                if self._write_comments(scl):
+                    wl("")
+                # No __init__ since this isn't a real class (yet), and requires manual casting to work.
+                self.write_grpc_stub_methods(service, scl, is_async=True)
+            wl("")
+
             # The service definition interface
             wl(
                 "class {}Servicer(metaclass={}):",
@@ -765,11 +826,13 @@ def write_grpc_services(
                     wl("")
                 self.write_grpc_methods(service, scl)
             wl("")
+            server = self._import("grpc", "Server")
+            aserver = self._import("grpc.aio", "Server")
             wl(
                 "def add_{}Servicer_to_server(servicer: {}Servicer, server: {}) -> None: ...",
                 service.name,
                 service.name,
-                self._import("grpc", "Server"),
+                f"{self._import('typing', 'Union')}[{server}, {aserver}]",
             )
             wl("")
 
@@ -960,6 +1023,7 @@ def generate_mypy_grpc_stubs(
             relax_strict_optional_primitives,
             grpc=True,
         )
+        pkg_writer.write_grpc_async_hacks()
         pkg_writer.write_grpc_services(fd.service, [d.FileDescriptorProto.SERVICE_FIELD_NUMBER])
 
         assert name == fd.name
diff --git a/run_test.sh b/run_test.sh
index 4d08092a..5c6d32cd 100755
--- a/run_test.sh
+++ b/run_test.sh
@@ -4,7 +4,7 @@ RED="\033[0;31m"
 NC='\033[0m'
 
 PY_VER_MYPY_PROTOBUF=${PY_VER_MYPY_PROTOBUF:=3.10.6}
-PY_VER_MYPY_PROTOBUF_SHORT=$(echo $PY_VER_MYPY_PROTOBUF | cut -d. -f1-2)
+PY_VER_MYPY_PROTOBUF_SHORT=$(echo "$PY_VER_MYPY_PROTOBUF" | cut -d. -f1-2)
 PY_VER_MYPY=${PY_VER_MYPY:=3.8.13}
 PY_VER_UNIT_TESTS="${PY_VER_UNIT_TESTS:=3.8.13}"
 
@@ -45,16 +45,16 @@ MYPY_VENV=venv_$PY_VER_MYPY
 (
     eval "$(pyenv init --path)"
     eval "$(pyenv init -)"
-    pyenv shell $PY_VER_MYPY
+    pyenv shell "$PY_VER_MYPY"
 
     if [[ -z $SKIP_CLEAN ]] || [[ ! -e $MYPY_VENV ]]; then
         python3 --version
         python3 -m pip --version
         python -m pip install virtualenv
-        python3 -m virtualenv $MYPY_VENV
-        $MYPY_VENV/bin/python3 -m pip install -r mypy_requirements.txt
+        python3 -m virtualenv "$MYPY_VENV"
+        "$MYPY_VENV"/bin/python3 -m pip install -r mypy_requirements.txt
     fi
-    $MYPY_VENV/bin/mypy --version
+    "$MYPY_VENV"/bin/mypy --version
 )
 
 # Create unit tests venvs
@@ -63,14 +63,14 @@ for PY_VER in $PY_VER_UNIT_TESTS; do
         UNIT_TESTS_VENV=venv_$PY_VER
         eval "$(pyenv init --path)"
         eval "$(pyenv init -)"
-        pyenv shell $PY_VER
+        pyenv shell "$PY_VER"
 
         if [[ -z $SKIP_CLEAN ]] || [[ ! -e $UNIT_TESTS_VENV ]]; then
             python -m pip install virtualenv
-            python -m virtualenv $UNIT_TESTS_VENV
-            $UNIT_TESTS_VENV/bin/python -m pip install -r test_requirements.txt
+            python -m virtualenv "$UNIT_TESTS_VENV"
+            "$UNIT_TESTS_VENV"/bin/python -m pip install -r test_requirements.txt
         fi
-        $UNIT_TESTS_VENV/bin/py.test --version
+        "$UNIT_TESTS_VENV"/bin/py.test --version
     )
 done
 
@@ -79,19 +79,19 @@ MYPY_PROTOBUF_VENV=venv_$PY_VER_MYPY_PROTOBUF
 (
     eval "$(pyenv init --path)"
     eval "$(pyenv init -)"
-    pyenv shell $PY_VER_MYPY_PROTOBUF
+    pyenv shell "$PY_VER_MYPY_PROTOBUF"
 
     # Create virtualenv + Install requirements for mypy-protobuf
     if [[ -z $SKIP_CLEAN ]] || [[ ! -e $MYPY_PROTOBUF_VENV ]]; then
         python -m pip install virtualenv
-        python -m virtualenv $MYPY_PROTOBUF_VENV
-        $MYPY_PROTOBUF_VENV/bin/python -m pip install -e .
+        python -m virtualenv "$MYPY_PROTOBUF_VENV"
+        "$MYPY_PROTOBUF_VENV"/bin/python -m pip install -e .
     fi
 )
 
 # Run mypy-protobuf
 (
-    source $MYPY_PROTOBUF_VENV/bin/activate
+    source "$MYPY_PROTOBUF_VENV"/bin/activate
 
     # Confirm version number
     test "$(protoc-gen-mypy -V)" = "mypy-protobuf 3.4.0"
@@ -138,22 +138,22 @@ MYPY_PROTOBUF_VENV=venv_$PY_VER_MYPY_PROTOBUF
 
 for PY_VER in $PY_VER_UNIT_TESTS; do
     UNIT_TESTS_VENV=venv_$PY_VER
-    PY_VER_MYPY_TARGET=$(echo $PY_VER | cut -d. -f1-2)
+    PY_VER_MYPY_TARGET=$(echo "$PY_VER" | cut -d. -f1-2)
 
     # Generate GRPC protos for mypy / tests
     (
-        source $UNIT_TESTS_VENV/bin/activate
+        source "$UNIT_TESTS_VENV"/bin/activate
         find proto/testproto/grpc -name "*.proto" -print0 | xargs -0 python -m grpc_tools.protoc "${PROTOC_ARGS[@]}" --grpc_python_out=test/generated
     )
 
     # Run mypy on unit tests / generated output
     (
-        source $MYPY_VENV/bin/activate
+        source "$MYPY_VENV"/bin/activate
         export MYPYPATH=$MYPYPATH:test/generated
 
         # Run mypy
-        MODULES=( "-m" "test" )
-        mypy --custom-typeshed-dir="$CUSTOM_TYPESHED_DIR" --python-executable=$UNIT_TESTS_VENV/bin/python --python-version="$PY_VER_MYPY_TARGET" "${MODULES[@]}"
+        MODULES=( -m test.test_generated_mypy -m test.test_grpc_usage -m test.test_grpc_async_usage )
+        mypy --custom-typeshed-dir="$CUSTOM_TYPESHED_DIR" --python-executable="$UNIT_TESTS_VENV"/bin/python --python-version="$PY_VER_MYPY_TARGET" "${MODULES[@]}"
 
         # Run stubtest. Stubtest does not work with python impl - only cpp impl
         API_IMPL="$(python3 -c "import google.protobuf.internal.api_implementation as a ; print(a.Type())")"
@@ -173,12 +173,12 @@ for PY_VER in $PY_VER_UNIT_TESTS; do
             cut -d: -f1,3- "$MYPY_OUTPUT/mypy_output" > "$MYPY_OUTPUT/mypy_output.omit_linenos"
         }
 
-        call_mypy $PY_VER "${NEGATIVE_MODULES[@]}"
+        call_mypy "$PY_VER" "${NEGATIVE_MODULES[@]}"
         if ! diff "$MYPY_OUTPUT/mypy_output" "test_negative/output.expected.$PY_VER_MYPY_TARGET" || ! diff "$MYPY_OUTPUT/mypy_output.omit_linenos" "test_negative/output.expected.$PY_VER_MYPY_TARGET.omit_linenos"; then
             echo -e "${RED}test_negative/output.expected.$PY_VER_MYPY_TARGET didnt match. Copying over for you. Now rerun${NC}"
 
             # Copy over all the mypy results for the developer.
-            call_mypy $PY_VER "${NEGATIVE_MODULES[@]}"
+            call_mypy "$PY_VER" "${NEGATIVE_MODULES[@]}"
             cp "$MYPY_OUTPUT/mypy_output" test_negative/output.expected.3.8
             cp "$MYPY_OUTPUT/mypy_output.omit_linenos" test_negative/output.expected.3.8.omit_linenos
             exit 1
@@ -187,7 +187,7 @@ for PY_VER in $PY_VER_UNIT_TESTS; do
 
     (
         # Run unit tests.
-        source $UNIT_TESTS_VENV/bin/activate
+        source "$UNIT_TESTS_VENV"/bin/activate
         PYTHONPATH=test/generated py.test --ignore=test/generated -v
     )
 done
diff --git a/stubtest_allowlist.txt b/stubtest_allowlist.txt
index 4c1aaab4..11c39f81 100644
--- a/stubtest_allowlist.txt
+++ b/stubtest_allowlist.txt
@@ -48,6 +48,10 @@ testproto.readme_enum_pb2._?MyEnum(EnumTypeWrapper)?
 testproto.nested.nested_pb2.AnotherNested._?NestedEnum(EnumTypeWrapper)?
 testproto.nested.nested_pb2.AnotherNested.NestedMessage._?NestedEnum2(EnumTypeWrapper)?
 
+# Our fake async stubs are not there at runtime (yet)
+testproto.grpc.dummy_pb2_grpc.DummyServiceAsyncStub
+testproto.grpc.import_pb2_grpc.SimpleServiceAsyncStub
+
 # Part of an "EXPERIMENTAL API" according to comment. Not documented.
 testproto.grpc.dummy_pb2_grpc.DummyService
 testproto.grpc.import_pb2_grpc.SimpleService
diff --git a/test/generated/testproto/grpc/dummy_pb2_grpc.pyi b/test/generated/testproto/grpc/dummy_pb2_grpc.pyi
index 7d432172..758c58d4 100644
--- a/test/generated/testproto/grpc/dummy_pb2_grpc.pyi
+++ b/test/generated/testproto/grpc/dummy_pb2_grpc.pyi
@@ -5,12 +5,22 @@ https://github.com/vmagamedov/grpclib/blob/master/tests/dummy.proto"""
 import abc
 import collections.abc
 import grpc
+import grpc.aio
 import testproto.grpc.dummy_pb2
+import typing
+
+_T = typing.TypeVar('_T')
+
+class _MaybeAsyncIterator(collections.abc.AsyncIterator[_T], collections.abc.Iterator[_T], metaclass=abc.ABCMeta):
+    ...
+
+class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext):  # type: ignore
+    ...
 
 class DummyServiceStub:
     """DummyService"""
 
-    def __init__(self, channel: grpc.Channel) -> None: ...
+    def __init__(self, channel: typing.Union[grpc.Channel, grpc.aio.Channel]) -> None: ...
     UnaryUnary: grpc.UnaryUnaryMultiCallable[
         testproto.grpc.dummy_pb2.DummyRequest,
         testproto.grpc.dummy_pb2.DummyReply,
@@ -32,6 +42,30 @@ class DummyServiceStub:
     ]
     """StreamStream"""
 
+class DummyServiceAsyncStub:
+    """DummyService"""
+
+    UnaryUnary: grpc.aio.UnaryUnaryMultiCallable[
+        testproto.grpc.dummy_pb2.DummyRequest,
+        testproto.grpc.dummy_pb2.DummyReply,
+    ]
+    """UnaryUnary"""
+    UnaryStream: grpc.aio.UnaryStreamMultiCallable[
+        testproto.grpc.dummy_pb2.DummyRequest,
+        testproto.grpc.dummy_pb2.DummyReply,
+    ]
+    """UnaryStream"""
+    StreamUnary: grpc.aio.StreamUnaryMultiCallable[
+        testproto.grpc.dummy_pb2.DummyRequest,
+        testproto.grpc.dummy_pb2.DummyReply,
+    ]
+    """StreamUnary"""
+    StreamStream: grpc.aio.StreamStreamMultiCallable[
+        testproto.grpc.dummy_pb2.DummyRequest,
+        testproto.grpc.dummy_pb2.DummyReply,
+    ]
+    """StreamStream"""
+
 class DummyServiceServicer(metaclass=abc.ABCMeta):
     """DummyService"""
 
@@ -39,29 +73,29 @@ class DummyServiceServicer(metaclass=abc.ABCMeta):
     def UnaryUnary(
         self,
         request: testproto.grpc.dummy_pb2.DummyRequest,
-        context: grpc.ServicerContext,
-    ) -> testproto.grpc.dummy_pb2.DummyReply:
+        context: _ServicerContext,
+    ) -> typing.Union[testproto.grpc.dummy_pb2.DummyReply, collections.abc.Awaitable[testproto.grpc.dummy_pb2.DummyReply]]:
         """UnaryUnary"""
     @abc.abstractmethod
     def UnaryStream(
         self,
         request: testproto.grpc.dummy_pb2.DummyRequest,
-        context: grpc.ServicerContext,
-    ) -> collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyReply]:
+        context: _ServicerContext,
+    ) -> typing.Union[collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyReply], collections.abc.AsyncIterator[testproto.grpc.dummy_pb2.DummyReply]]:
         """UnaryStream"""
     @abc.abstractmethod
     def StreamUnary(
         self,
-        request_iterator: collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyRequest],
-        context: grpc.ServicerContext,
-    ) -> testproto.grpc.dummy_pb2.DummyReply:
+        request_iterator: _MaybeAsyncIterator[testproto.grpc.dummy_pb2.DummyRequest],
+        context: _ServicerContext,
+    ) -> typing.Union[testproto.grpc.dummy_pb2.DummyReply, collections.abc.Awaitable[testproto.grpc.dummy_pb2.DummyReply]]:
         """StreamUnary"""
     @abc.abstractmethod
     def StreamStream(
         self,
-        request_iterator: collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyRequest],
-        context: grpc.ServicerContext,
-    ) -> collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyReply]:
+        request_iterator: _MaybeAsyncIterator[testproto.grpc.dummy_pb2.DummyRequest],
+        context: _ServicerContext,
+    ) -> typing.Union[collections.abc.Iterator[testproto.grpc.dummy_pb2.DummyReply], collections.abc.AsyncIterator[testproto.grpc.dummy_pb2.DummyReply]]:
         """StreamStream"""
 
-def add_DummyServiceServicer_to_server(servicer: DummyServiceServicer, server: grpc.Server) -> None: ...
+def add_DummyServiceServicer_to_server(servicer: DummyServiceServicer, server: typing.Union[grpc.Server, grpc.aio.Server]) -> None: ...
diff --git a/test/generated/testproto/grpc/import_pb2_grpc.pyi b/test/generated/testproto/grpc/import_pb2_grpc.pyi
index 77eb4daf..eb58942c 100644
--- a/test/generated/testproto/grpc/import_pb2_grpc.pyi
+++ b/test/generated/testproto/grpc/import_pb2_grpc.pyi
@@ -3,14 +3,25 @@
 isort:skip_file
 """
 import abc
+import collections.abc
 import google.protobuf.empty_pb2
 import grpc
+import grpc.aio
 import testproto.test_pb2
+import typing
+
+_T = typing.TypeVar('_T')
+
+class _MaybeAsyncIterator(collections.abc.AsyncIterator[_T], collections.abc.Iterator[_T], metaclass=abc.ABCMeta):
+    ...
+
+class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext):  # type: ignore
+    ...
 
 class SimpleServiceStub:
     """SimpleService"""
 
-    def __init__(self, channel: grpc.Channel) -> None: ...
+    def __init__(self, channel: typing.Union[grpc.Channel, grpc.aio.Channel]) -> None: ...
     UnaryUnary: grpc.UnaryUnaryMultiCallable[
         google.protobuf.empty_pb2.Empty,
         testproto.test_pb2.Simple1,
@@ -26,6 +37,24 @@ class SimpleServiceStub:
         google.protobuf.empty_pb2.Empty,
     ]
 
+class SimpleServiceAsyncStub:
+    """SimpleService"""
+
+    UnaryUnary: grpc.aio.UnaryUnaryMultiCallable[
+        google.protobuf.empty_pb2.Empty,
+        testproto.test_pb2.Simple1,
+    ]
+    """UnaryUnary"""
+    UnaryStream: grpc.aio.UnaryUnaryMultiCallable[
+        testproto.test_pb2.Simple1,
+        google.protobuf.empty_pb2.Empty,
+    ]
+    """UnaryStream"""
+    NoComment: grpc.aio.UnaryUnaryMultiCallable[
+        testproto.test_pb2.Simple1,
+        google.protobuf.empty_pb2.Empty,
+    ]
+
 class SimpleServiceServicer(metaclass=abc.ABCMeta):
     """SimpleService"""
 
@@ -33,21 +62,21 @@ class SimpleServiceServicer(metaclass=abc.ABCMeta):
     def UnaryUnary(
         self,
         request: google.protobuf.empty_pb2.Empty,
-        context: grpc.ServicerContext,
-    ) -> testproto.test_pb2.Simple1:
+        context: _ServicerContext,
+    ) -> typing.Union[testproto.test_pb2.Simple1, collections.abc.Awaitable[testproto.test_pb2.Simple1]]:
         """UnaryUnary"""
     @abc.abstractmethod
     def UnaryStream(
         self,
         request: testproto.test_pb2.Simple1,
-        context: grpc.ServicerContext,
-    ) -> google.protobuf.empty_pb2.Empty:
+        context: _ServicerContext,
+    ) -> typing.Union[google.protobuf.empty_pb2.Empty, collections.abc.Awaitable[google.protobuf.empty_pb2.Empty]]:
         """UnaryStream"""
     @abc.abstractmethod
     def NoComment(
         self,
         request: testproto.test_pb2.Simple1,
-        context: grpc.ServicerContext,
-    ) -> google.protobuf.empty_pb2.Empty: ...
+        context: _ServicerContext,
+    ) -> typing.Union[google.protobuf.empty_pb2.Empty, collections.abc.Awaitable[google.protobuf.empty_pb2.Empty]]: ...
 
-def add_SimpleServiceServicer_to_server(servicer: SimpleServiceServicer, server: grpc.Server) -> None: ...
+def add_SimpleServiceServicer_to_server(servicer: SimpleServiceServicer, server: typing.Union[grpc.Server, grpc.aio.Server]) -> None: ...
diff --git a/test/test_grpc_async_usage.py b/test/test_grpc_async_usage.py
new file mode 100644
index 00000000..0293ee4b
--- /dev/null
+++ b/test/test_grpc_async_usage.py
@@ -0,0 +1,68 @@
+import typing
+
+import grpc.aio
+import pytest
+from testproto.grpc import dummy_pb2, dummy_pb2_grpc
+
+ADDRESS = "localhost:22223"
+
+
+class Servicer(dummy_pb2_grpc.DummyServiceServicer):
+    async def UnaryUnary(
+        self,
+        request: dummy_pb2.DummyRequest,
+        context: grpc.aio.ServicerContext,
+    ) -> dummy_pb2.DummyReply:
+        return dummy_pb2.DummyReply(value=request.value[::-1])
+
+    async def UnaryStream(
+        self,
+        request: dummy_pb2.DummyRequest,
+        context: grpc.aio.ServicerContext,
+    ) -> typing.AsyncIterator[dummy_pb2.DummyReply]:
+        for char in request.value:
+            yield dummy_pb2.DummyReply(value=char)
+
+    async def StreamUnary(
+        self,
+        request: typing.AsyncIterator[dummy_pb2.DummyRequest],
+        context: grpc.aio.ServicerContext,
+    ) -> dummy_pb2.DummyReply:
+        values = [data.value async for data in request]
+        return dummy_pb2.DummyReply(value="".join(values))
+
+    async def StreamStream(
+        self,
+        request: typing.AsyncIterator[dummy_pb2.DummyRequest],
+        context: grpc.aio.ServicerContext,
+    ) -> typing.AsyncIterator[dummy_pb2.DummyReply]:
+        async for data in request:
+            yield dummy_pb2.DummyReply(value=data.value.upper())
+
+
+def make_server() -> grpc.aio.Server:
+    server = grpc.aio.server()
+    servicer = Servicer()
+    server.add_insecure_port(ADDRESS)
+    dummy_pb2_grpc.add_DummyServiceServicer_to_server(servicer, server)
+    return server
+
+
+@pytest.mark.asyncio
+async def test_grpc() -> None:
+    server = make_server()
+    await server.start()
+    async with grpc.aio.insecure_channel(ADDRESS) as channel:
+        client: dummy_pb2_grpc.DummyServiceAsyncStub = dummy_pb2_grpc.DummyServiceStub(channel)  # type: ignore
+        request = dummy_pb2.DummyRequest(value="cprg")
+        result1 = await client.UnaryUnary(request)
+        result2 = client.UnaryStream(dummy_pb2.DummyRequest(value=result1.value))
+        result2_list = [r async for r in result2]
+        assert len(result2_list) == 4
+        result3 = client.StreamStream(dummy_pb2.DummyRequest(value=part.value) for part in result2_list)
+        result3_list = [r async for r in result3]
+        assert len(result3_list) == 4
+        result4 = await client.StreamUnary(dummy_pb2.DummyRequest(value=part.value) for part in result3_list)
+        assert result4.value == "GRPC"
+
+    await server.stop(None)
diff --git a/test_negative/output.expected.3.8 b/test_negative/output.expected.3.8
index e10b8200..9cf4d93c 100644
--- a/test_negative/output.expected.3.8
+++ b/test_negative/output.expected.3.8
@@ -85,13 +85,13 @@ test_negative/negative.py:216: error: "DummyReply" has no attribute "not_exists"
 test_negative/negative.py:253: error: Argument 1 of "UnaryUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "DummyRequest"  [override]
 test_negative/negative.py:253: note: This violates the Liskov substitution principle
 test_negative/negative.py:253: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py:253: error: Return type "Iterator[DummyReply]" of "UnaryUnary" incompatible with return type "DummyReply" in supertype "DummyServiceServicer"  [override]
+test_negative/negative.py:253: error: Return type "Iterator[DummyReply]" of "UnaryUnary" incompatible with return type "Union[DummyReply, Awaitable[DummyReply]]" in supertype "DummyServiceServicer"  [override]
 test_negative/negative.py:261: error: Argument 1 of "UnaryStream" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "DummyRequest"  [override]
 test_negative/negative.py:261: note: This violates the Liskov substitution principle
 test_negative/negative.py:261: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py:261: error: Return type "DummyReply" of "UnaryStream" incompatible with return type "Iterator[DummyReply]" in supertype "DummyServiceServicer"  [override]
-test_negative/negative.py:270: error: Argument 1 of "StreamUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "Iterator[DummyRequest]"  [override]
+test_negative/negative.py:261: error: Return type "DummyReply" of "UnaryStream" incompatible with return type "Union[Iterator[DummyReply], AsyncIterator[DummyReply]]" in supertype "DummyServiceServicer"  [override]
+test_negative/negative.py:270: error: Argument 1 of "StreamUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "_MaybeAsyncIterator[DummyRequest]"  [override]
 test_negative/negative.py:270: note: This violates the Liskov substitution principle
 test_negative/negative.py:270: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py:270: error: Return type "Iterator[DummyReply]" of "StreamUnary" incompatible with return type "DummyReply" in supertype "DummyServiceServicer"  [override]
-Found 80 errors in 1 file (checked 2 source files)
+test_negative/negative.py:270: error: Return type "Iterator[DummyReply]" of "StreamUnary" incompatible with return type "Union[DummyReply, Awaitable[DummyReply]]" in supertype "DummyServiceServicer"  [override]
+Found 80 errors in 1 file (checked 4 source files)
diff --git a/test_negative/output.expected.3.8.omit_linenos b/test_negative/output.expected.3.8.omit_linenos
index c6c3e50f..2d17a89a 100644
--- a/test_negative/output.expected.3.8.omit_linenos
+++ b/test_negative/output.expected.3.8.omit_linenos
@@ -85,13 +85,13 @@ test_negative/negative.py: error: "DummyReply" has no attribute "not_exists"  [a
 test_negative/negative.py: error: Argument 1 of "UnaryUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "DummyRequest"  [override]
 test_negative/negative.py: note: This violates the Liskov substitution principle
 test_negative/negative.py: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py: error: Return type "Iterator[DummyReply]" of "UnaryUnary" incompatible with return type "DummyReply" in supertype "DummyServiceServicer"  [override]
+test_negative/negative.py: error: Return type "Iterator[DummyReply]" of "UnaryUnary" incompatible with return type "Union[DummyReply, Awaitable[DummyReply]]" in supertype "DummyServiceServicer"  [override]
 test_negative/negative.py: error: Argument 1 of "UnaryStream" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "DummyRequest"  [override]
 test_negative/negative.py: note: This violates the Liskov substitution principle
 test_negative/negative.py: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py: error: Return type "DummyReply" of "UnaryStream" incompatible with return type "Iterator[DummyReply]" in supertype "DummyServiceServicer"  [override]
-test_negative/negative.py: error: Argument 1 of "StreamUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "Iterator[DummyRequest]"  [override]
+test_negative/negative.py: error: Return type "DummyReply" of "UnaryStream" incompatible with return type "Union[Iterator[DummyReply], AsyncIterator[DummyReply]]" in supertype "DummyServiceServicer"  [override]
+test_negative/negative.py: error: Argument 1 of "StreamUnary" is incompatible with supertype "DummyServiceServicer"; supertype defines the argument type as "_MaybeAsyncIterator[DummyRequest]"  [override]
 test_negative/negative.py: note: This violates the Liskov substitution principle
 test_negative/negative.py: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides
-test_negative/negative.py: error: Return type "Iterator[DummyReply]" of "StreamUnary" incompatible with return type "DummyReply" in supertype "DummyServiceServicer"  [override]
-Found 80 errors in 1 file (checked 2 source files)
+test_negative/negative.py: error: Return type "Iterator[DummyReply]" of "StreamUnary" incompatible with return type "Union[DummyReply, Awaitable[DummyReply]]" in supertype "DummyServiceServicer"  [override]
+Found 80 errors in 1 file (checked 4 source files)
diff --git a/test_requirements.txt b/test_requirements.txt
index 71875ba8..065e71b9 100644
--- a/test_requirements.txt
+++ b/test_requirements.txt
@@ -2,6 +2,7 @@
 # generated code.
 protobuf==4.21.12
 pytest==7.2.1
-grpc-stubs==1.24.12
+pytest-asyncio==0.20.3
+grpc-stubs==1.24.12.1
 grpcio-tools==1.51.1
 types-protobuf==4.21.0.5