gRPC Services

May 30, 2026 | 5 min read

gRPC Server

import grpc
from concurrent import futures

class ComputeServicer:
    def Process(self, request, context):
        result = compiled_compute(request.data)
        return ComputeResponse(result=result)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
server.add_insecure_port('[::]:50051')
server.start()

Compiled Handler Pattern

def compiled_compute(data: bytes) -> bytes:
    # CPU-intensive logic compiled natively
    processed = transform(data)
    return processed

Streaming

def ProcessStream(self, request_iterator, context):
    for request in request_iterator:
        yield ComputeResponse(result=compiled_compute(request.data))