# Usage example
class UserService(user_pb2_grpc.UserServiceServicer):
    
    def CreateUser(self, request, context):
        # Validate business rules
        if self.user_repository.email_exists(request.email):
            CustomErrorHandler.business_logic_error(
                context,
                error_code="DUPLICATE_EMAIL",
                message="Email address is already registered",
                error_context={"email": request.email},
                suggested_actions=[
                    "Use a different email address",
                    "Reset password if you forgot your account"
                ]
            )
            return user_pb2.User()
        
        # Continue with user creation...
```

## Error interceptors

Implement global error handling with interceptors:

```python title="error_interceptor.py"
import logging
import traceback
from grpc import ServicerContext, StatusCode

class ErrorInterceptor(grpc.ServerInterceptor):
    
    def __init__(self, logger=None):
        self.logger = logger or logging.getLogger(__name__)
    
    def intercept_service(self, continuation, handler_call_details):
        def error_wrapper(behavior):
            def wrapper(request, context):
                try:
                    return behavior(request, context)
                    
                except ValueError as e:
                    # Convert Python ValueError to gRPC INVALID_ARGUMENT
                    context.set_code(StatusCode.INVALID_ARGUMENT)
                    context.set_details(f"Invalid input: {str(e)}")
                    self.logger.warning(f"Validation error in {handler_call_details.method}: {e}")
                    return self._get_default_response(behavior)
                    
                except PermissionError as e:
                    context.set_code(StatusCode.PERMISSION_DENIED)
                    context.set_details("Access denied")
                    self.logger.warning(f"Permission denied in {handler_call_details.method}: {e}")
                    return self._get_default_response(behavior)
                    
                except TimeoutError as e:
                    context.set_code(StatusCode.DEADLINE_EXCEEDED)
                    context.set_details("Operation timed out")
                    self.logger.error(f"Timeout in {handler_call_details.method}: {e}")
                    return self._get_default_response(behavior)
                    
                except Exception as e:
                    # Log unexpected errors
                    self.logger.error(
                        f"Unexpected error in {handler_call_details.method}: {e}\n"
                        f"Traceback: {traceback.format_exc()}"
                    )
                    
                    context.set_code(StatusCode.INTERNAL)
                    context.set_details("Internal server error")
                    return self._get_default_response(behavior)
            
            return wrapper
        
        return grpc.unary_unary_rpc_method_handler(
            error_wrapper(continuation(handler_call_details).unary_unary)
        )
    
    def _get_default_response(self, behavior):
        """Return an empty response of the correct type."""
        # This would need to be implemented based on your service methods
        return None
```

## Client-side error handling

Handle errors on the client side:

```python title="client_error_handling.py"
import grpc
from google.rpc import status_pb2, error_details_pb2
from google.protobuf import any_pb2

def handle_grpc_errors(stub_method, request):
    """Generic error handling for gRPC client calls."""
    
    try:
        response = stub_method(request)
        return response, None
        
    except grpc.RpcError as e:
        error_info = {
            'code': e.code(),
            'details': e.details(),
            'status': e.code().name
        }
        
        # Extract rich error details if available
        metadata = dict(e.trailing_metadata())
        if 'grpc-status-details-bin' in metadata:
            try:
                status_detail = status_pb2.Status()
                status_detail.ParseFromString(metadata['grpc-status-details-bin'])
                
                error_info['rich_details'] = []
                for detail in status_detail.details:
                    # Try to unpack common error types
                    if detail.Is(error_details_pb2.RetryInfo.DESCRIPTOR):
                        retry_info = error_details_pb2.RetryInfo()
                        detail.Unpack(retry_info)
                        error_info['rich_details'].append({
                            'type': 'retry_info',
                            'retry_delay_seconds': retry_info.retry_delay.seconds
                        })
                    elif detail.Is(error_details_pb2.QuotaFailure.DESCRIPTOR):
                        quota_failure = error_details_pb2.QuotaFailure()
                        detail.Unpack(quota_failure)
                        error_info['rich_details'].append({
                            'type': 'quota_failure',
                            'violations': [
                                {
                                    'subject': v.subject,
                                    'description': v.description
                                } for v in quota_failure.violations
                            ]
                        })
                        
            except Exception:
                # If we can't parse rich details, that's okay
                pass
        
        return None, error_info