Getting Started with gRPC-Python - Streaming

1. Introduction

In this codelab, you'll use gRPC-Python to create a client and server that form the foundation of a route-mapping application written in Python.

By the end of the tutorial, you will have a client that connects to a remote server using gRPC to get information about features on a client's route, create a summary of a client's route, and exchange route information such as traffic updates with the server and other clients.

The service is defined in a Protocol Buffers file, which will be used to generate boilerplate code for the client and server so that they can communicate with each other, saving you time and effort in implementing that functionality.

This generated code takes care of not only the complexities of the communication between the server and client, but also data serialization and deserialization.

What you'll learn

• How to use Protocol Buffers to define a service API.
• How to build a gRPC-based client and server from a Protocol Buffers definition using automated code generation.
• An understanding of client-server streaming communication with gRPC.

This codelab is aimed at Python developers new to gRPC or seeking a refresher of gRPC, or anyone else interested in building distributed systems. No prior gRPC experience is required.

2. Before you begin

What you'll need

• Python 3.9 or higher. We recommend Python 3.13. For platform-specific installation instructions, see Python Setup and Usage. Alternatively, install a non-system Python using tools like uv or pyenv.
• pip to install Python packages.
• venv to create Python virtual environments.

The ensurepip and venv packages are part of the Python Standard Library and are typically available by default.

However, some Debian-based distributions (including Ubuntu) choose to exclude them when redistributing python. To install the packages, run:

sudo apt install python3-pip python3-venv

Get the code

To streamline your learning, this codelab offers a pre-built source code scaffold to help you get started. The following steps will guide you through completing the application, including gRPC code generation using the grpc_tools.protoc Protocol Buffer compiler plugin.

grpc-codelabs

The scaffold source code for this codelab is available in the codelabs/grpc-python-streaming/start_here directory. If you prefer not to implement the code yourself, the completed source code is available in the completed directory.

First, create the codelab working directory and cd into it:

mkdir grpc-python-streaming && cd grpc-python-streaming

Download and extract the codelab:

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/v1.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-1/codelabs/grpc-python-streaming/start_here

Alternatively, you can download the .zip file containing only the codelab directory and manually unzip it.

Download Zip

3. Define messages and services

Your first step is to define the application's gRPC service, its RPC method, and its request and response message types using Protocol Buffers. Your service will provide:

• RPC methods called ListFeatures, RecordRoute, and RouteChat that the server implements and the client calls.
• The message types Point, Feature, Rectangle, RouteNote, and RouteSummary, which are data structures exchanged between the client and server when calling the RPC methods.

These RPC methods and their message types will all be defined in the protos/route_guide.proto file of the provided source code.

Protocol Buffers are commonly known as protobufs. For more information on gRPC terminology, see gRPC's Core concepts, architecture, and lifecycle.

Define message types

In the protos/route_guide.proto file of the source code, first define the Point message type. A Point represents a latitude-longitude coordinate pair on a map. For this codelab, use integers for the coordinates:

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

The numbers 1 and 2 are unique ID numbers for each of the fields in the message structure.

Next, define the Feature message type. A Feature uses a string field for the name or postal address of something at a location specified by a Point:

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

So that multiple points within an area can be streamed to a client, you'll need a Rectangle message that represents a latitude-longitude rectangle, represented as two diagonally opposite points lo and hi:

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

Also, a RouteNote message that represents a message sent while at a given point:

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

Finally, you'll need a RouteSummary message. This message is received in response to a RecordRoute RPC, which is explained in the next section. It contains the number of individual points received, the number of detected features, and the total distance covered as the cumulative sum of the distance between each point.

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

Define service methods

To define a service, you specify a named service in your .proto file. The route_guide.proto file has a service structure named RouteGuide that defines one or more methods provided by the application's service.

When you define RPC methods inside your service definition, you specify their request and response types. In this section of the codelab, let's define:

ListFeatures

Obtains the Feature objects available within the given Rectangle. Results are streamed rather than returned at once as the rectangle may cover a large area and contain a huge number of features.

For this application, you'll use a server-side streaming RPC: the client sends a request to the server and gets a stream to read a sequence of messages back. The client reads from the returned stream until there are no more messages. As you can see in our example, you specify a server-side streaming method by placing the stream keyword before the response type.

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

Accepts a stream of Points on a route being traversed, returning a RouteSummary when traversal is completed.

A client-side streaming RPC is appropriate in this case: the client writes a sequence of messages and sends them to the server, again using a provided stream. Once the client has finished writing the messages, it waits for the server to read them all and return its response. You specify a client-side streaming method by placing the stream keyword before the request type.

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

Accepts a stream of RouteNotes sent while a route is being traversed, while receiving other RouteNotes (e.g. from other users).

This is exactly the kind of usecase for bidirectional streaming. A bidirectional streaming RPC where both sides send a sequence of messages using a read-write stream. The two streams operate independently, so clients and servers can read and write in whatever order they like: for example, the server could wait to receive all the client messages before writing its responses, or it could alternately read a message then write a message, or some other combination of reads and writes. The order of messages in each stream is preserved. You specify this type of method by placing the stream keyword before both the request and the response.

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. Generate the client and server code

Next, generate the boilerplate gRPC code for both the client and server from the .proto file using the protocol buffer compiler.

For gRPC Python code generation, we created grpcio-tools. It includes:

1. The regular protoc compiler that generates Python code from message definitions.
2. gRPC protobuf plugin that generates Python code (client and server stubs) from the service definitions.

We'll install the grpcio-tools Python package using pip. Let's create a new python virtual environment (venv) to isolate your project's dependencies from the system packages:

python3 -m venv --upgrade-deps .venv

To activate the virtual environment in bash/zsh shell:

source .venv/bin/activate

For Windows and non-standard shells, see the table at https://docs.python.org/3/library/venv.html#how-venvs-work.

Note: you'll need to activate the virtual environment every time you open a new shell. The activate script configures your $PATH variable to use python and pip from your venv. To deactivate the venv, simply run deactivate.

Next, install the grpcio-tools (this also installs the grpcio package):

pip install grpcio-tools

Use the following command to generate the Python boilerplate code:

python -m grpc_tools.protoc --proto_path=./protos  \
 --python_out=. --pyi_out=. --grpc_python_out=. \
 ./protos/route_guide.proto

This will generate the following files for the interfaces we defined in route_guide.proto:

1. route_guide_pb2.py contains the code that dynamically creates classes generated from the message definitions.
2. route_guide_pb2.pyi is a "stub file" or "type hint file" generated from the message definitions. It only contains the signatures with no implementation. Stub files can be used by IDEs to provide better autocompletion and error detection.
3. route_guide_pb2_grpc.py is generated from the service definitions and contains gRPC-specific classes and functions.

Fact: The _pb2 suffix doesn't imply the proto2 syntax. It indicates the version of Python Protocol Buffers API. Currently proto2, proto3, and Editions all use _pb2.py for their generated files.

gRPC-specific code contains:

1. RouteGuideStub, which can be used by a gRPC client to invoke RouteGuide RPCs.
2. RouteGuideServicer, which defines the interface for implementations of the RouteGuide service.
3. add_RouteGuideServicer_to_server function which is used to register a RouteGuideServicer to a gRPC server.

5. Create the server

First let's look at how you create a RouteGuide server. Creating and running a RouteGuide server breaks down into two work items:

• Implementing the servicer interface generated from our service definition with functions that perform the actual "work" of the service.
• Running a gRPC server to listen for requests from clients and transmit responses.

Let's look at route_guide_server.py.

Implement RouteGuide

route_guide_server.py has a RouteGuideServicer class that subclasses the generated class route_guide_pb2_grpc.RouteGuideServicer:

# RouteGuideServicer provides an implementation of the methods of the RouteGuide service.
class RouteGuideServicer(route_guide_pb2_grpc.RouteGuideServicer):

RouteGuideServicer implements all the RouteGuide service methods.

Server-side streaming RPC

ListFeatures is a response-streaming RPC that sends multiple Features to the client:

def ListFeatures(self, request, context):
    """List all features contained within the given Rectangle."""
    left = min(request.lo.longitude, request.hi.longitude)
    right = max(request.lo.longitude, request.hi.longitude)
    top = max(request.lo.latitude, request.hi.latitude)
    bottom = min(request.lo.latitude, request.hi.latitude)
    for feature in self.db:
        lat, lng = feature.location.latitude, feature.location.longitude
        if left <= lng <= right and bottom <= lat <= top:
            yield feature

Here the request message is a route_guide_pb2.Rectangle within which the client wants to find Features. Instead of returning a single response the method yields zero or more responses.

Client-side streaming RPC

The request-streaming method RecordRoute uses an iterator of request values and returns a single response value.

def RecordRoute(self, request_iterator, context):
    """Calculate statistics about the trip composed of Points."""
    point_count = 0
    feature_count = 0
    distance = 0.0
    prev_point = None

    start_time = time.time()
    for point in request_iterator:
        point_count += 1
        if get_feature(self.db, point):
            feature_count += 1
        if prev_point:
            distance += get_distance(prev_point, point)
        prev_point = point

    elapsed_time = time.time() - start_time
    return route_guide_pb2.RouteSummary(
        point_count=point_count,
        feature_count=feature_count,
        distance=int(distance),
        elapsed_time=int(elapsed_time),
    )

Bidirectional streaming RPC

Finally, let's look at our bidirectional streaming RPC RouteChat():

def RouteChat(self, request_iterator, context):
    """
    Receive a stream of message/location pairs, and responds with
    a stream of all previous messages for the given location.
    """
    prev_notes = []
    for new_note in request_iterator:
        for prev_note in prev_notes:
            if prev_note.location == new_note.location:
                yield prev_note
        prev_notes.append(new_note)

This method's semantics are a combination of those of the request-streaming method and the response-streaming method. It is passed an iterator of request values and is itself an iterator of response values.

Start the server

Once you have implemented all the RouteGuide methods, the next step is to start up a gRPC server so that clients can actually use your service:

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    route_guide_pb2_grpc.add_RouteGuideServicer_to_server(
        RouteGuideServicer(),
        server,
    )
    listen_addr = "localhost:50051"
    server.add_insecure_port(listen_addr)
    print(f"Starting server on {listen_addr}")
    server.start()
    server.wait_for_termination()

The server start() method is non-blocking. A new thread will be instantiated to handle requests. The thread calling server.start() will often not have any other work to do in the meantime. In this case, you can call server.wait_for_termination() to cleanly block the calling thread until the server terminates.

6. Create the client

Let's look at route_guide_client.py.

Create a stub

To call service methods, we first need to create a stub.

We instantiate the RouteGuideStub class of the route_guide_pb2_grpc module, generated from our .proto. In run() method:

with grpc.insecure_channel("localhost:50051") as channel:
    stub = route_guide_pb2_grpc.RouteGuideStub(channel)

Note that here channel is used as a context manager, and will be automatically closed once the interpreter leaves the with block.

Call service methods

For RPC methods that return a single response ("response-unary" methods), gRPC Python supports both synchronous (blocking) and asynchronous (non-blocking) control flow semantics. For response-streaming RPC methods, calls immediately return an iterator of response values. Calls to that iterator's next() method block until the response to be yielded from the iterator becomes available.

Server-side streaming RPC

Calling the response-streaming ListFeatures is similar to working with sequence types:

def guide_list_features(stub):
    _lo = route_guide_pb2.Point(latitude=400000000, longitude=-750000000)
    _hi = route_guide_pb2.Point(latitude=420000000, longitude=-730000000)
    rectangle = route_guide_pb2.Rectangle(
        lo=_lo,
        hi=_hi,
    )
    print("Looking for features between 40, -75 and 42, -73")

    features = stub.ListFeatures(rectangle)
    for feature in features:
        print(
            f"Feature called '{feature.name}'"
            f" at {format_point(feature.location)}"
        )

Client-side streaming RPC

Calling the request-streaming RecordRoute is similar to passing an iterator to a local method. Like the simple RPC above that also returns a single response, it can be called synchronously:

def guide_record_route(stub):
    feature_list = route_guide_resources.read_route_guide_database()
    route_iterator = generate_route(feature_list)

    route_summary = stub.RecordRoute(route_iterator)
    print(f"Finished trip with {route_summary.point_count} points")
    print(f"Passed {route_summary.feature_count} features")
    print(f"Traveled {route_summary.distance} meters")
    print(f"It took {route_summary.elapsed_time} seconds")

Bidirectional streaming RPC

Calling the bidirectionally-streaming RouteChat has (as is the case on the service-side) a combination of the request-streaming and response-streaming semantics.

Generate the request messages, and send them one by one using yield.

def generate_notes():
    home = route_guide_pb2.Point(latitude=1, longitude=1)
    work = route_guide_pb2.Point(latitude=2, longitude=2)
    notes = [
        make_route_note("Departing from home", home),
        make_route_note("Arrived at work", work),
        make_route_note("Having lunch at work", work),
        make_route_note("Departing from work", work),
        make_route_note("Arrived home", home),
    ]
    for note in notes:
        print(
            f"Sending RouteNote for {format_point(note.location)}:"
            f" {note.message}"
        )
        yield note
        # Sleep to simulate moving from one point to another.
        # Only for demonstrating the order of the messages.
        time.sleep(0.1)

Note: In this example, we sleep for a brief time before sending each message to illustrate the order in which messages are sent and received. In practice, you may design a protocol where these operations occur independently.

Receive and process server responses:

def guide_route_chat(stub):
    responses = stub.RouteChat(generate_notes())
    for response in responses:
        print(
            "< Found previous note at"
            f" {format_point(response.location)}: {response.message}"
        )

Call the helper methods

In run, execute the methods we just created, and pass them the stub.

print("-------------- ListFeatures --------------")
guide_list_features(stub)
print("-------------- RecordRoute --------------")
guide_record_route(stub)
print("-------------- RouteChat --------------")
guide_route_chat(stub)

7. Try it out

Run the server:

python route_guide_server.py

From a different terminal, activate the virtual environment again (source .venv/bin/activate), then run the client:

python route_guide_client.py

Let's take a look at the output.

ListFeatures

First, you'll find the list of features. Each feature is streamed from the server (server-side streaming RPC) as it discovers them to be within the requested rectangle:

-------------- ListFeatures --------------
Looking for features between 40, -75 and 42, -73
Feature called 'Patriots Path, Mendham, NJ 07945, USA' at (lat=407838351, lng=-746143763)
Feature called '101 New Jersey 10, Whippany, NJ 07981, USA' at (lat=408122808, lng=-743999179)
Feature called 'U.S. 6, Shohola, PA 18458, USA' at (lat=413628156, lng=-749015468)
Feature called '5 Conners Road, Kingston, NY 12401, USA' at (lat=419999544, lng=-740371136)
...

RecordRoute

Second, RecordRoute demonstrates the list of randomly visited points streamed from the client to the server (client-side streaming RPC):

-------------- RecordRoute --------------
Visiting point (lat=410395868, lng=-744972325)
Visiting point (lat=404310607, lng=-740282632)
Visiting point (lat=403966326, lng=-748519297)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=406589790, lng=-743560121)
Visiting point (lat=410322033, lng=-747871659)
Visiting point (lat=415464475, lng=-747175374)
Visiting point (lat=407586880, lng=-741670168)
Visiting point (lat=402647019, lng=-747071791)
Visiting point (lat=414638017, lng=-745957854)

After the client finished streaming all visited points, it'll receive a non-streaming response (a unary RPC) from the server. This response will contain a summary of the calculations performed on the client's full route.

Finished trip with 10 points
Passed 10 features
Traveled 654743 meters
It took 0 seconds

RouteChat

Last, the RouteChat output demonstrates bi-directional streaming. When the client is "visiting" home or work points, it records a note for the point by sending a RouteNote to the server. When a point has already been visited, the server streams back all previous notes for this point.

-------------- RouteChat --------------
Sending RouteNote for (lat=1, lng=1): Departing from home
Sending RouteNote for (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Having lunch at work
< Found previous note at (lat=2, lng=2): Arrived at work
Sending RouteNote for (lat=2, lng=2): Departing from work
< Found previous note at (lat=2, lng=2): Arrived at work
< Found previous note at (lat=2, lng=2): Having lunch at work
Sending RouteNote for (lat=1, lng=1): Arrived home
< Found previous note at (lat=1, lng=1): Departing from home

8. What's next

• Learn how gRPC works in Introduction to gRPC and Core concepts
• Work through the Basics tutorial
• Explore the Python API reference