diff --git a/nitrostack/core/app.py b/nitrostack/core/app.py index f6813b4..3ffc28f 100644 --- a/nitrostack/core/app.py +++ b/nitrostack/core/app.py @@ -73,6 +73,7 @@ resolve_http_engine, resolve_protocol_era_resolution, wire_mode_for_era, + supports_resource_subscriptions, ) from nitrostack.protocol.mrtr import InputRequiredResult, split_mrtr_from_arguments from nitrostack.protocol.cache_hints import ( @@ -950,10 +951,14 @@ def handle_server_discover(self, protocol_version: Optional[str] = None) -> Dict getattr(self, "protocol_era", None), self.server_config.protocol_version, ) + # Same source of truth as `NitroStackMcpServer.get_capabilities`, so + # `server/discover` and real capability negotiation never disagree. + resources_subscribe = supports_resource_subscriptions(self.protocol_era) return build_discover_result( server_name=self.server_config.name, server_version=self.server_config.version, protocol_version=version, + resources_subscribe=resources_subscribe, advertise_tasks=self._advertise_tasks_extension(), advertise_app=has_widgets, custom_extensions=self._custom_extensions(), @@ -985,6 +990,7 @@ def _list_endpoint_cache_meta(self) -> Dict[str, Any]: return build_list_endpoint_cache_hint_meta() def _setup_handlers(self, server: NitroStackMcpServer) -> None: + server.protocol_era = self.protocol_era async def _list_tools( ctx: ServerRequestContext, _params: Optional[PaginatedRequestParams], @@ -1056,7 +1062,7 @@ async def _read_resource(ctx: ServerRequestContext, params: ReadResourceRequestP return _read_contents_to_result(uri, await self._read_resource(uri)) async def _subscribe_resource(_ctx: ServerRequestContext, params: SubscribeRequestParams): - if rejects_deprecated_method("resources/subscribe", self.protocol_era): + if not supports_resource_subscriptions(self.protocol_era): message = deprecated_method_message("resources/subscribe") raise MCPError(types.METHOD_NOT_FOUND, message or "Not supported") if str(params.uri) not in self._resources: @@ -1064,6 +1070,8 @@ async def _subscribe_resource(_ctx: ServerRequestContext, params: SubscribeReque return types.EmptyResult() async def _unsubscribe_resource(_ctx: ServerRequestContext, _params: UnsubscribeRequestParams): + if not supports_resource_subscriptions(self.protocol_era): + raise MCPError(types.METHOD_NOT_FOUND, "Resource subscriptions are not supported in this era") return types.EmptyResult() async def _list_prompts(_ctx: ServerRequestContext, _params: Optional[PaginatedRequestParams]): diff --git a/nitrostack/core/mcp_server.py b/nitrostack/core/mcp_server.py index b383ff2..1c8c490 100644 --- a/nitrostack/core/mcp_server.py +++ b/nitrostack/core/mcp_server.py @@ -8,9 +8,19 @@ from typing import Any, Callable, Dict, Optional import mcp.types as types +from mcp import MCPError from mcp.server.lowlevel import Server as LowLevelServer from mcp.server.lowlevel.server import NotificationOptions from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler +from nitrostack.protocol.errors import JsonRpcErrorCode +from nitrostack.protocol.version import ( + MODERN_PROTOCOL_VERSION, + ProtocolEra, + supports_resource_subscriptions, + supported_protocol_versions_for_era, +) +from nitrostack.protocol.meta import flatten_request_meta_object +from nitrostack.protocol.jsonrpc import HeaderBodyMismatchError, validate_protocol_version_header_meta class NitroStackMcpServer(LowLevelServer): @@ -33,12 +43,41 @@ def __init__(self, name: str, version: Optional[str] = None): self.has_task_support: bool = False self.http_engine: Optional[str] = None self.sessionful: bool = False + self.protocol_era: Optional[ProtocolEra] = None + self.middleware.append(self._enforce_protocol_era) self.discover_handler: Optional[Callable[[], dict[str, Any]]] = None self.initialize_handler: Optional[Callable[[Optional[str]], dict[str, Any]]] = None # In-process tests still look up handlers by request type. self.request_handlers: Dict[Any, Callable[..., Any]] = {} self.notification_handlers: Dict[Any, Callable[..., Any]] = {} + async def _enforce_protocol_era(self, context, call_next): + """Apply era policy inside the official runner, after its JSON-RPC parsing.""" + if self.protocol_era == "modern" and context.method in ( + "initialize", "notifications/initialized", + ): + raise MCPError(types.METHOD_NOT_FOUND, f"Method not found: {context.method}") + request_headers = getattr(getattr(context, "request", None), "headers", {}) + meta = flatten_request_meta_object(context.meta) + header_version = request_headers.get("mcp-protocol-version") + meta_version = meta.get("io.modelcontextprotocol/protocolVersion") + try: + validate_protocol_version_header_meta(header_version, meta_version) + except HeaderBodyMismatchError as exc: + raise MCPError(int(exc.code), exc.message) from None + claimed_version = header_version or meta_version + # The official SDK uses a 2025-03-26 default when a legacy-compatible + # HTTP request omits the header. Do not reject that implicit fallback. + unsupported_claim = self.protocol_era is not None and claimed_version and ( + claimed_version not in supported_protocol_versions_for_era(self.protocol_era) + ) + modern_legacy_request = ( + self.protocol_era == "modern" and context.protocol_version != MODERN_PROTOCOL_VERSION + ) + if unsupported_claim or modern_legacy_request: + raise MCPError(int(JsonRpcErrorCode.UNSUPPORTED_PROTOCOL_VERSION), "Unsupported protocol version") + return await call_next(context) + def handle_server_discover(self) -> dict[str, Any]: """Answer ``server/discover`` from this server instance.""" if self.discover_handler is None: @@ -67,7 +106,7 @@ def get_capabilities( ) if caps.resources is not None: - caps.resources.subscribe = True + caps.resources.subscribe = supports_resource_subscriptions(self.protocol_era) if self.has_task_support: caps.tasks = types.ServerTasksCapability( diff --git a/nitrostack/protocol/discovery.py b/nitrostack/protocol/discovery.py index d15fc0a..4360261 100644 --- a/nitrostack/protocol/discovery.py +++ b/nitrostack/protocol/discovery.py @@ -36,11 +36,17 @@ def build_discover_result( custom_extensions: Optional[dict[str, str]] = None, tools_list_changed: bool = True, resources_list_changed: bool = True, + resources_subscribe: bool = False, prompts_list_changed: bool = True, ttl_ms: int = DEFAULT_LIST_CACHE_TTL_MS, cache_scope: Literal["public", "private"] = "private", ) -> dict[str, Any]: - """Build the ``server/discover`` result for the mounted HTTP engine.""" + """Build the ``server/discover`` result for the mounted HTTP engine. + + ``resources_subscribe`` must match ``NitroStackMcpServer.get_capabilities`` + for the same era, so ``server/discover`` and real capability negotiation + never disagree on ``resources.subscribe``. + """ versions = list(supported_versions or SUPPORTED_PROTOCOL_VERSIONS) extensions: dict[str, dict[str, str]] = {} if advertise_app: @@ -53,7 +59,7 @@ def build_discover_result( capabilities: dict[str, Any] = { "tools": {"listChanged": tools_list_changed}, - "resources": {"subscribe": False, "listChanged": resources_list_changed}, + "resources": {"subscribe": resources_subscribe, "listChanged": resources_list_changed}, "prompts": {"listChanged": prompts_list_changed}, } if extensions: diff --git a/nitrostack/protocol/version.py b/nitrostack/protocol/version.py index 24cf3f0..93d2828 100644 --- a/nitrostack/protocol/version.py +++ b/nitrostack/protocol/version.py @@ -207,6 +207,11 @@ def needs_sessionful_engine(era: ProtocolEra) -> bool: return era == "legacy" +def supports_resource_subscriptions(era: Optional[ProtocolEra]) -> bool: + """Only the legacy compatibility era serves resource subscriptions.""" + return era == "legacy" + + def http_engine_for_era(era: ProtocolEra) -> HttpEngine: """ Select the /mcp HTTP engine for an era. diff --git a/nitrostack/transports/http.py b/nitrostack/transports/http.py index 7d958fb..e651624 100644 --- a/nitrostack/transports/http.py +++ b/nitrostack/transports/http.py @@ -34,6 +34,7 @@ from starlette.applications import Starlette from starlette.middleware import Middleware +from starlette.middleware.cors import CORSMiddleware from starlette.responses import HTMLResponse, JSONResponse, Response from starlette.routing import Mount, Route from starlette.types import ASGIApp, Receive, Scope, Send @@ -41,6 +42,7 @@ from nitrostack.transports.landing_page import render_landing_page from nitrostack.widgets.preview_page import render_preview_page from nitrostack.core.di import DIContainer +from nitrostack.transports.cors import configured_cors_origins from pydantic_core import PydanticUndefined from mcp.server.sse import SseServerTransport from mcp.server.streamable_http_manager import StreamableHTTPSessionManager @@ -56,10 +58,10 @@ from nitrostack.protocol.constants import LEGACY_SESSION_HEADER from nitrostack.transports.headers import ( CORS_ALLOW_HEADER_NAMES, + CORS_EXPOSE_HEADER_NAMES, SSE_SUBSCRIPTIONS_PATH, strip_legacy_session_headers_asgi, ) -from nitrostack.transports.cors import McpCorsMiddleware from nitrostack.transports.subscriptions import subscriptions_listen_endpoint from nitrostack.transports.proxy import public_url_for_request from nitrostack.core.log_emitter import install_log_handler @@ -490,6 +492,7 @@ def build_http_app( if server is not None: server.http_engine = http_engine server.sessionful = http_engine == "sessionful" + server.protocol_era = protocol_era security_settings = None if not enable_cors: @@ -662,7 +665,7 @@ async def json_version(request): return JSONResponse( { "Browser": meta["name"], - "Protocol-Version": "2025-06-18", + "Protocol-Version": protocol_version_for_era(protocol_era), "User-Agent": f"NitroStack/{meta['version']}", "webSocketDebuggerUrl": "", "transport": "mcp", @@ -741,14 +744,23 @@ async def lifespan(app): Middleware(ExactEndpointSlashMiddleware, endpoint=endpoint), ] if enable_cors: - # CORS policy lives in transports/cors.py (MCP_CORS_ALLOWED_ORIGINS, Mcp-* headers). - # Mcp-Session-Id is exposed only on the sessionful (legacy) engine. + expose_headers = list(CORS_EXPOSE_HEADER_NAMES) + if http_engine == "sessionful": + expose_headers.append(LEGACY_SESSION_HEADER) + # Single CORS layer for `/mcp`: the sidecar that used to add a second + # one has been removed (#22), so `MCP_CORS_ALLOWED_ORIGINS` is honored + # here and nowhere else. + allow_origins = list(configured_cors_origins()) or ["*"] middleware.insert( 0, Middleware( - McpCorsMiddleware, - expose_session_id=http_engine == "sessionful", - fallback_allow_headers=CORS_ALLOW_HEADERS, + CORSMiddleware, + allow_origins=allow_origins, + allow_methods=["GET", "POST", "DELETE", "OPTIONS"], + # SEP-2243 permits application-defined Mcp-Param-* names. + # Starlette mirrors requested names when '*' is enabled. + allow_headers=[*CORS_ALLOW_HEADERS, "*"], + expose_headers=expose_headers, ), ) diff --git a/nitrostack/transports/stdio.py b/nitrostack/transports/stdio.py index d3fc9bd..940d406 100644 --- a/nitrostack/transports/stdio.py +++ b/nitrostack/transports/stdio.py @@ -2,19 +2,20 @@ import contextlib import uuid import contextvars -from typing import TYPE_CHECKING, Any, Generator +from typing import TYPE_CHECKING, Any, Generator, Mapping import anyio -from mcp.server.runner import serve_loop +from anyio.abc import ObjectReceiveStream, ObjectSendStream +import mcp.types as types from mcp.server.stdio import stdio_server +from mcp.shared.message import SessionMessage +from mcp_types import PROTOCOL_VERSION_META_KEY, UNSUPPORTED_PROTOCOL_VERSION -from nitrostack.protocol.version import ProtocolEra +from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION, ProtocolEra from nitrostack.runtime.request_ctx import stdio_connection_id if TYPE_CHECKING: from mcp.server.lowlevel import Server - from mcp.shared._stream_protocols import ReadStream, WriteStream - from mcp.shared.message import SessionMessage class SafeStdoutWrapper: @@ -50,20 +51,55 @@ def safe_stdio_transport() -> Generator[None, None, None]: class _StdioConnectionStream: - """Preserve sender context while binding a server-owned connection identity.""" + """Apply era policy and preserve the trusted connection identity and sender context.""" - def __init__(self, stream: Any, connection_id: str) -> None: + def __init__(self, stream: Any, connection_id: str, write_stream: Any, era: ProtocolEra) -> None: self._stream = stream self._connection_id = connection_id + self._write_stream = write_stream + self._era = era self.last_context = None async def receive(self): - message = await self._stream.receive() + while True: + message = await self._stream.receive() + error = self._era_error(message) + if error is None: + break + await self._write_stream.send(SessionMessage(types.JSONRPCError( + jsonrpc="2.0", id=message.message.id, error=error, + ))) sender_context = getattr(self._stream, "last_context", None) self.last_context = sender_context.copy() if sender_context is not None else contextvars.copy_context() self.last_context.run(stdio_connection_id.set, self._connection_id) return message + def _era_error(self, message: SessionMessage | Exception) -> types.ErrorData | None: + # Server.run negotiates from the first request. Refuse incompatible requests + # before negotiation so a rejected handshake cannot select the wrong era. + # Accepted requests, notifications, responses and parse errors are passed + # unchanged to the official server; this adapter never dispatches handlers. + if not isinstance(message, SessionMessage) or not isinstance(message.message, types.JSONRPCRequest): + return None + request = message.message + if self._era == "modern" and request.method == "initialize": + return types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found: initialize") + params = request.params or {} + meta = params.get("_meta") + version = meta.get(PROTOCOL_VERSION_META_KEY) if isinstance(meta, Mapping) else None + if self._era == "modern" and version != MODERN_PROTOCOL_VERSION: + return types.ErrorData( + code=UNSUPPORTED_PROTOCOL_VERSION, + message="Modern stdio requests require a supported protocol version in params._meta", + data={"supportedVersions": [MODERN_PROTOCOL_VERSION]}, + ) + if self._era == "legacy" and version == MODERN_PROTOCOL_VERSION: + return types.ErrorData( + code=types.INVALID_REQUEST, + message="Legacy stdio does not accept modern protocol envelopes", + ) + return None + def __aiter__(self): return self @@ -86,62 +122,25 @@ async def aclose(self): async def serve_stdio_streams( server: "Server[Any]", - read_stream: "ReadStream[SessionMessage | Exception]", - write_stream: "WriteStream[SessionMessage]", + read_stream: "ObjectReceiveStream[SessionMessage | Exception]", + write_stream: "ObjectSendStream[SessionMessage]", era: ProtocolEra, ) -> None: """Serve one connection with a trusted identity shared by its requests.""" connection_id = uuid.uuid4().hex token = stdio_connection_id.set(connection_id) try: - await _serve_stdio_connection( - server, _StdioConnectionStream(read_stream, connection_id), write_stream, era + await server.run( + _StdioConnectionStream(read_stream, connection_id, write_stream, era), + write_stream, + server.create_initialization_options(), ) finally: stdio_connection_id.reset(token) -async def _serve_stdio_connection( - server: "Server[Any]", - read_stream: "ReadStream[SessionMessage | Exception]", - write_stream: "WriteStream[SessionMessage]", - era: ProtocolEra, -) -> None: - """Drive official mcp 2.x over an already-open stdio stream pair. - - ``auto`` uses the official dual-era loop (2026 envelope or 2025 handshake). - ``modern`` rejects ``initialize``. ``legacy`` keeps the handshake loop. - """ - init_options = server.create_initialization_options() - if era == "auto": - await server.run(read_stream, write_stream, init_options) - return - - async with server.lifespan(server) as lifespan_state: - try: - if era == "modern": - from mcp.server.runner import _serve_modern_stream - - await _serve_modern_stream( - server, - read_stream, - write_stream, - lifespan_state=lifespan_state, - raise_exceptions=False, - ) - else: - await serve_loop( - server, - read_stream, - write_stream, - lifespan_state=lifespan_state, - init_options=init_options, - ) - finally: - await write_stream.aclose() - - async def run_stdio(server: "Server[Any]", era: ProtocolEra) -> None: """Serve official mcp 2.x on process stdin/stdout for the active era.""" + server.protocol_era = era async with stdio_server() as (read_stream, write_stream): await serve_stdio_streams(server, read_stream, write_stream, era) diff --git a/tests/http_probe.py b/tests/http_probe.py new file mode 100644 index 0000000..8679f19 --- /dev/null +++ b/tests/http_probe.py @@ -0,0 +1,106 @@ +"""Shared probes of the actual official MCP HTTP mount (no sidecar doubles).""" + +import asyncio +import json +import os +from contextlib import contextmanager +from unittest.mock import patch + +from mcp_types import PROTOCOL_VERSION_META_KEY +from pydantic import BaseModel +from starlette.testclient import TestClient + +from nitrostack import DIContainer, module, resource, tool +from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app +from nitrostack.protocol.version import LEGACY_PROTOCOL_VERSION, MODERN_PROTOCOL_VERSION + + +class EchoInput(BaseModel): + value: str = "ok" + + +def make_app(era="auto"): + """Resolve era deterministically without leaking process environment or DI.""" + DIContainer.reset() + + class ProbeTools: + @tool(name="echo", description="Echo and expose verified context", input_schema=EchoInput, + task_support="optional") + async def echo(self, input, context): + return {"value": input.value, "user": context.user, + "protocol_version": context.protocol_version, + "mirrored": context.mcp_param_headers} + + @resource(uri="probe://info", name="Probe", description="Readable probe") + async def info(self, context): + return "probe resource" + + @module(name="http-probe", providers=[ProbeTools]) + class ProbeModule: + pass + + @mcp_app(module=ProbeModule, server=ServerConfig(name="http-probe", protocol_era=era)) + class ProbeApp: + pass + + with patch.dict(os.environ, {"NITRO_MCP_PROTOCOL_VERSION": era, "MCP_STATELESS": ""}): + return asyncio.run(McpApplicationFactory.create(ProbeApp)) + + +def envelope(method="tools/list", params=None, *, version=MODERN_PROTOCOL_VERSION, modern=True): + params = dict(params or {}) + if modern: + params["_meta"] = { + PROTOCOL_VERSION_META_KEY: version, + "io.modelcontextprotocol/clientInfo": {"name": "http-probe", "version": "1"}, + "io.modelcontextprotocol/clientCapabilities": {}, + **params.get("_meta", {}), + } + return {"jsonrpc": "2.0", "id": "probe", "method": method, "params": params} + + +def headers(method="tools/list", *, name=None, version=MODERN_PROTOCOL_VERSION): + result = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream", + "MCP-Protocol-Version": version, "Mcp-Method": method} + if name is not None: + result["Mcp-Name"] = name + return result + + +def decode(response): + """Handle JSON or a completed request's SSE response explicitly.""" + if "text/event-stream" in response.headers.get("content-type", ""): + messages = [json.loads(line[5:].strip()) for line in response.text.splitlines() + if line.startswith("data:")] + return next(message for message in messages if "id" in message) + return response.json() + + +def post(client, method="tools/list", params=None, *, modern=True, extra_headers=None, + version=MODERN_PROTOCOL_VERSION): + return client.post("/mcp", json=envelope(method, params, modern=modern, version=version), + headers={**headers(method, name=(params or {}).get("name") or (params or {}).get("uri"), + version=version), **(extra_headers or {})}) + + +@contextmanager +def http_client(era="auto", *, app=None, json_response=True, **options): + app = app or make_app(era) + http = app.get_combined_app(json_response=json_response, **options) + try: + with TestClient(http) as client: + if era == "legacy": + response = post(client, "initialize", { + "protocolVersion": LEGACY_PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "legacy-probe", "version": "1"}, + }, modern=False, version=LEGACY_PROTOCOL_VERSION) + assert response.status_code == 200, response.text + assert "result" in decode(response), response.text + session = response.headers.get("mcp-session-id") + assert session + client.headers["mcp-session-id"] = session + client.post("/mcp", json={"jsonrpc": "2.0", "method": "notifications/initialized"}, + headers=headers("notifications/initialized", version=LEGACY_PROTOCOL_VERSION)) + yield app, http, client + finally: + DIContainer.reset() diff --git a/tests/test_mcp20_blueprint.py b/tests/test_mcp20_blueprint.py index 7170ba3..2c34e03 100644 --- a/tests/test_mcp20_blueprint.py +++ b/tests/test_mcp20_blueprint.py @@ -3,17 +3,15 @@ from __future__ import annotations import asyncio -import json import os import sys from unittest.mock import patch import mcp.types as types import pytest -from nitrostack.runtime.request_ctx import Experimental, RequestContext, RequestParamsMeta, request_ctx +from nitrostack.runtime.request_ctx import Experimental, RequestContext, request_ctx from mcp import MCPError as McpError from pydantic import BaseModel, Field -from starlette.testclient import TestClient sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) @@ -23,7 +21,7 @@ from nitrostack.core.di import DIContainer from nitrostack.core.errors import ResourceNotFoundError, TaskNotFoundError from nitrostack.core.task import TaskManager -from nitrostack.protocol.constants import LEGACY_SESSION_HEADER, MAX_CIMD_BYTES +from nitrostack.protocol.constants import MAX_CIMD_BYTES from nitrostack.protocol.deprecated import deprecated_method_message from nitrostack.protocol.discovery import build_discover_result from nitrostack.protocol.errors import JsonRpcErrorCode @@ -40,11 +38,23 @@ protocol_version_matches_blueprint, verify_package_layout, ) -from nitrostack.runtime.stateless import assert_stateless_headers from nitrostack.tasks.eviction import should_evict_terminal_task from nitrostack.tasks.types import TaskAccessContext, TaskEntry, TaskWireData, datetime_to_ms, utc_now from nitrostack.tasks.authorization import check_task_access from nitrostack.transports.cors import cors_preflight_response_headers +from tests.http_probe import decode, http_client, post + + +def test_blueprint_discover_uses_official_http_without_session(): + with http_client("auto") as (app, _, client): + response = post(client, "server/discover") + assert response.status_code == 200, response.text + result = decode(response)["result"] + assert MODERN_PROTOCOL_VERSION in result["supportedVersions"] + assert result["capabilities"]["resources"]["subscribe"] is False + assert result["capabilities"]["tools"]["listChanged"] is True + assert "mcp-session-id" not in response.headers + assert app.mcp_server.get_capabilities().resources.subscribe is False class EchoInput(BaseModel): @@ -95,36 +105,11 @@ def test_discover_advertises_modern_protocol_and_extensions(self): def test_post_mcp_stateless_no_session_header(self): - @injectable() - class PingController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="BlueprintHttp", controllers=[PingController]) - class HttpModule: - pass - - @mcp_app(module=HttpModule, server=ServerConfig(name="blueprint-http", stateless=True)) - class HttpApp: - pass - - app = asyncio.run(McpApplicationFactory.create(HttpApp)) - http_app = app.get_combined_app(stateless=True, json_response=True) - with TestClient(http_app) as client: - resp = client.post( - "/mcp", - headers={**JSON_HEADERS, "Mcp-Method": "tools/call", "Mcp-Name": "echo"}, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - assert resp.status_code == 200 - assert LEGACY_SESSION_HEADER not in {k.title() for k in resp.headers.keys()} - assert resp.json()["result"]["content"][0]["text"] == "ok" + with http_client("auto") as (_, _, client): + response = post(client, "tools/call", {"name": "echo", "arguments": {"value": "ok"}}) + assert response.status_code == 200, response.text + assert "mcp-session-id" not in response.headers + assert decode(response)["result"]["structuredContent"]["value"] == "ok" def test_cors_preflight_includes_mcp_headers(self): headers = cors_preflight_response_headers({"Origin": "https://app.example.com"}) diff --git a/tests/test_mcp20_deprecated.py b/tests/test_mcp20_deprecated.py index 27944b0..6fda893 100644 --- a/tests/test_mcp20_deprecated.py +++ b/tests/test_mcp20_deprecated.py @@ -1,77 +1,11 @@ -"""Deprecated-method policy is the same on every modern entry point.""" - -from __future__ import annotations - -import asyncio -import json -import os -import sys +"""Retired methods are rejected by the official handler and HTTP transport.""" import mcp.types as types import pytest -from mcp import MCPError as McpError -from pydantic import BaseModel, Field -from starlette.testclient import TestClient - -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -from nitrostack import ExecutionContext, injectable, module, tool -from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app +from tests.http_probe import decode, http_client, make_app, post from nitrostack.core.di import DIContainer -from nitrostack.protocol.deprecated import ( - DEPRECATED_MODERN_METHODS, - deprecated_method_message, - rejects_deprecated_method, -) -from nitrostack.protocol.errors import JsonRpcErrorCode -from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION - -INITIALIZE_BODY = { - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, -} - -JSON_HEADERS = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", -} - - -class EchoInput(BaseModel): - value: str = Field(default="") - - -def setup_function() -> None: - DIContainer.reset() - - -def teardown_function() -> None: - DIContainer.reset() - - -def _echo_http_app(era: str): - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name=f"Deprecated{era.title()}", controllers=[EchoController]) - class EchoModule: - pass - - @mcp_app(module=EchoModule, server=ServerConfig(name=f"deprecated-{era}", protocol_era=era)) - class EchoApp: - pass - - app = asyncio.run(McpApplicationFactory.create(EchoApp)) - return app, app.get_combined_app(json_response=True) +from nitrostack.protocol.deprecated import DEPRECATED_MODERN_METHODS, deprecated_method_message, rejects_deprecated_method def test_initialize_is_not_in_retired_method_table(): @@ -81,93 +15,54 @@ def test_initialize_is_not_in_retired_method_table(): @pytest.mark.parametrize("method", sorted(DEPRECATED_MODERN_METHODS)) -def test_retired_methods_error_only_on_modern(method: str): - assert rejects_deprecated_method(method, "modern") is True - assert rejects_deprecated_method(method, "auto") is False - assert rejects_deprecated_method(method, "legacy") is False - - -async def _asgi_json_get(app, headers: dict[str, str]) -> tuple[int, dict]: - status = 0 - chunks: list[bytes] = [] - sent_request = False - finished = asyncio.Event() - - async def receive(): - nonlocal sent_request - if not sent_request: - sent_request = True - return {"type": "http.request", "body": b"", "more_body": False} - await finished.wait() - return {"type": "http.disconnect"} - - async def send(message): - nonlocal status - if message["type"] == "http.response.start": - status = message["status"] - elif message["type"] == "http.response.body": - chunks.append(message.get("body") or b"") - if not message.get("more_body", False): - finished.set() - - scope = { - "type": "http", - "asgi": {"version": "3.0"}, - "http_version": "1.1", - "method": "GET", - "scheme": "http", - "path": "/mcp", - "raw_path": b"/mcp", - "query_string": b"", - "headers": [ - (key.lower().encode("latin-1"), value.encode("latin-1")) - for key, value in headers.items() - ], - "client": ("testclient", 50000), - "server": ("test", 80), - } - - async def _http(): - await app(scope, receive, send) - await asyncio.wait_for(finished.wait(), timeout=2) - return status, json.loads(b"".join(chunks) or b"{}") - - started = asyncio.Event() - messages: asyncio.Queue = asyncio.Queue() - await messages.put({"type": "lifespan.startup"}) - - async def life_receive(): - return await messages.get() - - async def life_send(message): - if message["type"] == "lifespan.startup.complete": - started.set() - - task = asyncio.create_task(app({"type": "lifespan"}, life_receive, life_send)) - await asyncio.wait_for(started.wait(), timeout=2) - try: - return await _http() - finally: - await messages.put({"type": "lifespan.shutdown"}) - await asyncio.wait_for(task, timeout=2) - - -def test_auto_official_handler_hides_tasks_list(monkeypatch): - # Tasks are hidden on the 2026 wire (auto included), so there is no handler. - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) - monkeypatch.delenv("MCP_STATELESS", raising=False) - app, _ = _echo_http_app("auto") +def test_retired_method_policy(method): + assert rejects_deprecated_method(method, "modern") + assert not rejects_deprecated_method(method, "auto") + assert not rejects_deprecated_method(method, "legacy") + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +@pytest.mark.parametrize("method", ["resources/subscribe", "resources/unsubscribe"]) +@pytest.mark.parametrize("modern", [True, False]) +def test_sessionless_resource_subscriptions_are_rejected(era, method, modern): + with http_client(era) as (_, _, client): + response = post(client, method, {"uri": "probe://info"}, modern=modern, + version="2026-07-28" if modern else "2025-06-18") + assert response.status_code == (404 if modern else 200), response.text + error = decode(response)["error"] + assert error["code"] == (types.METHOD_NOT_FOUND if modern or era == "auto" else -32022) + + +@pytest.mark.parametrize("method", ["resources/subscribe", "resources/unsubscribe"]) +def test_legacy_resource_subscription_compatibility(method): + with http_client("legacy") as (_, _, client): + response = post(client, method, {"uri": "probe://info"}, modern=False, version="2025-06-18") + assert response.status_code == 200, response.text + assert decode(response)["result"] == {} + + +@pytest.mark.parametrize("method", ["tasks/list", "tasks/result", "logging/setLevel"]) +def test_modern_retired_methods_reach_official_dispatch(method): + params = {"tasks/list": {}, "tasks/result": {"taskId": "missing"}, "logging/setLevel": {"level": "info"}}[method] + with http_client("modern") as (_, _, client): + response = post(client, method, params) + assert response.status_code == 404, response.text + assert decode(response)["error"]["code"] == types.METHOD_NOT_FOUND + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_sessionless_official_handler_hides_tasks_list(era): + app = make_app(era) try: assert types.ListTasksRequest not in app.mcp_server.request_handlers finally: DIContainer.reset() -def test_modern_official_handler_hides_tasks_list(monkeypatch): - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) - monkeypatch.delenv("MCP_STATELESS", raising=False) - app, _ = _echo_http_app("modern") - try: - assert types.ListTasksRequest not in app.mcp_server.request_handlers - finally: - DIContainer.reset() +@pytest.mark.parametrize("era", ["auto", "modern"]) +@pytest.mark.parametrize("method", ["tasks/list", "tasks/get", "tasks/result", "tasks/cancel", "tasks/update"]) +def test_sessionless_http_hides_task_methods(era, method): + with http_client(era) as (_, _, client): + response = post(client, method, {"taskId": "missing"} if method != "tasks/list" else {}) + assert response.status_code == 404, response.text + assert decode(response)["error"]["code"] == types.METHOD_NOT_FOUND diff --git a/tests/test_mcp20_jsonrpc_wire.py b/tests/test_mcp20_jsonrpc_wire.py index d7c442e..637b235 100644 --- a/tests/test_mcp20_jsonrpc_wire.py +++ b/tests/test_mcp20_jsonrpc_wire.py @@ -1,6 +1,5 @@ """Tests for MCP 2.0 JSON-RPC wire specification.""" -import asyncio import json import pytest @@ -31,6 +30,45 @@ split_params_and_meta, ) from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION +from tests.http_probe import decode, http_client, post + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_official_http_rejects_unknown_protocol_header(era): + with http_client(era) as (_, _, client): + response = post(client, version="1999-01-01") + assert response.status_code == 400, response.text + assert "result" not in decode(response) + + +@pytest.mark.parametrize("era,version", [ + ("auto", "2025-06-18"), ("auto", "2025-11-25"), ("auto", "2026-07-28"), + ("modern", "2026-07-28"), +]) +def test_official_http_accepts_supported_protocol(era, version): + with http_client(era) as (_, _, client): + response = post(client, modern=version == MODERN_PROTOCOL_VERSION, version=version) + assert response.status_code == 200, response.text + assert decode(response)["result"]["tools"][0]["name"] == "echo" + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +@pytest.mark.parametrize("wrong_header", [{"Mcp-Method": "tools/list"}, {"Mcp-Name": "wrong"}]) +def test_official_http_checks_method_and_name_headers(era, wrong_header): + with http_client(era) as (_, _, client): + response = post(client, "tools/call", {"name": "echo", "arguments": {}}, + extra_headers=wrong_header) + assert response.status_code == 400, response.text + assert decode(response)["error"]["code"] == HEADER_BODY_MISMATCH + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_protocol_header_and_envelope_must_agree_in_official_runner(era): + with http_client(era) as (_, _, client): + response = post(client, extra_headers={"MCP-Protocol-Version": "2025-06-18"}) + # The 2025 header selects the SDK's legacy HTTP error mapping. + assert response.status_code == 200, response.text + assert decode(response)["error"]["code"] == HEADER_BODY_MISMATCH class TestErrorCodes: diff --git a/tests/test_mcp20_stateless_http.py b/tests/test_mcp20_stateless_http.py index 154a57e..840a5b3 100644 --- a/tests/test_mcp20_stateless_http.py +++ b/tests/test_mcp20_stateless_http.py @@ -1,13 +1,12 @@ """Tests for MCP 2.0 stateless HTTP transport.""" -import asyncio import json import pytest from nitrostack.protocol.constants import LEGACY_SESSION_HEADER from nitrostack.protocol.contracts import MCP_CACHE_HINT_KEY -from nitrostack.protocol.discovery import DISCOVER_RESULT_TYPE, SERVER_DISCOVER_METHOD, build_discover_result +from nitrostack.protocol.discovery import DISCOVER_RESULT_TYPE, build_discover_result from nitrostack.protocol.jsonrpc import ( HEADER_BODY_MISMATCH, PARSE_ERROR, @@ -16,12 +15,8 @@ parse_jsonrpc_request, validate_header_body_method, ) -from nitrostack.protocol.method_contract import ( - MODERN_METHOD_CONTRACTS, - mcp_name_field, - mcp_name_is_required, -) from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION +from nitrostack.protocol.method_contract import MODERN_METHOD_CONTRACTS from nitrostack.transports.cors import build_cors_headers, cors_preflight_response_headers, resolve_allowed_origin from nitrostack.transports.headers import ( MAX_MCP_PARAM_VALUE_BYTES, @@ -39,6 +34,127 @@ strip_legacy_session_headers_asgi, ) from nitrostack.transports.sse import format_sse_message, sse_notification +from tests.http_probe import decode, envelope, headers, http_client, post + + +@pytest.mark.parametrize("era", ["legacy", "auto", "modern"]) +@pytest.mark.parametrize("json_response", [True, False]) +def test_official_transport_tool_and_resource_roundtrip(era, json_response): + version = "2025-06-18" if era == "legacy" else MODERN_PROTOCOL_VERSION + with http_client(era, json_response=json_response) as (_, http, client): + assert http.state.streamable_http_manager_count == 1 + assert http.state.sessionful == (era == "legacy") + response = post(client, "tools/call", {"name": "echo", "arguments": {"value": "wire"}}, + modern=era != "legacy", version=version) + assert response.status_code == 200, response.text + assert decode(response)["result"]["structuredContent"]["value"] == "wire" + response = post(client, "resources/read", {"uri": "probe://info"}, + modern=era != "legacy", version=version) + assert response.status_code == 200, response.text + assert decode(response)["result"]["contents"][0]["text"] == "probe resource" + if era != "legacy": + assert "mcp-session-id" not in response.headers + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +@pytest.mark.parametrize("body", [b"", b"not-json", b"{}"]) +def test_header_only_ping_cannot_synthesize_a_success(era, body): + with http_client(era) as (_, _, client): + response = client.post("/mcp", content=body, headers=headers("ping")) + assert response.status_code == 400, response.text + assert "result" not in decode(response) + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_unknown_method_is_answered_by_official_dispatch(era): + with http_client(era) as (_, _, client): + response = post(client, "unknown/probe") + assert response.status_code == 404, response.text + assert decode(response)["error"]["code"] == -32601 + + +@pytest.mark.parametrize("era", ["legacy", "auto", "modern"]) +def test_http_health_version_slash_and_era(era): + with http_client(era) as (_, http, client): + health = client.get("/mcp/health") + assert health.status_code == 200 + assert http.state.protocol_era == era + version = client.get("/json/version").json()["Protocol-Version"] + assert version == ("2025-06-18" if era == "legacy" else MODERN_PROTOCOL_VERSION) + response = client.post("/mcp/", json=envelope(), headers=headers()) + if era != "legacy": + assert response.status_code == 200, response.text + assert decode(response)["result"]["tools"][0]["name"] == "echo" + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_allowlisted_cors_applies_once_to_success_and_errors(era, monkeypatch): + monkeypatch.setenv("MCP_CORS_ALLOWED_ORIGINS", "https://allowed.example") + with http_client(era) as (_, _, client): + for method in ["tools/list", "unknown/probe"]: + response = post(client, method, extra_headers={"Origin": "https://allowed.example"}) + assert response.headers.get_list("access-control-allow-origin") == ["https://allowed.example"] + denied = post(client, extra_headers={"Origin": "https://denied.example"}) + assert "access-control-allow-origin" not in denied.headers + preflight = client.options("/mcp", headers={"Origin": "https://allowed.example", + "Access-Control-Request-Method": "POST"}) + assert preflight.status_code == 200 + assert preflight.headers.get_list("access-control-allow-origin") == ["https://allowed.example"] + + +def test_verified_http_identity_and_argument_header_mirroring(): + from nitrostack.auth.jwt import JWTService + from nitrostack.core.di import DIContainer + + with http_client("auto") as (_, _, client): + jwt = JWTService() + DIContainer.get_instance().register_value(JWTService, jwt) + token = jwt.create_token({"sub": "alice", "tenant_id": "one"}) + response = post(client, "tools/call", { + "name": "echo", "arguments": {}, "_meta": {"userId": "mallory"}, + }, extra_headers={"Authorization": "Bearer " + token, "Mcp-Param-value": "from-header"}) + result = decode(response)["result"]["structuredContent"] + assert result["value"] == "from-header" + assert result["mirrored"] == {"value": "from-header"} + assert result["user"] == "alice" + assert result["protocol_version"] == MODERN_PROTOCOL_VERSION + kept = post(client, "tools/call", {"name": "echo", "arguments": {"value": "from-body"}}, + extra_headers={"Mcp-Param-value": "from-header"}) + assert decode(kept)["result"]["structuredContent"]["value"] == "from-body" + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_incoming_session_id_cannot_attach_sessionless_calls(era): + with http_client(era) as (_, _, client): + first = post(client, "tools/call", {"name": "echo", "arguments": {"value": "first"}}, + extra_headers={"Mcp-Session-Id": "forged"}) + second = post(client, "tools/call", {"name": "echo", "arguments": {"value": "second"}}, + extra_headers={"Mcp-Session-Id": "forged"}) + for response, expected in [(first, "first"), (second, "second")]: + assert response.status_code == 200, response.text + assert "mcp-session-id" not in response.headers + assert decode(response)["result"]["structuredContent"]["value"] == expected + + +def test_method_contract_table_still_covers_modern_surface(): + methods = {row.method for row in MODERN_METHOD_CONTRACTS} + assert {"server/discover", "resources/read", "prompts/get", "completion/complete", + "tools/call"} <= methods + assert not any(method.startswith("tasks/") for method in methods) + + +def test_response_echo_uses_supported_request_version(): + echoed = build_mcp_echo_headers( + {"MCP-Protocol-Version": "2025-06-18", "Mcp-Method": "tools/call"}, + protocol_version=MODERN_PROTOCOL_VERSION, + supported_versions={MODERN_PROTOCOL_VERSION, "2025-06-18"}, + ) + assert echoed["MCP-Protocol-Version"] == "2025-06-18" + assert echoed["Mcp-Method"] == "tools/call" + assert handled_protocol_version( + {"MCP-Protocol-Version": "1999-01-01"}, fallback=MODERN_PROTOCOL_VERSION, + supported={MODERN_PROTOCOL_VERSION, "2025-06-18"}, + ) == MODERN_PROTOCOL_VERSION class TestRequestHeaders: @@ -212,89 +328,6 @@ def test_oversized_param_is_detected(self): assert first_oversized_mcp_param({"Mcp-Param-City": "Boston"}) is None - def test_http_header_fills_missing_tool_argument(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> dict: - return { - "value": input.value, - "mirrored": context.mcp_param_headers, - } - - @module(name="McpParamHttp", controllers=[EchoController]) - class McpParamModule: - pass - - @mcp_app(module=McpParamModule, server=ServerConfig(name="mcp-param-http")) - class McpParamApp: - pass - - app = asyncio.run(McpApplicationFactory.create(McpParamApp)) - http_app = app.get_combined_app(json_response=True) - json_headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "Mcp-Method": "tools/call", - "Mcp-Name": "echo", - "MCP-Protocol-Version": "2025-06-18", - } - with TestClient(http_app) as client: - filled = client.post( - "/mcp", - headers={**json_headers, "Mcp-Param-value": "from-header"}, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": "echo", "arguments": {}}, - }, - ) - kept = client.post( - "/mcp", - headers={**json_headers, "Mcp-Param-value": "from-header"}, - json={ - "jsonrpc": "2.0", - "id": 2, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "from-body"}}, - }, - ) - assert filled.status_code == 200, filled.text - filled_body = filled.json()["result"] - filled_payload = filled_body.get("structuredContent") or json.loads( - filled_body["content"][0]["text"] - ) - assert filled_payload["value"] == "from-header" - assert filled_payload["mirrored"]["value"] == "from-header" - assert kept.status_code == 200, kept.text - kept_body = kept.json()["result"] - kept_payload = kept_body.get("structuredContent") or json.loads( - kept_body["content"][0]["text"] - ) - assert kept_payload["value"] == "from-body" - assert kept_payload["mirrored"]["value"] == "from-header" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - class TestCors: def test_cors_allow_methods(self): headers = build_cors_headers() @@ -391,1425 +424,3 @@ def test_sse_task_status_notification(self): {"method": "notifications/tasks/status", "params": {"taskId": "abc"}} ) assert b"notifications/tasks/status" in frame - - -class TestOptionsScopedToMcpPath: - def test_options_mcp_is_204_health_is_forwarded(self): - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - from pydantic import BaseModel, Field - - class EchoInput(BaseModel): - value: str = Field(default="") - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="OptionsHttp", controllers=[EchoController]) - class OptionsModule: - pass - - @mcp_app(module=OptionsModule, server=ServerConfig(name="options-http", stateless=True)) - class OptionsApp: - pass - - app = asyncio.run(McpApplicationFactory.create(OptionsApp)) - http_app = app.get_combined_app(stateless=True, json_response=True) - with TestClient(http_app) as client: - mcp_opt = client.options("/mcp") - health_opt = client.options("/mcp/health") - bare_health = client.options("/health") - oauth_opt = client.options("/oauth/v2/register") - health_preflight = client.options( - "/mcp/health", - headers={ - "Origin": "https://app.example.com", - "Access-Control-Request-Method": "GET", - }, - ) - oauth_preflight = client.options( - "/oauth/v2/register", - headers={ - "Origin": "https://app.example.com", - "Access-Control-Request-Method": "POST", - }, - ) - call = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "Mcp-Method": "tools/call", - "Mcp-Name": "echo", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - assert mcp_opt.status_code == 204 - assert "Access-Control-Allow-Origin" in mcp_opt.headers - assert health_opt.status_code != 204 - assert bare_health.status_code != 204 - assert oauth_opt.status_code != 204 - assert health_preflight.status_code != 204 - assert oauth_preflight.status_code != 204 - assert call.status_code == 200 - assert call.json()["result"]["content"][0]["text"] == "ok" - finally: - DIContainer.reset() - - def test_options_mcp_is_not_sidecar_204_when_cors_off(self): - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - from pydantic import BaseModel, Field - - class EchoInput(BaseModel): - value: str = Field(default="") - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="OptionsCorsOffHttp", controllers=[EchoController]) - class OptionsCorsOffModule: - pass - - @mcp_app(module=OptionsCorsOffModule, server=ServerConfig(name="options-cors-off")) - class OptionsCorsOffApp: - pass - - app = asyncio.run(McpApplicationFactory.create(OptionsCorsOffApp)) - http_app = app.get_combined_app(stateless=True, json_response=True, enable_cors=False) - with TestClient(http_app) as client: - mcp_opt = client.options("/mcp") - assert mcp_opt.status_code != 204 - finally: - DIContainer.reset() - - def test_options_allow_headers_include_2026_mcp_names(self, monkeypatch): - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - from pydantic import BaseModel, Field - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - monkeypatch.setenv("MCP_CORS_ALLOWED_ORIGINS", "https://app.example.com") - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="CorsAllowHeadersHttp", controllers=[EchoController]) - class CorsAllowHeadersModule: - pass - - @mcp_app(module=CorsAllowHeadersModule, server=ServerConfig(name="cors-allow-headers")) - class CorsAllowHeadersApp: - pass - - app = asyncio.run(McpApplicationFactory.create(CorsAllowHeadersApp)) - http_app = app.get_combined_app(json_response=True) - with TestClient(http_app) as client: - allowed = client.options( - "/mcp", - headers={ - "Origin": "https://app.example.com", - "Access-Control-Request-Method": "POST", - "Access-Control-Request-Headers": ( - "Mcp-Name, Mcp-Method, MCP-Protocol-Version, Mcp-Param-City" - ), - }, - ) - unknown = client.options( - "/mcp", - headers={ - "Origin": "https://evil.example", - "Access-Control-Request-Method": "POST", - "Access-Control-Request-Headers": "Mcp-Name", - }, - ) - assert allowed.status_code == 204 - allow = allowed.headers.get("access-control-allow-headers", "") - assert "Mcp-Name" in allow - assert "Mcp-Method" in allow - assert "MCP-Protocol-Version" in allow - assert "Mcp-Param-City" in allow - assert allowed.headers.get("access-control-allow-origin") == "https://app.example.com" - assert unknown.headers.get("access-control-allow-origin") != "https://evil.example" - finally: - DIContainer.reset() - monkeypatch.delenv("MCP_CORS_ALLOWED_ORIGINS", raising=False) - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) - - -class TestProviderToolDiscovery: - def test_provider_tools_are_registered(self): - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.decorators import ToolConfig - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - assert ToolConfig(name="x", description="d", input_schema={}).task_support == "forbidden" - - DIContainer.reset() - try: - @injectable() - class ToolProvider: - @tool(name="from_provider", description="provider", input_schema=EchoInput) - async def from_provider(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="ProviderTools", providers=[ToolProvider], controllers=[]) - class ProviderModule: - pass - - @mcp_app(module=ProviderModule, server=ServerConfig(name="provider-tools")) - class ProviderApp: - pass - - app = asyncio.run(McpApplicationFactory.create(ProviderApp)) - assert "from_provider" in app._tools - assert app._tools["from_provider"].config.task_support == "forbidden" - finally: - DIContainer.reset() - - -class TestNitroMcpProtocolVersionEnv: - def test_modern_era_enables_stateless_without_mcp_stateless(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "2026-07-28") - monkeypatch.delenv("MCP_STATELESS", raising=False) - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="EraHttp", controllers=[EchoController]) - class EraModule: - pass - - @mcp_app(module=EraModule, server=ServerConfig(name="era-http")) - class EraApp: - pass - - app = asyncio.run(McpApplicationFactory.create(EraApp)) - http_app = app.get_combined_app(json_response=True) - with TestClient(http_app) as client: - mcp_opt = client.options("/mcp") - call = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "Mcp-Method": "ping", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "ping", - }, - ) - assert mcp_opt.status_code == 204 - assert call.status_code == 200 - assert call.json()["result"] == {} - assert app.protocol_era == "modern" - inner = getattr(http_app, "app", http_app) - assert inner.state.protocol_era == "modern" - assert inner.state.wire_mode == "reject" - assert inner.state.stateless is True - assert inner.state.http_engine == "sessionless" - assert app.mcp_server.http_engine == "sessionless" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_auto_era_keeps_era_and_does_not_force_stateless(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="AutoEraHttp", controllers=[EchoController]) - class AutoEraModule: - pass - - @mcp_app(module=AutoEraModule, server=ServerConfig(name="auto-era-http")) - class AutoEraApp: - pass - - app = asyncio.run(McpApplicationFactory.create(AutoEraApp)) - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert app.protocol_era == "auto" - assert state.protocol_era == "auto" - assert state.wire_mode == "stateless" - assert state.stateless is True - assert state.http_engine == "sessionless" - assert state.streamable_http_manager_count == 1 - assert app.mcp_server.http_engine == "sessionless" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_legacy_era_uses_sessionful_engine(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "legacy") - monkeypatch.delenv("MCP_STATELESS", raising=False) - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="LegacyEngineHttp", controllers=[EchoController]) - class LegacyEngineModule: - pass - - @mcp_app(module=LegacyEngineModule, server=ServerConfig(name="legacy-engine-http")) - class LegacyEngineApp: - pass - - app = asyncio.run(McpApplicationFactory.create(LegacyEngineApp)) - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert app.protocol_era == "legacy" - assert state.http_engine == "sessionful" - assert state.sessionful is True - assert state.stateless is False - assert state.session_manager.stateless is False - assert app.mcp_server.http_engine == "sessionful" - assert app.mcp_server.sessionful is True - assert "echo" in app._tools - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -def _http_app_state(asgi): - current = asgi - seen: set[int] = set() - while current is not None and id(current) not in seen: - seen.add(id(current)) - state = getattr(current, "state", None) - if state is not None and getattr(state, "protocol_era", None) is not None: - return state - current = getattr(current, "app", None) - raise AssertionError("HTTP app is missing protocol era state") - - -class TestEraSessionfulCoexistence: - def _echo_app(self, monkeypatch, era_value: str): - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", era_value) - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name=f"EraCoexist{era_value.title()}", controllers=[EchoController]) - class EraModule: - pass - - @mcp_app(module=EraModule, server=ServerConfig(name=f"era-coexist-{era_value}")) - class EraApp: - pass - - return asyncio.run(McpApplicationFactory.create(EraApp)) - - def test_auto_stateless_false_does_not_start_sessionful_manager(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - try: - app = self._echo_app(monkeypatch, "auto") - http_app = app.get_combined_app(json_response=True, stateless=False) - state = _http_app_state(http_app) - assert state.http_engine == "sessionless" - assert state.sessionful is False - assert state.session_manager.stateless is True - assert state.streamable_http_manager_count == 1 - assert app.mcp_server.sessionful is False - assert "echo" in app._tools - - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - assert init.status_code == 200, init.text - assert "result" in init.json() - init_headers = {key.lower(): value for key, value in init.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in init_headers - assert not state.session_manager._server_instances - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_modern_stateless_false_still_rejects_initialize(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - try: - app = self._echo_app(monkeypatch, "modern") - http_app = app.get_combined_app(json_response=True, stateless=False) - state = _http_app_state(http_app) - assert state.http_engine == "sessionless" - assert state.sessionful is False - assert "echo" in app._tools - - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - assert init.status_code == 200 - assert init.json()["error"]["code"] == -32601 - init_headers = {key.lower(): value for key, value in init.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in init_headers - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_legacy_initialize_creates_session_without_modern_headers(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - try: - app = self._echo_app(monkeypatch, "legacy") - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert state.http_engine == "sessionful" - assert state.sessionful is True - assert state.streamable_http_manager_count == 1 - assert "echo" in app._tools - - json_headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - } - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers=json_headers, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - session_id = init.headers.get("mcp-session-id") or init.headers.get( - "Mcp-Session-Id" - ) - assert init.status_code == 200, init.text - assert session_id - client.post( - "/mcp", - headers={**json_headers, LEGACY_SESSION_HEADER: session_id}, - json={"jsonrpc": "2.0", "method": "notifications/initialized"}, - ) - call = client.post( - "/mcp", - headers={**json_headers, LEGACY_SESSION_HEADER: session_id}, - json={ - "jsonrpc": "2.0", - "id": 2, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - assert call.status_code == 200, call.text - assert call.json()["result"]["content"][0]["text"] == "ok" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -class TestUnsetEnvDefault: - def _echo_app(self): - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - - class EchoInput(BaseModel): - value: str = Field(default="") - - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="UnsetEnvDefault", controllers=[EchoController]) - class UnsetEnvModule: - pass - - @mcp_app(module=UnsetEnvModule, server=ServerConfig(name="unset-env-default")) - class UnsetEnvApp: - pass - - return asyncio.run(McpApplicationFactory.create(UnsetEnvApp)) - - def test_unset_env_http_is_auto_sessionless(self, monkeypatch, caplog): - import logging - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - try: - with caplog.at_level(logging.INFO, logger="nitrostack.core.app"): - app = self._echo_app() - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert app.protocol_era == "auto" - assert app.protocol_era_source == "default" - assert state.protocol_era == "auto" - assert state.http_engine == "sessionless" - assert "protocol era=auto (source=default)" in caplog.text - - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - assert init.status_code == 200, init.text - assert "result" in init.json() - assert init.json()["result"]["protocolVersion"] == "2025-06-18" - init_headers = {key.lower(): value for key, value in init.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in init_headers - assert not state.session_manager._server_instances - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - os.environ.pop("MCP_STATELESS", None) - - def test_mcp_stateless_false_still_forces_legacy(self, monkeypatch, caplog): - import logging - import os - - from nitrostack.core.di import DIContainer - - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) - monkeypatch.setenv("MCP_STATELESS", "false") - DIContainer.reset() - try: - with caplog.at_level(logging.INFO, logger="nitrostack.core.app"): - app = self._echo_app() - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert app.protocol_era == "legacy" - assert app.protocol_era_source == "mcp_stateless" - assert state.protocol_era == "legacy" - assert state.http_engine == "sessionful" - assert "protocol era=legacy (source=mcp_stateless)" in caplog.text - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - os.environ.pop("MCP_STATELESS", None) - - -class TestAutoEraOneMcpDualClients: - def test_auto_serves_initialize_and_discover_on_one_mcp(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="DualClientHttp", controllers=[EchoController]) - class DualClientModule: - pass - - @mcp_app(module=DualClientModule, server=ServerConfig(name="dual-client-http")) - class DualClientApp: - pass - - app = asyncio.run(McpApplicationFactory.create(DualClientApp)) - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert state.protocol_era == "auto" - assert state.streamable_http_manager_count == 1 - assert state.session_manager.stateless is True - - json_headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - } - modern_headers = { - **json_headers, - "MCP-Protocol-Version": MODERN_PROTOCOL_VERSION, - } - - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers=json_headers, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - initialized = client.post( - "/mcp", - headers=json_headers, - json={"jsonrpc": "2.0", "method": "notifications/initialized"}, - ) - discover = client.post( - "/mcp", - headers={**modern_headers, "Mcp-Method": "server/discover"}, - json={ - "jsonrpc": "2.0", - "id": 2, - "method": "server/discover", - "params": { - "_meta": { - "io.modelcontextprotocol/protocolVersion": MODERN_PROTOCOL_VERSION, - "io.modelcontextprotocol/clientCapabilities": {}, - } - }, - }, - ) - call = client.post( - "/mcp", - headers={ - **json_headers, - "MCP-Protocol-Version": "2025-06-18", - "Mcp-Method": "tools/call", - "Mcp-Name": "echo", - }, - json={ - "jsonrpc": "2.0", - "id": 3, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - - assert init.status_code == 200, init.text - init_body = init.json()["result"] - assert init_body["protocolVersion"] == "2025-06-18" - assert "serverInfo" in init_body - assert "capabilities" in init_body - init_headers = {key.lower(): value for key, value in init.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in init_headers - assert not state.session_manager._server_instances - - assert initialized.status_code == 202, initialized.text - if initialized.content: - assert initialized.json() == {} - initialized_headers = { - key.lower(): value for key, value in initialized.headers.items() - } - assert LEGACY_SESSION_HEADER.lower() not in initialized_headers - assert not state.session_manager._server_instances - - assert discover.status_code == 200, discover.text - discover_body = discover.json()["result"] - assert "capabilities" in discover_body - assert discover_body["resultType"] == "complete" - - assert call.status_code == 200, call.text - assert call.json()["result"]["content"][0]["text"] == "ok" - call_headers = {key.lower(): value for key, value in call.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in call_headers - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -class TestModernEraRejectsLegacyWire: - def test_modern_initialize_is_jsonrpc_error_without_session(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "modern") - monkeypatch.delenv("MCP_STATELESS", raising=False) - - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="ModernRejectHttp", controllers=[EchoController]) - class ModernRejectModule: - pass - - @mcp_app(module=ModernRejectModule, server=ServerConfig(name="modern-reject-http")) - class ModernRejectApp: - pass - - app = asyncio.run(McpApplicationFactory.create(ModernRejectApp)) - http_app = app.get_combined_app(json_response=True) - state = _http_app_state(http_app) - assert state.wire_mode == "reject" - - headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - } - initialize_body = { - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - } - initialized_body = {"jsonrpc": "2.0", "method": "notifications/initialized"} - - with TestClient(http_app) as client: - init = client.post("/mcp", headers=headers, json=initialize_body) - initialized = client.post( - "/mcp", headers=headers, json=initialized_body - ) - - assert init.status_code == 200 - assert init.json()["error"]["code"] == -32601 - assert init.json()["error"]["message"] == "Method not found: initialize" - init_headers = {key.lower(): value for key, value in init.headers.items()} - assert LEGACY_SESSION_HEADER.lower() not in init_headers - assert not state.session_manager._server_instances - - # Rejecting notifications/initialized and an incoming session id on - # the modern wire are separate rules, tracked in #27. - assert initialized.status_code == 202 - assert not state.session_manager._server_instances - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -class TestHealthAdvertisesEra: - def _health_body(self, monkeypatch, era_value: str) -> dict: - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", era_value) - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="HealthEraHttp", controllers=[EchoController]) - class HealthEraModule: - pass - - @mcp_app(module=HealthEraModule, server=ServerConfig(name="health-era-http")) - class HealthEraApp: - pass - - app = asyncio.run(McpApplicationFactory.create(HealthEraApp)) - http_app = app.get_combined_app(json_response=True) - with TestClient(http_app) as client: - response = client.get("/mcp/health") - assert response.status_code == 200 - return response.json() - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_modern_health_reports_modern_era(self, monkeypatch): - body = self._health_body(monkeypatch, "modern") - assert body["status"] == "ok" - assert body["transport"] == "streamable-http" - assert body["protocolEra"] == "modern" - assert body["protocolVersion"] == "2026-07-28" - assert body["statelessCapable"] is True - assert "stateless" in body - assert "uptimeSeconds" in body - - def test_auto_health_reports_auto_not_modern(self, monkeypatch): - body = self._health_body(monkeypatch, "auto") - assert body["protocolEra"] == "auto" - assert body["protocolEra"] != "modern" - assert body["statelessCapable"] is True - - def test_legacy_health_reports_legacy_era(self, monkeypatch): - body = self._health_body(monkeypatch, "legacy") - assert body["protocolEra"] == "legacy" - assert body["protocolVersion"] == "2025-06-18" - assert body["statelessCapable"] is False - - -class TestTrustedProxyPublicUrl: - def _http_app(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.delenv("MCP_STATELESS", raising=False) - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - DIContainer.reset() - - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="ProxyHttp", controllers=[EchoController]) - class ProxyModule: - pass - - @mcp_app(module=ProxyModule, server=ServerConfig(name="proxy-http")) - class ProxyApp: - pass - - app = asyncio.run(McpApplicationFactory.create(ProxyApp)) - return app.get_combined_app(json_response=True) - - def test_health_ignores_untrusted_forwarded_host(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - try: - http_app = self._http_app(monkeypatch) - with TestClient(http_app, client=("8.8.8.8", 4321)) as client: - response = client.get( - "/mcp/health", - headers={ - "Host": "internal:3000", - "X-Forwarded-Host": "mcp.example.com", - "X-Forwarded-Proto": "https", - }, - ) - docs = client.get( - "/", - headers={ - "Host": "internal:3000", - "X-Forwarded-Host": "mcp.example.com", - "X-Forwarded-Proto": "https", - }, - ) - oauth = client.get( - "/.well-known/oauth-authorization-server", - headers={ - "Host": "internal:3000", - "X-Forwarded-Host": "mcp.example.com", - "X-Forwarded-Proto": "https", - }, - ) - assert response.status_code == 200 - assert "mcp.example.com" not in response.json()["publicUrl"] - assert "mcp.example.com" not in docs.text - assert "mcp.example.com" not in oauth.json()["error_description"] - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - def test_health_honors_trusted_forwarded_host(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.5") - try: - http_app = self._http_app(monkeypatch) - with TestClient(http_app, client=("10.0.0.5", 4321)) as client: - headers = { - "Host": "internal:3000", - "X-Forwarded-Host": "mcp.example.com", - "X-Forwarded-Proto": "https", - } - response = client.get("/mcp/health", headers=headers) - docs = client.get("/", headers=headers) - oauth = client.get("/.well-known/oauth-authorization-server", headers=headers) - version = client.get("/json/version", headers=headers) - assert response.status_code == 200 - assert response.json()["publicUrl"] == "https://mcp.example.com/mcp" - assert "https://mcp.example.com/mcp" in docs.text - assert "https://mcp.example.com/mcp" in oauth.json()["error_description"] - assert version.json()["publicUrl"] == "https://mcp.example.com/mcp" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - os.environ.pop("TRUSTED_PROXIES", None) - - -class TestHeaderTriggeredPing: - def _http_app(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.delenv("MCP_STATELESS", raising=False) - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - DIContainer.reset() - - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="HeaderPingHttp", controllers=[EchoController]) - class HeaderPingModule: - pass - - @mcp_app(module=HeaderPingModule, server=ServerConfig(name="header-ping-http")) - class HeaderPingApp: - pass - - app = asyncio.run(McpApplicationFactory.create(HeaderPingApp)) - return app.get_combined_app(json_response=True) - -class TestEnvelopeOnHandlerContext: - def test_tool_reads_trace_and_protocol_version_not_spoofed_user(self, monkeypatch): - import os - - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EmptyInput(BaseModel): - pass - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - try: - @injectable() - class EnvelopeController: - @tool(name="envelope", description="read envelope", input_schema=EmptyInput) - async def envelope(self, input: EmptyInput, context: ExecutionContext) -> dict: - return { - "protocolVersion": context.protocol_version, - "trace": context.rpc_meta.trace if context.rpc_meta else None, - "user": context.user, - } - - @module(name="EnvelopeHttp", controllers=[EnvelopeController]) - class EnvelopeModule: - pass - - @mcp_app(module=EnvelopeModule, server=ServerConfig(name="envelope-http")) - class EnvelopeApp: - pass - - app = asyncio.run(McpApplicationFactory.create(EnvelopeApp)) - http_app = app.get_combined_app(json_response=True) - headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "MCP-Protocol-Version": "2025-06-18", - "Mcp-Method": "tools/call", - "Mcp-Name": "envelope", - } - with TestClient(http_app) as client: - response = client.post( - "/mcp", - headers=headers, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": "envelope", - "arguments": {}, - "_meta": { - "trace": {"id": "http-span"}, - "userId": "spoofed", - }, - }, - }, - ) - - assert response.status_code == 200, response.text - result = response.json()["result"] - body = result.get("structuredContent") or json.loads(result["content"][0]["text"]) - assert body["protocolVersion"] == "2025-06-18" - assert body["trace"] == {"id": "http-span"} - assert body["user"] is None - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -class TestAuthFromEnvelopeAndHeaders: - def test_tool_user_comes_from_header_jwt_not_unsigned_meta(self, monkeypatch): - import os - - from pydantic import BaseModel - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.auth.jwt import JWTService - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EmptyInput(BaseModel): - pass - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - jwt = JWTService() - DIContainer.get_instance().register_value(JWTService, jwt) - token = jwt.create_token({"sub": "alice", "tenant_id": "acme"}) - try: - @injectable() - class AuthController: - @tool(name="whoami", description="identity", input_schema=EmptyInput) - async def whoami(self, input: EmptyInput, context: ExecutionContext) -> dict: - return {"user": context.user} - - @module(name="AuthHttp", controllers=[AuthController]) - class AuthModule: - pass - - @mcp_app(module=AuthModule, server=ServerConfig(name="auth-http")) - class AuthApp: - pass - - app = asyncio.run(McpApplicationFactory.create(AuthApp)) - http_app = app.get_combined_app(json_response=True) - with TestClient(http_app) as client: - response = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "MCP-Protocol-Version": "2025-06-18", - "Mcp-Method": "tools/call", - "Mcp-Name": "whoami", - "Authorization": f"Bearer {token}", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": "whoami", - "arguments": {}, - "_meta": { - "userId": "eve", - "tenantId": "evil", - "io.modelcontextprotocol/auth": {"userId": "mallory"}, - }, - }, - }, - ) - - assert response.status_code == 200, response.text - result = response.json()["result"] - body = result.get("structuredContent") or json.loads(result["content"][0]["text"]) - assert body["user"] == "alice" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - -class TestIncomingSessionIdRejection: - def _echo_app(self, monkeypatch, era_value: str): - import os - - from pydantic import BaseModel, Field - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", era_value) - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="SessionRejectHttp", controllers=[EchoController]) - class SessionRejectModule: - pass - - @mcp_app(module=SessionRejectModule, server=ServerConfig(name="session-reject-http")) - class SessionRejectApp: - pass - - return asyncio.run(McpApplicationFactory.create(SessionRejectApp)) - - def test_legacy_post_accepts_session_id_after_initialize(self, monkeypatch): - import os - - from starlette.testclient import TestClient - - from nitrostack.core.di import DIContainer - - try: - app = self._echo_app(monkeypatch, "legacy") - http_app = app.get_combined_app(json_response=True) - json_headers = { - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - } - with TestClient(http_app) as client: - init = client.post( - "/mcp", - headers=json_headers, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", - "capabilities": {}, - "clientInfo": {"name": "legacy-client", "version": "1.0"}, - }, - }, - ) - session_id = init.headers.get("mcp-session-id") or init.headers.get( - "Mcp-Session-Id" - ) - assert init.status_code == 200, init.text - assert session_id - client.post( - "/mcp", - headers={**json_headers, LEGACY_SESSION_HEADER: session_id}, - json={"jsonrpc": "2.0", "method": "notifications/initialized"}, - ) - call = client.post( - "/mcp", - headers={ - **json_headers, - LEGACY_SESSION_HEADER: session_id, - "Mcp-Method": "tools/call", - "Mcp-Name": "echo", - }, - json={ - "jsonrpc": "2.0", - "id": 2, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - assert call.status_code == 200, call.text - assert call.json()["result"]["content"][0]["text"] == "ok" - finally: - DIContainer.reset() - os.environ.pop("NITRO_MCP_PROTOCOL_VERSION", None) - - - - -class TestSep2243AllModernMethods: - def test_table_covers_modern_surface(self): - methods = {row.method for row in MODERN_METHOD_CONTRACTS} - assert "server/discover" in methods - assert "resources/read" in methods - assert "prompts/get" in methods - assert "completion/complete" in methods - # Tasks are hidden on the 2026 wire until tasks/update exists. - assert not any(m.startswith("tasks/") for m in methods) - assert "tools/call" in methods - - -class TestResponseEchoHeaders: - def test_echo_uses_supported_request_version(self): - headers = build_mcp_echo_headers( - {"MCP-Protocol-Version": "2025-06-18", "Mcp-Method": "tools/call"}, - protocol_version=MODERN_PROTOCOL_VERSION, - supported_versions={"2026-07-28", "2025-06-18"}, - ) - assert headers["MCP-Protocol-Version"] == "2025-06-18" - assert headers["Mcp-Method"] == "tools/call" - assert handled_protocol_version( - {"MCP-Protocol-Version": "1999-01-01"}, - fallback=MODERN_PROTOCOL_VERSION, - supported={"2026-07-28", "2025-06-18"}, - ) == MODERN_PROTOCOL_VERSION - -class TestCorsExposeHeaders: - def test_http_mcp_response_exposes_echo_headers_not_session_id(self, monkeypatch): - from pydantic import BaseModel, Field - from starlette.testclient import TestClient - - from nitrostack import ExecutionContext, injectable, module, tool - from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app - from nitrostack.core.di import DIContainer - - class EchoInput(BaseModel): - value: str = Field(default="") - - monkeypatch.setenv("NITRO_MCP_PROTOCOL_VERSION", "auto") - monkeypatch.delenv("MCP_STATELESS", raising=False) - DIContainer.reset() - try: - @injectable() - class EchoController: - @tool(name="echo", description="echo", input_schema=EchoInput) - async def echo(self, input: EchoInput, context: ExecutionContext) -> str: - return input.value - - @module(name="CorsExposeHttp", controllers=[EchoController]) - class CorsExposeModule: - pass - - @mcp_app(module=CorsExposeModule, server=ServerConfig(name="cors-expose")) - class CorsExposeApp: - pass - - app = asyncio.run(McpApplicationFactory.create(CorsExposeApp)) - http_app = app.get_combined_app(json_response=True) - origin = "https://app.example.com" - with TestClient(http_app) as client: - response = client.post( - "/mcp", - headers={ - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - "Origin": origin, - "MCP-Protocol-Version": "2025-06-18", - "Mcp-Method": "tools/call", - "Mcp-Name": "echo", - }, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": "echo", "arguments": {"value": "ok"}}, - }, - ) - preflight = client.options( - "/mcp", - headers={ - "Origin": origin, - "Access-Control-Request-Method": "POST", - }, - ) - assert response.status_code == 200, response.text - expose = response.headers.get("access-control-expose-headers", "") - assert "MCP-Protocol-Version" in expose - assert "Mcp-Method" in expose - assert "Mcp-Session-Id" not in expose - assert preflight.status_code == 204 - preflight_expose = preflight.headers.get("access-control-expose-headers", "") - assert "MCP-Protocol-Version" in preflight_expose - assert "Mcp-Method" in preflight_expose - assert "Mcp-Session-Id" not in preflight_expose - finally: - DIContainer.reset() - monkeypatch.delenv("NITRO_MCP_PROTOCOL_VERSION", raising=False) diff --git a/tests/test_mcp20_stdio.py b/tests/test_mcp20_stdio.py index 94a4b23..333be84 100644 --- a/tests/test_mcp20_stdio.py +++ b/tests/test_mcp20_stdio.py @@ -135,8 +135,9 @@ def test_modern_stdio_rejects_initialize(monkeypatch): }, ) ) + # Reject the handshake before it can select the legacy connection style. assert "error" in payload - assert "initialize" in payload["error"]["message"] + assert payload["error"]["code"] == -32601 def test_legacy_stdio_still_answers_initialize(monkeypatch): diff --git a/tests/test_mcp22_official_http_owner.py b/tests/test_mcp22_official_http_owner.py index a552078..7e12f8c 100644 --- a/tests/test_mcp22_official_http_owner.py +++ b/tests/test_mcp22_official_http_owner.py @@ -5,10 +5,13 @@ from pathlib import Path import httpx +import nitrostack from nitrostack import module from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app +SDK_ROOT = Path(nitrostack.__file__).resolve().parent + @module(name="issue22_probe", controllers=[]) class Issue22ProbeModule: @@ -44,17 +47,19 @@ async def run(): def test_combined_app_has_no_jsonrpc_sidecar_call(): - source = Path("nitrostack/core/app.py").read_text(encoding="utf-8") + source = (SDK_ROOT / "core" / "app.py").read_text(encoding="utf-8") assert "wrap_stateless_transport" not in source assert "return http_app" in source def test_sidecar_modules_and_ping_helpers_are_not_production_code(): - assert not Path("nitrostack/transports/middleware.py").exists() - assert not Path("nitrostack/transports/dispatch.py").exists() + assert not (SDK_ROOT / "transports" / "middleware.py").exists() + assert not (SDK_ROOT / "transports" / "dispatch.py").exists() + sources = list(SDK_ROOT.rglob("*.py")) + assert sources, f"No production Python sources found in {SDK_ROOT}" production = "\n".join( path.read_text(encoding="utf-8") - for path in Path("nitrostack").rglob("*.py") + for path in sources ) for symbol in ( "wrap_stateless_transport", @@ -268,9 +273,10 @@ def test_cors_honours_the_configured_origin_allowlist(monkeypatch): headers={**_HEADERS, "Origin": "https://app.example.com"}, json=_rpc("ping"), ) - assert allowed.status_code == 204 + assert allowed.status_code == 200 assert allowed.headers["access-control-allow-origin"] == "https://app.example.com" assert "mcp-param-city" in allowed.headers["access-control-allow-headers"].lower() + assert denied.status_code == 400 assert denied.headers.get("access-control-allow-origin") != "*" assert denied.headers.get("access-control-allow-origin") != "https://evil.example.com" assert post.headers["access-control-allow-origin"] == "https://app.example.com" @@ -283,10 +289,8 @@ def test_cors_exposes_session_id_only_on_the_legacy_wire(monkeypatch): for era, exposed in (("legacy", True), ("auto", False)): app = _build(era) with TestClient(app) as client: - response = client.options( - "/mcp", - headers={"Origin": "http://x.test", "Access-Control-Request-Method": "POST"}, - ) + response = client.get("/mcp/health", headers={"Origin": "http://x.test"}) + assert response.status_code == 200 expose = response.headers.get("access-control-expose-headers", "") assert ("Mcp-Session-Id" in expose) is exposed, era diff --git a/tests/test_mcp23_http_factory.py b/tests/test_mcp23_http_factory.py new file mode 100644 index 0000000..f304d93 --- /dev/null +++ b/tests/test_mcp23_http_factory.py @@ -0,0 +1,391 @@ +"""Regression tests for Issue #23: single CORS layer, `resources.subscribe` +agreement between discover and real capability negotiation, era-driven +`/json/version`, and no private `mcp.server.runner` imports in stdio. +""" + +from __future__ import annotations + +import asyncio +import ast +import json +import os +import sys +from contextlib import asynccontextmanager +from pathlib import Path + +import anyio +import httpx +import pytest +import nitrostack +from mcp.shared.memory import create_client_server_memory_streams +from mcp.shared.message import SessionMessage +from mcp_types import PROTOCOL_VERSION_META_KEY, jsonrpc_message_adapter +from pydantic import BaseModel, Field + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from nitrostack import DIContainer, ExecutionContext, injectable, module, tool +from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app +from nitrostack.protocol.version import LEGACY_PROTOCOL_VERSION, MODERN_PROTOCOL_VERSION +from nitrostack.transports.stdio import serve_stdio_streams +from nitrostack.runtime.request_ctx import stdio_connection_id +from tests.http_probe import decode, http_client, make_app, post + +CLIENT_META = { + "io.modelcontextprotocol/clientInfo": {"name": "issue23-test", "version": "1.0"}, + "io.modelcontextprotocol/clientCapabilities": {}, +} +SDK_ROOT = Path(nitrostack.__file__).resolve().parent + + +def setup_function() -> None: + DIContainer.reset() + + +def teardown_function() -> None: + DIContainer.reset() + + +class EchoInput(BaseModel): + value: str = Field(default="ok") + + +def _era_app(era: str, name: str): + @injectable() + class EchoController: + @tool(name="echo", description="echo", input_schema=EchoInput) + async def echo(self, input: EchoInput, context: ExecutionContext) -> str: + return input.value + + @module(name=f"issue23_{name}", controllers=[EchoController]) + class EchoModule: + pass + + @mcp_app(module=EchoModule, server=ServerConfig(name=name, protocol_era=era)) + class EchoApp: + pass + + return asyncio.run(McpApplicationFactory.create(EchoApp)) + + +async def _send(http_app, method: str, path: str, *, headers: dict | None = None, payload: dict | None = None): + transport = httpx.ASGITransport(app=http_app) + async with http_app.router.lifespan_context(http_app): + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + return await client.request( + method, + path, + headers=headers or {}, + content=json.dumps(payload) if payload is not None else None, + ) + + +# 1. OPTIONS /mcp: exactly one CORS policy, not two fighting layers. +def test_single_cors_layer_on_options_mcp(): + app = _era_app("auto", "issue23-cors") + http_app = app.get_combined_app(json_response=True) + response = asyncio.run( + _send( + http_app, + "OPTIONS", + "/mcp", + headers={ + "Origin": "http://example.com", + "Access-Control-Request-Method": "POST", + }, + ) + ) + allow_origin = response.headers.get_list("access-control-allow-origin") + assert len(allow_origin) == 1 + + +# 2. server/discover and NitroStackMcpServer.get_capabilities must never disagree. +@pytest.mark.parametrize("era", ["legacy", "auto", "modern"]) +def test_discover_and_capabilities_agree_on_resources_subscribe(era): + app = _era_app(era, f"issue23-subscribe-{era}") + discover = app.handle_server_discover() + caps = app.mcp_server.get_capabilities() + assert discover["capabilities"]["resources"]["subscribe"] == caps.resources.subscribe + assert caps.resources.subscribe == (era == "legacy") + extra_server = app.create_configured_mcp_server() + assert extra_server.protocol_era == era + assert extra_server.get_capabilities().resources.subscribe == (era == "legacy") + if era == "auto": + assert app.handle_sessionless_initialize()["capabilities"]["resources"]["subscribe"] is False + + +# 3. No runner implementation imports or private MCP modules in stdio.py. +def test_stdio_module_has_no_private_mcp_runner_import(): + source = (SDK_ROOT / "transports" / "stdio.py").read_text(encoding="utf-8") + tree = ast.parse(source) + modules = [] + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + modules.append(node.module) + if node.module == "mcp.server": + modules.extend(f"{node.module}.{alias.name}" for alias in node.names) + elif isinstance(node, ast.Import): + modules.extend(alias.name for alias in node.names) + mcp_modules = [name for name in modules if name.split(".")[0] == "mcp"] + assert mcp_modules + assert not [name for name in mcp_modules if name == "mcp.server.runner" or + name.startswith("mcp.server.runner.") or + any(part.startswith("_") for part in name.split("."))] + + +# 4. /json/version tracks era instead of a hardcoded 2025 date. +@pytest.mark.parametrize( + "era, expected_version", + [ + ("legacy", LEGACY_PROTOCOL_VERSION), + ("auto", MODERN_PROTOCOL_VERSION), + ("modern", MODERN_PROTOCOL_VERSION), + ], +) +def test_json_version_matches_era(era, expected_version): + app = _era_app(era, f"issue23-jsonversion-{era}") + http_app = app.get_combined_app(json_response=True) + response = asyncio.run(_send(http_app, "GET", "/json/version")) + assert response.status_code == 200 + assert response.json()["Protocol-Version"] == expected_version + + +# 5. Modern-era stdio boots through the public Server.run entry point. +def test_modern_stdio_serves_tool_call(): + app = _era_app("modern", "issue23-modern-stdio") + + async def roundtrip(): + async with create_client_server_memory_streams() as (client, server): + client_read, client_write = client + server_read, server_write = server + + async def run_server(): + await serve_stdio_streams(app.mcp_server, server_read, server_write, "modern") + + async with anyio.create_task_group() as tg: + tg.start_soon(run_server) + await client_write.send( + SessionMessage( + jsonrpc_message_adapter.validate_python( + { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "echo", + "arguments": {"value": "hi"}, + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META}, + }, + } + ) + ) + ) + response = await asyncio.wait_for(client_read.receive(), timeout=2) + tg.cancel_scope.cancel() + return response.message.model_dump(by_alias=True, mode="json") + + payload = asyncio.run(roundtrip()) + assert "result" in payload + assert payload["result"]["content"][0]["text"] == "hi" + + +# 5b. Official notification handling must keep the connection usable, +# proven by a normal request still working +# on the same stream right after a notification. +def test_modern_stdio_notification_does_not_break_the_stream(): + app = _era_app("modern", "issue23-modern-notify") + + async def roundtrip(): + async with create_client_server_memory_streams() as (client, server): + client_read, client_write = client + server_read, server_write = server + + async def run_server(): + await serve_stdio_streams(app.mcp_server, server_read, server_write, "modern") + + async with anyio.create_task_group() as tg: + tg.start_soon(run_server) + await client_write.send( + SessionMessage( + jsonrpc_message_adapter.validate_python( + { + "jsonrpc": "2.0", + "method": "notifications/cancelled", + "params": { + "requestId": 999, + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION}, + }, + } + ) + ) + ) + await client_write.send( + SessionMessage( + jsonrpc_message_adapter.validate_python( + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "echo", + "arguments": {"value": "still-alive"}, + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META}, + }, + } + ) + ) + ) + response = await asyncio.wait_for(client_read.receive(), timeout=2) + tg.cancel_scope.cancel() + return response.message.model_dump(by_alias=True, mode="json") + + payload = asyncio.run(roundtrip()) + assert payload["result"]["content"][0]["text"] == "still-alive" + + +async def _stdio_exchange(server, era, messages, expected_responses): + """Keep one stream alive through handshake, notifications, and requests.""" + async with create_client_server_memory_streams() as (client, streams): + client_read, client_write = client + server_read, server_write = streams + async with anyio.create_task_group() as group: + group.start_soon(serve_stdio_streams, server, server_read, server_write, era) + for message in messages: + await client_write.send(SessionMessage(jsonrpc_message_adapter.validate_python(message))) + responses = [] + while len(responses) < expected_responses: + frame = await asyncio.wait_for(client_read.receive(), timeout=3) + data = frame.message.model_dump(by_alias=True, mode="json") + if "id" in data: + responses.append(data) + group.cancel_scope.cancel() + return responses + + +@pytest.mark.parametrize("era,style", [ + ("legacy", "legacy"), ("auto", "legacy"), ("auto", "modern"), ("modern", "modern"), +]) +def test_stdio_era_roundtrip_and_initialization_capabilities(era, style): + app = make_app(era) + messages = [] + if style == "legacy": + messages.extend([ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { + "protocolVersion": LEGACY_PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "probe", "version": "1"}, + }}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + ]) + params = {"name": "echo", "arguments": {"value": "stdio"}} + if style == "modern": + params["_meta"] = {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META} + messages.append({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": params}) + responses = asyncio.run(_stdio_exchange(app.mcp_server, era, messages, 2 if style == "legacy" else 1)) + if style == "legacy": + assert responses[0]["result"]["capabilities"]["resources"]["subscribe"] == (era == "legacy") + assert responses[-1]["result"]["structuredContent"]["value"] == "stdio" + + +def test_modern_stdio_rejects_initialize_and_then_serves_a_request(): + app = make_app("modern") + messages = [ + {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { + "protocolVersion": LEGACY_PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "probe", "version": "1"}, + }}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": { + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META}, + }}, + ] + responses = asyncio.run(_stdio_exchange(app.mcp_server, "modern", messages, 2)) + assert responses[0]["error"]["code"] == -32601 + assert responses[1]["result"]["tools"][0]["name"] == "echo" + + +def test_legacy_stdio_rejects_modern_request_without_poisoning_initialization(): + app = make_app("legacy") + messages = [ + {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": { + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META}, + }}, + {"jsonrpc": "2.0", "id": 2, "method": "initialize", "params": { + "protocolVersion": LEGACY_PROTOCOL_VERSION, "capabilities": {}, + "clientInfo": {"name": "probe", "version": "1"}, + }}, + {"jsonrpc": "2.0", "method": "notifications/initialized"}, + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { + "name": "echo", "arguments": {"value": "legacy-after-rejection"}, + }}, + ] + responses = asyncio.run(_stdio_exchange(app.mcp_server, "legacy", messages, 3)) + assert responses[0]["error"]["code"] == -32600 + assert responses[1]["result"]["protocolVersion"] == LEGACY_PROTOCOL_VERSION + assert responses[2]["result"]["structuredContent"]["value"] == "legacy-after-rejection" + + +@pytest.mark.parametrize("meta", [None, [], {PROTOCOL_VERSION_META_KEY: "1900-01-01"}]) +def test_modern_stdio_invalid_version_does_not_select_legacy_connection(meta): + app = make_app("modern") + messages = [ + {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {"_meta": meta}}, + {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": { + "_meta": {PROTOCOL_VERSION_META_KEY: MODERN_PROTOCOL_VERSION, **CLIENT_META}, + }}, + ] + responses = asyncio.run(_stdio_exchange(app.mcp_server, "modern", messages, 2)) + assert "error" in responses[0] + assert responses[0]["error"]["data"]["supportedVersions"] == [MODERN_PROTOCOL_VERSION] + assert responses[1]["result"]["tools"][0]["name"] == "echo" + + +@pytest.mark.parametrize("era", ["legacy", "auto", "modern"]) +def test_stdio_public_runner_closes_lifespan_and_stream_on_eof(era): + app = make_app(era) + events = [] + + @asynccontextmanager + async def lifespan(server): + events.append("start") + try: + yield {} + finally: + events.append("stop") + + app.mcp_server.lifespan = lifespan + + async def scenario(): + async with create_client_server_memory_streams() as (client, streams): + client_read, client_write = client + + async def serve(): + await serve_stdio_streams(app.mcp_server, *streams, era) + assert stdio_connection_id.get() is None + + task = asyncio.create_task(serve()) + try: + await client_write.aclose() + await asyncio.wait_for(task, timeout=3) + with pytest.raises(anyio.EndOfStream): + await asyncio.wait_for(client_read.receive(), timeout=3) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + assert stdio_connection_id.get() is None + + asyncio.run(scenario()) + assert events == ["start", "stop"] + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_http_initialize_respects_era(era): + with http_client(era) as (_, _, client): + response = post(client, "initialize", {"protocolVersion": LEGACY_PROTOCOL_VERSION, + "capabilities": {}, "clientInfo": {"name": "probe", "version": "1"}}, + modern=False, version=LEGACY_PROTOCOL_VERSION) + assert response.status_code == 200, response.text + data = decode(response) + if era == "modern": + assert data["error"]["code"] == -32601 + else: + assert data["result"]["capabilities"]["resources"]["subscribe"] is False diff --git a/tests/test_mcp25_hide_tasks.py b/tests/test_mcp25_hide_tasks.py index 3e35d15..0043e7d 100644 --- a/tests/test_mcp25_hide_tasks.py +++ b/tests/test_mcp25_hide_tasks.py @@ -177,8 +177,15 @@ def _tools_list_http(era: str) -> dict: """Run a real tools/list over the mounted /mcp app and return the tool named ``taskable``.""" response = _http_request( _app(era), - {"jsonrpc": "2.0", "id": 3, "method": "tools/list", "params": {}}, - extra_headers={"Mcp-Method": "tools/list"} if era == "modern" else None, + { + "jsonrpc": "2.0", "id": 3, "method": "tools/list", + "params": {"_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": {"name": "hide-tasks-test", "version": "1"}, + "io.modelcontextprotocol/clientCapabilities": {}, + }}, + }, + extra_headers={"Mcp-Method": "tools/list", "MCP-Protocol-Version": "2026-07-28"}, ) assert response.status_code == 200, response.text tools = response.json()["result"]["tools"] diff --git a/tests/test_transports.py b/tests/test_transports.py index ad88163..e976123 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -165,9 +165,11 @@ def test_http_health_and_cors(): preflight = client.options( "/mcp", - headers={"Origin": "http://example.com", "Access-Control-Request-Method": "POST"}, + headers={"Origin": "http://example.com", "Access-Control-Request-Method": "POST", + "Access-Control-Request-Headers": + "Mcp-Session-Id, Mcp-Name, Mcp-Method, MCP-Protocol-Version"}, ) - assert preflight.status_code == 204 + assert preflight.status_code == 200 assert preflight.headers.get("access-control-allow-origin") == "*" allow_headers = preflight.headers.get("access-control-allow-headers", "") assert "Mcp-Session-Id" in allow_headers