↓ Skip to main content
Git Push and Run

Serving Protocol Buffers from FastAPI: Binary Endpoints and gRPC Gateway

·6 mins
Python Fastapi Protobuf Grpc
Table of Contents
FastAPI is excellent for rapid API development. Protobuf is excellent for compact, typed binary payloads. You can combine them without any magic Pydantic integration – just read raw bytes, parse with the generated class, and return bytes with the right Content-Type.

If you search for “FastAPI protobuf” you will find many posts that try to use Pydantic models as a bridge to protobuf serialization. There is no such bridge in Pydantic. The correct approach is simpler and more direct: FastAPI endpoints can accept and return raw bytes. The protobuf Python library handles the actual encoding. This post walks through a complete working implementation, content negotiation, and testing.

Setup
#

Install the required packages:

install dependencies
pip install fastapi uvicorn protobuf betterproto grpcio-tools
  • protobuf is the official Google Python runtime for .proto-generated classes
  • betterproto is an alternative that generates Python dataclasses instead of the older imperative style
  • grpcio-tools bundles protoc and the Python gRPC plugin so you can compile .proto files without a separate protoc installation

Define and compile a proto schema
#

protos/user.proto
syntax = "proto3";

package user;

message Address {
  string street  = 1;
  string city    = 2;
  string country = 3;
}

message User {
  string  id      = 1;
  string  name    = 2;
  string  email   = 3;
  int32   age     = 4;
  Address address = 5;
}

message CreateUserRequest {
  User user = 1;
}

Compile to Python using grpcio-tools (no separate protoc needed):

compile proto
python -m grpc_tools.protoc \
  -I protos \
  --python_out=gen \
  --grpc_python_out=gen \
  protos/user.proto

This produces gen/user_pb2.py (message types) and gen/user_pb2_grpc.py (service stubs). For betterproto, add --python_betterproto_out=gen instead of --python_out.

Note

Add gen/ to .gitignore or commit the generated files – be consistent across your team. Generated files can drift if team members use different protoc versions.

Correct FastAPI endpoint: raw bytes in, raw bytes out
#

PROTOBUF does not exist in Pydantic. The correct pattern is to use Request directly and return Response with the protobuf content type:

app/main.py
from fastapi import FastAPI, Request, Response, HTTPException
from gen.user_pb2 import User, CreateUserRequest

app = FastAPI()

PROTOBUF_CONTENT_TYPE = "application/x-protobuf"


@app.post("/users", response_class=Response)
async def create_user(request: Request) -> Response:
    content_type = request.headers.get("content-type", "")
    if content_type != PROTOBUF_CONTENT_TYPE:
        raise HTTPException(status_code=415, detail="expected application/x-protobuf")

    body = await request.body()

    try:
        req = CreateUserRequest()
        req.ParseFromString(body)
    except Exception as exc:
        raise HTTPException(status_code=400, detail=f"invalid protobuf payload: {exc}")

    # Business logic here -- for this example, echo the user back with an ID assigned.
    req.user.id = "usr-generated-001"

    return Response(
        content=req.user.SerializeToString(),
        media_type=PROTOBUF_CONTENT_TYPE,
    )
Important

ParseFromString mutates the object in place and returns the number of bytes consumed, not the parsed object. The common mistake is writing user = User().ParseFromString(body) which assigns an integer, not a User.

Client sending protobuf
#

client.py
import requests
from gen.user_pb2 import User, CreateUserRequest

user = User(
    name="Alice",
    email="alice@example.com",
    age=31,
    address={"street": "10 Downing Street", "city": "London", "country": "UK"},
)
req = CreateUserRequest(user=user)

response = requests.post(
    "http://localhost:8000/users",
    data=req.SerializeToString(),
    headers={"Content-Type": "application/x-protobuf"},
)
response.raise_for_status()

created = User()
created.ParseFromString(response.content)
print(f"created user id: {created.id}")

Content negotiation: JSON and protobuf from the same endpoint
#

You can serve both formats from a single handler by inspecting the Accept header. This is useful during a migration when some clients have already adopted protobuf and others have not:

app/negotiation.py
import json
from fastapi import FastAPI, Request, Response
from google.protobuf.json_format import MessageToDict
from gen.user_pb2 import User

app = FastAPI()


@app.get("/users/{user_id}", response_class=Response)
async def get_user(user_id: str, request: Request) -> Response:
    # Fetch user from storage -- simplified here.
    user = User(
        id=user_id,
        name="Alice",
        email="alice@example.com",
        age=31,
    )

    accept = request.headers.get("accept", "application/json")

    if "application/x-protobuf" in accept:
        return Response(
            content=user.SerializeToString(),
            media_type="application/x-protobuf",
        )

    # Default to JSON using the protobuf JSON serializer for field-name consistency.
    return Response(
        content=json.dumps(MessageToDict(user, preserving_proto_field_name=True)),
        media_type="application/json",
    )
Tip

Use google.protobuf.json_format.MessageToDict with preserving_proto_field_name=True when converting to JSON. This keeps snake_case field names consistent with the proto definition rather than converting them to camelCase.

Using betterproto for cleaner dataclasses
#

The default protoc --python_out generates a verbose imperative API. betterproto generates Python dataclasses that feel more idiomatic:

app/betterproto_example.py
# With betterproto-generated code, the API becomes dataclass-style.
# Compile with: python -m grpc_tools.protoc ... --python_betterproto_out=gen_bt

import betterproto
from gen_bt.user import User, Address, CreateUserRequest


async def handle_create(body: bytes) -> bytes:
    req = CreateUserRequest().parse(body)
    req.user.id = "usr-generated-001"
    return bytes(req.user)  # betterproto objects serialize with bytes()


# In a FastAPI handler:
# body = await request.body()
# result_bytes = await handle_create(body)
# return Response(content=result_bytes, media_type="application/x-protobuf")
from gen.user_pb2 import User

# Parse
user = User()
user.ParseFromString(raw_bytes)

# Serialize
raw_bytes = user.SerializeToString()

Pros: official Google support, stable, widely compatible. Cons: generated code is verbose, not dataclass-style.

from gen_bt.user import User

# Parse
user = User().parse(raw_bytes)

# Serialize
raw_bytes = bytes(user)

Pros: clean dataclasses, type hints, async-friendly, no descriptor pool magic. Cons: third-party, some proto3 edge cases differ from the reference implementation.

Testing protobuf endpoints with pytest
#

tests/test_users.py
import pytest
from fastapi.testclient import TestClient
from gen.user_pb2 import User, CreateUserRequest
from app.main import app

client = TestClient(app)


def test_create_user_protobuf():
    user = User(name="Alice", email="alice@example.com", age=31)
    req = CreateUserRequest(user=user)

    response = client.post(
        "/users",
        content=req.SerializeToString(),
        headers={"content-type": "application/x-protobuf"},
    )

    assert response.status_code == 200
    assert response.headers["content-type"] == "application/x-protobuf"

    created = User()
    created.ParseFromString(response.content)
    assert created.name == "Alice"
    assert created.id != ""


def test_create_user_wrong_content_type():
    response = client.post(
        "/users",
        content=b"{}",
        headers={"content-type": "application/json"},
    )
    assert response.status_code == 415

Alternative: gRPC-Gateway pattern
#

If you need both a gRPC interface and a REST/JSON interface from the same service, the gRPC-Gateway pattern is worth knowing. Instead of two separate applications, you define HTTP annotations in your .proto file and a gateway proxy translates REST calls to gRPC:

flowchart LR
    Browser["Browser / curl\n(JSON + REST)"] --> GW["gRPC-Gateway\n(grpc-gateway proxy)"]
    GRPCClient["gRPC Client\n(Go / Python / Java)"] --> SVC
    GW -->|transcodes JSON to protobuf| SVC["gRPC Service\n(your Go server)"]

For Python services this is typically handled by running grpc-gateway as a sidecar or using grpcio directly alongside FastAPI. The FastAPI approach in this post is simpler and appropriate when you own all clients and want a single binary endpoint.

**Using Pydantic for protobuf serialization.** Pydantic has no protobuf support. There is no `PROTOBUF` import. Trying to use Pydantic models as a bridge requires manually copying every field, which defeats the purpose of code generation. **Assigning the return value of ParseFromString.** `ParseFromString` returns an integer (bytes consumed), not the message object. Always call it on an existing object and use that object afterwards. **Not setting Content-Type on the request.** Sending protobuf bytes without `Content-Type: application/x-protobuf` will confuse any middleware that expects JSON. Your server should explicitly reject mismatched content types rather than silently trying to parse garbage. **Using REST semantics with protobuf.** Protobuf endpoints do not need `GET` vs `POST` to distinguish read vs write operations -- the message type carries that intent. If you are putting protobuf inside REST conventions, consider whether gRPC would be a cleaner fit. **Not handling the empty body case.** `ParseFromString(b"")` on a proto3 message succeeds and returns an all-defaults message. Decide explicitly whether an empty body is a valid request or an error.

If you want to go deeper on any of this, I offer 1:1 coaching sessions for engineers working on AI integration, cloud architecture, and platform engineering. Book a session (50 EUR / 60 min) or reach out at manuel.fedele+website@gmail.com.

Related