fix: unify HTTP factory and migrate tests to official MCP transport

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,14 +1062,16 @@ 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:
                 raise ResourceNotFoundError(str(params.uri))
             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]):

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(

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:

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.

nitrostack/transports/http.py

@@ -34,13 +34,15 @@
 
 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
 
 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,
             ),
         )
 

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)