nemo_gym.mcp_auto_exposure

Serve an unmodified resources server’s FastAPI tool routes over MCP.

A resources server sets expose_tools_over_mcp: true in its config and its plain POST /<tool> routes are advertised and callable over an MCP /mcp endpoint — no decorators, no handler changes. The handlers keep their request: Request parameter and their request.session[SESSION_ID_KEY] reads exactly as written; this module never touches them.

run_webserver calls :func:maybe_auto_expose after building the app, so exposure is automatic for any server that sets the flag. A server tailors what it exposes by overriding one method, mcp_tools(harvested, catchall): the default returns the auto-harvested typed POST routes, and an override may filter them (exclude a route), append catch-all-backed tools (dispatcher servers whose per-tool schemas live in data: harvested + [catchall.tool(name, input_schema, description)]), or return None/[] to expose nothing. A server can narrow one rollout’s token to a subset of tools by overriding mcp_allowed_tools_for_session(seed_body); the token minted by that /seed_session response then lists and calls only those tools.

Dispatch is direct: the route’s handler runs exactly once per MCP call, invoked with a fabricated Request whose .session is materialized directly — no middleware, no routing, no second app pass. Where that cannot be proven equivalent to a real HTTP request (the server installs custom middleware, or a handler uses a shape direct dispatch does not reproduce — FastAPI dependency injection, multiple body models, …), exposure refuses loudly at startup, naming the route and the reason: a wrong dispatch would corrupt rollouts silently, while a startup error is a small fix.

MCP-side engine: the official SDK’s public low-level mcp.server.lowlevel.Server + StreamableHTTPSessionManager — no private-attribute access.

How to read this file

Two timelines run here, and every function belongs to exactly one. Keep them apart while reading.

STARTUP (runs once at boot, per opted-in server): maybe_auto_expose (flag gate) -> install_auto_exposure, which (after refusing a server that already serves /mcp):

harvest_tools — walks the app’s POST routes, calls bind_route on each, and builds the {tool name: MCPTool} map (advertisement + a direct binding per tool);
wraps endpoints — _wrap_seed_session makes the /seed_session response hand the client a session token; _wrap_verify normalizes MCP-namespaced tool-call names for scoring only;
mounts /mcp — registers the list_tools/call_tool handlers and attaches the SDK ASGI app.

PER-CALL (runs on every MCP tools/call): a POST /mcp request -> session_claims reads the token (session id + any allow-list) -> call_direct(binding) runs the route’s own handler exactly once with a fabricated Request and returns its JSON-able payload.

Reading order: start with the dataclasses DirectBinding, BindResult, and MCPTool, then bind_route (startup classify), then call_direct (per-call dispatch), then harvest_tools (startup build), then install_auto_exposure (startup wire-up). Everything else is a helper for those five.

Module Contents

Classes

Name	Description
`BindResult`	bind_route’s verdict for one route.
`DirectBinding`	Everything needed to invoke one route handler directly: resolved once at STARTUP by bind_route,
`DirectDispatchError`	Wraps a handler-visible failure, carrying the HTTP status and detail the plain route would have
`MCPTool`	Holds one exposed tool: its tools/list advertisement plus how to dispatch it (binding +
`_CatchAll`	Holds the single parameterized catch-all route; built during STARTUP harvesting and handed to

Functions

Name	Description
`_mint_session_metadata`	Mint the session token embedded in a /seed_session response (see `_wrap_seed_session`).
`_parse_session_token`	Decode a Gym MCP session token into `(session id, token allow-list)`.
`_swap_route`	-
`_take_route`	-
`_to_result`	Per the SDK’s call_tool contract, a `(content, structuredContent)` tuple populates both
`_validate_tools`	Validate the final tool list from `mcp_tools()`: legal name, not reserved, unique, dispatchable.
`_wrap_seed_session`	Replace the /seed_session route with a wrapper that appends the MCP session token to its response.
`_wrap_verify`	Wrap the app’s current /verify endpoint so MCP-namespaced tool-call names are normalized for
`bind_route`	Classify one route’s handler signature for direct dispatch (see :class:`BindResult`).
`call_direct`	Invoke the handler resolved at startup once and return its JSON-able payload.
`harvest_tools`	Scan app.routes once; return {tool name -> MCPTool}. Also runs the server-level middleware gate.
`install_auto_exposure`	Harvest the tool routes, wire the /seed_session token, and mount the /mcp endpoint.
`maybe_auto_expose`	Install MCP auto-exposure iff the server opts in (`expose_tools_over_mcp: true` in the config).

Data

_GYM_MIDDLEWARE_MODULES

_MCP_TOOL_NAME_RE

_PATH_PARAM_RE

API

class nemo_gym.mcp_auto_exposure.BindResult(
    binding: typing.Optional[nemo_gym.mcp_auto_exposure.DirectBinding],
    reasons: list[str],
    body_model: typing.Optional[type[pydantic.BaseModel]]
)

Dataclass

bind_route’s verdict for one route.

binding is None for a handler shape that is not directly dispatchable, with reasons saying why. body_model is resolved even when binding is None, so the harvested tools/list schema stays typed for a route the mcp_tools() override may drop.

binding

Optional[DirectBinding]

body_model

Optional[type[BaseModel]]

reasons

list[str]

class nemo_gym.mcp_auto_exposure.DirectBinding(
    endpoint: typing.Callable,
    path: str,
    request_params: tuple[str, ...] = (),
    body_param: typing.Optional[str] = None,
    body_model: typing.Optional[type[pydantic.BaseModel]] = None,
    path_param: typing.Optional[str] = None,
    return_model: typing.Optional[type[pydantic.BaseModel]] = None,
    body_is_dict: bool = False,
    is_coroutine: bool = False
)

Dataclass

Everything needed to invoke one route handler directly: resolved once at STARTUP by bind_route, read per-call by call_direct.

body_is_dict

bool = False

body_model

Optional[type[BaseModel]] = None

body_param

Optional[str] = None

endpoint

Callable

is_coroutine

bool = False

path

str

path_param

Optional[str] = None

request_params

tuple[str, ...] = ()

return_model

Optional[type[BaseModel]] = None

class nemo_gym.mcp_auto_exposure.DirectDispatchError(
    status: int,
    detail: str
)

Exception

Bases: Exception

Wraps a handler-visible failure, carrying the HTTP status and detail the plain route would have returned, so call_tool can surface the same text in the MCP isError result.

class nemo_gym.mcp_auto_exposure.MCPTool(
    name: str,
    tool: mcp.types.Tool,
    binding: typing.Optional[nemo_gym.mcp_auto_exposure.DirectBinding] = None,
    path_value: typing.Optional[str] = None,
    path: typing.Optional[str] = None,
    reasons: tuple[str, ...] = ()
)

Dataclass

Holds one exposed tool: its tools/list advertisement plus how to dispatch it (binding + optional catch-all path value); built during STARTUP harvesting, read per-call by call_tool.

binding

Optional[DirectBinding] = None

name

str

path

Optional[str] = None

path_value

Optional[str] = None

reasons

tuple[str, ...] = ()

tool

Tool

class nemo_gym.mcp_auto_exposure._CatchAll(
    server: typing.Any,
    route: fastapi.routing.APIRoute
)

Holds the single parameterized catch-all route; built during STARTUP harvesting and handed to mcp_tools() overrides so they can mint catch-all-backed tools.

tool(name, input_schema, description) binds one MCP tool to that route with its path param set to name (workplace’s POST /{path} pattern), reusing the same direct-dispatch binding.

_binding

Optional[DirectBinding] = None

nemo_gym.mcp_auto_exposure._CatchAll.tool(
    name: str,
    input_schema: typing.Optional[dict] = None,
    description: typing.Optional[str] = None
) -> nemo_gym.mcp_auto_exposure.MCPTool

nemo_gym.mcp_auto_exposure._mint_session_metadata(
    server: typing.Any,
    serializer: itsdangerous.URLSafeSerializer,
    request: fastapi.Request,
    seed_body: dict
) -> dict

Mint the session token embedded in a /seed_session response (see _wrap_seed_session).

Assigns the rollout its session id, asks the server for any per-session tool narrowing, and signs sid + allow-list into the token the agent replays on every tools/call.

nemo_gym.mcp_auto_exposure._parse_session_token(
    serializer: itsdangerous.URLSafeSerializer,
    token: typing.Optional[str],
    required: bool
) -> tuple[typing.Optional[str], typing.Optional[frozenset]]

Decode a Gym MCP session token into (session id, token allow-list).

required callers (tools/call) raise on a missing/forged token; optional callers (tools/list) fall back to (None, None). A tools of None means the session is unrestricted.

nemo_gym.mcp_auto_exposure._swap_route(
    app: fastapi.FastAPI,
    idx: int,
    path: str,
    endpoint: typing.Callable
) -> None

nemo_gym.mcp_auto_exposure._take_route(
    app: fastapi.FastAPI,
    path: str,
    why: str
) -> tuple[int, fastapi.routing.APIRoute]

nemo_gym.mcp_auto_exposure._to_result(
    payload: typing.Any
)

Per the SDK’s call_tool contract, a (content, structuredContent) tuple populates both CallToolResult fields while a bare content list leaves structuredContent unset — and only a JSON object maps to structuredContent.

nemo_gym.mcp_auto_exposure._validate_tools(
    server: typing.Any,
    selected: typing.Optional[list[nemo_gym.mcp_auto_exposure.MCPTool]]
) -> dict[str, nemo_gym.mcp_auto_exposure.MCPTool]

Validate the final tool list from mcp_tools(): legal name, not reserved, unique, dispatchable.

nemo_gym.mcp_auto_exposure._wrap_seed_session(
    app: fastapi.FastAPI,
    mint_metadata: typing.Callable[[Request, dict], dict]
) -> None

Replace the /seed_session route with a wrapper that appends the MCP session token to its response.

The wrapper is a **kwargs function, but FastAPI decides which dependencies to inject by reading the endpoint’s signature — so **kwargs alone would receive nothing. The fix is signature surgery: build a parameter list that (a) presents the original handler’s params (with string annotations resolved to real types, so FastAPI still validates the body model) and (b) guarantees a Request param, since the wrapper needs the live Request to read the session and raw body when minting the token. That list is then stamped onto __signature__/__annotations__ so FastAPI injects exactly those arguments into **kwargs; passthrough records the original param names to forward to the real handler (the injected Request, if added, is not among them).

nemo_gym.mcp_auto_exposure._wrap_verify(
    app: fastapi.FastAPI,
    server: typing.Any
) -> None

Wrap the app’s current /verify endpoint so MCP-namespaced tool-call names are normalized for scoring only. Verification runs against a deep copy with bare names; the reward response’s echoed names are restored to what the model emitted (matched by call_id), so persisted rollout artifacts keep transport provenance. Wrapping whatever handler the route holds at install time covers servers that strip and re-register /verify with their own handler.

nemo_gym.mcp_auto_exposure.bind_route(
    route: fastapi.routing.APIRoute
) -> nemo_gym.mcp_auto_exposure.BindResult

Classify one route’s handler signature for direct dispatch (see :class:BindResult). Public introspection only.

Annotation resolution matches FastAPI’s own: inspect.signature first (it honors a factory-set __signature__ — some servers rewrite it with the real body model while __annotations__ still says Any), falling back to get_type_hints only for deferred string annotations (from __future__ import annotations).

Pure (no app state is mutated), and called once per route: by harvest_tools for each typed POST route, and by _CatchAll.tool the first time an override mints a catch-all-backed tool. This is the deferred-validation pattern: a route the mcp_tools() override drops is never required to be dispatchable, so bind failures must surface only for tools actually exposed — harvest keeps the binding (None for an undispatchable route) and the reasons on the MCPTool, and _validate_tools raises with those stored reasons iff such a tool is exposed.

nemo_gym.mcp_auto_exposure.call_direct(
    app: fastapi.FastAPI,
    binding: nemo_gym.mcp_auto_exposure.DirectBinding,
    session_id: str,
    arguments: dict,
    path_value: typing.Optional[str] = None
) -> typing.Any

async

Invoke the handler resolved at startup once and return its JSON-able payload.

Replicates what skipping Gym’s own stack would otherwise lose: SessionMiddleware + add_session_id become scope["session"] = {SESSION_ID_KEY: sid} (handlers only read request.session); the exception middleware’s status-carrying text is reproduced by pre-formatting HTTPException / ValidationError / ClientResponseError into DirectDispatchError.

nemo_gym.mcp_auto_exposure.harvest_tools(
    app: fastapi.FastAPI,
    server: typing.Any
) -> dict[str, nemo_gym.mcp_auto_exposure.MCPTool]

Scan app.routes once; return {tool name -> MCPTool}. Also runs the server-level middleware gate.

Each non-parameterized typed POST route becomes a harvested tool. The single parameterized catch-all route (if any) is offered to the server via mcp_tools(harvested, catchall), whose return value is the final tool list — the default returns harvested unchanged; an override may filter it, append catchall.tool(...) entries, or return None/[] to expose nothing.

nemo_gym.mcp_auto_exposure.install_auto_exposure(
    server: typing.Any,
    app: fastapi.FastAPI
) -> dict[str, nemo_gym.mcp_auto_exposure.MCPTool]

Harvest the tool routes, wire the /seed_session token, and mount the /mcp endpoint.

server is any resources server built exactly as on main; app is the FastAPI app its unmodified setup_webserver() returned. Returns the tool map.

nemo_gym.mcp_auto_exposure.maybe_auto_expose(
    server: typing.Any,
    app: fastapi.FastAPI
) -> typing.Optional[dict[str, nemo_gym.mcp_auto_exposure.MCPTool]]

Install MCP auto-exposure iff the server opts in (expose_tools_over_mcp: true in the config).

Called by run_webserver after the app is fully built, so every route is present. Returns the tool map (for tests/introspection), or None when the server did not opt in.

nemo_gym.mcp_auto_exposure.LOG = logging.getLogger(__name__)

nemo_gym.mcp_auto_exposure.MCP_URL_PATH = '/mcp'

nemo_gym.mcp_auto_exposure.PERMISSIVE_SCHEMA: dict = {'type': 'object', 'additionalProperties': True}

nemo_gym.mcp_auto_exposure.TOKEN_HEADER = NEMO_GYM_MCP_SESSION_TOKEN_HEADER

nemo_gym.mcp_auto_exposure._GYM_MIDDLEWARE_MODULES = frozenset({'nemo_gym.server_utils'})

nemo_gym.mcp_auto_exposure._MCP_TOOL_NAME_RE = re.compile('[A-Za-z0-9_-]+')

nemo_gym.mcp_auto_exposure._PATH_PARAM_RE = re.compile('{([^}:]+)(?::[^}]*)?}')