protocols-grpc-implementation

Implementing gRPC APIs with Protocol Buffers

gRPC Implementation

Scope: gRPC services, Protocol Buffers, streaming, error handling, performance optimization Lines: ~350 Last Updated: 2025-10-27

When to Use This Skill

Activate this skill when:

• Implementing gRPC services from scratch
• Designing Protocol Buffer schemas
• Implementing streaming RPCs (unary, server, client, bidirectional)
• Adding error handling and status codes
• Implementing interceptors and middleware
• Optimizing gRPC performance
• Setting up load balancing for gRPC
• Securing gRPC with TLS and authentication
• Debugging gRPC services

Core Concepts

gRPC Architecture

gRPC (gRPC Remote Procedure Call): High-performance RPC framework using HTTP/2 and Protocol Buffers.

Key characteristics:

• HTTP/2 transport: Multiplexing, header compression, binary framing
• Protocol Buffers: Efficient binary serialization (smaller payloads than JSON)
• Code generation: Client and server code generated from .proto files
• Streaming: Native support for streaming requests and responses
• Multi-language: Official support for 10+ languages
• Type-safe: Strong typing enforced by Protocol Buffers

Architecture components:

Client → Stub (Generated) → HTTP/2 → Server → Service Implementation
         ↓                              ↓
    Interceptors                   Interceptors
         ↓                              ↓
    Metadata                       Metadata

---

Protocol Buffers Basics

Message Definition

// users.proto
syntax = "proto3";

package users.v1;

option go_package = "github.com/example/users/v1;usersv1";

// User represents a user in the system
message User {
  string id = 1;              // Field number (used for encoding)
  string email = 2;
  string name = 3;
  int32 age = 4;
  bool is_active = 5;
  repeated string tags = 6;   // List of strings
  google.protobuf.Timestamp created_at = 7;
}

// CreateUserRequest with nested message
message CreateUserRequest {
  string email = 1;
  string name = 2;
  Profile profile = 3;        // Nested message

  message Profile {
    string bio = 1;
    string avatar_url = 2;
  }
}

message CreateUserResponse {
  User user = 1;
  string message = 2;
}

Key concepts:

• Field numbers (1, 2, 3): Used for encoding (never reuse deleted numbers)
• Types: string, int32, int64, bool, bytes, float, double
• repeated: List/array of values
• Nested messages: Messages within messages
• Imports: import "google/protobuf/timestamp.proto";

Service Definition (RPC Methods)

// Four types of RPCs
service UserService {
  // Unary: Single request → Single response
  rpc GetUser(GetUserRequest) returns (GetUserResponse);

  // Server streaming: Single request → Stream of responses
  rpc ListUsers(ListUsersRequest) returns (stream User);

  // Client streaming: Stream of requests → Single response
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

  // Bidirectional streaming: Stream of requests ↔ Stream of responses
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message GetUserRequest {
  string id = 1;
}

message GetUserResponse {
  User user = 1;
}

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

message CreateUsersResponse {
  repeated User users = 1;
  int32 created_count = 2;
}

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

---

RPC Types and Implementation

1. Unary RPC (Request-Response)

Pattern: Client sends one request, server returns one response

.proto definition:

rpc GetUser(GetUserRequest) returns (GetUserResponse);

Python server (grpcio):

import grpc
from concurrent import futures
import users_pb2
import users_pb2_grpc

class UserService(users_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        # request: GetUserRequest
        # context: grpc.ServicerContext (metadata, peer, etc.)

        user_id = request.id

        # Fetch from database
        user = db.get_user(user_id)

        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f'User {user_id} not found')
            return users_pb2.GetUserResponse()

        # Return response
        return users_pb2.GetUserResponse(
            user=users_pb2.User(
                id=user['id'],
                email=user['email'],
                name=user['name']
            )
        )

# Start server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
users_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

Python client:

import grpc
import users_pb2
import users_pb2_grpc

# Create channel
channel = grpc.insecure_channel('localhost:50051')
stub = users_pb2_grpc.UserServiceStub(channel)

# Call RPC
try:
    response = stub.GetUser(users_pb2.GetUserRequest(id='123'))
    print(f'User: {response.user.name}')
except grpc.RpcError as e:
    print(f'Error: {e.code()} - {e.details()}')

2. Server Streaming RPC

Pattern: Client sends one request, server streams multiple responses

.proto definition:

rpc ListUsers(ListUsersRequest) returns (stream User);

Python server:

def ListUsers(self, request, context):
    page_size = request.page_size or 50

    # Stream users (yield multiple responses)
    for user in db.list_users(limit=page_size):
        yield users_pb2.User(
            id=user['id'],
            email=user['email'],
            name=user['name']
        )

Python client:

# Receive stream
response_stream = stub.ListUsers(users_pb2.ListUsersRequest(page_size=10))

for user in response_stream:
    print(f'User: {user.name} ({user.email})')

3. Client Streaming RPC

Pattern: Client streams multiple requests, server returns one response

.proto definition:

rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

Python server:

def CreateUsers(self, request_iterator, context):
    created_users = []

    # Receive stream of requests
    for request in request_iterator:
        user = db.create_user(email=request.email, name=request.name)
        created_users.append(user)

    return users_pb2.CreateUsersResponse(
        users=created_users,
        created_count=len(created_users)
    )

Python client:

def request_generator():
    # Generate stream of requests
    for i in range(5):
        yield users_pb2.CreateUserRequest(
            email=f'user{i}@example.com',
            name=f'User {i}'
        )

# Send stream
response = stub.CreateUsers(request_generator())
print(f'Created {response.created_count} users')

4. Bidirectional Streaming RPC

Pattern: Client and server both stream messages independently

.proto definition:

rpc Chat(stream ChatMessage) returns (stream ChatMessage);

Python server:

def Chat(self, request_iterator, context):
    # Concurrent reading and writing
    for message in request_iterator:
        # Process incoming message
        user_id = message.user_id
        content = message.content

        # Broadcast to other users (example)
        response = users_pb2.ChatMessage(
            user_id='system',
            content=f'{user_id} said: {content}',
            timestamp=Timestamp()
        )
        yield response

Python client:

def message_generator():
    messages = ['Hello', 'How are you?', 'Goodbye']
    for msg in messages:
        yield users_pb2.ChatMessage(
            user_id='user123',
            content=msg,
            timestamp=Timestamp()
        )

# Bidirectional stream
responses = stub.Chat(message_generator())

for response in responses:
    print(f'{response.user_id}: {response.content}')

---

Error Handling

gRPC Status Codes

from grpc import StatusCode

# Common status codes
StatusCode.OK               # 0: Success
StatusCode.CANCELLED        # 1: Operation cancelled
StatusCode.INVALID_ARGUMENT # 3: Invalid argument
StatusCode.NOT_FOUND        # 5: Resource not found
StatusCode.ALREADY_EXISTS   # 6: Resource already exists
StatusCode.PERMISSION_DENIED # 7: Permission denied
StatusCode.UNAUTHENTICATED  # 16: Missing authentication
StatusCode.INTERNAL         # 13: Internal error
StatusCode.UNAVAILABLE      # 14: Service unavailable
StatusCode.DEADLINE_EXCEEDED # 4: Deadline exceeded

Server Error Handling

def GetUser(self, request, context):
    try:
        user_id = request.id

        # Validate input
        if not user_id:
            context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
            context.set_details('User ID is required')
            return users_pb2.GetUserResponse()

        # Check authentication
        if not context.metadata().get('authorization'):
            context.set_code(grpc.StatusCode.UNAUTHENTICATED)
            context.set_details('Authentication required')
            return users_pb2.GetUserResponse()

        # Fetch user
        user = db.get_user(user_id)

        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f'User {user_id} not found')
            return users_pb2.GetUserResponse()

        return users_pb2.GetUserResponse(user=user)

    except Exception as e:
        context.set_code(grpc.StatusCode.INTERNAL)
        context.set_details(f'Internal error: {str(e)}')
        return users_pb2.GetUserResponse()

Client Error Handling

try:
    response = stub.GetUser(
        users_pb2.GetUserRequest(id='123'),
        timeout=5  # Deadline in seconds
    )
    print(f'User: {response.user.name}')

except grpc.RpcError as e:
    # Handle specific errors
    if e.code() == grpc.StatusCode.NOT_FOUND:
        print('User not found')
    elif e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
        print('Request timeout')
    elif e.code() == grpc.StatusCode.UNAUTHENTICATED:
        print('Authentication required')
    else:
        print(f'Error: {e.code()} - {e.details()}')

---

Interceptors (Middleware)

Server Interceptor

class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        # Before RPC
        method = handler_call_details.method
        print(f'[Server] Received: {method}')

        # Continue to actual RPC
        response = continuation(handler_call_details)

        # After RPC
        print(f'[Server] Completed: {method}')
        return response

# Use interceptor
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[LoggingInterceptor()]
)

Client Interceptor

class AuthInterceptor(grpc.UnaryUnaryClientInterceptor):
    def __init__(self, token):
        self.token = token

    def intercept_unary_unary(self, continuation, client_call_details, request):
        # Add authentication metadata
        metadata = []
        if client_call_details.metadata:
            metadata = list(client_call_details.metadata)
        metadata.append(('authorization', f'Bearer {self.token}'))

        # Update call details
        new_details = client_call_details._replace(metadata=metadata)

        # Continue with modified metadata
        return continuation(new_details, request)

# Use interceptor
channel = grpc.insecure_channel('localhost:50051')
intercepted_channel = grpc.intercept_channel(
    channel,
    AuthInterceptor(token='secret-token')
)
stub = users_pb2_grpc.UserServiceStub(intercepted_channel)

---

Common Patterns and Best Practices

Deadlines (Timeouts)

Always set deadlines to prevent hanging requests:

# Client-side deadline
response = stub.GetUser(
    request,
    timeout=5  # 5 seconds
)

# Server-side deadline check
def GetUser(self, request, context):
    if context.time_remaining() < 1:  # Less than 1 second left
        context.set_code(grpc.StatusCode.DEADLINE_EXCEEDED)
        return users_pb2.GetUserResponse()

Metadata (Headers)

# Client: Send metadata
metadata = [
    ('authorization', 'Bearer token123'),
    ('request-id', 'abc-123')
]
response = stub.GetUser(request, metadata=metadata)

# Server: Read metadata
def GetUser(self, request, context):
    metadata = dict(context.invocation_metadata())
    auth_token = metadata.get('authorization')
    request_id = metadata.get('request-id')

Connection Management

# Reuse channels (don't create per-request)
channel = grpc.insecure_channel('localhost:50051')
stub = users_pb2_grpc.UserServiceStub(channel)

# Close when done
channel.close()

# Connection options
channel = grpc.insecure_channel(
    'localhost:50051',
    options=[
        ('grpc.max_receive_message_length', 10 * 1024 * 1024),  # 10 MB
        ('grpc.max_send_message_length', 10 * 1024 * 1024),     # 10 MB
        ('grpc.keepalive_time_ms', 30000),                       # 30 seconds
    ]
)

---

Anti-Patterns

❌ Not setting deadlines: Requests can hang forever ✅ Set timeouts: stub.GetUser(request, timeout=5)

❌ Creating channels per request: Expensive (TCP connection overhead) ✅ Reuse channels: Create once, use many times

❌ Ignoring status codes: All errors return same generic message ✅ Handle specific codes: Check grpc.StatusCode.NOT_FOUND, etc.

❌ Large messages: Sending 100 MB+ messages ✅ Use streaming: Break into smaller chunks

❌ No error handling in streams: Stream errors kill connection ✅ Wrap in try/except: Handle grpc.RpcError

❌ Missing field numbers: Can't decode old messages ✅ Never reuse field numbers: Mark as reserved

❌ Breaking changes: Removing required fields ✅ Use proto evolution: Add fields, use reserved, deprecate

---

Related Skills

• protobuf-schema-design.md - Advanced Protocol Buffers patterns
• grpc-load-balancing.md - Client-side and proxy load balancing
• grpc-security.md - TLS, authentication, authorization
• http2-fundamentals.md - Understanding HTTP/2 (gRPC transport)
• api-rest-design.md - Comparing REST vs gRPC

---

Level 3: Resources

Overview

This skill includes comprehensive Level 3 resources for deep gRPC implementation knowledge and practical tools.

Resources include:

• REFERENCE.md (2,303 lines): Complete technical reference covering all gRPC concepts
• 3 executable scripts: Proto validation, client generation, server testing
• 8 production examples: Complete implementations across multiple languages

REFERENCE.md

Location: skills/protocols/grpc-implementation/resources/REFERENCE.md

Comprehensive technical reference (2,303 lines) covering:

Core Topics:

• gRPC fundamentals and architecture
• Protocol Buffers (syntax, types, evolution)
• Service definitions and RPC types
• Error handling and status codes
• Interceptors and middleware
• Metadata and headers
• Deadlines and timeouts
• Load balancing strategies
• Security (TLS, authentication)
• Performance optimization
• Streaming patterns
• Testing strategies
• Debugging tools
• Migration strategies
• Anti-patterns
• Language-specific guides (Python, Go, Node.js)

Key Sections:

1. Fundamentals: HTTP/2 transport, Protocol Buffers, code generation
2. Protocol Buffers: Complete syntax guide, field types, schema evolution
3. RPC Types: All four types with complete implementations (unary, server streaming, client streaming, bidirectional)
4. Error Handling: Status codes, rich error details, retry strategies
5. Interceptors: Server and client interceptors for cross-cutting concerns
6. Load Balancing: Client-side and proxy-based strategies
7. Security: TLS/mTLS, token-based auth, OAuth 2.0
8. Performance: Connection pooling, compression, keepalive
9. Tools: grpcurl, buf, grpc-gateway, health probes
10. Anti-Patterns: Common mistakes and solutions

Format: Markdown with extensive code examples in Python, Go, and Node.js

Scripts

Three production-ready executable scripts in resources/scripts/:

1. validate_proto.py (687 lines)

Purpose: Validate Protocol Buffer definitions

Features:

• Parse and validate .proto syntax
• Check naming conventions (PascalCase messages, snake_case fields)
• Detect breaking changes between versions
• Validate service definitions
• Check best practices (reserved fields, deprecation)
• Output as JSON or human-readable text

Usage:

# Basic validation
./validate_proto.py --proto-file api.proto

# JSON output
./validate_proto.py --proto-file api.proto --json

# Check breaking changes
./validate_proto.py --proto-file api_v2.proto --check-breaking --baseline api_v1.proto

# Save report
./validate_proto.py --proto-file api.proto --json > validation-report.json

Categories checked:

• Syntax: Proto syntax errors, duplicate field numbers, reserved ranges
• Naming: Convention violations (PascalCase, snake_case)
• Best practices: Missing options, field number efficiency, request/response patterns
• Breaking changes: Deleted messages/services, changed types, reused field numbers

2. generate_client.py (906 lines)

Purpose: Generate gRPC client code and examples

Features:

• Generate client code for Python, Go, Node.js
• Include example usage code
• Add error handling patterns
• Generate interceptor templates
• Create testing helpers
• Support multiple output formats

Usage:

# Generate Python client
./generate_client.py --proto-file api.proto --language python --output-dir ./client

# Generate Go client
./generate_client.py --proto-file api.proto --language go --output-dir ./client

# JSON output (list generated files)
./generate_client.py --proto-file api.proto --language python --json

Generated files:

• Python: client.py, example.py, interceptor.py, test_client.py
• Go: client.go, example.go
• Node.js: client.js, example.js

Includes: Error handling, retry logic, connection management, interceptors

3. test_grpc_server.sh (643 lines)

Purpose: Test gRPC server endpoints and performance

Features:

• Test all four RPC types
• Measure latency and throughput
• Test error handling
• Validate metadata
• Generate test reports (JSON or text)
• Performance benchmarking

Usage:

# Test all methods
./test_grpc_server.sh --server localhost:50051 --proto-file api.proto

# Test specific method
./test_grpc_server.sh --server localhost:50051 --proto-file api.proto --method UserService/GetUser

# With metadata (authentication)
./test_grpc_server.sh --server localhost:50051 --proto-file api.proto --metadata authorization:"Bearer token123"

# JSON output for CI/CD
./test_grpc_server.sh --server localhost:50051 --proto-file api.proto --json > report.json

# Performance test (100 iterations)
./test_grpc_server.sh --server localhost:50051 --proto-file api.proto --iterations 100

Requirements: grpcurl, jq, bc

Metrics: Latency (min, avg, p50, p95, p99, max), throughput (req/sec), success rate

Examples

Eight production-ready examples in resources/examples/:

1. protos/service.proto

Complete service definition demonstrating:

• All four RPC types (unary, server streaming, client streaming, bidirectional)
• Domain models (User, UserStatus)
• Request/response messages
• Enums and oneof types
• Well-known types (Timestamp, Empty)
• Complete CRUD operations plus streaming

Services: UserService with 10 methods covering all RPC patterns

2. python/server.py

Complete Python gRPC server implementation:

• All four RPC types implemented
• In-memory database (example)
• Error handling with proper status codes
• Input validation
• Metadata handling
• Authentication checks
• Comprehensive logging
• Context management

Key features: Unary CRUD, server streaming (ListUsers, WatchUserChanges), client streaming (CreateUsers, UploadUserData), bidirectional (Chat, CollaborativeEdit)

3. python/client.py

Complete Python gRPC client implementation:

• Test functions for all RPC types
• Error handling with retries
• Timeout configuration
• Metadata support
• Comprehensive examples
• Test suite for error cases

Tests: Unary RPCs, server streaming, client streaming, bidirectional streaming, error handling

4. go/server.go

Go gRPC server implementation (template):

• Type-safe implementation
• Context handling
• Error handling with status package
• Connection management

Format: Production-ready Go server template

5. nodejs/server.js

Node.js gRPC server implementation (template):

• Async/await patterns
• Callback-based RPCs
• Error handling
• Proto loading with grpc-js

Format: Production-ready Node.js server template

6. interceptors/auth_interceptor.py

Authentication interceptor example:

• Server interceptor: JWT validation, public methods whitelist, UNAUTHENTICATED errors
• Client interceptor: Automatic token addition, metadata management
• Token generation utilities
• Complete JWT flow example

Features: JWT authentication, token validation, expiration handling, public endpoint support

7. streaming/bidirectional_chat.py

Bidirectional streaming example (template):

• Chat room implementation
• Concurrent send/receive
• Connection management
• Message broadcasting

Pattern: Full duplex communication for real-time chat

8. docker/docker-compose.yml

Production Docker deployment:

• gRPC server container
• gRPC client container
• grpcurl for testing
• Health checks
• Service networking
• Volume mounts for code
• Automatic code generation

Services: grpc-server (Python), grpc-client (Python), grpcurl (testing)

Features: Health checks, service discovery, logging, production notes for TLS/auth

Quick Start

1. Validate your .proto file:

cd skills/protocols/grpc-implementation/resources/scripts
./validate_proto.py --proto-file ../examples/protos/service.proto --json

2. Generate client code:

./generate_client.py --proto-file ../examples/protos/service.proto --language python --output-dir ./client

3. Run examples:

cd ../examples

# Generate stubs
python -m grpc_tools.protoc -I./protos --python_out=./python --grpc_python_out=./python protos/service.proto

# Start server
python python/server.py

# Run client (in another terminal)
python python/client.py

4. Test with Docker:

cd docker
docker-compose up

5. Test server:

cd scripts
./test_grpc_server.sh --server localhost:50051 --proto-file ../examples/protos/service.proto --json

File Structure

skills/protocols/grpc-implementation/
├── grpc-implementation.md (this file)
└── resources/
    ├── REFERENCE.md (2,303 lines)
    ├── scripts/
    │   ├── validate_proto.py (687 lines) - Proto validation
    │   ├── generate_client.py (906 lines) - Client generation
    │   └── test_grpc_server.sh (643 lines) - Server testing
    └── examples/
        ├── protos/
        │   └── service.proto - Complete service definition
        ├── python/
        │   ├── server.py - Python server (all RPC types)
        │   └── client.py - Python client (all RPC types)
        ├── go/
        │   └── server.go - Go server template
        ├── nodejs/
        │   └── server.js - Node.js server template
        ├── interceptors/
        │   └── auth_interceptor.py - JWT authentication
        ├── streaming/
        │   └── bidirectional_chat.py - Chat example
        └── docker/
            └── docker-compose.yml - Docker deployment

Resources Summary

| Category | Item | Lines | Description | |----------|------|-------|-------------| | Reference | REFERENCE.md | 2,303 | Complete technical reference | | Scripts | validate_proto.py | 687 | Proto validation tool | | | generate_client.py | 906 | Client code generator | | | test_grpc_server.sh | 643 | Server testing tool | | Examples | service.proto | 156 | Complete service definition | | | python/server.py | 450 | Python server implementation | | | python/client.py | 280 | Python client implementation | | | auth_interceptor.py | 240 | JWT authentication example | | | docker-compose.yml | 90 | Docker deployment |

Total: 5,755 lines of production-ready resources

---

Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)