fix(mcp): support MCP 1 and 2 tool exposure

nemo_gym/mcp_auto_exposure.py

@@ -71,12 +71,14 @@
 import json
 import logging
 import re
+from collections.abc import Awaitable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from types import UnionType
 from typing import Any, Callable, Optional, Union, get_args, get_origin, get_type_hints
 from uuid import uuid4
 
+import jsonschema
 import mcp.types as types
 from aiohttp import ClientResponseError
 from fastapi import FastAPI, Request
@@ -756,13 +758,57 @@ def _parse_session_token(
     return payload["sid"], None if allowed is None else frozenset(allowed)
 
 
-def _to_result(payload: 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."""
+def _to_result(payload: Any, *, output_schema: Optional[dict] = None) -> types.CallToolResult:
+    """Preserve text content and object-only structured content on both SDK versions."""
+    structured = payload if isinstance(payload, dict) else None
+    if output_schema is not None:
+        if structured is None:
+            raise ValueError("Output validation error: outputSchema defined but no structured output returned")
+        try:
+            jsonschema.validate(instance=structured, schema=output_schema)
+        except jsonschema.ValidationError as exc:
+            raise ValueError(f"Output validation error: {exc.message}") from exc
     text = payload if isinstance(payload, str) else JSONResponse(payload).body.decode("utf-8")
-    content = [types.TextContent(type="text", text=text)]
-    return (content, payload) if isinstance(payload, dict) else content
+    return types.CallToolResult(
+        content=[types.TextContent(type="text", text=text)], structuredContent=structured, isError=False
+    )
+
+
+def _make_mcp_server(
+    name: str,
+    list_tools: Callable[[Optional[Request]], Awaitable[list[types.Tool]]],
+    call_tool: Callable[[Optional[Request], str, dict], Awaitable[types.CallToolResult]],
+) -> _LowLevelMCPServer:
+    """Adapt only SDK registration and request context; Gym dispatch stays version independent."""
+    if "on_list_tools" in inspect.signature(_LowLevelMCPServer).parameters:
+        from mcp.server.context import ServerRequestContext
+
+        async def on_list_tools(
+            ctx: ServerRequestContext, params: Optional[types.PaginatedRequestParams]
+        ) -> types.ListToolsResult:
+            return types.ListToolsResult(tools=await list_tools(ctx.request))
+
+        async def on_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
+            try:
+                return await call_tool(ctx.request, params.name, params.arguments or {})
+            except Exception as exc:
+                # MCP 1's call_tool decorator performs this translation. MCP 2's low-level
+                # callbacks otherwise turn tool failures into JSON-RPC protocol errors.
+                return types.CallToolResult(content=[types.TextContent(type="text", text=str(exc))], isError=True)
+
+        return _LowLevelMCPServer(name, on_list_tools=on_list_tools, on_call_tool=on_call_tool)
+
+    mcp_server = _LowLevelMCPServer(name)
+
+    @mcp_server.list_tools()
+    async def legacy_list_tools() -> list[types.Tool]:
+        return await list_tools(mcp_server.request_context.request)
+
+    @mcp_server.call_tool(validate_input=False)
+    async def legacy_call_tool(name: str, arguments: dict) -> types.CallToolResult:
+        return await call_tool(mcp_server.request_context.request, name, arguments)
+
+    return mcp_server
 
 
 def session_token_serializer(server: Any) -> URLSafeSerializer:
@@ -805,22 +851,15 @@ def install_auto_exposure(server: Any, app: FastAPI) -> dict[str, MCPTool]:
     _wrap_seed_session(app, mint_metadata)
     _wrap_verify(app, server)
 
-    mcp_server = _LowLevelMCPServer(server.config.name or type(server).__name__)
-
-    # mcp_server.request_context is a contextvar the SDK populates only while one of the handlers
-    # below is on the stack — exactly when session_claims runs, which is why it stays local.
-    def session_claims(required: bool = True) -> tuple[Optional[str], Optional[frozenset]]:
-        ctx_request = mcp_server.request_context.request  # the POST /mcp starlette Request
+    def session_claims(ctx_request: Optional[Request], *, required: bool) -> tuple[Optional[str], Optional[frozenset]]:
         token = ctx_request.headers.get(NEMO_GYM_MCP_SESSION_TOKEN_HEADER) if ctx_request is not None else None
         return _parse_session_token(serializer, token, required)
 
-    @mcp_server.list_tools()
-    async def list_tools() -> list[types.Tool]:
-        _, allowed = session_claims(required=False)
+    async def list_tools(ctx_request: Optional[Request]) -> list[types.Tool]:
+        _, allowed = session_claims(ctx_request, required=False)
         return [t.tool for t in tools.values() if allowed is None or t.name in allowed]
 
-    @mcp_server.call_tool(validate_input=False)
-    async def call_tool(name: str, arguments: dict):
+    async def call_tool(ctx_request: Optional[Request], name: str, arguments: dict) -> types.CallToolResult:
         tool = tools.get(name)
         if tool is None:
             # Models hallucinate tool names, so answer the miss cheaply with a plain isError result
@@ -832,15 +871,16 @@ async def call_tool(name: str, arguments: dict):
                 ],
                 isError=True,
             )
-        session_id, allowed = session_claims(required=True)
+        session_id, allowed = session_claims(ctx_request, required=True)
         if allowed is not None and name not in allowed:
             raise ValueError(f"Tool {name!r} is not allowed for this session.")
         try:
             payload = await call_direct(app, tool.binding, session_id, arguments, path_value=tool.path_value)
         except DirectDispatchError as exc:
             raise ValueError(f"HTTP {exc.status} from POST /{name}: {exc.detail}")
-        return _to_result(payload)
+        return _to_result(payload, output_schema=tool.tool.model_dump(by_alias=True).get("outputSchema"))
 
+    mcp_server = _make_mcp_server(server.config.name or type(server).__name__, list_tools, call_tool)
     manager = StreamableHTTPSessionManager(
         app=mcp_server,
         event_store=None,

pyproject.toml

@@ -133,9 +133,9 @@ dependencies = [
     "fastapi!=0.142.*",
 
     # MCP Python SDK: Used by Gym-owned Streamable HTTP MCP resources servers.
-    # Updated Tue Jul 29, 2026 with mcp>=1.28.1,<2
+    # Updated Tue Oct 13, 2026: MCP 1.30.0 and 2.3.0 compatibility validated.
     # License: MIT https://github.com/modelcontextprotocol/python-sdk/blob/main/LICENSE
-    "mcp>=1.28.1,<2",
+    "mcp>=1.28.1,<3",
 
     # anyio: transitive dependency of mcp/fastapi/starlette; pinned directly since
     # mcp's own floor (>=4.10.0) resolves to a version with a known CVE.
@@ -508,10 +508,10 @@ exclude-dependencies = [
     # (missing #include <cstdint> in src/args.cc). Excluded as a transitive dep
     # of unbabel-comet; wmt_translation does not import fasttext at runtime.
     "fasttext-wheel",
+    # cryptography must remain available: MCP 2 imports it at startup via pyjwt[crypto].
     # Exclude unwanted dependencies from mlflow
     "alembic",
     "blinker",
-    "cryptography",
     "flask",
     "flask-cors",
     "graphene",

resources_servers/gdpval/tests/test_tools.py

@@ -63,7 +63,7 @@ def test_benchmark_tools_are_the_tools_the_server_serves(server):
             "type": "function",
             "name": t.name,
             "description": t.tool.description,
-            "parameters": t.tool.inputSchema,
+            "parameters": t.tool.model_dump(by_alias=True)["inputSchema"],
             "strict": None,
         }
         for t in tools.values()

resources_servers/newton_bench/requirements.txt

@@ -4,3 +4,5 @@ scipy>=1.7
 sympy>=1.10
 fix-busted-json>=0.0.18
 pandas>=1.5.3
+# NewtonBench imports dotenv directly; do not rely on optional transitive dependencies.
+python-dotenv>=1.0.0

resources_servers/ns_tools/requirements.txt

@@ -1,3 +1,5 @@
 -e nemo-gym[dev] @ ../../
 ## NeMo-Skills tools subpackage as of June 12, 2026
 nemo-skills-tools @ git+https://github.com/NVIDIA-NeMo/Skills.git@da85a881d972e6fec847b90cf553a0bf9bf10638#subdirectory=tools
+# The pinned NeMo-Skills client imports the MCP 1 Streamable HTTP API.
+mcp>=1.28.1,<2

responses_api_agents/hermes_agent/requirements.txt

@@ -1,2 +1,4 @@
 -e nemo-gym[dev,sandbox] @ ../../
 hermes-agent @ git+https://github.com/cmunley1/hermes-agent@26bb847a
+# The pinned Hermes client requires the MCP 1 Streamable HTTP API.
+mcp>=1.28.1,<2

responses_api_agents/hermes_agent/sandbox.py

@@ -100,7 +100,7 @@ async def _runtime_installed(self) -> bool:
         session in this sandbox is reused.
         """
         check = await self.session.sandbox.exec(
-            f"{quote(_SANDBOX_PYTHON)} -c 'import run_agent, mcp'",
+            f"{quote(_SANDBOX_PYTHON)} -c 'import run_agent; from mcp.client.streamable_http import streamablehttp_client'",
             cwd=self.session.workdir,
             timeout_s=120,
         )
@@ -114,7 +114,7 @@ async def _install_runtime(self, timeout: float, uv_cache_dir: str) -> None:
         install = await self.session.sandbox.exec(
             f"chmod 755 {quote(_SANDBOX_UV)} && rm -rf {venv} && "
             f"{quote(_SANDBOX_UV)} venv {venv} --python 3.13 && "
-            f"{quote(_SANDBOX_UV)} pip install --python {quote(_SANDBOX_PYTHON)} {quote(_HERMES_REQUIREMENT)}",
+            f"{quote(_SANDBOX_UV)} pip install --python {quote(_SANDBOX_PYTHON)} {quote(_HERMES_REQUIREMENT)} 'mcp>=1.28.1,<2'",
             cwd=self.session.workdir,
             timeout_s=timeout,
         )

responses_api_agents/hermes_agent/tests/test_app.py

@@ -224,6 +224,7 @@ async def exec_(command, **_kwargs):
         commands = [call.args[0] for call in sandbox.exec.await_args_list]
         install = [command for command in commands if "pip install" in command]
         assert len(install) == 1 and "rm -rf" in install[0]
+        assert "mcp>=1.28.1,<2" in install[0]
         # The import is checked before installing and again after.
         assert sum("import run_agent" in command for command in commands) == 2
         assert sandbox.upload.await_args_list[0].args[0] == Path("/usr/bin/uv")

responses_api_agents/osworld_agent/uv.toml

@@ -33,7 +33,6 @@ exclude-dependencies = [
     "fasttext-wheel",
     "alembic",
     "blinker",
-    "cryptography",
     "flask",
     "flask-cors",
     "graphene",

tests/unit_tests/test_mcp_auto_exposure.py

@@ -26,7 +26,8 @@
 import logging
 import subprocess
 import sys
-from contextlib import contextmanager
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import asynccontextmanager, contextmanager
 from types import SimpleNamespace
 from typing import Any, Optional
 from unittest.mock import MagicMock
@@ -704,7 +705,7 @@ async def gated(body: EchoBody, ok: bool = Depends(lambda: True)):
             return app
 
         def mcp_tools(self, harvested, catchall):
-            harvested_schemas.update({t.name: t.tool.inputSchema for t in harvested})
+            harvested_schemas.update({t.name: t.tool.model_dump(by_alias=True)["inputSchema"] for t in harvested})
             return [t for t in harvested if t.name not in ("end_session", "gated")]
 
     server = _server(DroppedTyped, "excl")
@@ -956,3 +957,161 @@ def test_normalize_tool_name_without_server_name_strips_first_namespace():
     # with a server name, only that server's prefix is stripped
     assert normalize_tool_name("mcp__store__append", "store") == "append"
     assert normalize_tool_name("mcp__other__append", "store") == "mcp__other__append"
+
+
+@pytest.mark.parametrize("payload", [{"answer": 42}, [1, 2], "hello", 7, None])
+def test_result_content_and_structured_content(payload):
+    class Results(Store):
+        def setup_webserver(self):
+            app = super().setup_webserver()
+
+            @app.post("/result")
+            async def result():
+                return payload
+
+            return app
+
+    with _mcp(Results) as (client, token):
+        result = _call(client, "result", {}, token)
+        assert result.get("isError") is not True
+        text = result["content"][0]["text"]
+        assert (text if isinstance(payload, str) else json.loads(text)) == payload
+        assert result.get("structuredContent") == (payload if isinstance(payload, dict) else None)
+
+
+@pytest.mark.parametrize("payload", [{"answer": 42}, {"answer": "wrong"}, "not an object"])
+def test_output_schema_validation_is_preserved(payload):
+    class OutputSchema(Store):
+        def setup_webserver(self):
+            app = super().setup_webserver()
+
+            @app.post("/result")
+            async def result():
+                return payload
+
+            return app
+
+        def mcp_tools(self, harvested, catchall):
+            tools = super().mcp_tools(harvested, catchall)
+            for tool in tools:
+                if tool.name == "result":
+                    data = tool.tool.model_dump(by_alias=True)
+                    data["outputSchema"] = {
+                        "type": "object",
+                        "properties": {"answer": {"type": "integer"}},
+                        "required": ["answer"],
+                    }
+                    tool.tool = type(tool.tool).model_validate(data)
+            return tools
+
+    with _mcp(OutputSchema) as (client, token):
+        result = _call(client, "result", {}, token)
+        if payload == {"answer": 42}:
+            assert result.get("isError") is not True
+            assert result["structuredContent"] == payload
+        else:
+            assert result["isError"] is True
+            assert "Output validation error:" in result["content"][0]["text"]
+
+
+def test_omitted_arguments_and_pydantic_coercion():
+    class Coercing(Store):
+        def setup_webserver(self):
+            app = super().setup_webserver()
+
+            @app.post("/coerce")
+            async def coerce(body: CountBody):
+                return {"count": body.count}
+
+            return app
+
+    with _mcp(Coercing) as (client, token):
+        result = _rpc(client, "tools/call", {"name": "raw_step"}, token)["result"]
+        assert result["structuredContent"] == {"echo": {}}
+        # Gym delegates input validation/coercion to Pydantic, not SDK JSON Schema validation.
+        result = _call(client, "coerce", {"count": "42"}, token)
+        assert result["structuredContent"] == {"count": 42}
+
+
+class CountBody(BaseModel):
+    count: int
+
+
+def test_concurrent_calls_use_each_requests_signed_session():
+    with _mcp(SessionScoped) as (client, _):
+        tokens = [
+            client.post("/seed_session", json={"allowed_tools": names}).json()["mcp"]["headers"][TOKEN_HEADER]
+            for names in (["append"], ["raw_step"])
+        ]
+
+        def run(index):
+            token = tokens[index % 2]
+            expected = "append" if index % 2 == 0 else "raw_step"
+            assert [tool["name"] for tool in _list(client, token)] == [expected]
+            allowed = _call(client, expected, {"value": str(index)}, token)
+            assert allowed.get("isError") is not True
+            denied = _call(client, "raw_step" if expected == "append" else "append", {}, token)
+            assert denied["isError"] is True
+            assert "not allowed" in denied["content"][0]["text"]
+
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            list(pool.map(run, range(12)))
+
+
+def test_original_lifespan_state_and_shutdown_are_preserved():
+    events = []
+    server = _server()
+    app = server.setup_webserver()
+
+    @asynccontextmanager
+    async def lifespan(app):
+        events.append("start")
+        try:
+            yield {"marker": "original lifespan"}
+        finally:
+            events.append("stop")
+
+    app.router.lifespan_context = lifespan
+    maybe_auto_expose(server, app)
+    with TestClient(app) as client:
+        assert client.app_state["marker"] == "original lifespan"
+        assert events == ["start"]
+        token = _seed(client)
+        _handshake(client)
+        assert _call(client, "raw_step", {}, token)["structuredContent"] == {"echo": {}}
+    assert events == ["start", "stop"]
+
+
+async def test_sdk_client_discovers_and_calls_over_streamable_http():
+    from importlib.metadata import version
+
+    from mcp import ClientSession
+    from mcp.client.streamable_http import streamable_http_client
+
+    # Exercise each SDK's real HTTP transport against the ASGI app without a network listener.
+    # MCP 2 uses httpx2; MCP 1 uses httpx. These clients are test-only.
+    if version("mcp").startswith("2."):
+        import httpx2 as http_client_module
+    else:
+        import httpx as http_client_module
+
+    server = _server(SessionScoped)
+    app = server.setup_webserver()
+    maybe_auto_expose(server, app)
+    async with app.router.lifespan_context(app):
+        async with http_client_module.AsyncClient(
+            transport=http_client_module.ASGITransport(app=app), base_url="http://testserver"
+        ) as http_client:
+            seeded = await http_client.post("/seed_session", json={"allowed_tools": ["append"]})
+            http_client.headers[TOKEN_HEADER] = seeded.json()["mcp"]["headers"][TOKEN_HEADER]
+            async with streamable_http_client("http://testserver/mcp", http_client=http_client) as streams:
+                async with ClientSession(streams[0], streams[1]) as session:
+                    await session.initialize()
+                    listed = await session.list_tools()
+                    assert [tool.name for tool in listed.tools] == ["append"]
+                    result = (await session.call_tool("append", {"value": "sdk"})).model_dump(by_alias=True)
+                    assert result["structuredContent"] == {"values": ["sdk"]}
+                    assert result["isError"] is False
+                    denied = (await session.call_tool("raw_step", {})).model_dump(by_alias=True)
+                    assert denied["isError"] is True
+                    assert "not allowed" in denied["content"][0]["text"]