open-feature · tuhinkanti · Jul 22, 2026 · Aug 17, 2026 · Sep 3, 2026 · Sep 4, 2026
@@ -92,6 +92,10 @@ The default options can be defined in the FlagdProvider constructor.
 | max_cache_size           | FLAGD_MAX_CACHE_SIZE           | int                        | 1000                          | rpc                 |
 | retry_backoff_ms         | FLAGD_RETRY_BACKOFF_MS         | int                        | 1000                          | rpc                 |
 | offline_flag_source_path | FLAGD_OFFLINE_FLAG_SOURCE_PATH | str                        | null                          | in-process          |
+| sync_metadata_disabled   | -                              | bool                       | null                          | in-process          |
+| fatal_status_codes       | FLAGD_FATAL_STATUS_CODES       | sequence of gRPC status code names | empty                 | rpc & in-process    |
+| channel_credentials      | -                              | `grpc.ChannelCredentials` (including mTLS) | null                | rpc & in-process    |
+| client_interceptors      | -                              | sequence of gRPC client interceptors | null                | rpc & in-process    |
 
 > [!NOTE]
 > The `selector` configuration is only used in **in-process** mode for filtering flag configurations. See [Selector Handling](#selector-handling-in-process-mode-only) for migration guidance.
@@ -106,6 +110,67 @@ The default options can be defined in the FlagdProvider constructor.
 > [!NOTE]
 > Some configurations are only applicable for RPC resolver.
 
+### Mutual TLS
+
+Pass custom `grpc.ChannelCredentials` to `channel_credentials` when the flagd server requires mutual TLS (mTLS). The provider uses these credentials for both resolver types and gives them precedence over `tls` and `cert_path`.
+
+```python
+import grpc
+
+from openfeature.contrib.provider.flagd import FlagdProvider
+
+credentials = grpc.ssl_channel_credentials(
+    root_certificates=ca_certificate,
+    private_key=client_private_key,
+    certificate_chain=client_certificate,
+)
+
+provider = FlagdProvider(channel_credentials=credentials)
+```
+
+### Custom gRPC interceptors
+
+`client_interceptors` are synchronous gRPC client interceptors applied to the channel in the order provided. Use them for infrastructure concerns such as custom headers or credentials. Flagd-specific options like `selector` stay first-class and do not need a custom interceptor.
+
+Metadata keys added by an interceptor must be valid lowercase gRPC metadata keys. If an interceptor adds `flagd-selector` while `selector` is set, the request contains duplicate keys. gRPC permits duplicate metadata keys.
+
+`grpc.aio` interceptors are not supported. Passing an object that does not implement one of the synchronous client interceptor interfaces raises `TypeError` when the provider creates its channel.
+Exceptions raised while opening a sync or event stream are logged and retried after `retry_backoff_max_ms`; a persistently failing interceptor prevents stream updates.
+
+```python
+import grpc
+from openfeature.contrib.provider.flagd import FlagdProvider
+from openfeature.contrib.provider.flagd.config import ResolverType
+
+
+class _ClientCallDetails(grpc.ClientCallDetails):
+    def __init__(self, details, metadata):
+        self.method = details.method
+        self.timeout = details.timeout
+        self.metadata = metadata
+        self.credentials = details.credentials
+        self.wait_for_ready = details.wait_for_ready
+        self.compression = details.compression
+
+
+class DisableEnvoyTimeout(grpc.UnaryStreamClientInterceptor):
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        metadata = list(client_call_details.metadata or [])
+        metadata.append(("x-envoy-upstream-rq-timeout-ms", "0"))
+        details = _ClientCallDetails(client_call_details, metadata)
+        return continuation(details, request)
+
+
+provider = FlagdProvider(
+    resolver_type=ResolverType.IN_PROCESS,
+    client_interceptors=[DisableEnvoyTimeout()],
+)
+```
+
+See also:
+- https://grpc.github.io/grpc/python/grpc.html#client-side-interceptor
+- https://grpc.github.io/grpc/python/grpc.html#grpc.intercept_channel
+
 ### Selector Handling (In-Process Mode Only)
 
 > [!IMPORTANT]

@@ -57,6 +57,22 @@ class CacheType(Enum):
 
 T = typing.TypeVar("T")
 
+ClientInterceptor: typing.TypeAlias = (
+    grpc.UnaryUnaryClientInterceptor
+    | grpc.UnaryStreamClientInterceptor
+    | grpc.StreamUnaryClientInterceptor
+    | grpc.StreamStreamClientInterceptor
+)
+
+
+def apply_client_interceptors(
+    channel: grpc.Channel,
+    client_interceptors: typing.Sequence[ClientInterceptor],
+) -> grpc.Channel:
+    if not client_interceptors:
+        return channel
+    return grpc.intercept_channel(channel, *client_interceptors)
+
 
 def str_to_bool(val: str) -> bool:
     return val.lower() == "true"
@@ -105,6 +121,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         channel_credentials: grpc.ChannelCredentials | None = None,
         sync_metadata_disabled: bool | None = None,
         fatal_status_codes: list[str] | None = None,
+        client_interceptors: typing.Sequence[ClientInterceptor] | None = None,
     ):
         self.host = env_or_default(ENV_VAR_HOST, DEFAULT_HOST) if host is None else host
 
@@ -278,3 +295,10 @@ def __init__(  # noqa: PLR0913, PLR0915
         # Disabling will prevent static context from flagd being used in evaluations.
         # GetMetadata and this option will be removed.
         self.sync_metadata_disabled = sync_metadata_disabled
+
+        # gRPC client interceptors applied to the channel (rpc and in-process).
+        # Use this for infrastructure concerns such as custom headers or
+        # credentials; flagd-specific options (e.g. selector) stay first-class.
+        self.client_interceptors: tuple[ClientInterceptor, ...] = (
+            tuple(client_interceptors) if client_interceptors is not None else ()
+        )
@@ -33,7 +33,7 @@
 from openfeature.provider import AbstractProvider
 from openfeature.provider.metadata import Metadata
 
-from .config import CacheType, Config, ResolverType
+from .config import CacheType, ClientInterceptor, Config, ResolverType
 from .resolvers import AbstractResolver, GrpcResolver, InProcessResolver
 from .sync_metadata_hook import SyncMetadataHook
 
@@ -66,6 +66,7 @@ def __init__(  # noqa: PLR0913
         channel_credentials: grpc.ChannelCredentials | None = None,
         sync_metadata_disabled: bool | None = None,
         fatal_status_codes: list[str] | None = None,
+        client_interceptors: typing.Sequence[ClientInterceptor] | None = None,
     ):
         """
         Create an instance of the FlagdProvider
@@ -83,6 +84,8 @@ def __init__(  # noqa: PLR0913
         :param stream_deadline_ms: the maximum time to wait before a request times out
         :param keep_alive_time: the number of milliseconds to keep alive
         :param resolver_type: the type of resolver to use
+        :param channel_credentials: custom gRPC channel credentials, including mTLS credentials
+        :param client_interceptors: gRPC client interceptors applied to the channel. Metadata keys added by interceptors must be valid lowercase gRPC metadata keys. An interceptor that adds ``flagd-selector`` can duplicate the provider's selector metadata.
         """
         if deadline_ms is None and timeout is not None:
             deadline_ms = timeout * 1000
@@ -113,6 +116,7 @@ def __init__(  # noqa: PLR0913
             channel_credentials=channel_credentials,
             sync_metadata_disabled=sync_metadata_disabled,
             fatal_status_codes=fatal_status_codes,
+            client_interceptors=client_interceptors,
         )
         self.enriched_context: dict = {}
 

@@ -28,7 +28,7 @@
     evaluation_pb2_grpc,
 )
 
-from ..config import CacheType, Config
+from ..config import CacheType, Config, apply_client_interceptors
 from ..flag_type import FlagType
 from .types import GrpcMultiCallableArgs
 
@@ -117,7 +117,14 @@ def _generate_channel(self, config: Config) -> grpc.Channel:
                 ),
             ),
         ]
-        if config.tls:
+        if config.channel_credentials is not None:
+            channel = grpc.secure_channel(
+                target,
+                credentials=config.channel_credentials,
+                options=options,
+            )
+
+        elif config.tls:
             credentials = grpc.ssl_channel_credentials()
             if config.cert_path:
                 with open(config.cert_path, "rb") as f:
@@ -135,7 +142,7 @@ def _generate_channel(self, config: Config) -> grpc.Channel:
                 options=options,
             )
 
-        return channel
+        return apply_client_interceptors(channel, config.client_interceptors)
 
     def initialize(self, evaluation_context: EvaluationContext) -> None:
         self.connect()
@@ -289,6 +296,11 @@ def listen(self) -> None:
                 logger.exception(
                     f"Could not parse flag data using flagd syntax: {message=}"
                 )
+            except Exception:
+                if self.active:
+                    logger.exception("Unexpected EventStream error, reconnecting")
+                else:
+                    logger.debug("EventStream ended during shutdown", exc_info=True)
             if self.active:
                 self._wait_before_reconnect()
 

@@ -21,7 +21,7 @@
     sync_pb2_grpc,
 )
 
-from ....config import Config
+from ....config import Config, apply_client_interceptors
 from ...types import GrpcMultiCallableArgs
 from ..connector import FlagStateConnector
 from ..flags import FlagStore
@@ -122,7 +122,7 @@ def _generate_channel(self, config: Config) -> grpc.Channel:
                 options=options,
             )
 
-        return channel
+        return apply_client_interceptors(channel, config.client_interceptors)
 
     def initialize(self, context: EvaluationContext) -> None:
         self.connect()
@@ -293,27 +293,37 @@ def _handle_rpc_error(self, e: grpc.RpcError) -> bool:
     def _wait_before_reconnect(self) -> None:
         self._shutdown_event.wait(self.retry_backoff_max_seconds)
 
+    def _listen_once(
+        self, call_args: GrpcMultiCallableArgs, request_args: dict
+    ) -> bool:
+        try:
+            context_values_response = self._fetch_metadata()
+            request = sync_pb2.SyncFlagsRequest(**request_args)
+            logger.debug("Setting up gRPC sync flags connection")
+            for flag_rsp in self.stub.SyncFlags(request, **call_args):
+                if self._handle_flag_response(flag_rsp, context_values_response):
+                    return True
+        except grpc.RpcError as e:
+            if self._handle_rpc_error(e):
+                return True
+        except json.JSONDecodeError:
+            logger.exception("Could not parse JSON flag data from SyncFlags endpoint")
+        except ParseError:
+            logger.exception("Could not parse flag data using flagd syntax")
+        except Exception:
+            if self.active:
+                logger.exception("Unexpected SyncFlags stream error, reconnecting")
+            else:
+                logger.debug("SyncFlags stream ended during shutdown", exc_info=True)
+        return False
+
     def listen(self) -> None:
         call_args = self.generate_grpc_call_args()
         request_args = self._create_request_args()
 
         while self.active:
-            try:
-                context_values_response = self._fetch_metadata()
-                request = sync_pb2.SyncFlagsRequest(**request_args)
-                logger.debug("Setting up gRPC sync flags connection")
-                for flag_rsp in self.stub.SyncFlags(request, **call_args):
-                    if self._handle_flag_response(flag_rsp, context_values_response):
-                        return
-            except grpc.RpcError as e:
-                if self._handle_rpc_error(e):
-                    return
-            except json.JSONDecodeError:
-                logger.exception(
-                    "Could not parse JSON flag data from SyncFlags endpoint"
-                )
-            except ParseError:
-                logger.exception("Could not parse flag data using flagd syntax")
+            if self._listen_once(call_args, request_args):
+                return
             if self.active:
                 self._wait_before_reconnect()
 

@@ -1,3 +1,6 @@
+from unittest.mock import Mock
+
+import grpc
 import pytest
 
 # not sure if we still need this test, as this is also covered with gherkin tests.
@@ -44,6 +47,22 @@ def test_return_default_values_rpc():
     assert config.retry_backoff_ms == DEFAULT_RETRY_BACKOFF
     assert config.stream_deadline_ms == DEFAULT_STREAM_DEADLINE
     assert config.tls is DEFAULT_TLS
+    assert config.client_interceptors == ()
+
+
+def test_client_interceptors_passthrough():
+    interceptor = Mock(spec=grpc.UnaryUnaryClientInterceptor)
+    config = Config(resolver=ResolverType.IN_PROCESS, client_interceptors=[interceptor])
+    assert config.client_interceptors == (interceptor,)
+
+
+def test_positional_fatal_status_codes_backwards_compatible():
+    # fatal_status_codes stays ahead of client_interceptors so callers that
+    # passed it positionally keep working. It is the 22nd positional parameter.
+    leading_args = [None] * 21
+    config = Config(*leading_args, ["UNAVAILABLE", "DATA_LOSS"])
+    assert config.fatal_status_codes == ["UNAVAILABLE", "DATA_LOSS"]
+    assert config.client_interceptors == ()
 
 
 def test_return_default_values_in_process():