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

[saikishore-p]

Description

Adds dapr.ext.grpc.aio, an asyncio-native App backed by a grpc.aio server, so Dapr callback handlers can be async def.

dapr.ext.grpc.App runs on grpc.server() with a 10-thread pool, so handlers must be synchronous. The async client (dapr.aio.clients.DaprClient) has been at parity with the sync client for a while, so an asyncio application can call Dapr asynchronously but cannot serve Dapr callbacks asynchronously. This PR closes that gap; there are no client-side changes.

This follows the implementation notes left on #695:

Just like we have dapr.ext.grpc which export App and Rule, we should have dapr.ext.grpc.aio which export the same. The only usage difference should be this single import.

Swapping the import is the only change an app needs, beyond async def handlers and awaiting run()/stop():

-from dapr.ext.grpc import App, InvokeMethodRequest, InvokeMethodResponse
+from dapr.ext.grpc.aio import App, InvokeMethodRequest, InvokeMethodResponse

 app = App()

 @app.method(name='my-method')
-def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse:
+async def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse:
     ...

-app.run(13551)
+asyncio.run(app.run(13551))

Reviewing this

Two commits, and the first is the one worth scrutiny:

1. refactor: extract shared bases — touches only the three existing synchronous .py files, provably behaviour-preserving.
2. feat: add dapr.ext.grpc.aio — purely additive apart from docs.

Why the refactor

Of the servicer's 456 lines, roughly 250 — register_topic's routing, _get_topic_callback, the bulk entry builders — contain no async content at all, and that is exactly the code that has churned most recently (SubscriptionMessage delivery, _route_map, bulk events). Copying it is what left #829 unmergeable after six months: it was written before all of those landed, and the copy silently fell behind.

So the two servicers are siblings over a shared base:

_CallbackServicerBase       # registries, topic routing, request → event translation
├── _CallbackServicer       # synchronous gRPC entry points
└── _AioCallbackServicer    # asyncio gRPC entry points

Neither subclasses the other, so there is no sync/async mixing in one MRO and no misleading isinstance. This differs from the repo's other aio modules (DaprGrpcClientAsync, dapr.ext.workflow.aio.DaprWorkflowClient), which are standalone copies — but their duplication is inherent, since every method body differs by an await. Most of this servicer's body doesn't.

Compatibility: nothing is removed. Attribute lookup walks the MRO, so every method still resolves on _CallbackServicer, instance state is untouched, and isinstance against both proto servicer types still holds — including for code reaching into internals like app._servicer._registered_topics, which examples/pubsub-simple does. The only observable delta is class-__dict__ reflection on a doubly-private class. Of the 19 methods, 12 move byte-identical; the 7 entry points delegate to 10 extracted helpers. The 48 pre-existing tests/ext/grpc tests pass unmodified against commit 1 alone.

tests/ext/grpc/aio/test_servicer.py::AsyncParityTests then locks the arrangement in: every RPC the sync servicer implements must be mirrored on the asyncio servicer as a coroutine function, and the registration helpers must stay shared rather than be reimplemented. A future RPC added to only one side fails the build.

_AioHealthCheckServicer is a standalone copy — at ~30 lines, a shared base would cost more than it saves.

If you'd rather not carry the refactor, dropping commit 1 and inlining the shared methods into a standalone _AioCallbackServicer is mechanical — happy to do that instead.

Behaviour notes

• Full current surface, not a 2025 snapshot: service invocation, pub/sub with SubscriptionMessage delivery and the cloudevents deprecation path, topic rules, dead letter topics, disable_topic_validation, bulk topic events, job events on both the stable and alpha services, input bindings, health checks, and DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES.
• AppCallbackAlphaServicer is registered on the aio server, so OnJobEventAlpha1 and OnBulkTopicEventAlpha1 are actually reachable.
• send_initial_metadata is awaited. It is a coroutine on grpc.aio's context but a plain method on the sync one; calling it without await silently drops response headers.
• The server is built on the first run()/start(), not in __init__, because grpc.aio.server() binds to whichever event loop is current when it is called. add_external_service() therefore queues its registration and replays it at creation, so it still works before startup.
• start() is added as a non-blocking alternative to run(), for serving the app alongside other work on the same loop (e.g. an ASGI lifespan handler). run() is start() plus wait_for_termination().
• stop(grace=None) takes a grace period. There is no __del__ hook, since a coroutine can't be awaited from one.
• Plain functions are still accepted as handlers — results are awaited only when awaitable — so register_health_check(lambda: None) keeps working. Documented as running inline on the event loop.

Testing

• 78 unit tests under tests/ext/grpc/aio/, mirroring the sync suite, plus test_server.py, which runs a real grpc.aio server on an ephemeral port and drives it through generated stubs — covering invoke (data, content type, response headers, invocation metadata), topic events, bindings, job events, health checks, UNIMPLEMENTED routing, and lifecycle: cancellation during and after startup, concurrent starts, event-loop binding, bind failure, a cancelled graceful drain, and a slow handler not blocking a concurrent one.
• Two new examples with tests/examples/ coverage: invoke-simple-async and pubsub-simple-async.

Run locally on Python 3.10 (the lint/type floor and the bottom of the CI matrix) against a real Dapr 1.18.4 runtime — sidecar, placement, scheduler and Redis:

Suite	Command	Result
Unit	pytest -m "not e2e" ./tests --ignore=tests/integration --ignore=tests/examples	1713 passed
gRPC extension	pytest tests/ext/grpc/	126 passed
New asyncio tests	pytest tests/ext/grpc/aio/	78 passed
Integration	pytest tests/integration/ -m "not dapr_head"	89 passed
Examples	pytest tests/examples/	43 passed, 1 pre-existing failure
Types	mypy	clean, 186 source files
Lint / format	ruff check / ruff format --check	clean
unittest runner	python -m unittest discover ./tests/ext/grpc	OK

The single example failure is test_configuration, which fails identically on unmodified main in the same environment; it does not import dapr.ext.grpc and is untouched by this PR. dapr.ext.grpc.* is not in mypy's ignore list, so the new code is fully type-checked.

The first commit was also verified in isolation: with only refactor: extract shared bases... applied, the 48 pre-existing tests/ext/grpc tests pass unmodified and the full unit suite reports 1635 — exactly the pre-asyncio baseline.

pubsub-simple-async is the server-side callback model; the existing pubsub-streaming-async is the client-side streaming API. Both READMEs now cross-reference each other so the distinction is clear.

Issue reference

Please reference the issue this PR will close: #695

Some history, since it isn't obvious from the issue: #695 was closed as completed on 2026-04-01, but nothing in the repo implements it. The prior attempt, #829, was auto-closed for inactivity after 60 days. That PR was written against the old ext/dapr-ext-grpc/ layout and predates SubscriptionMessage, the _route_map routing rewrite, bulk topic events and the stable OnJobEvent, so it could not simply be rebased — this is a fresh implementation against current main.

We hit this in production and ended up vendoring our own asyncio callback server to work around it, which is what prompted picking the work back up.

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Extended the documentation