Skip to content

datarobot_genai.drmcputils.feature_flags

feature_flags

DataRobot entitlements-backed feature flag evaluation.

Lives in drtools so consumers (tools, global-mcp, any drtools-only package) can evaluate per-user entitlements without depending on the heavier drmcp/drmcpbase packages or their module-load side effects.

This evaluator is per-user: it requires a request-scoped :class:~datarobot.rest.RESTClientObject (e.g. from :func:datarobot_genai.drmcputils.clients.datarobot.request_user_dr_client) and caches results per (flag, principal). It is the building block for per-user, live tool gating (e.g. hiding the sandbox tool unless an entitlement is set).

It is intentionally separate from :mod:datarobot_genai.drmcp.core.feature_flags, which evaluates the application-static MCP-container account against drmcpbase for dynamic tool/prompt registration. The two may be consolidated later.

FeatureFlag dataclass

Source code in datarobot_genai/drmcputils/feature_flags.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
@dataclass
class FeatureFlag:
    name: str
    enabled: bool

    @classmethod
    def create(
        cls,
        feature_flag_name: str,
        *,
        client: RESTClientObject,
    ) -> FeatureFlag:
        """Evaluate a DR entitlement against the principal owning ``client``.

        ``client`` is required at this layer. If you need the historical
        "use the application-static client by default" behavior, use
        ``datarobot_genai.drmcp.core.feature_flags.FeatureFlag`` instead.
        """
        response = client.post(
            "entitlements/evaluate/",
            json={"entitlements": [{"name": feature_flag_name}]},
        )
        feature_flag_info = response.json()["entitlements"][0]
        return cls(
            name=feature_flag_info["name"],
            enabled=bool(feature_flag_info["value"]),
        )

    @classmethod
    def is_enabled(
        cls,
        feature_flag_name: str,
        *,
        client: RESTClientObject,
        ttl_seconds: float = _DEFAULT_TTL_SECONDS,
    ) -> bool:
        """Return the cached entitlement state, keyed by ``(flag, principal)``.

        Pass ``ttl_seconds=0`` to bypass the cache. The principal portion of
        the key is a hash of the client's token, so different users do not
        share cache entries.
        """
        cache_key = (feature_flag_name, _principal_key(client))
        now = time.monotonic()
        if ttl_seconds > 0:
            cached = _eval_cache.get(cache_key)
            if cached is not None and (now - cached[0]) < ttl_seconds:
                return cached[1]
        enabled = cls.create(feature_flag_name, client=client).enabled
        if ttl_seconds > 0:
            _prune_expired(now, ttl_seconds)
            _eval_cache[cache_key] = (now, enabled)
        return enabled

    @classmethod
    def is_mcp_tools_gallery_support_enabled(cls, *, client: RESTClientObject) -> bool:
        return cls.is_enabled("ENABLE_MCP_TOOLS_GALLERY_SUPPORT", client=client)

create classmethod 

create(feature_flag_name: str, *, client: RESTClientObject) -> FeatureFlag

Evaluate a DR entitlement against the principal owning client.

client is required at this layer. If you need the historical "use the application-static client by default" behavior, use datarobot_genai.drmcp.core.feature_flags.FeatureFlag instead.

Source code in datarobot_genai/drmcputils/feature_flags.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@classmethod
def create(
    cls,
    feature_flag_name: str,
    *,
    client: RESTClientObject,
) -> FeatureFlag:
    """Evaluate a DR entitlement against the principal owning ``client``.

    ``client`` is required at this layer. If you need the historical
    "use the application-static client by default" behavior, use
    ``datarobot_genai.drmcp.core.feature_flags.FeatureFlag`` instead.
    """
    response = client.post(
        "entitlements/evaluate/",
        json={"entitlements": [{"name": feature_flag_name}]},
    )
    feature_flag_info = response.json()["entitlements"][0]
    return cls(
        name=feature_flag_info["name"],
        enabled=bool(feature_flag_info["value"]),
    )

is_enabled classmethod 

is_enabled(feature_flag_name: str, *, client: RESTClientObject, ttl_seconds: float = _DEFAULT_TTL_SECONDS) -> bool

Return the cached entitlement state, keyed by (flag, principal).

Pass ttl_seconds=0 to bypass the cache. The principal portion of the key is a hash of the client's token, so different users do not share cache entries.

Source code in datarobot_genai/drmcputils/feature_flags.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@classmethod
def is_enabled(
    cls,
    feature_flag_name: str,
    *,
    client: RESTClientObject,
    ttl_seconds: float = _DEFAULT_TTL_SECONDS,
) -> bool:
    """Return the cached entitlement state, keyed by ``(flag, principal)``.

    Pass ``ttl_seconds=0`` to bypass the cache. The principal portion of
    the key is a hash of the client's token, so different users do not
    share cache entries.
    """
    cache_key = (feature_flag_name, _principal_key(client))
    now = time.monotonic()
    if ttl_seconds > 0:
        cached = _eval_cache.get(cache_key)
        if cached is not None and (now - cached[0]) < ttl_seconds:
            return cached[1]
    enabled = cls.create(feature_flag_name, client=client).enabled
    if ttl_seconds > 0:
        _prune_expired(now, ttl_seconds)
        _eval_cache[cache_key] = (now, enabled)
    return enabled

is_tool_feature_enabled 

is_tool_feature_enabled(feature_flag_name: str | None, *, evaluator: Callable[[str], bool]) -> bool

Decide whether a feature-flag-gated tool should be exposed.

Shared gating policy for every MCP tool registry: drmcp's static-account registry and global-mcp's per-user registry both call this, so the "no flag → expose; flag set → evaluate; lookup error → fail closed" decision lives once in the drtools layer instead of being duplicated per server.

evaluator performs the actual entitlement check for one flag name. Only the calling registry knows which principal/client to evaluate against (the static container account for drmcp, the requesting user for global-mcp), so it supplies that via the closure rather than this layer reaching for a client it cannot choose correctly.

Parameters:

Name Type Description Default
feature_flag_name str | None

The entitlement gating the tool, or None if ungated.

 required
evaluator Callable[[str], bool]

Callable returning whether feature_flag_name is enabled.

 required
Returns
``True`` if the tool should be registered. ``False`` if the flag is
disabled or its evaluation raised (fail-closed, so a lookup failure
never accidentally exposes a gated tool).
Source code in datarobot_genai/drmcputils/feature_flags.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def is_tool_feature_enabled(
    feature_flag_name: str | None,
    *,
    evaluator: Callable[[str], bool],
) -> bool:
    """Decide whether a feature-flag-gated tool should be exposed.

    Shared gating policy for every MCP tool registry: ``drmcp``'s static-account
    registry and ``global-mcp``'s per-user registry both call this, so the
    "no flag → expose; flag set → evaluate; lookup error → fail closed" decision
    lives once in the drtools layer instead of being duplicated per server.

    ``evaluator`` performs the actual entitlement check for one flag name. Only
    the calling registry knows which principal/client to evaluate against (the
    static container account for ``drmcp``, the requesting user for global-mcp),
    so it supplies that via the closure rather than this layer reaching for a
    client it cannot choose correctly.

    Args:
        feature_flag_name: The entitlement gating the tool, or ``None`` if ungated.
        evaluator: Callable returning whether ``feature_flag_name`` is enabled.

    Returns
    -------
        ``True`` if the tool should be registered. ``False`` if the flag is
        disabled or its evaluation raised (fail-closed, so a lookup failure
        never accidentally exposes a gated tool).
    """
    if feature_flag_name is None:
        return True
    try:
        return evaluator(feature_flag_name)
    except Exception:
        logger.debug(
            "feature flag %s evaluation failed; gating tool off (fail-closed)",
            feature_flag_name,
            exc_info=True,
        )
        return False