Add asyncio support to the gRPC app extension (dapr.ext.grpc.aio)

AGENTS.md

@@ -57,7 +57,7 @@ Extensions are bundled into the core `dapr` wheel and exposed as installable ext
 | Extra | Import path | Purpose | Active development |
 |-------|-------------|---------|--------------------|
 | `dapr[workflow]` | `dapr.ext.workflow` | Durable workflow orchestration (durabletask vendored internally) | **High**, major focus area |
-| `dapr[grpc]` | `dapr.ext.grpc` | gRPC server for Dapr callbacks (methods, pub/sub, bindings, jobs) | Moderate |
+| `dapr[grpc]` | `dapr.ext.grpc`, `dapr.ext.grpc.aio` | gRPC server for Dapr callbacks (methods, pub/sub, bindings, jobs), sync and asyncio | Moderate |
 | `dapr[fastapi]` | `dapr.ext.fastapi` | FastAPI integration for pub/sub and actors | Moderate |
 | `dapr[flask]` | `dapr.ext.flask` | Flask integration for pub/sub and actors (legacy `flask_dapr` import path is a deprecated shim) | Low |
 | `dapr[langgraph]` | `dapr.ext.langgraph` | LangGraph checkpoint persistence to Dapr state store | Moderate |

dapr/ext/grpc/AGENTS.md

@@ -7,16 +7,26 @@ The gRPC extension provides a **server-side callback framework** for Dapr applic
 ```
 dapr/ext/grpc/
 ├── __init__.py                    # Public API exports
-├── app.py                         # App class — main entry point
-├── _servicer.py                   # _CallbackServicer — internal routing
-├── _health_servicer.py            # _HealthCheckServicer
+├── app.py                         # App class — main entry point (sync)
+├── _servicer.py                   # _CallbackServicerBase (shared) + _CallbackServicer (sync)
+├── _health_servicer.py            # _HealthCheckServicerBase (shared) + _HealthCheckServicer
+├── aio/                           # asyncio-native variant, same public surface
+│   ├── __init__.py                #   Public API exports
+│   ├── app.py                     #   App class backed by grpc.aio
+│   ├── _servicer.py               #   _AioCallbackServicer — async gRPC entry points
+│   └── _health_servicer.py        #   _AioHealthCheckServicer
 └── py.typed
 
 tests/ext/grpc/
 ├── test_app.py                    # Decorator registration tests
 ├── test_servicer.py               # Routing, handlers, bulk events
 ├── test_health_servicer.py        # Health check tests
-└── test_topic_event_response.py   # Response status tests
+├── test_topic_event_response.py   # Response status tests
+└── aio/
+    ├── test_app.py                # Decorator registration, lazy server creation, lifecycle
+    ├── test_servicer.py           # Async handlers, sync/aio parity guards
+    ├── test_health_servicer.py    # Async and sync health check callbacks
+    └── test_server.py             # End-to-end over a real grpc.aio server
 ```
 
 Installed via the `grpc` extra on core dapr: `pip install "dapr[grpc]"`.
@@ -42,6 +52,8 @@ from dapr.ext.grpc import (
 
 Note: `InvokeMethodRequest`, `InvokeMethodResponse`, `BindingRequest`, `TopicEventResponse`, `Job`, `JobEvent`, and failure policies are actually defined in the core SDK (`dapr/clients/grpc/`) and re-exported here.
 
+`dapr.ext.grpc.aio` exports the same names. Swapping the import is the only change an app needs, beyond `async def` handlers and awaiting `run()`/`stop()`.
+
 ## App class (`app.py`)
 
 The central entry point. Creates a gRPC server and provides decorators for handler registration.
@@ -88,9 +100,44 @@ app.register_health_check(lambda: None)  # Not a decorator — direct registrati
 - `TopicEventResponse('success'|'retry'|'drop')` → explicit status
 - `None` → defaults to SUCCESS
 
+## Asyncio app (`aio/`)
+
+`dapr.ext.grpc.aio.App` is the asyncio counterpart, backed by `grpc.aio.server()`. Same decorators, same registration semantics, same wire behavior.
+
+```python
+import asyncio
+from dapr.ext.grpc.aio import App
+
+app = App()
+
+@app.subscribe(pubsub_name='pubsub', topic='orders')
+async def handle_event(event: SubscriptionMessage) -> TopicEventResponse:
+    ...
+
+asyncio.run(app.run(3010))
+```
+
+Differences from the synchronous `App`, all of them forced by the async runtime:
+
+- **`run()` and `stop()` are coroutines.** `stop(grace=None)` takes a grace period (the sync `stop()` is always immediate). There is no `__del__` hook, because a coroutine cannot be awaited from one — `run()` instead stops the server in a `finally`, so a cancelled app does not leave its port bound.
+- **`start()` exists** as a non-blocking alternative to `run()`, for serving the app alongside other work on the same loop (e.g. from an ASGI lifespan handler). `run()` is `start()` plus `wait_for_termination()`.
+- **The server is built lazily**, on the first `run()`/`start()` call, not in `__init__`. `grpc.aio.server()` binds to whichever event loop is current when it is called, so building it in `__init__` would attach it to the wrong loop. `add_external_service()` therefore queues its registration and replays it when the server is created, and raises if called once the app is running.
+- **`start()` and `stop()` are serialised** by an `asyncio.Lock`. grpc.aio segfaults if a `stop()` call is *concurrently in flight* with a `start()` call, so the two must never overlap; the lock also means a restart waits for an in-progress drain rather than binding a second server to the same port. (Stopping a server whose own `start()` has already unwound — the cleanup path in `_start`'s `except` — is sequential, not concurrent, and is safe. `stop()` on a server that never started returns cleanly; on one whose `start()` was *cancelled part-way* it raises `InvalidStateError`, which that path suppresses.)
+- **The lock is built per running loop**, not in `__init__`: an `asyncio.Lock` binds to the loop of its first *contended* acquire, so one built at construction raises "bound to a different event loop" on a second `asyncio.run()` — and silently stops excluding anything before that, because the uncontended path returns before the loop check.
+- **Decorators return the handler**, so the decorated name stays bound. The sync decorators return `None`.
+- **Handlers may be plain functions.** Results are awaited only when awaitable, so `register_health_check(lambda: None)` still works. A plain handler runs inline on the event loop and must not block.
+
+### Sharing with the sync implementation
+
+`_CallbackServicerBase` (in `_servicer.py`) holds everything that does not invoke a user handler: the handler registries, topic routing, and the request→event translation. `_CallbackServicer` and `_AioCallbackServicer` are **siblings** on top of it — neither subclasses the other — and each supplies only the gRPC entry points.
+
+This keeps the churn-prone routing logic (`_get_topic_callback`, `register_topic`, the bulk entry builders) in one place while leaving the two servicers free to differ where they must. `tests/ext/grpc/aio/test_servicer.py::AsyncParityTests` enforces the arrangement: every RPC the sync servicer implements must be mirrored on the aio servicer as a coroutine function, and the registration helpers must stay shared rather than be reimplemented.
+
+`_HealthCheckServicerBase` splits the health servicer the same way: registration in the base, the gRPC entry point in each sibling.
+
 ## Internal routing (`_servicer.py`)
 
-`_CallbackServicer` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces. It maintains internal registries:
+`_CallbackServicerBase` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces; `_CallbackServicer` adds the synchronous entry points. It maintains internal registries:
 
 - `_invoke_method_map` — method name → handler
 - `_topic_map` — topic key → handler
@@ -124,6 +171,13 @@ app.register_health_check(lambda: None)  # Not a decorator — direct registrati
 uv run python -m unittest discover -v ./tests/ext/grpc
 ```
 
+`unittest discover` covers the whole tree including `aio/` — `IsolatedAsyncioTestCase` and
+`subTest` are both native unittest. pytest runs the same tests:
+
+```bash
+uv run pytest ./tests/ext/grpc
+```
+
 Test patterns:
 - `test_app.py` — decorator registration, health check registration
 - `test_servicer.py` — handler invocation with mock gRPC context, return type handling (str, bytes, proto, response object), topic subscriptions, bulk events, bindings, duplicate registration errors
@@ -132,8 +186,8 @@ Test patterns:
 
 ## Key details
 
-- **Synchronous only**: Uses `grpc.server()` with `ThreadPoolExecutor(10)`. No async handler support.
+- **Sync app threading**: `dapr.ext.grpc.App` uses `grpc.server()` with `ThreadPoolExecutor(10)`. For `async def` handlers use `dapr.ext.grpc.aio.App`, which serves on a `grpc.aio` event loop instead.
 - **Default port**: 3010 (from `dapr.conf.global_settings.GRPC_APP_PORT`)
-- **Topic handler event type**: inferred from the handler annotation. Annotating the event parameter with `dapr.ext.grpc.SubscriptionMessage` — the same SDK-owned type the streaming subscription API (`DaprClient.subscribe`) delivers, with `metadata()` populated from the gRPC invocation metadata — delivers that type. Unannotated or otherwise-annotated handlers receive the DEPRECATED `cloudevents.sdk.event.v1.Event` and `subscribe()` emits a `DeprecationWarning` at registration. Deprecation timeline: 1.20 delivers `SubscriptionMessage` to unannotated handlers (legacy only via explicit `v1.Event` annotation), 1.21 drops `cloudevents` from the `grpc` extra (import becomes conditional), 1.22 removes the legacy path entirely (same release the `flask_dapr` shim goes away). New code must annotate with `SubscriptionMessage`. (Internally the choice is plumbed through `_CallbackServicer.register_topic(legacy_cloudevent=...)`.)
+- **Topic handler event type**: inferred from the handler annotation. Annotating the event parameter with `dapr.ext.grpc.SubscriptionMessage` — the same SDK-owned type the streaming subscription API (`DaprClient.subscribe`) delivers, with `metadata()` populated from the gRPC invocation metadata — delivers that type. Unannotated or otherwise-annotated handlers receive the DEPRECATED `cloudevents.sdk.event.v1.Event` and `subscribe()` emits a `DeprecationWarning` at registration. Deprecation timeline: 1.20 delivers `SubscriptionMessage` to unannotated handlers (legacy only via explicit `v1.Event` annotation), 1.21 drops `cloudevents` from the `grpc` extra (import becomes conditional), 1.22 removes the legacy path entirely (same release the `flask_dapr` shim goes away). New code must annotate with `SubscriptionMessage`. (Internally the choice is plumbed through `_CallbackServicerBase.register_topic(legacy_cloudevent=...)`.)
 - **Duplicate registration**: Registering the same method/topic/binding name twice raises `ValueError`
 - **Missing handlers**: Calling an unregistered method/topic/binding raises `NotImplementedError` (gRPC UNIMPLEMENTED)

dapr/ext/grpc/README.md

@@ -12,5 +12,32 @@ pip install "dapr[grpc]"
 from dapr.ext.grpc import App
 ```
 
+An asyncio-native app backed by a `grpc.aio` server is available under
+`dapr.ext.grpc.aio`. It exposes the same names and decorators; handlers may be
+`async def` and are awaited, and `run()`/`stop()` are coroutines:
+
+```python
+import asyncio
+
+from dapr.ext.grpc.aio import App, InvokeMethodRequest, InvokeMethodResponse
+
+app = App()
+
+
+@app.method(name='my-method')
+async def my_method(request: InvokeMethodRequest) -> InvokeMethodResponse:
+    ...
+
+
+asyncio.run(app.run(50051))
+```
+
+Plain (non-async) handlers are still accepted, but they run inline on the event
+loop, so they must not block. See the [`invoke-simple-async`][invoke-async] and
+[`pubsub-simple-async`][pubsub-async] examples.
+
 See the root [README](../../../README.md) for migration steps from the legacy
-`dapr-ext-grpc` distribution.
+`dapr-ext-grpc` distribution.
+
+[invoke-async]: ../../../examples/invoke-simple-async
+[pubsub-async]: ../../../examples/pubsub-simple-async

dapr/ext/grpc/_health_servicer.py

@@ -1,33 +1,49 @@
-from typing import Callable, Optional
+from typing import Awaitable, Callable, Optional
 
 import grpc
 
+from dapr.ext.grpc._servicer import _ServicerContext
 from dapr.proto import appcallback_service_v1
 from dapr.proto.runtime.v1.appcallback_pb2 import HealthCheckResponse
 
 HealthCheckCallable = Optional[Callable[[], None]]
+# The asyncio servicer awaits its callback, so it also accepts an awaitable-returning one.
+AsyncHealthCheckCallable = Callable[[], Optional[Awaitable[None]]]
 
 
-class _HealthCheckServicer(appcallback_service_v1.AppCallbackHealthCheckServicer):
-    """The implementation of HealthCheck Server.
+class _HealthCheckServicerBase(appcallback_service_v1.AppCallbackHealthCheckServicer):
+    """Health check registration shared by the sync and asyncio servicers.
 
-    :class:`App` provides useful decorators to register method, topic, input bindings.
+    Mirrors the :class:`_CallbackServicerBase` arrangement: everything that does not invoke
+    the callback lives here, and each servicer supplies only the gRPC entry point.
     """
 
     def __init__(self):
-        self._health_check_cb: Optional[HealthCheckCallable] = None
+        self._health_check_cb: Optional[AsyncHealthCheckCallable] = None
 
-    def register_health_check(self, cb: HealthCheckCallable) -> None:
+    def register_health_check(self, cb: Optional[AsyncHealthCheckCallable]) -> None:
         if not cb:
             raise ValueError('health check callback must be defined')
         self._health_check_cb = cb
 
-    def HealthCheck(self, request, context):
-        """Health check."""
-
+    def _require_health_check_cb(self, context: _ServicerContext) -> AsyncHealthCheckCallable:
+        """Returns the registered callback, or marks the RPC UNIMPLEMENTED if there is none."""
         if not self._health_check_cb:
             context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore
             context.set_details('Method not implemented!')
             raise NotImplementedError('Method not implemented!')
-        self._health_check_cb()
+        return self._health_check_cb
+
+
+class _HealthCheckServicer(_HealthCheckServicerBase):
+    """The synchronous HealthCheck servicer.
+
+    Shares registration with the asyncio servicer via their common base; only this gRPC entry
+    point differs, calling the callback directly rather than awaiting it.
+    """
+
+    def HealthCheck(self, request, context):
+        """Health check."""
+        health_check = self._require_health_check_cb(context)
+        health_check()
         return HealthCheckResponse()