Skip to content

Adding Interceptors to grpc.aio Clients and Servers

An interceptor is gRPC's middleware: one object that sees every call on a channel or a server, before and after the handler. The two jobs it does best are the ones nobody wants to write per method — attaching authentication and trace metadata on the client, and turning unhandled exceptions into sensible statuses on the server. That second one is not optional. Without it, a servicer that raises reaches the caller as UNKNOWN: Unexpected <class 'RuntimeError'>: unhandled bug, verified below — a useless status and an information leak in one message. With a mapping interceptor, the same failure became INTERNAL: internal error while the real exception went to the logs.

Prerequisites

Where an interceptor sits 5 stages from client interceptor to back out, both ways. Where an interceptor sits client interceptor adds metadata the wire HTTP/2 stream server interceptor wraps the handler servicer runs your code back out, both ways timing, status, logs Interceptors run for every call, so everything they do is multiplied by your request rate.

1. Write a server interceptor that cannot leak

grpc.aio.ServerInterceptor has one method. It receives the call details, awaits a continuation to get the real handler, and returns a handler — usually a wrapped one:

class ExceptionMapping(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        handler = await continuation(handler_call_details)
        if handler is None or not handler.unary_unary:
            return handler                             # leave other shapes alone

        inner = handler.unary_unary

        async def wrapper(request, context):
            started = time.perf_counter()
            try:
                reply = await inner(request, context)
                RPC_LATENCY.labels(handler_call_details.method, "OK").observe(
                    time.perf_counter() - started)
                return reply
            except grpc.RpcError:
                raise                                  # already a status: pass it through
            except Exception:
                log.exception("unhandled error in %s", handler_call_details.method)
                await context.abort(grpc.StatusCode.INTERNAL, "internal error")

        return grpc.unary_unary_rpc_method_handler(
            wrapper,
            request_deserializer=handler.request_deserializer,
            response_serializer=handler.response_serializer)

Three details are load-bearing. except grpc.RpcError: raise lets a deliberate context.abort() through unchanged — without it, every intentional NOT_FOUND becomes INTERNAL. The serializers must be carried across, or the replacement handler cannot decode anything. And the client gets a generic message while the real exception, with its traceback, goes to the log where it belongs.

Register it when building the server:

server = grpc.aio.server(interceptors=[ExceptionMapping()])

Verify: a servicer that raises produces INTERNAL at the client and a logged traceback on the server.

2. Handle all four call shapes, or return the handler unchanged

RpcMethodHandler has four mutually exclusive attributes — unary_unary, unary_stream, stream_unary, stream_stream — and exactly one is set. An interceptor that assumes unary_unary and builds a new handler from it breaks every streaming method on the server, usually with an error that points at the servicer rather than the interceptor.

The safe pattern is to dispatch explicitly:

        if handler.unary_unary:
            return grpc.unary_unary_rpc_method_handler(wrap_unary(handler.unary_unary), ...)
        if handler.unary_stream:
            return grpc.unary_stream_rpc_method_handler(wrap_stream(handler.unary_stream), ...)
        return handler                                 # the shapes you do not wrap

Note the handler is None check too: continuation returns None for a method the server does not implement, and gRPC turns that into UNIMPLEMENTED. An interceptor that dereferences it crashes on every mistyped method name a client sends.

Wrapping a streaming handler means wrapping an async generator, which is a different function shape — so most services implement the unary case carefully and leave streams alone until they need them.

Verify: streaming methods still work after the interceptor is added.

Writing a server interceptor 5 ordered steps. Writing a server interceptor intercept_service(continuation, details) one coroutine handler = await continuation(details) the real handler check handler.unary_unary etc. four shapes exist wrap the behaviour time it, catch it, log it return a same-shaped handler with the same serializers Return the handler unchanged for shapes you do not handle, or streaming RPCs break.

3. Add metadata from a client interceptor

Client interceptors are per call shape as well; UnaryUnaryClientInterceptor covers the common case:

class AuthInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    async def intercept_unary_unary(self, continuation, client_call_details, request):
        metadata = grpc.aio.Metadata(*(client_call_details.metadata or ()))
        metadata.add("authorization", f"Bearer {await self.token()}")
        details = client_call_details._replace(metadata=metadata)
        return await continuation(details, request)


channel = grpc.aio.insecure_channel(target, interceptors=[AuthInterceptor()])

Verified end to end: the server read Bearer token-123 from context.invocation_metadata() without any call site passing it. client_call_details is a named tuple, so _replace is the documented way to modify it, and the metadata must be rebuilt rather than mutated in place.

This is also where trace context belongs — injecting the current span's headers so the server can continue the trace, as covered in tracing asyncio services with OpenTelemetry. The OpenTelemetry gRPC instrumentation is itself a pair of interceptors.

Verify: the server sees the metadata for every call, including ones added after the interceptor was written.

4. Keep interceptors cheap

An interceptor runs on every RPC, on the event loop, so its cost is multiplied by your request rate. Three rules keep that cost invisible:

  • No I/O on the hot path. A token fetched per call is a round trip added to every call; cache it and refresh in the background.
  • No per-call logging at INFO. A log line per RPC is the highest-volume thing your service writes; record metrics instead, and log only failures.
  • Bounded work. Serialising the request to inspect it, hashing it, or building a rich span for every call all cost real time. Sample instead of doing it always.

The order of interceptors is the order they are given, with the first one outermost. Put exception mapping outermost so it also catches failures raised inside the other interceptors, and metrics inside it so the recorded status matches what the client actually receives.

Verify: adding the interceptors does not move p50 latency measurably.

What belongs in an interceptor A grid of 5 rows by 2 columns. What belongs in an interceptor concern side why there auth tokens client every call gets one, no call sites change trace context both inject on the client, extract on the server latency and status metrics server one place that sees every RPC exception to status server stops UNKNOWN reaching clients retries client, carefully never for streams that have delivered Verified: without the mapping interceptor, an unhandled exception reached the client as UNKNOWN with its repr.

5. Know when a decorator is the better tool

Interceptors are for concerns that apply to every call. Behaviour specific to one method — validating a particular field, caching one expensive lookup, enforcing a per-method rate limit — is clearer as a decorator or as code in the servicer, where a reader can see it.

The trade is discoverability against coverage. An interceptor cannot be forgotten when a method is added, which is exactly why auth and error mapping belong there; a decorator is visible at the definition, which is why business rules belong in the method. A service that puts business logic in interceptors becomes one where nobody can answer "what happens when I call this?" without reading the server construction code.

Verify: each interceptor's purpose is expressible as "for every RPC", without exceptions.

Interceptor or decorator? 2 columns contrasting interceptor, per-method decorator. Interceptor or decorator? interceptor one place, every call new methods are covered automatically works for calls with no handler, like unimplemented sees the method name and metadata must handle all four call shapes per-method decorator explicit, per method nothing hidden from the reader easy to skip one by accident no shape dispatch needed duplicated across a large service Use interceptors for cross-cutting concerns and decorators for behaviour specific to one method.

Verification

Interceptors are correctly built when:

  • Unhandled exceptions become INTERNAL, with the detail logged rather than returned.
  • Deliberate abort() statuses pass through unchanged.
  • All four handler shapes are handled or returned untouched, and None is checked.
  • Client metadata is added by rebuilding client_call_details with _replace.
  • No interceptor does I/O per call, and none logs per successful call.
  • Ordering is deliberate: error mapping outermost, metrics inside it.

Pitfalls & edge cases

  • Assuming unary_unary. Streaming methods break, with an error that blames the servicer.
  • Not checking handler is None. Unimplemented method names crash the interceptor instead of returning UNIMPLEMENTED.
  • Dropping the serializers. The wrapped handler cannot decode requests or encode replies.
  • Swallowing grpc.RpcError. Deliberate statuses become INTERNAL and clients lose their error semantics.
  • Retrying inside a client interceptor. Safe only for unary, idempotent calls; a stream that has delivered messages cannot be replayed.
  • Mutating client_call_details in place. It is a named tuple; use _replace.

Frequently Asked Questions

How do I add middleware to a grpc.aio server?

Implement grpc.aio.ServerInterceptor with an async intercept_service, await the continuation to get the real handler, wrap its behaviour, and return a handler of the same shape with the same serializers. Pass instances to grpc.aio.server(interceptors=[...]).

How do I stop gRPC returning UNKNOWN for unhandled exceptions?

Add a server interceptor that catches Exception, logs it, and calls context.abort(grpc.StatusCode.INTERNAL, "internal error"). Let grpc.RpcError through unchanged so deliberate statuses survive. Without it, clients see UNKNOWN with the exception's repr, which leaks internals.

How do I add an auth token to every gRPC call in Python?

Write a UnaryUnaryClientInterceptor that rebuilds client_call_details with _replace and an extra metadata entry, then pass it to grpc.aio.insecure_channel(..., interceptors=[...]). Verified, the server then reads the header from context.invocation_metadata() with no change to call sites.

Why did my gRPC streaming methods break after adding an interceptor?

Because the interceptor built a unary_unary handler for every method. Exactly one of unary_unary, unary_stream, stream_unary and stream_stream is set on each handler; check which, wrap that one, and return the handler unchanged for shapes you do not support.

Do gRPC interceptors slow down calls?

Only as much as what they do. They run on the event loop for every RPC, so a token fetch, a synchronous log write or a full request serialisation per call is multiplied by your request rate. Cache credentials, record metrics rather than logs, and sample expensive work.