A Beginner’s Guide to gRPC in Python — With Sync & Async Microservices

“If you’re building modern microservices, it’s time to go beyond REST.”

What is gRPC?

gRPC (Google Remote Procedure Call) is a high-performance framework for communication between services. Unlike REST (which uses JSON over HTTP), gRPC:

• Uses Protocol Buffers (protobuf) — a compact, binary data format
• Supports strong typing and code generation
• Works over HTTP/2 (streaming, multiplexing, efficient)

Think of it as a faster, type-safe alternative to REST APIs.

Why Use gRPC in Microservices?

Microservices often need to call each other frequently. REST can be slow and verbose.

With gRPC, you get: ✅ Fast, binary serialization ✅ Type-safe contracts between services ✅ Streaming & bi-directional communication ✅ Auto-generated code for client and server

What We’ll Build

We’ll create two Python microservices that communicate using gRPC:

                    ┌──────────────────────┐
                    │     User Service     │
                    │ (Async gRPC Server)  │
                    │   Port: 50051        │
                    └──────────▲───────────┘
                               │  async gRPC call
                               │
                    ┌──────────┴───────────┐
                    │    Order Service     │
                    │ (Async gRPC Server)  │
                    │   Port: 50052        │
                    └──────────▲───────────┘
                               │
                     ┌─────────┴──────────┐
                     │   Test Client      │
                     │ Calls OrderService │
                     └────────────────────┘

1. We request an order with order_id and user_id to order_server.
2. Request reach order_server and send a call to user_server to get user details
3. Return final result

We’ll implement both synchronous and asynchronous (asyncio) versions.

📦 GitHub Repo: Link 🐍 Tech Stack: Python · gRPC · asyncio · Protocol Buffers

Prerequisites

• Python 3.7 or higher
• pip version 9.0.1 or higher

Setup env

• Setup virtualenv

# Create a virtual environment named "venv"
python -m venv venv

# Activate the virtual environment
.\venv\Scripts\Activate.ps1

# Upgrade pip
python -m pip install --upgrade pip

• Install gRPC and gRPC tools

pip install grpcio
pip install grpcio-tools

• Clone the git repo: 👉 GitHub Repo: Link

Project Structure

grpc-microservices/
│
├── user_service/
│   ├── protos/user.proto
│   ├── server.py              # Sync gRPC Server
│   ├── async_server.py        # Async gRPC Server
│   ├── user_pb2.py
│   ├── user_pb2_grpc.py
│
└── order_service/
    ├── protos/order.proto
    ├── server.py              # Sync gRPC Server
    ├── async_server.py        # Async gRPC Server
    ├── order_to_user.py       # Sync client to user service
    ├── async_order_to_user.py # Async client to user service
    ├── test_client.py         # Sync client
    └── async_test_client.py   # Async client

Step 1 — Define the Protobuf files

user_service/protos/user.proto

syntax = "proto3";

package user;

service UserService {
    rpc GetUser (UserRequest) returns (UserResponse);
}

message UserRequest {
    int32 user_id = 1;
}

message UserResponse {
    int32 user_id = 1;
    string name = 2;
    string email = 3;
}

order_service/protos/order.proto

syntax = "proto3";

package order;

service OrderService {
    rpc GetOrder (OrderRequest) returns (OrderResponse);
}

message OrderRequest {
    int32 order_id = 1;
    int32 user_id = 2;
    int32 quantity = 3;
}

message OrderResponse {
    int32 order_id = 1;
    string product = 2;
    int32 quantity = 3;
    int32 price = 4;
    string user_name = 5;
    string user_email = 6;
}

Then generate Python code:

cd user_service
python -m grpc_tools.protoc -I protos --python_out=. --grpc_python_out=. protos/user.proto
cd ../order_service
python -m grpc_tools.protoc -I protos --python_out=. --grpc_python_out=. protos/order.proto

Step 2 — The Synchronous Version

Setup server side by implementing — UserServiceServicer and OrderServiceServicer

User Service (user_service/server.py)

import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        #Simulate user lookup
        user = {
            1: {"name": "Ashik", "email": "ashik@example.com"},
            2: {"name": "Rony", "email": "rony@example.com"}
        }
        user = user.get(request.user_id, {"name": "Unknown user", "email": ""})
        return user_pb2.UserResponse(
            user_id=request.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('localhost:50051')
    print("User service running on port 50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

Order Service (order_service/server.py)

from concurrent import futures
import grpc

import order_pb2
import order_pb2_grpc

import order_to_user

class OrderService(order_pb2_grpc.OrderServiceServicer):
    def GetOrder(self, request, context):
        # Simulate order db
        order = {
            1: {"name": "TV", "price": 10000},
            2: {"name": "Shirt", "price": 800},
            3: {"name": "Chair", "price": 2000}
        }
        user = order_to_user.get_user(request.user_id)
        order = order.get(request.order_id, {"name": "Unknown product", "price": 0})
        return order_pb2.OrderResponse(
            order_id = request.order_id,
            product = order["name"],
            quantity = request.quantity,
            price = request.quantity * order["price"],
            user_name = user.name,
            user_email = user.email
        )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    order_pb2_grpc.add_OrderServiceServicer_to_server(OrderService(), server)
    server.add_insecure_port('localhost:50052')
    print("Order service running on port 50052")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve()

Step 3 — The Async Version (High Performance)

Async User Service (user_service/async_server.py)

import asyncio
import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    async def GetUser(self, request, context):
        #Simulate user lookup
        await asyncio.sleep(0.1)
        user = {
            1: {"name": "Ashik", "email": "ashik@example.com"},
            2: {"name": "Rony", "email": "rony@example.com"}
        }
        user = user.get(request.user_id, {"name": "Unknown user", "email": ""})
        return user_pb2.UserResponse(
            user_id=request.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('localhost:50051')
    print("User service running on port 50051")
    await server.start()
    await server.wait_for_termination()

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

Async Order Service (order_service/async_server.py)

import asyncio
from concurrent import futures
import grpc

import order_pb2
import order_pb2_grpc

import async_order_to_user

class OrderService(order_pb2_grpc.OrderServiceServicer):
    async def GetOrder(self, request, context):
        # Simulate order db
        await asyncio.sleep(0.1)
        order = {
            1: {"name": "TV", "price": 10000},
            2: {"name": "Shirt", "price": 800},
            3: {"name": "Chair", "price": 2000}
        }
        user = await async_order_to_user.get_user(request.user_id)
        order = order.get(request.order_id, {"name": "Unknown product", "price": 0})
        return order_pb2.OrderResponse(
            order_id = request.order_id,
            product = order["name"],
            quantity = request.quantity,
            price = request.quantity * order["price"],
            user_name = user.name,
            user_email = user.email
        )

async def serve():
    server = grpc.aio.server()
    order_pb2_grpc.add_OrderServiceServicer_to_server(OrderService(), server)
    server.add_insecure_port('localhost:50052')
    print("Order service running on port 50052")
    await server.start()
    await server.wait_for_termination()

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

Step 4 — Implement channel to order services

Channel for sync grpc:

*order_service/test_client.py*

import grpc
import order_pb2
import order_pb2_grpc

def run():
    with grpc.insecure_channel('localhost:50052') as channel:
        stub = order_pb2_grpc.OrderServiceStub(channel)
        response = stub.GetOrder(order_pb2.OrderRequest(
            order_id=3,
            user_id=2,
            quantity=4
        ))
    print(response)

if __name__ == "__main__":
    run()

Channel for async grpc:

*order_service/async_test_client.py*

import asyncio
import grpc
import order_pb2
import order_pb2_grpc

async def run():
    async with grpc.aio.insecure_channel('localhost:50052') as channel:
        stub = order_pb2_grpc.OrderServiceStub(channel)
        response = await stub.GetOrder(order_pb2.OrderRequest(
            order_id=3,
            user_id=2,
            quantity=4
        ))
    print(response)

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

Step 5— Implement channel between order service and user service

Channel for sync grpc:

*order_service/order_to_user.py*

import grpc
import sys
sys.path.append('../user_service')
import user_pb2
import user_pb2_grpc

def get_user(user_id):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = stub.GetUser(user_pb2.UserRequest(user_id=user_id))
    return response

Channel for async grpc:

*order_service/async_order_to_user.py*

import grpc
import sys
sys.path.append('../user_service')
import user_pb2
import user_pb2_grpc

async def get_user(user_id):
    async with grpc.aio.insecure_channel('localhost:50051') as channel:
        stub = user_pb2_grpc.UserServiceStub(channel)
        response = await stub.GetUser(user_pb2.UserRequest(user_id=user_id))
    return response

Step 6— Testing the Async Service

Run both async servers:

# Terminal 1
cd user_service
python async_server.py

# Terminal 2
cd order_service
python async_server.py

Then run the async client:

cd order_service
python async_test_client.py

Step 7— Testing the sync Service

Stop async servers, run both sync servers:

# Terminal 1
cd user_service
python server.py

# Terminal 2
cd order_service
python server.py

Then run the sync client:

cd order_service
python test_client.py

Synchronous vs Asynchronous gRPC

Conclusion

gRPC may seem intimidating at first, but once you get it running, you’ll realize it’s a clean, efficient, and powerful way to connect microservices.

This project is a simple foundation — you can now add databases, authentication, or even real-time streaming RPCs.

Hope you learned something new today! Happy coding!