gRPC Complete Guide 2026: Protocol Buffers, Python and TypeScript

// user_service.proto
syntax = "proto3";
package userservice.v1;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

option go_package = "github.com/mycompany/proto/userservice/v1";

// Enums
enum UserRole {
  USER_ROLE_UNSPECIFIED = 0;
  USER_ROLE_USER = 1;
  USER_ROLE_ADMIN = 2;
  USER_ROLE_MODERATOR = 3;
}

// Messages
message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  UserRole role = 4;
  bool active = 5;
  google.protobuf.Timestamp created_at = 6;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  string password = 3;
  UserRole role = 4;
}

message GetUserRequest {
  int64 id = 1;
}

message ListUsersRequest {
  int32 page = 1;
  int32 page_size = 2;
  string filter = 3;
}

message ListUsersResponse {
  repeated User users = 1;
  int32 total = 2;
  int32 page = 3;
}

// Service definition
service UserService {
  // Unary RPC
  rpc GetUser(GetUserRequest) returns (User);
  rpc CreateUser(CreateUserRequest) returns (User);

  // Server streaming — server sends multiple messages
  rpc ListUsers(ListUsersRequest) returns (stream User);

  // Client streaming — client sends multiple messages
  rpc UploadUsers(stream CreateUserRequest) returns (ListUsersResponse);

  // Bidirectional streaming
  rpc SyncUsers(stream User) returns (stream User);
}

pip install grpcio grpcio-tools

# Generate Python code from .proto
python -m grpc_tools.protoc   -I.   --python_out=.   --grpc_python_out=.   user_service.proto

# server.py
import grpc
from concurrent import futures
import user_service_pb2 as pb2
import user_service_pb2_grpc as pb2_grpc
from datetime import datetime
from google.protobuf import timestamp_pb2

class UserServiceServicer(pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users: dict[int, pb2.User] = {}
        self._next_id = 1

    def GetUser(self, request, context):
        user = self.users.get(request.id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"User {request.id} not found")
            return pb2.User()
        return user

    def CreateUser(self, request, context):
        # Validation
        if not request.email or "@" not in request.email:
            context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
            context.set_details("Invalid email address")
            return pb2.User()

        ts = timestamp_pb2.Timestamp()
        ts.GetCurrentTime()

        user = pb2.User(
            id=self._next_id,
            name=request.name,
            email=request.email,
            role=request.role or pb2.USER_ROLE_USER,
            active=True,
            created_at=ts,
        )
        self.users[self._next_id] = user
        self._next_id += 1
        return user

    def ListUsers(self, request, context):
        for user in list(self.users.values()):
            if context.is_active():
                yield user  # streaming!

def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[LoggingInterceptor()],
    )
    pb2_grpc.add_UserServiceServicer_to_server(UserServiceServicer(), server)
    server.add_insecure_port("0.0.0.0:50051")
    server.start()
    print("gRPC server started on :50051")
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

# client.py
import grpc
import user_service_pb2 as pb2
import user_service_pb2_grpc as pb2_grpc

def run():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = pb2_grpc.UserServiceStub(channel)

        # Unary call
        user = stub.CreateUser(pb2.CreateUserRequest(
            name="Alice",
            email="alice@example.com",
            role=pb2.USER_ROLE_ADMIN,
        ))
        print(f"Created: {user.id} {user.name}")

        # Get user
        fetched = stub.GetUser(pb2.GetUserRequest(id=user.id))
        print(f"Fetched: {fetched.name}")

        # Server streaming
        for u in stub.ListUsers(pb2.ListUsersRequest(page=1, page_size=10)):
            print(f"  {u.id}: {u.name}")

# With metadata (auth header)
metadata = [("authorization", "Bearer " + jwt_token)]
user = stub.GetUser(pb2.GetUserRequest(id=1), metadata=metadata)

# With timeout
user = stub.GetUser(pb2.GetUserRequest(id=1), timeout=5.0)

# Handle errors
try:
    user = stub.GetUser(pb2.GetUserRequest(id=999))
except grpc.RpcError as e:
    print(f"Error: {e.code()} - {e.details()}")

if __name__ == "__main__":
    run()

npm install @grpc/grpc-js @grpc/proto-loader
npm install --save-dev ts-proto

// server.ts
import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';
import path from 'path';

const PROTO_PATH = path.join(__dirname, 'user_service.proto');
const packageDef = protoLoader.loadSync(PROTO_PATH);
const proto = grpc.loadPackageDefinition(packageDef) as any;

const getUserService = {
  getUser: (call: grpc.ServerUnaryCall<any, any>, callback: grpc.sendUnaryData<any>) => {
    const { id } = call.request;
    const user = db.users.findById(id);
    if (!user) {
      callback({ code: grpc.status.NOT_FOUND, message: `User ${id} not found` });
      return;
    }
    callback(null, user);
  },
};

const server = new grpc.Server();
server.addService(proto.userservice.v1.UserService.service, getUserService);
server.bindAsync('0.0.0.0:50051', grpc.ServerCredentials.createInsecure(), () => {
  server.start();
  console.log('gRPC server on :50051');
});