Skip to content

Streaming RPCs with grpc.aio

Streaming is where gRPC earns its keep over an HTTP API, and where grpc.aio reads best: a server-streaming method is an async generator, a client-streaming call takes an async generator, and a bidirectional call is two independent flows over one HTTP/2 stream. The parts that need care are not the syntax but the semantics — how much the sender may run ahead of the receiver (measured below: 6,222 messages produced while the client had read 50), what a failure halfway through a stream means, and the fact that a streaming call cannot be retried as a unit. This guide covers all three shapes against grpcio 1.84.

Prerequisites

The four RPC shapes and what each is for A grid of 4 rows by 2 columns. The four RPC shapes and what each is for shape python side use for unary reply = await stub.Method(req) ordinary request and response server streaming async for reply in stub.Method(req) feeds, tails, large results client streaming await stub.Method(async_gen()) uploads, batched writes bidirectional write() and read() on one call chat, control channels All four are the same HTTP/2 stream underneath; only who may send, and when, differs.

1. Server streaming: yield from the servicer

Declare returns (stream Reply) and the servicer becomes an async generator:

    async def ServerStream(self, request, context):
        try:
            async for record in query(request):
                yield echo_pb2.Reply(text=record.text, index=record.id)
        except asyncio.CancelledError:
            log.info("client went away mid-stream")
            raise                                      # always re-raise

The client consumes it with async for:

async for reply in stub.ServerStream(echo_pb2.Request(text="s", count=20)):
    handle(reply)

Twenty messages arrived in 0.20 s with a deliberate 10 ms gap between them — the client sees each one as it is produced rather than waiting for the whole set, which is the entire point for a feed, a log tail or a large result set.

The except asyncio.CancelledError is not optional in production code. A client that disconnects or times out cancels the servicer at its current yield, and any resource the generator holds — a database cursor, a file handle — must be released there. This is the same discipline as streaming HTTP responses.

Verify: the client receives messages incrementally, and abandoning the stream cancels the servicer.

2. Know how far the sender runs ahead

HTTP/2 flow control provides back-pressure, but its window is large. With 1 KB messages and a client that stalled after 50:

client read 50, server had produced 6222

Roughly 6 MB of messages were produced and buffered before the sender was throttled. For a cheap generator that is fine. For one where each message costs a database query, an API call or real computation, it means the server does 6,000 units of work for a client that may already be gone.

Where each message is expensive, pace it explicitly against the client's progress — an application-level acknowledgement on a bidirectional stream, or a request that includes a batch size the client asks for again. Tuning grpc.http2.bdp_probe and window options is possible but obscure; an explicit protocol is clearer and portable.

Verify: with an instrumented generator, compare messages produced against messages consumed under a slow client.

How far the server runs ahead of a slow reader 2 bars comparing client had read with the others. How far the server runs ahead of a slow reader client had read 50 messages server had produced 6,222 messages Messages of about 1 KB; the gap is the HTTP/2 flow-control window, not a bug. Back-pressure exists, but the window is megabytes: do not rely on it to pace expensive work.

3. Client streaming: pass an async generator

rpc ClientStream (stream Request) returns (Reply) takes an async iterable and returns one reply:

async def uploads():
    async for chunk in read_chunks(path):
        yield echo_pb2.Request(text=chunk)

reply = await stub.ClientStream(uploads())             # one awaited call

Verified with five requests carrying counts 0–4: the server summed them and returned 10. The servicer side iterates the incoming stream:

    async def ClientStream(self, request_iterator, context):
        total = 0
        async for request in request_iterator:
            total += request.count
        return echo_pb2.Reply(text="sum", index=total)

The generator is consumed as the call proceeds, so a large upload never exists in memory whole — the reason to prefer this over a single huge message, alongside the 4 MB default message limit.

Verify: the reply reflects every message sent, and memory stays flat for a large upload.

4. Bidirectional: read and write independently

async for works for bidirectional calls too, but the explicit API is what makes them useful, because the two directions are independent:

call = stub.BiDi()                                     # note: no request argument
await call.write(echo_pb2.Request(text="one", count=1))
first = await call.read()
await call.write(echo_pb2.Request(text="two", count=2))
second = await call.read()
await call.done_writing()                              # half-close: no more requests
assert await call.read() is grpc.aio.EOF               # the sentinel, not None
status = await call.code()                             # StatusCode.OK

Verified exactly as written: two request-and-reply exchanges, then EOF and OK. Three details matter. done_writing() half-closes the request side, which many servers wait for before finishing. The end of the response stream is the grpc.aio.EOF sentinel, not None — a truthiness check gets this wrong. And await call.code() gives the final status after the stream ends, which is where a server-side failure surfaces.

For genuinely concurrent duplex traffic — a control channel where either side may speak at any time — run the reader and writer as separate tasks under a TaskGroup, so neither blocks the other.

Verify: interleaved writes and reads complete, and the stream ends with EOF followed by a status.

Driving a bidirectional stream explicitly 5 ordered steps. Driving a bidirectional stream explicitly call = stub.BiDi() no request argument await call.write(req) as many as you like await call.read() interleaved freely await call.done_writing() half-close read until grpc.aio.EOF then await call.code() read() returns the EOF sentinel, not None, when the server has finished.

5. Handle failures with the stream's semantics in mind

Two properties distinguish a streaming call from a unary one, and both affect error handling.

A stream can fail after partial success. The client has already processed 500 messages when the server aborts. There is no rollback, so the protocol must let the client recover: a sequence number or cursor in each message, so a retry can resume from where it stopped, is the standard answer. Without one, the only safe retry is to start over and deduplicate.

A stream cannot be retried by generic middleware. gRPC's built-in retry policy applies to calls with no messages received yet; once the server has sent something, the call is not replayable. Interceptors that transparently retry unary calls must leave streams alone.

On the server side, aborting mid-stream is done the same way as anywhere:

            if not authorised(record):
                await context.abort(grpc.StatusCode.PERMISSION_DENIED, "not yours")

but the client sees it after the messages it already received — so anything it did with them has happened. Where partial results are unacceptable, send a terminal "complete" message as the last element and have the client treat its absence as failure.

Verify: a server-side abort after N messages leaves the client able to resume rather than having to restart.

Does this call need streaming? A decision on What does the data look like with 3 outcomes. Does this call need streaming? What does the data look like? one request, one answer unary simplest, and retryable many results, or produced over time server streaming bounded memory both ends both sides send, interleaved bidirectional a protocol of your own A streaming call is not retryable as a unit: partial progress is visible to the caller.

Verification

Streaming RPCs are used correctly when:

  • Servicers re-raise CancelledError and release resources in the process.
  • Producers are paced when each message is expensive, not left to HTTP/2 windows.
  • Uploads stream rather than being packed into one large message.
  • Bidirectional calls half-close with done_writing() and check for grpc.aio.EOF.
  • The final status is read after a stream ends, not assumed to be OK.
  • Resumption is designed in, with sequence numbers or cursors.

Pitfalls & edge cases

  • Checking if not reply for the end of a stream. The end is the grpc.aio.EOF sentinel; a valid empty message is falsy too.
  • Forgetting done_writing(). The server waits for a half-close that never comes, and the call hangs.
  • Relying on flow control to pace expensive work. Measured, the server ran 6,172 messages ahead.
  • Retrying a stream that already delivered messages. Duplicates, or re-executed side effects.
  • A blocking call inside a streaming servicer. It stalls every other RPC in the worker, not just this stream.
  • Very many concurrent streams. Each is an HTTP/2 stream against grpc.max_concurrent_streams; beyond it, calls queue invisibly.

Frequently Asked Questions

How do I write a server-streaming gRPC method in Python?

Declare it as returns (stream Reply) in the .proto and make the servicer method an async generator that yields reply messages. The client consumes it with async for. Wrap the generator body so asyncio.CancelledError releases resources and is re-raised, because a client disconnect cancels it.

How does back-pressure work in gRPC streaming?

Through HTTP/2 flow control, with a window measured in megabytes. Measured with 1 KB messages, the server had produced 6,222 while the client had read 50. That is fine for cheap messages and wasteful when each one costs real work, so pace expensive producers with an application-level acknowledgement.

How do I know when a grpc.aio stream has ended?

read() returns the grpc.aio.EOF sentinel. Compare with is grpc.aio.EOF rather than testing truthiness, since a legitimate empty message is also falsy. After the stream ends, await call.code() gives the final status.

Can I retry a streaming gRPC call?

Not once messages have been delivered. gRPC's retry policy covers calls where nothing has been received yet; after that the call is not replayable. Design the protocol so the client can resume — a sequence number or cursor per message — instead of relying on transparent retries.

What is done_writing() for in a bidirectional gRPC call?

It half-closes the request side, telling the server no more requests are coming. Many servers finish only after that signal, so omitting it makes the call hang. The response stream then ends with the EOF sentinel and a final status.