{"library":"grpc-interceptor","title":"gRPC Interceptor","description":"grpc-interceptor is a Python library that simplifies the implementation of gRPC interceptors. It provides base classes and utility interceptors that offer direct access to request, response, and service context objects, which are typically harder to access with standard `grpc` library interceptors. The library emphasizes a small, readable codebase and minimal dependencies, primarily `grpcio`. It is actively maintained with regular minor releases, currently at version 0.15.4.","status":"active","version":"0.15.4","language":"en","source_language":"en","source_url":"https://github.com/d5h-foss/grpc-interceptor","tags":["grpc","interceptor","middleware","asyncio","exceptions","server","client"],"install":[{"cmd":"pip install grpc-interceptor","lang":"bash","label":"Install core library"},{"cmd":"pip install grpc-interceptor[testing]","lang":"bash","label":"Install with testing framework"}],"dependencies":[{"reason":"Core dependency for gRPC functionality. Version >=1.32.0 is required for async features.","package":"grpcio","optional":false},{"reason":"Included with the testing framework extra.","package":"protobuf","optional":true}],"imports":[{"note":"Do not confuse with the native `grpc.ServerInterceptor` class, which has a different API.","wrong":"from grpc import ServerInterceptor","symbol":"ServerInterceptor","correct":"from grpc_interceptor import ServerInterceptor"},{"symbol":"AsyncServerInterceptor","correct":"from grpc_interceptor import AsyncServerInterceptor"},{"note":"Do not confuse with the native `grpc.ClientInterceptor` class, which has a different API.","wrong":"from grpc import ClientInterceptor","symbol":"ClientInterceptor","correct":"from grpc_interceptor import ClientInterceptor"},{"symbol":"ExceptionToStatusInterceptor","correct":"from grpc_interceptor import ExceptionToStatusInterceptor"},{"symbol":"AsyncExceptionToStatusInterceptor","correct":"from grpc_interceptor import AsyncExceptionToStatusInterceptor"},{"symbol":"GrpcException","correct":"from grpc_interceptor.exceptions import GrpcException"},{"note":"A specific `GrpcException` subclass for NOT_FOUND status.","symbol":"NotFound","correct":"from grpc_interceptor.exceptions import NotFound"}],"quickstart":{"code":"import grpc\nfrom concurrent import futures\nfrom grpc_interceptor import ServerInterceptor\nfrom grpc_interceptor.exceptions import GrpcException, NotFound\n\n# Assuming a generated protobuf service like my_pb2_grpc and my_pb2\n# For demonstration, we'll mock these:\nclass MockRequest:\n    def __init__(self, name):\n        self.name = name\nclass MockResponse:\n    def __init__(self, message):\n        self.message = message\nclass MockServicerContext:\n    def __init__(self):\n        self.code = grpc.StatusCode.OK\n        self.details = ''\n    def set_code(self, code):\n        self.code = code\n    def set_details(self, details):\n        self.details = details\n    def abort(self, code, details):\n        self.set_code(code)\n        self.set_details(details)\n        raise GrpcException(code, details)\n\nclass CustomExceptionInterceptor(ServerInterceptor):\n    def intercept(self, method, request, context, method_name):\n        try:\n            return method(request, context)\n        except GrpcException as e:\n            context.set_code(e.status_code)\n            context.set_details(e.details)\n            raise # Re-raise to let gRPC handle it after context is set\n        except Exception as e:\n            # Catch other unexpected exceptions\n            context.set_code(grpc.StatusCode.INTERNAL)\n            context.set_details(f\"An unexpected error occurred: {e}\")\n            raise\n\nclass MyServiceServicer:\n    def SayHello(self, request, context):\n        if request.name == \"Error\":\n            raise NotFound(\"Name not found!\")\n        return MockResponse(message=f\"Hello, {request.name}!\")\n\ndef serve():\n    # In a real application, you'd use generated stubs and an actual gRPC server\n    interceptors = [CustomExceptionInterceptor()]\n    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), interceptors=interceptors)\n    \n    # In a real app, you'd add your service here:\n    # my_pb2_grpc.add_MyServiceServicer_to_server(MyServiceServicer(), server)\n    \n    # For this example, we'll simulate the service call through the interceptor\n    service_instance = MyServiceServicer()\n    print(\"Simulating RPC calls through the interceptor:\")\n\n    # Successful call\n    req_success = MockRequest(\"World\")\n    ctx_success = MockServicerContext()\n    try:\n        res_success = interceptors[0].intercept(service_instance.SayHello, req_success, ctx_success, \"/MyService/SayHello\")\n        print(f\"Success: {res_success.message} (Status: {ctx_success.code.name})\")\n    except Exception as e:\n        print(f\"Unexpected error in successful call: {e}\")\n\n    # Error call\n    req_error = MockRequest(\"Error\")\n    ctx_error = MockServicerContext()\n    try:\n        interceptors[0].intercept(service_instance.SayHello, req_error, ctx_error, \"/MyService/SayHello\")\n    except GrpcException as e:\n        print(f\"Caught expected error: {ctx_error.details} (Status: {ctx_error.code.name})\")\n    except Exception as e:\n        print(f\"Caught unexpected error type: {e}\")\n\nif __name__ == '__main__':\n    serve()\n","lang":"python","description":"This quickstart demonstrates how to create a custom server interceptor using `grpc-interceptor` to handle exceptions and set gRPC status codes. It includes a basic `ServerInterceptor` subclass that catches `GrpcException` (like `NotFound`) and maps them to gRPC status codes. The example simulates a gRPC call flow to show both successful and error handling paths."},"warnings":[{"fix":"Upgrade your Python environment to 3.7 or newer.","message":"Version 0.15.0 and later dropped support for Python 3.6.0.","severity":"breaking","affected_versions":">=0.15.0"},{"fix":"Ensure `grpcio` is updated to version 1.32.0 or newer (e.g., `pip install --upgrade grpcio`).","message":"Asynchronous server interceptors (AsyncServerInterceptor and AsyncExceptionToStatusInterceptor) introduced in v0.15.0 require grpcio >= 1.32.0. Using older grpcio versions will prevent async features from working.","severity":"breaking","affected_versions":">=0.15.0"},{"fix":"Avoid relying on `MethodName`'s internal type; treat it as an opaque string or object with expected attributes rather than a `NamedTuple`.","message":"The internal type of `MethodName` changed from a `NamedTuple` in version 0.14.1. Code relying on `MethodName` being iterable or having `NamedTuple`-specific behavior might break.","severity":"gotcha","affected_versions":">=0.14.1"},{"fix":"Always import interceptor base classes from `grpc_interceptor` (e.g., `from grpc_interceptor import ServerInterceptor`) when using this library.","message":"Do not confuse `grpc_interceptor.ServerInterceptor` or `grpc_interceptor.ClientInterceptor` with the native `grpc.ServerInterceptor` or `grpc.ClientInterceptor` classes. They have different APIs and intended usage.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Check for the `__aiter__` attribute to determine if the result is an `async_generator`. For the alternate API, you may need to wrap the context object to capture `await context.write(...)` calls.","message":"For async server streaming RPCs, an alternate API was introduced where the RPC method might return `None` instead of an `async_generator`. If your interceptor logic expects an `async_generator` in all streaming cases, it may need adjustment.","severity":"gotcha","affected_versions":">=0.15.0"},{"fix":"Upgrade to `grpc-interceptor` v0.15.3 or newer to ensure correct status code propagation from `context.abort` calls.","message":"Prior to v0.15.3, calling `context.abort` from an interceptor might have resulted in the wrong gRPC status code being set for the RPC. This was fixed in v0.15.3.","severity":"gotcha","affected_versions":"<0.15.3"},{"fix":"Be aware that interceptors will now correctly only apply to registered methods. If you relied on interceptors running for unregistered methods, adjust your logic accordingly, or upgrade to benefit from this fix.","message":"As of v0.15.4, interceptors will be skipped for RPC methods that are not registered in the gRPC server. Previous versions might have invoked interceptors even for unregistered methods, potentially leading to unexpected behavior.","severity":"gotcha","affected_versions":"<0.15.4"}],"env_vars":null,"last_verified":"2026-04-04T00:00:00.000Z","next_check":"2026-07-03T00:00:00.000Z"}