Streaming

fastgrpc supports all four gRPC streaming modes through Python's standard AsyncIterator (or Iterator for sync handlers).

Mode	Request	Response
Unary-unary	single value	single value
Server-streaming	single value	AsyncIterator[T]
Client-streaming	AsyncIterator[T]	single value
Bidirectional	AsyncIterator[T]	AsyncIterator[T]

Server streaming

The server pushes a stream of messages back to a single client request.

from typing import AsyncIterator
from dataclasses import dataclass
from fastgrpc import rpc, service

@dataclass
class Message:
    text: str

@service
class ChatService:
    @rpc
    async def stream_messages(self, room_id: str) -> AsyncIterator[Message]:
        async for msg in db.watch(room_id):
            yield Message(text=msg.text)

Generated proto:

service ChatService {
  rpc StreamMessages (StreamMessagesRequest) returns (stream Message);
}

Client streaming

The client pushes a stream; the server returns a single response.

@service
class UploadService:
    @rpc
    async def upload_chunks(self, chunks: AsyncIterator[Chunk]) -> UploadResult:
        total = 0
        async for chunk in chunks:
            total += len(chunk.data)
        return UploadResult(bytes_received=total)

Generated proto:

service UploadService {
  rpc UploadChunks (stream Chunk) returns (UploadResult);
}

Note: when the only non-self parameter is an AsyncIterator[T], fastgrpc uses T directly as the request type rather than synthesizing a {Method}Request wrapper.

Bidirectional streaming

Both sides push streams.

@service
class ChatService:
    @rpc
    async def chat(self, messages: AsyncIterator[Message]) -> AsyncIterator[Message]:
        async for msg in messages:
            yield Message(text=f"echo: {msg.text}")

Generated proto:

service ChatService {
  rpc Chat (stream Message) returns (stream Message);
}

Sync streaming

For sync services, use Iterator and def instead of AsyncIterator and async def:

from typing import Iterator

@service
class ChatService:
    @rpc
    def stream_messages(self, room_id: str) -> Iterator[Message]:
        for msg in db.watch_sync(room_id):
            yield Message(text=msg.text)

Calling streaming endpoints

With grpcurl:

# server streaming — multiple JSON objects come back, one per line
grpcurl -plaintext -d '{"room_id": "general"}' \
  127.0.0.1:50051 fastgrpc.ChatService/StreamMessages