PyPIWeb Backendupdated 08 Aug 2026

grpc-interceptor

Item: grpc-interceptor
Rating: 3.3
Author: Keyoor

grpc-interceptor is a small helper layer over Python gRPC's interceptor APIs. Its server base class passes the decoded request, service context, method name, and next handler directly to one intercept method, and its client base class unifies all four synchronous RPC shapes. It also supplies exception-to-status interceptors and a testing extra. It does not replace grpcio, generate stubs, or define a cross-language middleware standard.

grpc-interceptor · v0.15.4 · 6.1M/wk · ★ 151 · pushed 11 Apr 2024 · d5h-foss/grpc-interceptor · docs · reviewed by Keyoor

Verdict

grpc-interceptor makes Python server middleware noticeably easier to read, especially for exception mapping. Its stalled release history and missing asyncio client support make it a selective convenience, not an automatic addition to every grpcio service.

API stability4/5The package is tiny and its principal ServerInterceptor, AsyncServerInterceptor, ClientInterceptor, ClientCallDetails, exception classes, and method-name helpers have straightforward signatures. That apparent stability partly reflects low recent activity rather than repeated compatibility work, and the wrapper still depends on grpcio's handler behavior underneath.

Docs4/5The README and Read the Docs pages show complete server, client, exception, streaming, asyncio, and testing examples, and they clearly call out the lack of asyncio client interceptors. Because the abstraction is small, the documentation is enough to become productive, though advanced retry, cancellation, rich status, and streaming edge cases still send users to grpcio documentation and tests.

Maintenance2/5The latest PyPI release, 0.15.4, dates to 2023-11-16, and GitHub shows the last push on 2024-04-11 with 151 stars and 8 open issues and PRs. The package is stable and small rather than marked abandoned, but more than two years without a release is meaningful for a wrapper around grpcio internals and asyncio behavior.

Ecosystem3/5It works directly with grpcio servers and synchronous channels, covers all unary and streaming RPC shapes, offers asyncio server bases, and includes protobuf-backed testing helpers. The ecosystem stops there: there is no asyncio client wrapper, no code-generation layer, and no cross-language configuration, so broader gRPC tooling still comes from grpcio and OpenTelemetry packages.

Use it if

You write Python gRPC server middleware and want decoded requests instead of manipulating RpcMethodHandler objects
You need one synchronous client interceptor implementation to cover unary and streaming call shapes
You want service methods to raise typed GrpcException subclasses that become gRPC status codes
You maintain both synchronous and asyncio servers and can use separate interceptor classes for each runtime

Skip it if

You need an asyncio client interceptor: the README explicitly says the package does not provide one
grpcio's native interceptors already meet your needs: this library is a convenience wrapper, so another dependency and abstraction may buy very little
You need an actively released compatibility layer: 0.15.4 was published in November 2023 and the repository's last push was April 2024
You assume one intercept method erases streaming complexity: request iterators, response iterators, async generators, and exceptions raised during iteration still need separate handling
You need middleware shared across services written in several languages: these Python classes do not define portable policies or generated configuration

Setup reality

Install grpc-interceptor alongside grpcio; version 0.15.4 declares grpcio 1.49.1 or newer but below 2.0. The testing helpers require pip install grpc-interceptor[testing], which adds protobuf. Your .proto compilation and generated service classes still come from grpcio-tools or another build pipeline. Server interceptors are passed to grpc.server(..., interceptors=[...]) or grpc.aio.server(interceptors=[...]) when the server is created, so adding one means changing process startup and testing the full handler chain. Order matters because each interceptor receives the next interceptor or service method. The synchronous and asyncio bases are different classes, and an async interceptor must await unary handlers while preserving async iterators for streaming responses. The supplied ExceptionToStatusInterceptor only maps GrpcException subclasses automatically; unexpected exceptions remain UNKNOWN unless you set status_on_unknown_exception or override handle_exception. Exposing repr(exception) to clients can leak internals, so map errors deliberately. Client interception uses grpc.intercept_channel and the synchronous ClientInterceptor; there is no matching asyncio client helper. Metadata must be rebuilt through ClientCallDetails without dropping timeout, credentials, wait_for_ready, or compression. The method name arrives as /package.Service/Method and can be parsed with parse_method_name. Finally, the release age is a real operational concern: test against your pinned grpcio and Python versions before adoption, especially for asyncio and streaming behavior, because grpcio has continued evolving since this package's last release.

Patterns

Install exception-to-status mapping on a servermap-service-exceptions

from concurrent import futures
import grpc
from grpc_interceptor import ExceptionToStatusInterceptor

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[ExceptionToStatusInterceptor()],
)

Interceptors must be supplied when the server is created; the service registration and server start steps are still required.

Raise a typed gRPC error from service coderaise-not-found-status

from grpc_interceptor.exceptions import NotFound

def GetBook(self, request, context):
    book = repository.get(request.id)
    if book is None:
        raise NotFound(f'book {request.id} not found')
    return book

The exception becomes NOT_FOUND only when ExceptionToStatusInterceptor is installed in the server chain.

Time every synchronous RPCwrite-server-interceptor

from time import perf_counter
from grpc_interceptor import ServerInterceptor

class TimingInterceptor(ServerInterceptor):
    def intercept(self, method, request_or_iterator, context, method_name):
        started = perf_counter()
        try:
            return method(request_or_iterator, context)
        finally:
            print(method_name, perf_counter() - started)

For streaming responses this measures creation of the iterator, not consumption; wrap the returned iterator to time the full stream.

Observe errors raised during a response streamwrap-stream-responses

class StreamInterceptor(ServerInterceptor):
    def intercept(self, method, request_or_iterator, context, method_name):
        responses = method(request_or_iterator, context)
        if not hasattr(responses, '__iter__'):
            return responses
        def wrapped():
            try:
                yield from responses
            finally:
                print('stream closed', method_name)
        return wrapped()

A service response protobuf may also be iterable in unusual APIs; use generated method knowledge when deciding whether a handler streams.

Apply middleware to selected methodsfilter-by-method-name

from grpc_interceptor import ServerInterceptor, parse_method_name

class AdminAudit(ServerInterceptor):
    def intercept(self, method, request_or_iterator, context, method_name):
        parsed = parse_method_name(method_name)
        if parsed.service == 'AdminService':
            audit(parsed.method)
        return method(request_or_iterator, context)

The wire name is /package.Service/Method; parse it instead of relying on brittle string slicing.

Intercept asyncio server callswrite-async-server-interceptor

from grpc_interceptor import AsyncServerInterceptor

class AsyncAudit(AsyncServerInterceptor):
    async def intercept(self, method, request_or_iterator, context, method_name):
        result = method(request_or_iterator, context)
        if hasattr(result, '__aiter__'):
            return result
        response = await result
        await audit(method_name)
        return response

Async streaming handlers return async iterators rather than ordinary awaitables; preserve that distinction.

Map typed errors on an asyncio servermap-async-exceptions

from grpc import aio
from grpc_interceptor import AsyncExceptionToStatusInterceptor

server = aio.server(interceptors=[AsyncExceptionToStatusInterceptor()])

Use the async exception interceptor with grpc.aio; the synchronous interceptor calls a different context.abort API.

Choose a status for unexpected server errorsmap-unknown-exceptions

import grpc
from grpc_interceptor import ExceptionToStatusInterceptor

errors = ExceptionToStatusInterceptor(
    status_on_unknown_exception=grpc.StatusCode.INTERNAL
)

The default details use repr(exception), which can expose implementation data; override handle_exception for safe client messages and server logging.

Add metadata to synchronous client callsinject-client-metadata

import grpc
from grpc_interceptor import ClientCallDetails, ClientInterceptor

class TenantInterceptor(ClientInterceptor):
    def intercept(self, method, request_or_iterator, call_details):
        metadata = list(call_details.metadata or []) + [('x-tenant', 'acme')]
        details = ClientCallDetails(
            call_details.method, call_details.timeout, metadata,
            call_details.credentials, call_details.wait_for_ready, call_details.compression,
        )
        return method(request_or_iterator, details)

channel = grpc.intercept_channel(grpc.insecure_channel('localhost:50051'), TenantInterceptor())

Copy every ClientCallDetails field; dropping credentials, deadlines, readiness, or compression changes the call.

Control middleware order explicitlychain-server-interceptors

server = grpc.server(
    executor,
    interceptors=[RequestIdInterceptor(), AuthInterceptor(), ExceptionToStatusInterceptor()],
)

Interceptors wrap in list order; place exception handling where it can catch the failures you intend without hiding authentication policy.

Alternatives

Package	Registry	Pick it when
grpcio	PyPI	You can use the native ServerInterceptor and client interceptor interfaces without a helper layer
grpclib	PyPI	You want an asyncio-native pure-Python gRPC implementation and accept a different stack
opentelemetry-instrumentation-grpc	PyPI	Your goal is standard gRPC tracing rather than writing general-purpose middleware