PyPIWeb Backendupdated 22 Sept 2026

grpc-interceptor review

Q: Does grpc-interceptor install cleanly?

Yes. In a fresh container with an empty cache, pip install grpc-interceptor finished in 0.3s, leaving 3 packages and 18 MB on disk. pip-audit reported no known vulnerabilities.

Q: What does grpc-interceptor need to run?

Python >=3.7,<4.0, and nothing compiled: it is pure Python. In our run import grpc_interceptor succeeded in 0.27s, and the package ships py.typed for type checkers.

Q: grpc-interceptor or grpcio: which should you use?

grpcio: Use its native APIs when an extra interceptor abstraction is unnecessary. grpc-interceptor 0.15.4 installed in 0.3 seconds, used 18 MB, and imported in 0.27 seconds in our sandbox with 0 audit findings.

Q: When should you not use grpc-interceptor?

You need asyncio client middleware. The 0.15.4 documentation says no async counterpart to ClientInterceptor is implemented.

grpc-interceptor 0.15.4 gives Python gRPC middleware direct access to decoded requests, the service context, the RPC method name, and the next handler. It includes synchronous and asyncio server bases, a synchronous client base, status-aware exception classes, and a small end-to-end test service. grpcio still owns channels, servers, generated stubs, and protocol behavior. Version 0.15.4 skips custom interception when a requested method is not registered and brings sync and async server types into closer agreement. Async client interceptors remain absent.

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

Verdict

grpc-interceptor 0.15.4 installed in 0.3 seconds, used 18 MB, and imported in 0.27 seconds in our sandbox with 0 audit findings. Its shorter server API is still useful, but a November 2023 release and no asyncio client layer make it a poor default for an all-async stack.

We installed it

uv pip install grpc-interceptor (v0.15.4) on 22 Aug 2026, fresh sandbox (3 CPUs, 8 GB, Python 3.12, empty cache). How we test →

Lab card: what happened when we installed grpc-interceptor

Screenshot of grpc-interceptor documentation

Install	✓ · 0.3s	3 packages on disk · 18 MB
Import	✓	import grpc_interceptor in 0.27s · pure Python · py.typed · requires Python >=3.7,<4.0
Known vulns	0	(pip-audit)

Answers from our run

Does grpc-interceptor install cleanly?

Yes. In a fresh container with an empty cache, pip install grpc-interceptor finished in 0.3s, leaving 3 packages and 18 MB on disk. pip-audit reported no known vulnerabilities.

What does grpc-interceptor need to run?

Python >=3.7,<4.0, and nothing compiled: it is pure Python. In our run import grpc_interceptor succeeded in 0.27s, and the package ships py.typed for type checkers.

grpc-interceptor or grpcio: which should you use?

grpcio: Use its native APIs when an extra interceptor abstraction is unnecessary. grpc-interceptor 0.15.4 installed in 0.3 seconds, used 18 MB, and imported in 0.27 seconds in our sandbox with 0 audit findings.

When should you not use grpc-interceptor?

You need asyncio client middleware. The 0.15.4 documentation says no async counterpart to ClientInterceptor is implemented.

API stability4/5The 0.15.4 surface is small and explicit: server implementations override `intercept(method, request_or_iterator, context, method_name)`, client implementations receive `ClientCallDetails`, and exception types carry a gRPC status. Release notes describe type symmetry fixes rather than a redesign. The concern is that grpcio and asyncio behavior continue to evolve underneath this wrapper while grpc-interceptor has not released a version since 2023.

Docs4/5The README and Read the Docs explain unary servers, server streams, asyncio streams, synchronous clients, continuation futures, exception conversion, testing extras, and the missing async client feature. The async section also documents the hard case where `context.read()` and `context.write()` hide stream messages from an ordinary iterator wrapper. Retry deadlines, cancellation, rich status details, and production logging still require grpcio documentation and application-specific design.

Maintenance2/5PyPI still lists 0.15.4, published on 2023-11-16, as current. GitHub showed 151 stars, 8 open issues and pull requests together, an unarchived repository, and a last push on 2024-04-11. Slow releases can be reasonable for a compact wrapper, but this package follows grpcio handler behavior and Python's async APIs. Teams adopting it now should run compatibility tests against every grpcio upgrade instead of expecting quick wrapper releases.

Ecosystem3/5The package works with grpcio's synchronous clients and servers, adds an asyncio server base, covers the four RPC shapes on its synchronous client path, and offers a real in-process service for tests. Its scope stops there: it does not compile protobufs, instrument other languages, supply an async client interceptor, or replace grpcio. Compilation, tracing, load balancing, channel configuration, and cross-language policy remain in other packages or service infrastructure.

Use it if

Python server middleware needs decoded protobuf messages and `ServicerContext` in one short method signature.
Business code should raise `NotFound` or another `GrpcException` and map it to a gRPC status in one place.
A synchronous client interceptor must cover unary, client-streaming, server-streaming, and bidirectional calls.
Interceptor tests should exercise a real in-process gRPC service through the package's optional testing helpers.

Skip it if

You need asyncio client middleware. The 0.15.4 documentation says no async counterpart to `ClientInterceptor` is implemented.
grpcio's native interceptor API is acceptable to your team. This wrapper shortens access to requests and context but adds no wire-level capability.
Your dependency policy requires a recently released wrapper. Version 0.15.4 was published on 2023-11-16 and the repository's last push was 2024-04-11.
One unary interceptor body must also cover every stream shape without extra work. Response errors can happen during iterator consumption, after the service method has returned.
Your async services use the `context.read()` and `context.write()` streaming API and middleware must inspect each message. The docs say you need a custom context wrapper for that path.
Retry behavior must be ready-made and policy-aware. The client base exposes the continuation, while deadlines, cancellation, safe methods, backoff, and retryable status choices remain your responsibility.

Setup reality

Our grpc-interceptor 0.15.4 install finished in 0.3 seconds in a fresh Python 3.12 container. It produced 3 installed packages using 18 MB and declared 2 direct dependencies. import grpc_interceptor worked in 0.27 seconds, and pip-audit found 0 known vulnerabilities. The distribution is pure Python, supports Python 3.7 through the 3.x line, includes py.typed, and uses the MIT license.

The base install contains the interceptors. Install the testing extra for dummy_client, its generated protobuf messages, and the in-process test service. Application services still need .proto compilation and generated bindings, commonly supplied by grpcio-tools. Pass interceptors into grpc.server(...) or grpc.aio.server(...) during construction; the order in that list determines how they wrap one another.

Unary and streaming handlers need separate control flow. A synchronous streaming method returns an iterator whose work and exceptions occur while it is consumed. An asyncio unary handler returns an awaitable, while an async generator must be iterated rather than awaited. The alternative async context.read() and context.write() API can return None from a streaming method, so the documented __aiter__ check cannot expose each message there.

ExceptionToStatusInterceptor maps GrpcException subclasses to status codes and details. Unknown failures otherwise reach clients as UNKNOWN, or as the configured fallback status. Do not expose raw exception representations as public details. On the client side, rebuilding ClientCallDetails must retain the 6 fields for method, timeout, metadata, credentials, wait-for-ready, and compression. Version 0.15.4 also avoids intercepting unregistered methods, which matters for gRPC's missing-method response path.

Patterns

Map typed exceptions on a server install-status-mapper

from concurrent import futures
import grpc
from grpc_interceptor import ExceptionToStatusInterceptor

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

Interceptors are fixed when the server is created. Service registration and `server.start()` still happen separately.

Return a NOT_FOUND status raise-not-found

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

`NotFound` reaches the wire as `NOT_FOUND` only when `ExceptionToStatusInterceptor` or equivalent handling wraps the service.

Inspect a synchronous request write-server-interceptor

from grpc_interceptor import ServerInterceptor

class RequestAudit(ServerInterceptor):
    def intercept(self, method, request_or_iterator, context, method_name):
        audit(method_name, request_or_iterator)
        return method(request_or_iterator, context)

`request_or_iterator` is one protobuf message for unary input and an iterator for client-streaming input. Consuming the iterator changes what the service receives.

Observe a stream through completion wrap-response-stream

from grpc_interceptor import ServerInterceptor

class StreamAudit(ServerInterceptor):
    def intercept(self, method, request, context, method_name):
        response = method(request, context)
        if not hasattr(response, '__iter__'):
            return response

        def wrapped():
            try:
                yield from response
            finally:
                audit_closed(method_name)

        return wrapped()

Server-streaming work continues while the iterator is consumed. Timing only the call to `method(...)` measures generator creation rather than the whole RPC.

Split the canonical RPC method name parse-method-name

from grpc_interceptor import ServerInterceptor, parse_method_name

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

gRPC supplies names in `/package.Service/Method` form. `parse_method_name()` avoids custom slash and package parsing.

Preserve unary and async-generator handlers write-async-interceptor

from grpc_interceptor import AsyncServerInterceptor

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

An async generator is not awaitable, while a unary coroutine is. The `context.write()` streaming style needs different interception because the handler can return `None`.

Translate exceptions on grpc.aio map-async-exceptions

from grpc import aio
from grpc_interceptor import AsyncExceptionToStatusInterceptor

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

The async mapper is made for `grpc.aio.server`. The synchronous exception interceptor calls a different context API.

Choose a fallback status set-unknown-fallback

import grpc
from grpc_interceptor import ExceptionToStatusInterceptor

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

The fallback changes the status for exceptions outside the package's `GrpcException` family. Send a controlled details string so internal messages do not reach callers.

Add metadata to synchronous calls append-client-metadata

import grpc
from grpc_interceptor import ClientCallDetails, ClientInterceptor

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

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

`ClientCallDetails` has 6 values that affect the call. Preserve timeout, credentials, wait-for-ready, and compression when replacing metadata.

Retry through the continuation retry-client-call

from grpc_interceptor import ClientInterceptor

class OneRetry(ClientInterceptor):
    def intercept(self, method, request_or_iterator, call_details):
        future = method(request_or_iterator, call_details)
        try:
            future.result()
            return future
        except Exception:
            return method(request_or_iterator, call_details)

The continuation returns a future, unlike a normal blocking stub call. A production retry must check status, deadline, cancellation, and operation safety before repeating it.

Test status mapping end to end test-interceptor

import grpc
import pytest
from grpc_interceptor import ExceptionToStatusInterceptor
from grpc_interceptor.exceptions import NotFound
from grpc_interceptor.testing import DummyRequest, dummy_client, raises

def test_missing_item():
    cases = {'missing': raises(NotFound())}
    with dummy_client(
        special_cases=cases,
        interceptors=[ExceptionToStatusInterceptor()],
    ) as client:
        with pytest.raises(grpc.RpcError) as caught:
            client.Execute(DummyRequest(input='missing'))
        assert caught.value.code() == grpc.StatusCode.NOT_FOUND

`dummy_client` belongs to the optional `testing` extra and starts a real in-process gRPC service. It catches integration behavior that a mocked context can miss.

Make wrapping order explicit order-interceptors

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

List order controls which interceptor wraps the next handler. Test the chosen order because an exception mapper can catch failures raised by inner middleware.

Alternatives

Package	Registry	Pick it when
grpcio	PyPI	Use its native APIs when an extra interceptor abstraction is unnecessary.
grpclib	PyPI	Use it for a separate asyncio-oriented, pure-Python gRPC implementation.
opentelemetry-instrumentation-grpc	PyPI	Use it when standard client and server tracing is the only middleware requirement.

More web backend guides

urllib3 · requests · ws · anyio · httpx · undici · the whole shelf →

How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.