gRPC in Python, the FastAPI way: models in, models out

Here is what a gRPC handler usually looks like in Python:

# 1. write orders.proto # 2. python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. orders.proto # 3. commit orders_pb2.py and orders_pb2_grpc.py, regenerate them at every change import grpc import orders_pb2, orders_pb2_grpc class OrdersServicer(orders_pb2_grpc.OrdersServicer): async def Get(self, request, context): if request.order_number not in DB: await context.abort(grpc.StatusCode.NOT_FOUND, "no such order") return orders_pb2.Order(order_number=request.order_number, paid=True) 

It works. But the messages are not your domain models, nothing validates them, the generated files drift from the schema, and every new service starts with the same servicer-class ceremony. If you have written a FastAPI endpoint, you know it does not have to be like that.

This tutorial builds the same service with fastgrpc3, an open-source gRPC framework I wrote on top of my Rust protobuf codec, protoruf.
By the end you will have a service, its .proto contract, a typed client, tests, and the option to run it on a Rust HTTP/2 server — without writing a single line of protobuf by hand.

Install

pip install fastgrpc3 

Python 3.12 to 3.14. There is no protoc to install: protobuf is compiled and encoded in Rust by protoruf, which ships as a wheel.

A service is a function and two models

# orders.py from pydantic import BaseModel from fastgrpc3 import GrpcApp, GrpcError, GrpcService, Insecure, StatusCode class GetOrder(BaseModel): order_number: str class Order(BaseModel): order_number: str paid: bool = False DB = {"A": Order(order_number="A", paid=True)} service = GrpcService("Orders", package="orders.v1") @service.unary() async def Get(req: GetOrder) -> Order: if req.order_number not in DB: raise GrpcError(StatusCode.NOT_FOUND, "no such order") return DB[req.order_number] app = GrpcApp(address="[::]:50051", tls=Insecure()) app.add_service(service) app.enable_reflection() # so that grpcurl can discover the service 

The request is validated by Pydantic before your function runs; a payload that does not fit the model is answered with INVALID_ARGUMENT. The return value is checked against the response model. GrpcError becomes a real gRPC status, readable by any client in any language. Note that tls= is mandatory: Insecure() is a choice you make on purpose, not a default you forget.

Run it:

fastgrpc3 run orders:app 

And call it from the command line with grpcurl, which reads the contract through server reflection:

grpcurl -plaintext -d '{"order_number": "A"}' localhost:50051 orders.v1.Orders/Get # { # "orderNumber": "A", # "paid": true # } 

The .proto is generated from your models

Your models are the contract. Field numbers are what keeps protobuf compatible over time, so fastgrpc3 does not leave them to chance: fastgrpc3 lock pins them in a proto.lock file you commit. Adding a field later never renumbers the others, and a removed field's number stays reserved. Then export renders the contract as standard
.proto files, for the teams that call you in Go, Java or TypeScript:

fastgrpc3 lock orders:app # write proto.lock (commit it) fastgrpc3 export orders:app -o proto/ # proto/orders/v1/orders.proto fastgrpc3 check orders:app # CI: fail if proto.lock is out of date fastgrpc3 breaking orders:app --against main # CI: fail on a breaking change 

// proto/orders/v1/orders.proto syntax = "proto3"; package orders.v1; message GetOrder { string order_number = 1; } message Order { string order_number = 1; bool paid = 2; } service Orders { rpc Get(GetOrder) returns (Order); } 

Already have .proto files? It works the other way too: fastgrpc3 generate proto/ -o src/myapp/gen produces Pydantic models, the service contract and a typed client, all passing mypy --strict.

A typed client

# client.py import asyncio from fastgrpc3 import Channel, GrpcClient, GrpcError, Insecure from orders import GetOrder, Order class OrdersClient(GrpcClient): package = "orders.v1" service = "Orders" get = GrpcClient.unary("Get", GetOrder, Order) async def main() -> None: async with Channel("localhost:50051", tls=Insecure()) as channel: orders = OrdersClient(channel) print(await orders.get(GetOrder(order_number="A"))) # order_number='A' paid=True  try: await orders.get(GetOrder(order_number="Z")) except GrpcError as error: print(error.code.name, error.message) # NOT_FOUND no such order  if __name__ == "__main__": asyncio.run(main()) 

What you get back is an Order, not a protobuf message: your editor and mypy know its fields.

Dependencies, FastAPI-style

Real handlers need a database session, the current user, a settings object. Depends works the way you expect, including cleanup after the response for yield dependencies:

from collections.abc import AsyncIterator from fastgrpc3 import Context, Depends async def session() -> AsyncIterator[dict[str, Order]]: yield DB # open a real session here, close it after the response  def current_user(ctx: Context) -> str: return str(ctx.metadata.get("x-user", "anonymous")) @service.unary() async def Pay( req: GetOrder, db: dict[str, Order] = Depends(session), user: str = Depends(current_user), ) -> Order: order = db[req.order_number] order.paid = True return order 

Sub-dependencies, a per-call cache, @blocking for providers that do blocking I/O, and app.dependency_overrides for tests are all there. So are hooks, middlewares and exception handlers around every call.

Tests against a real server

# test_orders.py from fastgrpc3.testing import TestClient from orders import GetOrder, app from client import OrdersClient async def test_get() -> None: async with TestClient(app) as tc: order = await tc.client(OrdersClient).get(GetOrder(order_number="A")) assert order.paid 

TestClient starts the app on a free local port: you test the real wire, not a mock.

Ready for production

A few lines turn the service into something you can deploy:

from fastgrpc3 import TLS app = GrpcApp(address="[::]:50051", tls=TLS(cert="server.pem", key="server.key")) app.enable_health() # grpc.health.v1: Kubernetes probes, grpc-health-probe app.enable_reflection() app.enable_docs() # Swagger UI on :8080 (pip install "fastgrpc3[docs]") 

fastgrpc3 run orders:app --workers 4 # one process per core 

And then: one flag, 14× more calls per core

Everything above runs on grpc.aio, the standard asyncio server. It is solid, but on one CPU core it tops out around 4,400 calls per second, and most of that time is spent in the transport, not in your code.

So fastgrpc3 ships a second transport, written in Rust on tonic: HTTP/2, gRPC framing, TLS and flow control run in Rust, and your Python handlers receive the calls in batches. Same handlers, same dependencies, same errors:

pip install "fastgrpc3[tonic]" fastgrpc3 run orders:app --transport tonic 

On one core, unary calls, 64 concurrent clients:

That is 14× the throughput, 17× less CPU per call and an 8× lower tail latency — and the 62,240 is the load generator's ceiling, not the server's. On grpc.aio, the framework itself costs about 3 % over hand-written stubs. How the Rust transport gets there is a story of its own, which I tell in a separate article.

Under the hood: protoruf

None of this needs generated _pb2 classes because the protobuf layer is protoruf: it compiles .proto files in memory and its codec goes straight from protobuf bytes to Pydantic models, in Rust. Encoding a 20-field message takes about 1 µs — 55× faster than going through google.protobuf and Pydantic.

Try it

pip install "fastgrpc3[tonic]" 

• Docs: https://edwinalkins.github.io/fastgrpc3/
• GitHub: https://github.com/EdwinAlkins/fastgrpc3 (and the transport: https://github.com/EdwinAlkins/fastgrpc3-transport)

fastgrpc3 is at version 0.1: if you build something with it, or if it does not fit your use case, open an issue and tell me — and a ⭐ on GitHub helps other Python developers find it.