Interceptors

Beta software

Connect for Python is in beta. Join us on Slack if you have questions or feedback.

Interceptors are similar to the middleware or decorators you may be familiar with from other frameworks: they’re the primary way of extending Connect. They can modify the context, the request, the response, and any errors. Interceptors are often used to add logging, metrics, tracing, retries, and other functionality.

Take care when writing interceptors! They’re powerful, but overly complex interceptors can make debugging difficult.

Interceptors are protocol implementations

Connect interceptors are protocol implementations with the same signature as an RPC handler, along with a call_next Callable to continue with request processing. This allows writing interceptors in much the same way as any handler, making sure to call call_next when needing to call business logic - or not, if overriding the response within the interceptor itself.

Connect supports unary RPC and three stream types - because each has a different handler signature, we provide protocols corresponding to each.

ASGI

class UnaryInterceptor(Protocol):

    async def intercept_unary(
        self,
        call_next: Callable[[REQ, RequestContext], Awaitable[RES]],
        request: REQ,
        ctx: RequestContext,
    ) -> RES: ...

class ClientStreamInterceptor(Protocol):

    async def intercept_client_stream(
        self,
        call_next: Callable[[AsyncIterator[REQ], RequestContext], Awaitable[RES]],
        request: AsyncIterator[REQ],
        ctx: RequestContext,
    ) -> RES: ...

class ServerStreamInterceptor(Protocol):

    def intercept_server_stream(
        self,
        call_next: Callable[[REQ, RequestContext], AsyncIterator[RES]],
        request: REQ,
        ctx: RequestContext,
    ) -> AsyncIterator[RES]: ...

class BidiStreamInterceptor(Protocol):

    def intercept_bidi_stream(
        self,
        call_next: Callable[[AsyncIterator[REQ], RequestContext], AsyncIterator[RES]],
        request: AsyncIterator[REQ],
        ctx: RequestContext,
    ) -> AsyncIterator[RES]: ...

WSGI

class UnaryInterceptorSync(Protocol):

    def intercept_unary_sync(
        self,
        call_next: Callable[[REQ, RequestContext], RES],
        request: REQ,
        ctx: RequestContext,
    ) -> RES:

class ClientStreamInterceptorSync(Protocol):

    def intercept_client_stream_sync(
        self,
        call_next: Callable[[Iterator[REQ], RequestContext], RES],
        request: Iterator[REQ],
        ctx: RequestContext,
    ) -> RES:

class ServerStreamInterceptorSync(Protocol):

    def intercept_server_stream_sync(
        self,
        call_next: Callable[[REQ, RequestContext], Iterator[RES]],
        request: REQ,
        ctx: RequestContext,
    ) -> Iterator[RES]:

class BidiStreamInterceptorSync(Protocol):

    def intercept_bidi_stream_sync(
        self,
        call_next: Callable[[Iterator[REQ], RequestContext], Iterator[RES]],
        request: Iterator[REQ],
        ctx: RequestContext,
    ) -> Iterator[RES]:

A single class can implement as many of the protocols as needed.

An example

That’s a little abstract, so let’s consider an example: we’d like to apply a filter to our greeting service from the getting started documentation that says “Goodbye” instead of “Hello” to certain callers.

ASGI

from collections.abc import Awaitable, Callable

class GoodbyeInterceptor:
    def __init__(self, users: list[str]) -> None:
        self._users = users

    async def intercept_unary(
        self,
        call_next: Callable[[GreetRequest, RequestContext[GreetRequest, GreetResponse]], Awaitable[GreetResponse]],
        request: GreetRequest,
        ctx: RequestContext[GreetRequest, GreetResponse],
    ) -> GreetResponse:
        if request.name in self._users:
            return GreetResponse(greeting=f"Goodbye, {request.name}!")
        return await call_next(request, ctx)

WSGI

from collections.abc import Awaitable, Callable

class GoodbyeInterceptor:
    def __init__(self, users: list[str]) -> None:
        self._users = users

    def intercept_unary_sync(
        self,
        call_next: Callable[[GreetRequest, RequestContext[GreetRequest, GreetResponse]], Awaitable[GreetResponse]],
        request: GreetRequest,
        ctx: RequestContext[GreetRequest, GreetResponse],
    ) -> GreetResponse:
        if request.name in self._users:
            return GreetResponse(greeting=f"Goodbye, {request.name}!")
        return call_next(request, ctx)

To apply our new interceptor to handlers, we can pass it to the application with interceptors=.

ASGI

app = GreetServiceASGIApplication(Greeter(), interceptors=[GoodbyeInterceptor(["user1", "user2"])])

WSGI

app = GreetServiceWSGIApplication(Greeter(), interceptors=[GoodbyeInterceptor(["user1", "user2"])])

Client constructors also accept an interceptors= parameter.

Async

client = GreetServiceClient("http://localhost:8000", interceptors=[GoodbyeInterceptor(["user1", "user2"])])

Sync

client = GreetServiceClientSync("http://localhost:8000", interceptors=[GoodbyeInterceptor(["user1", "user2"])])

Metadata interceptors

Because the signature is different for each RPC type, we have an interceptor protocol for each to be able to intercept RPC messages. However, many interceptors, such as for metrics or tracing, only need access to headers and not messages. Connect provides a metadata interceptor protocol that can be implemented to work with any RPC type.

An interceptor timing each RPC and logging its duration may look like this:

ASGI

import logging
import time

logger = logging.getLogger(__name__)

class TimingInterceptor:
    async def on_start[REQ, RES](self, ctx: RequestContext[REQ, RES]) -> float:
        return time.perf_counter()

    async def on_end[REQ, RES](self, start: float, ctx: RequestContext[REQ, RES], error: Exception | None) -> None:
        duration = time.perf_counter() - start
        logger.info("%s took %.3fs", ctx.method.name, duration)

WSGI

import logging
import time

logger = logging.getLogger(__name__)

class TimingInterceptor:
    def on_start_sync[REQ, RES](self, ctx: RequestContext[REQ, RES]) -> float:
        return time.perf_counter()

    def on_end_sync[REQ, RES](self, start: float, ctx: RequestContext[REQ, RES], error: Exception | None) -> None:
        duration = time.perf_counter() - start
        logger.info("%s took %.3fs", ctx.method().name, duration)

on_start can return any value, which is passed to the optional on_end method. Here, we return the start time to compute the RPC’s duration.

Authentication

Don’t use interceptors to authenticate requests on the server. Handlers run unary interceptors after the request message has been read, decompressed, and deserialized. An interceptor-based check lets unauthenticated clients consume memory and CPU on your server. Instead, authenticate with standard ASGI or WSGI middleware, which runs before Connect reads the request body. Starlette’s AuthenticationMiddleware provides authentication middleware that works with any ASGI application, including Connect servers.