Server-Side gRPC Interceptors for Cross-Cutting Concerns

Welcome back! In our previous lesson, you mastered robust error handling by using standard gRPC status codes and the rich error model. This allowed you to communicate precisely what went wrong inside a specific RPC.

However, as you build more complex services, you'll encounter logic that doesn't belong to any single RPC but applies to all of them—concerns like logging every request, measuring performance, or checking for an authentication token. Placing this code in every method is inefficient and violates the DRY (Don't Repeat Yourself) principle.

This lesson addresses that exact problem. You will learn how to implement a gRPC interceptor on the server. Think of interceptors as middleware for your gRPC services, a centralized layer that processes every incoming request before it reaches your business logic. We'll use them to add powerful cross-cutting features like logging and authentication in a clean, maintainable way.

1. What are gRPC Interceptors?

A gRPC interceptor is a function that "intercepts" an RPC on the server (or client) side, allowing you to run code before and after the actual RPC handler is executed. They are essential for implementing cross-cutting concerns.

What is gRPC Interceptor? — This diagram illustrates the concept of interceptors. On both the client and server side, a request can pass through a chain of interceptors, each handling a specific task like logging or authentication, before the main application logic is invoked.

Common use cases for server-side interceptors include:

Authentication & Authorization: Validating credentials (like a JWT) sent with each request.
Logging: Recording details about every incoming call, its parameters, and its outcome.
Metrics & Tracing: Measuring RPC latency, counting errors, and propagating trace IDs for distributed tracing.
Request Validation: Performing initial validation on incoming requests before they hit the service logic.

Interceptors are chained together, forming a processing pipeline that every request flows through. This is analogous to how middleware functions work in web frameworks you're familiar with, like Django or FastAPI.

2. Implementing an Async Interceptor for Logging

While gRPC's core Python library provides the hooks for interceptors, a community library, grpc-interceptor, makes the process much more straightforward, especially for async servers.

First, let's install it:

pip install grpc-interceptor

To create an interceptor, you subclass AsyncServerInterceptor and override the intercept method. Let's build one for logging.

Reference — grpc-interceptor documentation

This is the documentation for the grpc-interceptor library. We will focus on the AsyncServerInterceptor base class.

Read the documentation for the AsyncServerInterceptor class. Pay close attention to the signature of the intercept method and the description of its parameters: method, request_or_iterator, context, and method_name. This is the core method you will be overriding.

Now, let's create a new file myservice/interceptors.py and add our first interceptor.




# myservice/interceptors.py

import time
import logging
import grpc
from grpc_interceptor import AsyncServerInterceptor

class LoggingInterceptor(AsyncServerInterceptor):
    async def intercept(
        self,
        method,
        request_or_iterator,
        context: grpc.aio.ServicerContext,
        method_name: str,
    ):
        """
        Intercepts an RPC to log its execution time and outcome.
        """
        logging.info(f"START - Received call to '{method_name}'")
        start_time = time.perf_counter()

        try:



            # This is the crucial part: we call the next handler in the chain,
            # which could be another interceptor or the final RPC handler.
            response = await method(request_or_iterator, context)
            



            # If the call succeeds, log the duration.
            end_time = time.perf_counter()
            duration = end_time - start_time
            logging.info(
                f"SUCCESS - Call to '{method_name}' completed in {duration:.4f}s"
            )
            return response

        except grpc.RpcError as e:



            # If an error occurs (either in a downstream interceptor or the RPC handler),
            # log the error and re-raise it.
            end_time = time.perf_counter()
            duration = end_time - start_time
            logging.error(
                f"FAILURE - Call to '{method_name}' failed after {duration:.4f}s with status {e.code()}"
            )
            raise

To use this interceptor, we modify our server startup logic in myservice/server.py.




# myservice/server.py
import grpc
import asyncio
from concurrent import futures



# ... other imports




# Import the new interceptor
from myservice.interceptors import LoggingInterceptor




# ... UserService class definition

async def serve():



    # 1. Instantiate the interceptor
    logging_interceptor = LoggingInterceptor()
    



    # 2. Pass it to the server constructor
    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[logging_interceptor],  # Pass as a list
    )

    user_service_pb2_grpc.add_UserServiceServicer_to_server(UserService(), server)
    
    listen_addr = "[::]:50051"
    server.add_insecure_port(listen_addr)
    logging.info(f"Starting server on {listen_addr}")
    await server.start()
    await server.wait_for_termination()




# ... main block

Now, restart your server and run any client from the previous lesson. You'll see new, detailed log messages for every RPC call, printed by your interceptor, proving it's successfully intercepting all traffic.

3. Implementing an Authentication Interceptor

Now for the main event: implementing an interceptor for authentication. The goal is to protect our RPCs by requiring a valid token in the request metadata.

The flow will be:

The client includes an authorization header in the metadata (e.g., 'authorization': 'Bearer my-secret-token').
Our AuthInterceptor will check for this header on every incoming call.
It will validate the token. If it's missing or invalid, the interceptor will reject the call with an UNAUTHENTICATED status code before it ever reaches our service logic.
If the token is valid, the call proceeds.

The resource below provides a great Python example of a JWT-based interceptor, which we will adapt for our needs.

How to Implement JWT Authentication in gRPC

This article from OneUptime provides a practical walkthrough of implementing JWT authentication with a Python gRPC interceptor.

Focus on the "Python JWT Interceptor" section. Study the AuthInterceptor class. Note how it inherits from grpc.ServerInterceptor (we will use AsyncServerInterceptor instead), how it checks for public methods, and how it extracts the token from the metadata. Pay special attention to the _unauthenticated_response method, which is how it aborts the call. We will use a slightly different but related technique.

A clean way to reject a call from an interceptor when using the grpc-interceptor library is to raise a special exception that corresponds to a gRPC status code. Let's see this in action.

1. Create the AuthInterceptor

First, add the special exception to your imports in myservice/interceptors.py. Then, define the AuthInterceptor.




# myservice/interceptors.py
# ... other imports




# Import the exception we'll raise for auth failures
from grpc_interceptor.exceptions import Unauthenticated

class AuthInterceptor(AsyncServerInterceptor):
    def __init__(self, public_methods: list[str]):
        """
        Initializes the interceptor with a list of public methods
        that do not require authentication.
        """
        self._public_methods = public_methods

    async def intercept(
        self,
        method,
        request_or_iterator,
        context: grpc.aio.ServicerContext,
        method_name: str,
    ):



        # If the method is public, skip authentication and proceed.
        if method_name in self._public_methods:
            return await method(request_or_iterator, context)

        metadata = context.invocation_metadata()
        auth_header = None
        for key, value in metadata:
            if key == "authorization":
                auth_header = value
                break

        if not auth_header or not auth_header.startswith("Bearer "):
            raise Unauthenticated("Missing or invalid authorization token.")




        # In a real application, you would use a library like PyJWT to
        # decode and verify the token. For this example, we'll just
        # check for a hardcoded secret.
        token = auth_header.split(" ")[1]
        if token != "my-secret-token":
            raise Unauthenticated("Invalid token.")




        # Token is valid, proceed with the call.
        logging.info(f"Authenticated call to '{method_name}'")
        return await method(request_or_iterator, context)

2. Handling the Exception

If we raise the Unauthenticated exception, something needs to catch it and translate it into the correct grpc.StatusCode.UNAUTHENTICATED. The grpc-interceptor library provides another interceptor for exactly this purpose: AsyncExceptionToStatusInterceptor.

Reference — grpc-interceptor documentation

Revisit the library's documentation to understand this crucial helper.

Read the documentation for AsyncExceptionToStatusInterceptor. Note that its purpose is to catch exceptions (like the ones we plan to raise) and correctly set the RPC status. This saves us from writing boilerplate error-handling logic.

3. Chaining the Interceptors

Now, we update myservice/server.py to chain all our interceptors together. The order is important! The exception handler should come first so it can wrap all subsequent interceptors and the RPC handler itself.




# myservice/server.py
# ... other imports




# Import all our interceptors
from myservice.interceptors import LoggingInterceptor, AuthInterceptor
from grpc_interceptor import AsyncServerInterceptor
from grpc_interceptor.exceptions import GrpcException
from grpc_interceptor.server import AsyncExceptionToStatusInterceptor


async def serve():



    # Define which methods do not require auth.
    # We don't have a login method yet, so for now we'll protect everything.
    public_methods = []
    



    # Instantiate all interceptors
    exception_handler = AsyncExceptionToStatusInterceptor()
    logging_interceptor = LoggingInterceptor()
    auth_interceptor = AuthInterceptor(public_methods=public_methods)
    



    # Chain them in the desired order of execution.
    # Requests will flow: exception_handler -> auth_interceptor -> logging_interceptor -> RPC
    interceptors = [
        exception_handler,
        auth_interceptor, 
        logging_interceptor
    ]
    
    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=interceptors,
    )



    # ... rest of the function

4. Updating the Client to Authenticate

Our server is now secure. Any client call without a valid token will be rejected. Let's update myservice/concurrent_client.py to send the token. You can pass metadata as a keyword argument to the stub method.




# myservice/concurrent_client.py
# ... imports

async def run_single_call_with_auth(stub, user_id, token=None):
    """Makes a single GetUser call, optionally with an auth token."""
    logging.info(f"\n--- Calling GetUser for '{user_id}' ---")
    
    metadata = []
    if token:
        logging.info(f"Sending token: {token[:10]}...")
        metadata.append(("authorization", f"Bearer {token}"))
    else:
        logging.info("Sending no token.")

    try:
        request = user_service_pb2.GetUserRequest(user_id=user_id)



        # Pass metadata to the stub call
        response = await stub.GetUser(request, metadata=metadata)
        logging.info(f"SUCCESS: Got user: {response.user.name}")
    except grpc.aio.AioRpcError as e:
        logging.error(f"FAILURE: RPC failed with status '{e.code()}': {e.details()}")

async def main():
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        stub = user_service_pb2_grpc.UserServiceStub(channel)




        # Test case 1: No token (should fail)
        await run_single_call_with_auth(stub, "user-1")




        # Test case 2: Invalid token (should fail)
        await run_single_call_with_auth(stub, "user-1", token="wrong-token")




        # Test case 3: Valid token (should succeed)
        await run_single_call_with_auth(stub, "user-1", token="my-secret-token")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

Restart your server, then run the updated client. The output will clearly demonstrate the AuthInterceptor at work:




# Client Output
--- Calling GetUser for 'user-1' ---
Sending no token.
FAILURE: RPC failed with status 'StatusCode.UNAUTHENTICATED': Missing or invalid authorization token.

--- Calling GetUser for 'user-1' ---
Sending token: wrong-toke...
FAILURE: RPC failed with status 'StatusCode.UNAUTHENTICATED': Invalid token.

--- Calling GetUser for 'user-1' ---
Sending token: my-secret-...
SUCCESS: Got user: Alice




# Server Log Output for the successful call
INFO:root:Authenticated call to '/user_service.UserService/GetUser'
INFO:root:START - Received call to '/user_service.UserService/GetUser'
INFO:root:SERVER received GetUser request for id: user-1
INFO:root:SUCCESS - Call to '/user_service.UserService/GetUser' completed in 0.1005s

Conclusion

In this lesson, you leveraged gRPC interceptors to cleanly separate cross-cutting concerns from your core service logic. This is a fundamental pattern for building maintainable and production-ready microservices.

Key Takeaways:

Interceptors for Cross-Cutting Concerns: Interceptors act as middleware, perfect for handling logic like logging, metrics, and authentication that applies to all RPCs.
Async Implementation: The grpc-interceptor library simplifies creating async interceptors by subclassing AsyncServerInterceptor.
Centralized Authentication: You built an AuthInterceptor that secures RPC methods by validating a token from the request metadata.
Exception-Based Flow Control: Raising specific exceptions from an interceptor (like Unauthenticated) and using AsyncExceptionToStatusInterceptor to catch them is a clean pattern for error handling.
Interceptor Chaining: You can create a pipeline of interceptors, and their order of execution matters.

Preview of the Next Lesson

So far, our clients have communicated directly with our gRPC service. In a real-world microservices architecture, this is rarely the case. Clients typically interact with a single, unified entry point called an API Gateway.

In the next module, you will learn to implement high-concurrency API endpoints using FastAPI and build a gateway that translates incoming REST requests into gRPC calls for our backend services. This is where the public_methods concept in our AuthInterceptor will become critical, as the gateway will handle user-facing authentication, while internal services rely on service-to-service security.