Skip to content

Building Async gRPC Services with grpc.aio

grpc.aio is the asyncio API inside the standard grpcio package, and it is a genuinely different implementation from the threaded one — servicer methods are coroutines, calls are awaited, and there is no thread pool sized by hand. That removes an entire class of tuning problem: 1,000 concurrent unary calls over a single channel completed in 0.15 s in the measurements below, on one loop, with no executor involved. What it adds is the need to keep everything in the request path non-blocking, and a set of status-code conventions that decide what your clients see when things go wrong. This guide builds both ends against grpcio 1.84.

Prerequisites

  • Python 3.11+ with grpcio and grpcio-tools (pip install grpcio grpcio-tools).
  • A .proto file and the generated modules; the examples use a small Echo service.
  • Loop discipline from Async HTTP Clients & Servers — the same rule about blocking calls applies.
From .proto to a running server 5 stages from write the .proto to start and await. From .proto to a running server write the .proto the contract protoc generates pb2 and pb2_grpc implement it async def per method register it on grpc.aio.server start and await the loop serves it The generated code is the source of truth for both sides; check it into the repo or generate it in CI.

1. Generate the stubs, then implement the servicer

The contract comes first:

syntax = "proto3";
package demo;

service Echo {
  rpc Unary (Request) returns (Reply);
}

message Request { string text = 1; int32 count = 2; }
message Reply   { string text = 1; int32 index = 2; }

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. echo.proto

That produces echo_pb2.py (messages) and echo_pb2_grpc.py (the servicer base class and the client stub). Generate them in CI or check them in, but do it once and consistently — the most common gRPC build problem is generated code that no longer matches the .proto.

The servicer is an ordinary async class:

class Echo(echo_pb2_grpc.EchoServicer):
    async def Unary(self, request, context):
        return echo_pb2.Reply(text=f"echo:{request.text}", index=0)

Every method is async def, and everything inside must be non-blocking. A synchronous database driver in a servicer blocks the loop for all concurrent RPCs exactly as it does in an ASGI handler — use asyncio.to_thread or an async client.

Verify: a unary call returns the expected message; on loopback it took 4.8 ms including channel setup.

2. Start and stop the server deliberately

async def serve(stop: asyncio.Event) -> None:
    server = grpc.aio.server(options=[
        ("grpc.max_concurrent_streams", 100),
        ("grpc.keepalive_time_ms", 30_000),
    ])
    echo_pb2_grpc.add_EchoServicer_to_server(Echo(), server)
    port = server.add_insecure_port("127.0.0.1:50051")     # 0 asks the OS for a free port
    await server.start()                                   # returns immediately
    try:
        await stop.wait()
    finally:
        await server.stop(grace=5)                         # let in-flight RPCs finish

start() does not block, so the server needs something to wait on — an Event set by a signal handler, as in graceful shutdown and signals. stop(grace) stops accepting new RPCs and gives running ones up to grace seconds; stop(None) cancels them immediately, which is what you want only when the grace period has already expired.

add_insecure_port returns the bound port, which is what makes "127.0.0.1:0" useful in tests — no port collisions in CI.

Verify: an RPC in flight when shutdown begins completes rather than failing.

A server that shuts down cleanly 5 ordered steps. A server that shuts down cleanly grpc.aio.server(options=...) limits and keepalive add_ServicerToServer generated helper add_insecure_port or credentials returns the bound port await server.start() non-blocking await server.stop(grace) drains in-flight RPCs stop(grace) lets running RPCs finish; stop(None) cancels them immediately.

3. Reuse the channel

A channel is a connection pool with HTTP/2 multiplexing, and creating one per call throws away both:

channel = grpc.aio.insecure_channel(target)            # once, at startup
stub = echo_pb2_grpc.EchoStub(channel)
...
await channel.close()                                  # once, at shutdown

With one channel, 1,000 concurrent unary calls took 0.15 s — they share a single HTTP/2 connection, each as its own stream. The per-call channel version pays a TCP and HTTP/2 handshake every time.

In an ASGI service this belongs in the lifespan: create the channel at startup, close it at shutdown, and reach it from handlers through state. The stub itself is cheap and stateless — create it wherever it is convenient.

Two channel options worth setting for long-lived clients: grpc.keepalive_time_ms so an idle connection through a NAT is not silently dropped (the same problem as TCP keepalive for async connections), and grpc.max_receive_message_length when messages can exceed the 4 MB default.

Verify: the client's connection count stays at one under concurrent load.

4. Return status codes, not exceptions

gRPC has a status code vocabulary, and using it is the difference between a client that can react and one that can only log. The measured behaviours:

context.abort(INVALID_ARGUMENT, "...")  -> client: INVALID_ARGUMENT, "text must not be 'bad'"
raise RuntimeError("unhandled bug")     -> client: UNKNOWN, "Unexpected <class 'RuntimeError'>: unhandled bug"
server not listening                    -> client: UNAVAILABLE

The middle row is the one to fix. An unhandled exception becomes UNKNOWN and leaks the exception's repr to the caller — an information disclosure as well as a useless status. Convert deliberately:

    async def Unary(self, request, context):
        if not request.text:
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "text is required")
        try:
            return await self.handle(request)
        except NotFoundError:
            await context.abort(grpc.StatusCode.NOT_FOUND, "no such record")

context.abort() raises, so nothing after it runs. For a service-wide safety net, a server interceptor that catches everything and maps it to INTERNAL is the standard approach — see adding interceptors to grpc.aio.

On the client, failures arrive as grpc.aio.AioRpcError:

try:
    reply = await stub.Unary(request, timeout=1.0)
except grpc.aio.AioRpcError as exc:
    if exc.code() is grpc.StatusCode.UNAVAILABLE:
        ...                                            # retryable: nothing was processed

UNAVAILABLE and DEADLINE_EXCEEDED are the retry candidates, with the same idempotency caveat as classifying retryable errors: a deadline means the outcome is unknown, not that nothing happened.

Verify: every failure path returns a specific code, and no client ever sees UNKNOWN.

How a failure reaches the client A grid of 4 rows by 2 columns. How a failure reaches the client server does client sees details context.abort(INVALID_ARGUMENT) INVALID_ARGUMENT your message, intact raises an unhandled exception UNKNOWN the repr of the exception, leaked takes longer than the deadline DEADLINE_EXCEEDED server task is cancelled is not listening UNAVAILABLE no server involved The second row is why every service needs an interceptor that converts exceptions to statuses.

5. Keep metadata and context in the right places

context carries everything about the call that is not the message: metadata, the deadline, the peer, and the cancellation state.

    async def Unary(self, request, context):
        metadata = dict(context.invocation_metadata())
        tenant = metadata.get("x-tenant-id")
        context.set_trailing_metadata((("server-version", VERSION),))

Metadata keys are lower-case; a key ending in -bin carries bytes. Use it for cross-cutting concerns — authentication tokens, trace context, tenant ids — and keep business data in the message, where the schema documents it.

context.time_remaining() returns the seconds left on the caller's deadline, or None when there is none — verified as 2.01 for a 2-second deadline. Passing that budget to downstream calls is what makes a gRPC system's timeouts coherent, and is covered in propagating gRPC deadlines and cancellation.

Verify: metadata set by the client is visible in context.invocation_metadata().

gRPC or HTTP and JSON? 2 columns contrasting gRPC, HTTP + JSON. gRPC or HTTP and JSON? gRPC a typed contract schema enforced by generated code streaming in both directions deadlines propagate automatically binary: not readable by eye HTTP + JSON a convention schema is documentation, if any streaming needs SSE or chunking timeouts are per client debuggable with curl Measured here: 1,000 concurrent unary calls over one channel completed in 0.15 s.

Verification

A grpc.aio service is production-ready when:

  • Generated code matches the .proto, built by CI rather than by hand.
  • No servicer method blocks the loop, proven by loop lag under load.
  • One channel per target is created at startup and closed at shutdown.
  • Every failure maps to a status code, with no UNKNOWN reaching clients.
  • Shutdown uses a grace period and in-flight RPCs complete.
  • Cross-cutting data travels as metadata, not smuggled into messages.

Pitfalls & edge cases

  • Mixing grpc and grpc.aio. The synchronous stub in an async program blocks the loop; the APIs are not interchangeable.
  • A channel per call. Throws away HTTP/2 multiplexing and pays a handshake each time.
  • The 4 MB message limit. Larger payloads need grpc.max_receive_message_length on both ends, or streaming.
  • Blocking work in a servicer. One slow call stalls every concurrent RPC in the process.
  • server.stop(None) on shutdown. Cancels in-flight RPCs immediately; give a grace period first.
  • Forgetting await on abort(). In grpc.aio it is a coroutine; calling it without await silently does nothing.

Frequently Asked Questions

How do I write an async gRPC server in Python?

Generate stubs with grpc_tools.protoc, subclass the generated servicer with async def methods, register it on a grpc.aio.server(), bind a port with add_insecure_port, and await server.start(). Keep everything inside the methods non-blocking, and stop with await server.stop(grace).

What is the difference between grpc and grpc.aio in Python?

They are two implementations in the same package. The synchronous one uses a thread pool you size; grpc.aio runs on the event loop with coroutine servicers and awaitable calls. Do not mix them in one program — a synchronous stub called from a coroutine blocks the loop.

Should I create a new gRPC channel for each call?

No. A channel is a connection with HTTP/2 multiplexing, so one channel serves many concurrent calls — 1,000 concurrent unary RPCs completed in 0.15 s over a single channel here. Create it at startup, close it at shutdown, and create cheap stubs from it as needed.

Why does my gRPC client see UNKNOWN instead of a real error?

Because the servicer raised an exception that nothing converted to a status. gRPC maps it to UNKNOWN and includes the exception's repr in the details, which leaks internals. Call context.abort with a specific StatusCode, and add a server interceptor that maps anything unexpected to INTERNAL.

How do I pass authentication tokens with grpc.aio?

As call metadata: stub.Method(request, metadata=(("authorization", "Bearer ..."),)), or from a client interceptor that adds it to every call. On the server, read it with dict(context.invocation_metadata()); keys are lower-case and -bin suffixed keys carry bytes.