A Unary RPC End to End

Implement the servicer, call the stub

On the server, subclass the generated servicer, implement each method (receiving the request message and a context), register it with a server, and listen on a port. On the client, create a channel to the server address and call methods on the generated stub, ideally with a deadline. Errors are reported by aborting with a status code on the server and raised as RpcError on the client.

A Python server and client

A sketch using grpcio; production servers need TLS.

# server.py
from concurrent import futures
import grpc
from gen.shop.v1 import order_pb2, order_pb2_grpc

class OrderService(order_pb2_grpc.OrderServiceServicer):
    def GetOrder(self, request, context):
        order = repo.find(request.id)
        if order is None:
            context.abort(grpc.StatusCode.NOT_FOUND, f"order {request.id} not found")
        return order_pb2.Order(id=order.id, total_cents=order.total_cents, currency="INR")

server = grpc.server(futures.ThreadPoolExecutor(max_workers=16))
order_pb2_grpc.add_OrderServiceServicer_to_server(OrderService(), server)
server.add_insecure_port("[::]:50051")   # local development only
server.start()
server.wait_for_termination()

# client.py
with grpc.insecure_channel("localhost:50051") as channel:
    stub = order_pb2_grpc.OrderServiceStub(channel)
    order = stub.GetOrder(order_pb2.GetOrderRequest(id="o-1001"), timeout=2.0)

Never ship insecure ports

add_insecure_port and insecure_channel are for local development; use TLS credentials everywhere else.

Quick check: How does a Python gRPC server report "not found"?

• Raise HTTP 404
• Return None
• context.abort with StatusCode.NOT_FOUND
• Print an error and return an empty message