Skip to content
OpenSmartRoute
Skillv1.0.0

grpc-expert

Expert-level gRPC, Protocol Buffers, microservices communication, and streaming. Use when the user mentions Protocol Buffers, microservices, RPC, or streaming, or when the task involves gRPC Fundament

by personamanagmentlayer(0) 0 installs
Free
Sign in to install

Free account. Installing gives you the manifest plus copy-paste snippets.

See reviews

About

Imported from personamanagmentlayer/pcl (stdlib/api/grpc-expert/SKILL.md). Install upstream with npx skills add personamanagmentlayer/pcl --skill grpc-expert. Copyright stays with the author.

gRPC Expert

Expert guidance for gRPC services, Protocol Buffers, microservices communication, and streaming patterns.

Core Concepts

gRPC Fundamentals

  • Protocol Buffers (protobuf)
  • Service definitions
  • RPC patterns (unary, server streaming, client streaming, bidirectional)
  • HTTP/2 transport
  • Code generation
  • Interceptors and middleware

Communication Patterns

  • Unary RPC (request-response)
  • Server streaming RPC
  • Client streaming RPC
  • Bidirectional streaming RPC
  • Deadline/timeout handling
  • Error handling and status codes

Production Features

  • Load balancing
  • Service discovery
  • Health checking
  • Authentication (TLS, tokens)
  • Monitoring and tracing
  • Retry policies

Protocol Buffer Definition

syntax = "proto3";

package user.v1;

import "google/protobuf/timestamp.proto";

service UserService {
  // Unary RPC
  rpc GetUser(GetUserRequest) returns (GetUserResponse);

  // Server streaming
  rpc ListUsers(ListUsersRequest) returns (stream User);

  // Client streaming
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

  // Bidirectional streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message User {
  string id = 1;
  string email = 2;
  string name = 3;
  google.protobuf.Timestamp created_at = 4;
  UserRole role = 5;
}

enum UserRole {
  USER_ROLE_UNSPECIFIED = 0;
  USER_ROLE_USER = 1;
  USER_ROLE_ADMIN = 2;
}

message GetUserRequest {
  string id = 1;
}

message GetUserResponse {
  User user = 1;
}

message ListUsersRequest {
  int32 page_size = 1;
  string page_token = 2;
}

message CreateUserRequest {
  string email = 1;
  string name = 2;
}

message CreateUsersResponse {
  repeated string user_ids = 1;
  int32 created_count = 2;
}

message ChatMessage {
  string user_id = 1;
  string message = 2;
  google.protobuf.Timestamp timestamp = 3;
}

Python gRPC Server

import grpc
from concurrent import futures
import logging
from typing import Iterator

import user_pb2
import user_pb2_grpc

class UserService(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users = {}

    def GetUser(self, request, context):
        """Unary RPC"""
        user_id = request.id

        if user_id not in self.users:
            context.abort(grpc.StatusCode.NOT_FOUND, f"User {user_id} not found")

        user = self.users[user_id]
        return user_pb2.GetUserResponse(user=user)

    def ListUsers(self, request, context):
        """Server streaming RPC"""
        page_size = request.page_size or 10

        for i, user in enumerate(self.users.values()):
            if i >= page_size:
                break
            yield user

    def CreateUsers(self, request_iterator, context):
        """Client streaming RPC"""
        created_ids = []

        for request in request_iterator:
            user_id = self._generate_id()
            user = user_pb2.User(
                id=user_id,
                email=request.email,
                name=request.name
            )
            self.users[user_id] = user
            created_ids.append(user_id)

        return user_pb2.CreateUsersResponse(
            user_ids=created_ids,
            created_count=len(created_ids)
        )

    def Chat(self, request_iterator, context):
        """Bidirectional streaming RPC"""
        for message in request_iterator:
            # Echo back with modification
            response = user_pb2.ChatMessage(
                user_id="server",
                message=f"Echo: {message.message}",
                timestamp=message.timestamp
            )
            yield response

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()
    print("Server started on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    logging.basicConfig()
    serve()

Go gRPC Server

package main

import (
    "context"
    "io"
    "log"
    "net"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    pb "example.com/user/v1"
)

type userServer struct {
    pb.UnimplementedUserServiceServer
    users map[string]*pb.User
}

func (s *userServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.GetUserResponse, error) {
    user, exists := s.users[req.Id]
    if !exists {
        return nil, status.Errorf(codes.NotFound, "user %s not found", req.Id)
    }

    return &pb.GetUserResponse{User: user}, nil
}

func (s *userServer) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
    pageSize := req.PageSize
    if pageSize == 0 {
        pageSize = 10
    }

    count := 0
    for _, user := range s.users {
        if count >= int(pageSize) {
            break
        }

        if err := stream.Send(user); err != nil {
            return err
        }
        count++
    }

    return nil
}

func (s *userServer) CreateUsers(stream pb.UserService_CreateUsersServer) error {
    var userIds []string

    for {
        req, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.CreateUsersResponse{
                UserIds:      userIds,
                CreatedCount: int32(len(userIds)),
            })
        }
        if err != nil {
            return err
        }

        userId := generateID()
        user := &pb.User{
            Id:    userId,
            Email: req.Email,
            Name:  req.Name,
        }

        s.users[userId] = user
        userIds = append(userIds, userId)
    }
}

func (s *userServer) Chat(stream pb.UserService_ChatServer) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        response := &pb.ChatMessage{
            UserId:    "server",
            Message:   "Echo: " + msg.Message,
            Timestamp: msg.Timestamp,
        }

        if err := stream.Send(response); err != nil {
            return err
        }
    }
}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    s := grpc.NewServer()
    pb.RegisterUserServiceServer(s, &userServer{
        users: make(map[string]*pb.User),
    })

    log.Println("Server listening on :50051")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

gRPC Client with Interceptors

import grpc
import user_pb2
import user_pb2_grpc
from typing import Callable

def logging_interceptor(
    method: Callable,
    request,
    call_details: grpc.ClientCallDetails
):
    """Unary-unary interceptor for logging"""
    print(f"Calling method: {call_details.method}")
    print(f"Request: {request}")

    response = method(request, call_details)

    print(f"Response: {response}")
    return response

class UserClient:
    def __init__(self, host: str = 'localhost:50051'):
        # Create channel with interceptor
        self.channel = grpc.insecure_channel(host)
        self.channel = grpc.intercept_channel(
            self.channel,
            logging_interceptor
        )
        self.stub = user_pb2_grpc.UserServiceStub(self.channel)

    def get_user(self, user_id: str) -> user_pb2.User:
        """Call unary RPC"""
        request = user_pb2.GetUserRequest(id=user_id)

        try:
            response = self.stub.GetUser(
                request,
                timeout=5.0  # 5 second timeout
            )
            return response.user
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")
            raise

    def list_users(self, page_size: int = 10):
        """Call server streaming RPC"""
        request = user_pb2.ListUsersRequest(page_size=page_size)

        try:
            for user in self.stub.ListUsers(request):
                yield user
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

    def create_users_batch(self, users: list):
        """Call client streaming RPC"""
        def request_generator():
            for user in users:
                yield user_pb2.CreateUserRequest(
                    email=user['email'],
                    name=user['name']
                )

        try:
            response = self.stub.CreateUsers(request_generator())
            return response
        except grpc.RpcError as e:
            print(f"RPC failed: {e.code()} - {e.details()}")

    def close(self):
        self.channel.close()

Best Practices

Design

  • Use semantic versioning for protobuf packages
  • Design backward-compatible changes
  • Use proper field numbering (never reuse)
  • Include metadata in messages
  • Use enums with explicit UNSPECIFIED value
  • Design for pagination in list operations

Performance

  • Enable HTTP/2 connection pooling
  • Use streaming for large data transfers
  • Implement proper timeouts
  • Use compression for large payloads
  • Batch operations when possible
  • Monitor and tune thread pool sizes

Production

  • Implement health checks
  • Use TLS for secure communication
  • Add authentication/authorization
  • Implement retry policies with backoff
  • Monitor gRPC metrics (latency, errors)
  • Use service discovery for dynamic endpoints

Anti-Patterns

❌ Breaking backward compatibility ❌ No timeout configuration ❌ Ignoring error status codes ❌ Not handling stream cancellation ❌ Over-sized messages ❌ No health checking ❌ Missing authentication

Resources

Use it

Copy one of these into your project. Installing also returns the manifest and these snippets.

yaml
targets:
  - https://api.opensmartroute.ai/api/v1/registry/personamanagmentlayer-pcl-grpc-expert/manifest   # or paste the manifest below

Manifest

An Open Capability Manifest: the router reads it to know what this does, what it costs and when to pick it.

personamanagmentlayer-pcl-grpc-expert.ocm.jsonjson
{
  "ocm": "1",
  "id": "personamanagmentlayer-pcl-grpc-expert",
  "kind": "skill",
  "name": "grpc-expert",
  "description": "Expert-level gRPC, Protocol Buffers, microservices communication, and streaming. Use when the user mentions Protocol Buffers, microservices, RPC, or streaming, or when the task involves gRPC Fundamentals, Communication Patterns, or Production Features.",
  "publisher": "personamanagmentlayer",
  "version": "1.0.0",
  "capabilities": {
    "domains": [
      "general"
    ],
    "tags": [
      "skill-md",
      "grpc",
      "protobuf",
      "microservices",
      "rpc",
      "streaming",
      "skills-sh"
    ],
    "languages": [
      "en"
    ]
  },
  "quality_prior": 0.6,
  "examples": [
    "Expert-level gRPC, Protocol Buffers, microservices communication, and streaming. Use when the user mentions Protocol Buffers, microservices, RPC, or streaming, or when the task involves gRPC Fundamentals, Communication Patterns, or Production Features."
  ],
  "primary": false,
  "metadata": {
    "source": {
      "provider": "skills.sh",
      "repository": "https://github.com/personamanagmentlayer/pcl",
      "path": "stdlib/api/grpc-expert/SKILL.md",
      "ref": "HEAD",
      "url": "https://github.com/personamanagmentlayer/pcl/blob/HEAD/stdlib/api/grpc-expert/SKILL.md",
      "key": "personamanagmentlayer/pcl/stdlib/api/grpc-expert/SKILL.md"
    },
    "allowed_tools": [
      "Read",
      "Write",
      "Edit"
    ]
  },
  "instructions": "# gRPC Expert\n\nExpert guidance for gRPC services, Protocol Buffers, microservices communication, and streaming patterns.\n\n## Core Concepts\n\n### gRPC Fundamentals\n\n- Protocol Buffers (protobuf)\n- Service definitions\n- RPC patterns (unary, server streaming, client streaming, bidirectional)\n- HTTP/2 transport\n- Code generation\n- Interceptors and middleware\n\n### Communication Patterns\n\n- Unary RPC (request-response)\n- Server streaming RPC\n- Client streaming RPC\n- Bidirectional streaming RPC\n- Deadline/timeout handling\n- Error handling and status codes\n\n### Production Features\n\n- Load balancing\n- S",
  "cost": {
    "context_tokens": 2471
  }
}

Fetch it by URL: GET /api/v1/registry/personamanagmentlayer-pcl-grpc-expert/manifest?version=1.0.0

Reviews

Star ratings from people who tried it. One review per account; edit yours any time.

No reviews yet. Install it, try it, and be the first to rate it.