Python gRPC

6 interview angles 6 min read source

Python gRPC

The Python side. grpcio is the C-backed implementation; grpcio-tools generates code from .proto files. Both sync and async APIs exist; async is the modern recommendation for new services.

Setup

pip install grpcio grpcio-tools

grpcio is the runtime; grpcio-tools is the protoc compiler + Python plugins. For pure Protobuf without gRPC, only protobuf is needed.

Generating code

mkdir -p generated
python -m grpc_tools.protoc \
    -I=./proto \
    --python_out=./generated \
    --grpc_python_out=./generated \
    --pyi_out=./generated \
    proto/user.proto

Generates three files per .proto:

user_pb2.py — message classes (User, GetUserRequest).
user_pb2_grpc.py — service stubs (UserServiceStub for clients, UserServiceServicer for servers).
user_pb2.pyi — type stubs for mypy / pyright (with --pyi_out).

Make this part of your build (Makefile, pyproject build hook, etc.) so generated files stay current.

Sync server

import grpc
from concurrent import futures
from generated import user_pb2, user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = db.get_user(request.id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details("User not found")
            return user_pb2.User()
        return user_pb2.User(id=user.id, name=user.name, email=user.email)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

The threadpool model: one thread per RPC. For ~10–100 concurrent RPCs, fine. For thousands of concurrent streams, switch to async.

Async server

import asyncio
import grpc
from generated import user_pb2, user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    async def GetUser(self, request, context):
        user = await db.get_user_async(request.id)
        if not user:
            await context.abort(grpc.StatusCode.NOT_FOUND, "User not found")
        return user_pb2.User(id=user.id, name=user.name, email=user.email)

async def serve():
    server = grpc.aio.server()
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    server.add_insecure_port("[::]:50051")
    await server.start()
    await server.wait_for_termination()

if __name__ == "__main__":
    asyncio.run(serve())

Each RPC method is a coroutine. The server handles many concurrent RPCs on one event loop.

For new services, prefer async — better resource use, simpler concurrency model, integrates with asyncpg / httpx / etc.

Sync client

import grpc
from generated import user_pb2, user_pb2_grpc

channel = grpc.insecure_channel("localhost:50051")
stub = user_pb2_grpc.UserServiceStub(channel)

response = stub.GetUser(user_pb2.GetUserRequest(id="42"))
print(response.name)

Always close the channel:

with grpc.insecure_channel("localhost:50051") as channel:
    stub = user_pb2_grpc.UserServiceStub(channel)
    response = stub.GetUser(...)

For long-running apps, keep one channel for the app’s lifetime — channels are heavy (TCP connection, HTTP/2 state). Don’t create one per request.

Async client

import asyncio
import grpc
from generated import user_pb2, user_pb2_grpc

async def main():
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = await stub.GetUser(user_pb2.GetUserRequest(id="42"))
        print(response.name)

asyncio.run(main())

TLS / secure channels

# Server
with open("server.key", "rb") as f: server_key = f.read()
with open("server.crt", "rb") as f: server_cert = f.read()

credentials = grpc.ssl_server_credentials([(server_key, server_cert)])
server.add_secure_port("[::]:50051", credentials)

# Client
with open("ca.crt", "rb") as f: ca = f.read()
credentials = grpc.ssl_channel_credentials(root_certificates=ca)
channel = grpc.secure_channel("server.example.com:50051", credentials)

For mTLS (mutual TLS, client cert verification):

# Server
credentials = grpc.ssl_server_credentials(
    [(server_key, server_cert)],
    root_certificates=ca,
    require_client_auth=True,
)

# Client
credentials = grpc.ssl_channel_credentials(
    root_certificates=ca,
    private_key=client_key,
    certificate_chain=client_cert,
)

mTLS is the standard for service-to-service authentication inside a mesh.

Channel options

options = [
    ("grpc.max_send_message_length", 50 * 1024 * 1024),       # 50 MB
    ("grpc.max_receive_message_length", 50 * 1024 * 1024),
    ("grpc.keepalive_time_ms", 10000),
    ("grpc.keepalive_timeout_ms", 5000),
    ("grpc.keepalive_permit_without_calls", 1),
    ("grpc.http2.max_pings_without_data", 0),
]
channel = grpc.insecure_channel("server:50051", options=options)

Default message size limit is 4 MB. Raise for services that handle large payloads (file chunks, embeddings). Limit at sender AND receiver — both must agree.

Server streaming — Python sync

class UserService(user_pb2_grpc.UserServiceServicer):
    def ListUpdates(self, request, context):
        for update in db.iter_updates(since=request.since):
            if not context.is_active():
                return
            yield user_pb2.Update(id=update.id, body=update.body)

The service method is a generator. yield each response.

Client streaming — Python sync

def UploadFile(self, request_iterator, context):
    total = 0
    for chunk in request_iterator:
        save_chunk(chunk.content)
        total += len(chunk.content)
    return user_pb2.UploadResult(bytes_received=total)

The method receives an iterator over incoming requests; returns one response.

Client side:

def chunk_generator(path):
    with open(path, "rb") as f:
        while data := f.read(64 * 1024):
            yield user_pb2.FileChunk(content=data)

response = stub.UploadFile(chunk_generator("big.zip"))

Bidi streaming — Python sync

def Chat(self, request_iterator, context):
    for msg in request_iterator:
        # process incoming
        reply = handle_message(msg)
        yield user_pb2.ChatMessage(text=reply)

For independent timing (server may send before/after each client message), use separate threads or async.

Async streaming

class UserService(user_pb2_grpc.UserServiceServicer):
    async def ListUpdates(self, request, context):
        async for update in db.iter_updates_async(since=request.since):
            yield user_pb2.Update(id=update.id, body=update.body)

# Client async
async with grpc.aio.insecure_channel("...") as channel:
    stub = user_pb2_grpc.UserServiceStub(channel)
    async for update in stub.ListUpdates(filter):
        print(update)

async for over the stream. Async streaming is cleaner than sync threadpool for high-fan-out scenarios.

Reflection — make grpcurl work

By default, clients need the .proto file to call your server. Enable reflection:

from grpc_reflection.v1alpha import reflection

SERVICE_NAMES = (
    user_pb2.DESCRIPTOR.services_by_name["UserService"].full_name,
    reflection.SERVICE_NAME,
)
reflection.enable_server_reflection(SERVICE_NAMES, server)

Now grpcurl -plaintext localhost:50051 list shows your services without a .proto file.

grpcurl -plaintext localhost:50051 user.UserService/GetUser -d '{"id": "42"}'

Don’t enable reflection in public-facing production (schema leak). Internal services and dev only.

Testing gRPC services

import pytest
import grpc
from generated import user_pb2, user_pb2_grpc

@pytest.fixture
def server_and_stub():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=2))
    user_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    port = server.add_insecure_port("[::]:0")    # 0 = OS picks a free port
    server.start()
    channel = grpc.insecure_channel(f"localhost:{port}")
    stub = user_pb2_grpc.UserServiceStub(channel)
    yield stub
    channel.close()
    server.stop(0)

def test_get_user(server_and_stub):
    response = server_and_stub.GetUser(user_pb2.GetUserRequest(id="42"))
    assert response.name == "Alice"

For mocking the service stub without a real server, the grpc-testing library or pytest fixtures with mock generators.

gRPC with FastAPI / Starlette

You’d typically run a gRPC service alongside or instead of REST, not in the same process. But for Python services that need both:

# Run gRPC in a thread / asyncio task alongside FastAPI
import asyncio

async def grpc_server():
    server = grpc.aio.server()
    # ... add services ...
    await server.start()
    await server.wait_for_termination()

async def main():
    config = uvicorn.Config("app:app", host="0.0.0.0", port=8000)
    server_task = asyncio.create_task(uvicorn.Server(config).serve())
    grpc_task = asyncio.create_task(grpc_server())
    await asyncio.gather(server_task, grpc_task)

In Kubernetes, simpler to run two separate deployments (REST and gRPC) and let the LB route.

grpcio-status — richer error info

For structured errors (with details, not just status codes), grpcio-status lets you attach Protobuf messages to error responses:

from grpc_status import rpc_status
from google.rpc import status_pb2, code_pb2, error_details_pb2

def GetUser(self, request, context):
    if not request.id:
        detail = error_details_pb2.BadRequest(
            field_violations=[error_details_pb2.BadRequest.FieldViolation(
                field="id", description="must not be empty",
            )],
        )
        rich_status = status_pb2.Status(
            code=code_pb2.INVALID_ARGUMENT,
            message="Invalid id",
            details=[any_pb2.Any.Pack(detail)],
        )
        context.abort_with_status(rpc_status.to_status(rich_status))

See 07_error_handling.md.

Common pitfalls

Creating a channel per request — defeats HTTP/2 multiplexing; pay TCP+TLS handshake on every RPC.
Default 4 MB message limit — surprises with large payloads. Raise on both ends.
No keepalive — connections die behind firewalls.
Sync server with blocking I/O — threadpool exhausted under load. Use async, or increase max_workers.
grpc_tools.protoc paths confusion — -I=./proto --python_out=./generated proto/user.proto. The -I path is the root for imports between .proto files.

Common interview confusions

“You write .proto and import directly.” — you generate _pb2.py and _pb2_grpc.py from the .proto, then import those.
“sync and async gRPC are the same.” — different APIs (grpc.server vs grpc.aio.server); methods are coroutines in async. Same wire protocol.
“grpc.insecure_channel is fine for production.” — only inside a trusted network (and even then, mTLS is best practice). Public-facing always TLS.

Interview angle

“How do you build a Python gRPC server?” — pip install grpcio grpcio-tools. Generate stubs with python -m grpc_tools.protoc. Subclass the generated *ServiceServicer, implement methods. grpc.server(executor) and add_*Servicer_to_server. Bind a port; start(); wait_for_termination().
“Sync vs async gRPC Python — which?” — async for new services (better resource use, integrates with asyncio ecosystem). Sync is simpler and fine for small services with low concurrency.
“How do you handle errors in a gRPC method?” — context.set_code(...) + set_details(...) then return an empty message, OR context.abort(code, message) which raises and never returns. Async uses await context.abort(...).
“What’s the message size limit and how do you change it?” — default 4 MB. Set grpc.max_send_message_length and grpc.max_receive_message_length in channel options on both client and server.
“How do you do mTLS in Python gRPC?” — grpc.ssl_server_credentials([(key, cert)], root_certificates=ca, require_client_auth=True) on server; grpc.ssl_channel_credentials(root_certificates=ca, private_key=client_key, certificate_chain=client_cert) on client.
“What’s gRPC reflection?” — a meta-service that exposes the .proto schema at runtime. Lets tools like grpcurl call your server without the .proto file. Enable via grpc_reflection.v1alpha.reflection.enable_server_reflection. Don’t enable in public production.