Skip to content

datarobot_genai.drmcpbase.routes.utils

utils

Shared gating helpers for DataRobot MCP HTTP route groups.

A route gate is an async predicate (Request) -> bool. When it returns False — or raises — the route responds 404 so the feature stays hidden (fail closed). global-mcp injects a per-user feature-flag check; user-mcp injects the static-account check; either may pass None to leave a route open.

resolve_catalog async 

resolve_catalog(mcp: Any, provider: CatalogProvider | None) -> Sequence[Any]

Return the tool catalog a describe-the-server route should report on.

These routes answer "what can this server do", not "what may this request call", so they must not be reshaped by the caller's x-datarobot-mcp-* session headers — a picker built from a filtered taxonomy cannot offer the categories you filtered out.

run_middleware=False does not achieve that on its own: in FastMCP it suppresses middleware but still runs registered catalog transforms, and DataRobot's transform is where the mode/allowlist/gate filtering lives. So a caller that reuses one HTTP client for its MCP session and these routes gets a taxonomy narrowed to its own allowlist, all-zero counts under mode=search, and a 500 under mode=code.

Servers that install the transform therefore pass a provider that resolves the catalog with the session filter neutralized (drmcpbase.fastmcp_transforms supplies one). Without a provider this falls back to the plain call — correct for any server with no transform installed, and unchanged from the previous behaviour.

Either way FastMCP's per-component auth checks are skipped (without_component_auth_checks): the scope checks run inside every list_tools() against the current request's token, so the catalog would otherwise shrink to what the REST caller may call — the opposite of describing the server.

Source code in datarobot_genai/drmcpbase/routes/utils.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
async def resolve_catalog(mcp: Any, provider: CatalogProvider | None) -> Sequence[Any]:
    """Return the tool catalog a describe-the-server route should report on.

    These routes answer "what can this server do", not "what may this request call", so
    they must not be reshaped by the caller's ``x-datarobot-mcp-*`` session headers — a
    picker built from a filtered taxonomy cannot offer the categories you filtered out.

    ``run_middleware=False`` does *not* achieve that on its own: in FastMCP it suppresses
    middleware but still runs registered catalog transforms, and DataRobot's transform is
    where the mode/allowlist/gate filtering lives. So a caller that reuses one HTTP client
    for its MCP session and these routes gets a taxonomy narrowed to its own allowlist,
    all-zero counts under ``mode=search``, and a 500 under ``mode=code``.

    Servers that install the transform therefore pass a *provider* that resolves the
    catalog with the session filter neutralized (``drmcpbase.fastmcp_transforms``
    supplies one). Without a provider this falls back to the plain call — correct for any
    server with no transform installed, and unchanged from the previous behaviour.

    Either way FastMCP's per-component ``auth`` checks are skipped
    (``without_component_auth_checks``): the scope checks run inside every
    ``list_tools()`` against the current request's token, so the catalog would otherwise
    shrink to what the REST caller may call — the opposite of describing the server.
    """
    with without_component_auth_checks():
        if provider is not None:
            return await provider()
        return await mcp.list_tools(run_middleware=False)

register_gated_get 

register_gated_get(mcp: Any, path: str, handler: RouteHandler, gate: RouteGate | None = None) -> None

Register a single GET path that runs gate (fail closed) before handler.

Source code in datarobot_genai/drmcpbase/routes/utils.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
def register_gated_get(
    mcp: Any,
    path: str,
    handler: RouteHandler,
    gate: RouteGate | None = None,
) -> None:
    """Register a single ``GET`` *path* that runs *gate* (fail closed) before *handler*."""

    @mcp.custom_route(path, methods=["GET"])
    async def route(request: Request) -> JSONResponse:
        if gate is not None and not await _gate_allows(gate, request):
            return JSONResponse({"detail": "Not Found"}, status_code=404)
        return await handler(request)