gRPC Interceptors, Metadata, and Auth

Updated 18 Aug 2026 6 interview angles 6 min read source

gRPC Interceptors, Metadata, and Auth

Interceptors are gRPC’s middleware. Metadata is the headers-equivalent — key-value pairs that ride along with every RPC. Together they’re how you do auth, logging, tracing, retry, rate limiting.

Metadata

Key-value pairs attached to RPCs. Like HTTP headers, but the gRPC name.

python

# Client — send metadata
metadata = [
    ("authorization", "Bearer eyJ..."),
    ("x-request-id", "abc-123"),
    ("x-tenant-id", "acme"),
]
response = stub.GetUser(request, metadata=metadata)

# Server — read metadata
def GetUser(self, request, context):
    md = dict(context.invocation_metadata())
    auth = md.get("authorization", "")
    request_id = md.get("x-request-id", "")
    ...

Rules:

Keys are lowercase, ASCII.
Values are strings, OR bytes for keys ending in -bin (binary metadata).
: prefix is reserved (used internally for HTTP/2 pseudo-headers like :method).
Don’t put PII unencrypted in metadata — gets logged by intermediaries.

Auth via metadata

The standard pattern:

python

# Client puts auth in metadata
def get_auth_metadata():
    token = get_current_token()
    return [("authorization", f"Bearer {token}")]

response = stub.GetUser(request, metadata=get_auth_metadata())

Server validates in an interceptor (see below) or in each method.

Auth via channel credentials

For automatic per-call auth (token attached to every RPC):

python

# Custom call credentials
class TokenCredentials(grpc.AuthMetadataPlugin):
    def __init__(self, token):
        self._token = token
    def __call__(self, context, callback):
        callback([("authorization", f"Bearer {self._token}")], None)

call_creds = grpc.metadata_call_credentials(TokenCredentials("eyJ..."))
ssl_creds = grpc.ssl_channel_credentials()
composite = grpc.composite_channel_credentials(ssl_creds, call_creds)

channel = grpc.secure_channel("server:50051", composite)

Now every RPC on this channel carries the token. Useful when the token doesn’t change per request.

For token refresh: AuthMetadataPlugin.__call__ can fetch a fresh token before each RPC.

Server-side interceptor

The middleware equivalent. Runs before each RPC.

python

import grpc

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        # handler_call_details.invocation_metadata is the metadata
        md = dict(handler_call_details.invocation_metadata)
        auth = md.get("authorization", "")

        if not auth.startswith("Bearer "):
            return self._unary_unary_terminator(grpc.StatusCode.UNAUTHENTICATED, "Missing token")

        try:
            user = validate_token(auth[7:])
        except InvalidTokenError:
            return self._unary_unary_terminator(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")

        # store on context (custom propagation)
        # gRPC doesn't have a clean per-RPC context object;
        # use ContextVar or pass via header re-reading in the handler
        return continuation(handler_call_details)

    def _unary_unary_terminator(self, code, message):
        def terminate(request, context):
            context.abort(code, message)
        return grpc.unary_unary_rpc_method_handler(terminate)

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

Interceptors are chained — multiple in the list, applied in order. Common ones: auth, logging, tracing, rate limiting.

Async interceptor

python

class AsyncAuthInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        md = dict(handler_call_details.invocation_metadata)
        if not validate(md.get("authorization", "")):
            ...
        return await continuation(handler_call_details)

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

Same API, async flavor.

Logging interceptor — the always-useful one

python

import time
import logging

class LoggingInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        method = handler_call_details.method
        start = time.perf_counter()

        handler = continuation(handler_call_details)

        def wrapped_handler(request, context):
            try:
                response = handler.unary_unary(request, context)
                elapsed = time.perf_counter() - start
                logging.info(f"{method} OK in {elapsed*1000:.1f}ms")
                return response
            except Exception as e:
                elapsed = time.perf_counter() - start
                logging.error(f"{method} FAILED in {elapsed*1000:.1f}ms: {e}")
                raise

        return grpc.unary_unary_rpc_method_handler(wrapped_handler)

Production servers should always have this (or OpenTelemetry, which generates spans).

Client-side interceptor

For adding metadata uniformly, logging client RPCs, retries:

python

class ClientLoggingInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        start = time.perf_counter()
        response = continuation(client_call_details, request)
        elapsed = time.perf_counter() - start
        logging.info(f"called {client_call_details.method} in {elapsed*1000:.1f}ms")
        return response

channel = grpc.intercept_channel(channel, ClientLoggingInterceptor())

For each RPC type (unary-unary, unary-stream, stream-unary, stream-stream), implement the corresponding interceptor interface.

Per-method auth — when interceptors are too coarse

python

def GetUser(self, request, context):
    md = dict(context.invocation_metadata())
    user = validate_token(md.get("authorization", "").removeprefix("Bearer "))
    if user.id != request.user_id and not user.is_admin:
        context.abort(grpc.StatusCode.PERMISSION_DENIED, "Cannot view other users")
    return ...

Resource-level auth (this user can see this specific resource) usually happens in the method, not the interceptor.

Context propagation — request ID, tracing

Distributed tracing typically rides on metadata:

text

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

W3C Trace Context standard. OpenTelemetry-instrumented gRPC clients/servers handle this automatically.

For app-level context (current user, tenant) across function calls within a request, Python’s contextvars:

python

from contextvars import ContextVar

current_user: ContextVar[User | None] = ContextVar("current_user", default=None)

class AuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        token = get_token(handler_call_details.invocation_metadata)
        user = validate_token(token)
        # context-local; visible to handler
        current_user.set(user)
        return continuation(handler_call_details)

def GetUser(self, request, context):
    user = current_user.get()
    if user is None:
        context.abort(grpc.StatusCode.UNAUTHENTICATED)
    ...

ContextVars work with asyncio and threadpool servers.

OpenTelemetry instrumentation

bash

pip install opentelemetry-instrumentation-grpc

python

from opentelemetry.instrumentation.grpc import GrpcInstrumentorServer, GrpcInstrumentorClient

GrpcInstrumentorServer().instrument()
GrpcInstrumentorClient().instrument()

Automatic spans for every RPC (in and out), with propagated trace IDs. Plug into Jaeger / Zipkin / DataDog / etc.

Rate limiting interceptor

python

class RateLimitInterceptor(grpc.ServerInterceptor):
    def __init__(self):
        self.bucket = TokenBucket(rate=100, capacity=200)

    def intercept_service(self, continuation, handler_call_details):
        if not self.bucket.consume(1):
            def reject(req, ctx):
                ctx.abort(grpc.StatusCode.RESOURCE_EXHAUSTED, "Rate limit exceeded")
            return grpc.unary_unary_rpc_method_handler(reject)
        return continuation(handler_call_details)

For per-client rate limiting, key the bucket by client identity from metadata.

mTLS for service-to-service auth

The standard pattern in service meshes (Istio, Linkerd) — every service has a cert, both sides authenticate. See Python gRPC for setup.

The cert identifies the calling service; you read it via:

python

def GetUser(self, request, context):
    auth_context = context.auth_context()
    peer_cn = auth_context.get("x509_common_name", [b""])[0].decode()
    # peer_cn = "users-service.acme.svc.cluster.local"

Combined with policy: “service A can call methods X, Y; service B can call X only.”

JWT as alternative to mTLS

Skip cert distribution; use signed tokens:

python

class JWTAuthInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        md = dict(handler_call_details.invocation_metadata)
        token = md.get("authorization", "").removeprefix("Bearer ")
        try:
            claims = jwt.decode(token, public_key, algorithms=["RS256"], audience="my-service")
        except jwt.InvalidTokenError:
            ...
        current_user.set(User(id=claims["sub"], role=claims["role"]))
        return continuation(handler_call_details)

Cheaper to operate (no cert rotation per service); less cryptographically strong than mTLS (token can be stolen and replayed).

Common pitfalls

Trusting metadata without validation — clients can send any metadata. Validate auth on the server.
Using mTLS but not checking the peer cert in app code — the TLS layer authenticates; the app must authorize.
Auth interceptor that returns the wrong status code — return UNAUTHENTICATED for missing/invalid creds, PERMISSION_DENIED for valid but disallowed.
Putting tokens in plaintext metadata over grpc.insecure_channel — anyone on the network can grab them. Use TLS.
Forgetting to install for the right RPC pattern — UnaryUnaryClientInterceptor doesn’t cover streaming. Implement all four flavors if you have all four.

Common interview confusions

“Interceptors run on the client only / server only.” — both have their own. Symmetric concept.
“Metadata is encrypted in TLS.” — yes in transit; not at the application layer. Don’t log it.
“UNAUTHENTICATED and PERMISSION_DENIED are the same.” — UNAUTHENTICATED = no/invalid credentials. PERMISSION_DENIED = authenticated but not allowed. Same as HTTP 401 vs 403.

Interview angle 6

“What’s gRPC metadata?” — key-value pairs attached to each RPC, equivalent to HTTP headers. Used for auth tokens, request IDs, trace context, tenant IDs. Set on the client, read on the server.
“How would you implement auth in a gRPC service?” — client sends Bearer token in metadata; server-side interceptor validates the token and attaches user info to a ContextVar (or per-RPC context). Resource-level checks happen in the method handler.
“What’s a gRPC interceptor?” — middleware: a function called before/around each RPC. Common uses: auth, logging, tracing, rate limiting. Server-side and client-side both exist.
“mTLS vs JWT for service-to-service auth?” — mTLS authenticates the calling service via its cert (strong, infrastructure-level). JWT is application-level signed tokens — simpler to manage but token theft is a risk. Service meshes do mTLS automatically; greenfield services often use JWT.
“How do you propagate request-scoped state in Python gRPC?” — contextvars.ContextVar. Set in an interceptor; readable across the call chain (works with asyncio and threadpool).
“How does OpenTelemetry integrate with gRPC?” — opentelemetry-instrumentation-grpc automatically adds spans on every RPC, propagates trace IDs via metadata (traceparent header per W3C Trace Context).