diff --git a/nitrostack/__init__.py b/nitrostack/__init__.py index 9f1c3d5..d918872 100644 --- a/nitrostack/__init__.py +++ b/nitrostack/__init__.py @@ -135,7 +135,6 @@ from nitrostack.tasks import InMemoryTaskStore, TaskAccessContext, TaskStore from nitrostack.tasks.authorization import check_task_access, extract_task_access_context from nitrostack.runtime import StatelessInvariants, assert_stateless_headers -from nitrostack.transports import wrap_stateless_transport, StatelessIngressPipeline __all__ = [ @@ -242,6 +241,4 @@ "extract_task_access_context", "StatelessInvariants", "assert_stateless_headers", - "wrap_stateless_transport", - "StatelessIngressPipeline", ] diff --git a/nitrostack/core/app.py b/nitrostack/core/app.py index a0bc83e..e857561 100644 --- a/nitrostack/core/app.py +++ b/nitrostack/core/app.py @@ -789,10 +789,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 = getattr(self, "protocol_era", None) != "modern" 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(), @@ -1769,37 +1773,10 @@ def get_combined_app( http_engine=http_engine, ) - if http_engine == "sessionless": - from nitrostack.transports.middleware import wrap_stateless_transport - - def _discover_handler(_request): - return self.handle_server_discover() - - def _initialize_handler(request): - params = getattr(request, "params", None) or {} - requested = params.get("protocolVersion") if isinstance(params, dict) else None - return self.handle_sessionless_initialize( - requested if isinstance(requested, str) else None - ) - - http_app = wrap_stateless_transport( - http_app, - server_name=self.server_config.name, - server_version=self.server_config.version, - protocol_version=protocol_version_for_era(era, self.server_config.protocol_version), - advertise_tasks=self._advertise_tasks_extension(), - advertise_app=any( - getattr(entry, "component", None) is not None - for entry in getattr(self, "_tools", {}).values() - ), - custom_extensions=self._custom_extensions(), - wire_mode=wire_mode, - protocol_era=era, - enable_cors=enable_cors, - discover_handler=_discover_handler, - initialize_handler=_initialize_handler, - ) - + # The official MCP SDK owns the complete Streamable HTTP request lifecycle + # mounted by ``build_http_app``. Do not wrap it in a second JSON-RPC + # dispatcher: a sidecar would answer ping/initialize/tools/call itself and + # could diverge from the protocol and session behaviour of the SDK. return http_app async def _run_stdio(self) -> None: diff --git a/nitrostack/core/mcp_server.py b/nitrostack/core/mcp_server.py index b383ff2..926bcc4 100644 --- a/nitrostack/core/mcp_server.py +++ b/nitrostack/core/mcp_server.py @@ -33,6 +33,7 @@ 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[str] = None 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. @@ -67,7 +68,10 @@ def get_capabilities( ) if caps.resources is not None: - caps.resources.subscribe = True + # `resources/subscribe` is only rejected on `modern` + # (see `rejects_deprecated_method`); `legacy` and `auto` both serve + # it live, so this must track era, not the sessionful/sessionless split. + caps.resources.subscribe = self.protocol_era != "modern" if self.has_task_support: caps.tasks = types.ServerTasksCapability( diff --git a/nitrostack/protocol/__init__.py b/nitrostack/protocol/__init__.py index 86dad4b..1aaf760 100644 --- a/nitrostack/protocol/__init__.py +++ b/nitrostack/protocol/__init__.py @@ -32,7 +32,6 @@ from nitrostack.protocol.extensions import MCPExtensionId from nitrostack.protocol.jsonrpc import ( JsonRpcWireError, - build_ping_response, build_tool_error_result, jsonrpc_error, jsonrpc_success, @@ -166,7 +165,6 @@ "build_sessionless_initialize_result", "build_discover_result", "parse_jsonrpc_request", - "build_ping_response", "jsonrpc_success", "jsonrpc_error", "build_tool_error_result", 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/jsonrpc.py b/nitrostack/protocol/jsonrpc.py index 24239e2..43c7e87 100644 --- a/nitrostack/protocol/jsonrpc.py +++ b/nitrostack/protocol/jsonrpc.py @@ -239,11 +239,6 @@ def jsonrpc_error(request_id: Any, code: int, message: str, data: Any = None) -> return {"jsonrpc": JSONRPC_VERSION, "id": request_id, "error": error} -def build_ping_response(request_id: Any) -> dict[str, Any]: - """Ping fast path.""" - return jsonrpc_success(request_id, {}) - - def build_tool_error_result(message: str, *, text_type: str = "text") -> dict[str, Any]: """ Tool business failure — JSON-RPC success with isError: true. diff --git a/nitrostack/transports/__init__.py b/nitrostack/transports/__init__.py index 6cf62b1..32d0742 100644 --- a/nitrostack/transports/__init__.py +++ b/nitrostack/transports/__init__.py @@ -1,12 +1,4 @@ -"""MCP transport adapters (stdio, stateless HTTP).""" - -from nitrostack.transports.dispatch import ( - DispatchStage, - IngressContext, - StatelessIngressPipeline, - is_task_wire_interception, -) -from nitrostack.transports.middleware import StatelessTransportMiddleware, wrap_stateless_transport +"""MCP transport adapters (stdio and the official Streamable HTTP engine).""" from nitrostack.transports.sse import format_sse_message, sse_connect_headers, sse_notification from nitrostack.transports.subscriptions import ( http_listen_requires_auth, @@ -15,12 +7,6 @@ ) __all__ = [ - "DispatchStage", - "IngressContext", - "StatelessIngressPipeline", - "StatelessTransportMiddleware", - "wrap_stateless_transport", - "is_task_wire_interception", "format_sse_message", "sse_connect_headers", "sse_notification", diff --git a/nitrostack/transports/dispatch.py b/nitrostack/transports/dispatch.py deleted file mode 100644 index f4005b9..0000000 --- a/nitrostack/transports/dispatch.py +++ /dev/null @@ -1,558 +0,0 @@ -"""Stateless HTTP ingress pipeline.""" - -from __future__ import annotations - -import json -from dataclasses import dataclass -from enum import Enum -from collections.abc import Awaitable -from typing import Any, Callable, Optional, Union - -from nitrostack.protocol.deprecated import ( - deprecated_method_message, - rejects_deprecated_method, -) -from nitrostack.protocol.discovery import ( - INITIALIZE_METHOD, - INITIALIZED_NOTIFICATION, - SERVER_DISCOVER_METHOD, - build_sessionless_initialize_result, -) -from nitrostack.protocol.errors import ERROR_CODE_MESSAGES, JsonRpcErrorCode -from nitrostack.protocol.jsonrpc import ( - HeaderBodyMismatchError, - InvalidRequestError, - JsonRpcParseError, - JsonRpcRequest, - JsonRpcWireError, - MethodNotFoundError, - build_ping_response, - jsonrpc_error, - jsonrpc_success, - parse_jsonrpc_request, - validate_header_body_method, - validate_header_body_name, - UnsupportedProtocolVersionError, - validate_protocol_version_header_meta, - validate_required_mcp_method, - validate_required_mcp_name, - validate_supported_protocol_version, -) -from nitrostack.protocol.method_contract import ( - mcp_method_is_required, - mcp_name_field, - mcp_name_is_required, -) -from nitrostack.protocol.meta import envelope_protocol_version -from nitrostack.protocol.version import ( - LEGACY_PROTOCOL_VERSION, - ProtocolEra, - WireMode, - accepts_sessionless_initialize, - protocol_era_for_wire_mode, - rejects_legacy_initialize, - supported_protocol_versions_for_era, -) -from nitrostack.runtime.stateless import ( - is_unsupported_protocol_version, - request_protocol_version, - sessionless_strips_incoming_session_id, -) -from nitrostack.transports.headers import ( - HEADER_MCP_METHOD, - HEADER_MCP_NAME, - HEADER_MCP_PROTOCOL_VERSION, - first_oversized_mcp_param, - get_header, - strip_legacy_session_headers, -) - -TaskDispatchHandler = Callable[[JsonRpcRequest], Awaitable[Optional[dict[str, Any]]]] -RegistryDispatchHandler = Callable[[JsonRpcRequest], Awaitable[Optional[dict[str, Any]]]] -DiscoverHandler = Callable[[JsonRpcRequest], Union[Awaitable[dict[str, Any]], dict[str, Any]]] -InitializeHandler = Callable[[JsonRpcRequest], Union[Awaitable[dict[str, Any]], dict[str, Any]]] - - -class DispatchStage(str, Enum): - """Six-step stateless HTTP ingress lifecycle.""" - - CORS_SECURITY = "cors_security" - BODY_PARSING = "body_parsing" - PING_FAST_PATH = "ping_fast_path" - TASK_INTERCEPTION = "task_interception" - REGISTRY_DISPATCH = "registry_dispatch" - RESPONSE_SERIALIZATION = "response_serialization" - - -TASK_METHOD_PREFIX = "tasks/" -TOOLS_CALL_METHOD = "tools/call" -PING_METHOD = "ping" -LEGACY_HANDSHAKE_METHODS = frozenset({"initialize", "notifications/initialized"}) - - -def is_header_only_ping(raw_body: bytes, request_headers: dict[str, str]) -> bool: - """True when ``Mcp-Method: ping`` and the body is empty or not JSON-RPC. - - A parsed JSON-RPC body is never header-only: header/body match still applies. - Other ``Mcp-Method`` values are not answered from the header alone. - """ - header_method = get_header(request_headers, HEADER_MCP_METHOD) - if header_method is None or header_method.strip() != PING_METHOD: - return False - if not raw_body or not raw_body.strip(): - return True - try: - parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - return True - return False - - -@dataclass -class IngressContext: - server_name: str - server_version: str - protocol_version: str - advertise_tasks: bool = True - advertise_app: bool = False - custom_extensions: Optional[dict[str, str]] = None - wire_mode: WireMode = "stateless" - protocol_era: Optional[ProtocolEra] = None - - def resolved_era(self) -> ProtocolEra: - if self.protocol_era is not None: - return self.protocol_era - return protocol_era_for_wire_mode(self.wire_mode) - - def accepts_sessionless_initialize(self) -> bool: - """True when this engine answers 2025 ``initialize`` without a session.""" - return accepts_sessionless_initialize(self.resolved_era()) - - def rejects_legacy_initialize(self) -> bool: - """True when this engine rejects 2025 ``initialize`` / ``initialized``.""" - return rejects_legacy_initialize(self.resolved_era()) - - -def prepare_sessionless_request_headers( - request_headers: dict[str, str], - wire_mode: WireMode, -) -> dict[str, str]: - """ - Drop client ``Mcp-Session-Id`` on sessionless engines (``modern`` / ``auto``). - - Obsolete session headers must not become a dependency; they are ignored rather - than rejected so stateless clients stay unambiguous. - """ - if not sessionless_strips_incoming_session_id(wire_mode): - return request_headers - return strip_legacy_session_headers(request_headers) - - -def reject_legacy_handshake( - request: JsonRpcRequest, - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """Era ``modern`` answers ``initialize`` / ``initialized`` as method-not-found.""" - return reject_legacy_handshake_method(request.method, request.id, era) - - -def reject_legacy_handshake_method( - method: str, - request_id: Any, - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """Era ``modern`` answers handshake methods as method-not-found.""" - if not rejects_legacy_initialize(era): - return None - if method not in LEGACY_HANDSHAKE_METHODS: - return None - return 200, MethodNotFoundError(method).to_response(request_id) - - -def reject_deprecated_method( - request: JsonRpcRequest, - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """Era ``modern`` answers retired 2025 methods as method-not-found.""" - return reject_deprecated_method_name(request.method, request.id, era) - - -def reject_deprecated_method_name( - method: str, - request_id: Any, - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """Same retired-method error used on POST, GET, and replay.""" - if not rejects_deprecated_method(method, era): - return None - message = deprecated_method_message(method) - return 200, jsonrpc_error( - request_id, - int(JsonRpcErrorCode.METHOD_NOT_FOUND), - message or f"Method not found: {method}", - ) - - -def reject_modern_method_policy( - method: str, - request_id: Any, - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """Handshake then retired-method policy. ``auto`` does not error here.""" - handshake = reject_legacy_handshake_method(method, request_id, era) - if handshake is not None: - return handshake - return reject_deprecated_method_name(method, request_id, era) - - -def reject_legacy_wire( - request: JsonRpcRequest, - request_headers: dict[str, str], - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """ - Era ``modern`` fails closed on 2025 handshake and 2025 protocol versions. - - Official v2 ``legacy: 'reject'`` is not mounted yet; this is the sidecar - stand-in. Handshake methods are method-not-found before header contracts. - Incoming session ids are rejected earlier. - """ - handshake = reject_legacy_handshake(request, era) - if handshake is not None: - return handshake - - if not rejects_legacy_initialize(era): - return None - - header_version = get_header(request_headers, HEADER_MCP_PROTOCOL_VERSION) - body_version = request.params.get("protocolVersion") - if header_version == LEGACY_PROTOCOL_VERSION or body_version == LEGACY_PROTOCOL_VERSION: - return 400, jsonrpc_error( - request.id, - int(JsonRpcErrorCode.UNSUPPORTED_PROTOCOL_VERSION), - ERROR_CODE_MESSAGES[JsonRpcErrorCode.UNSUPPORTED_PROTOCOL_VERSION], - ) - - return None - - -def is_task_wire_interception(method: str, params: dict[str, Any]) -> bool: - """Ingress step 4 — route to the task subsystem when matched.""" - if method.startswith(TASK_METHOD_PREFIX): - return True - return method == TOOLS_CALL_METHOD and bool(params.get("task")) - - -def reject_required_mcp_name( - request: JsonRpcRequest, - request_headers: dict[str, str], - wire_mode: WireMode, -) -> Optional[tuple[int, dict[str, Any]]]: - """Require or cross-check ``Mcp-Name`` on name-scoped methods.""" - if not mcp_name_is_required(request.method): - return None - field = mcp_name_field(request.method) or "name" - header_name = get_header(request_headers, HEADER_MCP_NAME) - body_name = request.params.get(field) - body_value = body_name if isinstance(body_name, str) else None - try: - if wire_mode == "stateless": - if header_name is not None: - validate_header_body_name(header_name, body_value) - elif not (body_value and body_value.strip()): - raise InvalidRequestError( - f"{field!r} is required in params for {request.method!r}" - ) - else: - validate_required_mcp_name( - header_name, - body_value, - body_label=field, - ) - except HeaderBodyMismatchError as exc: - return 400, exc.to_response(request.id) - except InvalidRequestError as exc: - return 400, exc.to_response(request.id) - return None - - -def reject_required_mcp_method( - request: JsonRpcRequest, - request_headers: dict[str, str], - wire_mode: WireMode, -) -> Optional[tuple[int, dict[str, Any]]]: - """Require or optionally cross-check ``Mcp-Method`` against the JSON-RPC method.""" - header_method = get_header(request_headers, HEADER_MCP_METHOD) - try: - if mcp_method_is_required(request.method, wire_mode): - validate_required_mcp_method(header_method, request.method) - else: - validate_header_body_method(header_method, request.method) - except HeaderBodyMismatchError as exc: - return 400, exc.to_response(request.id) - return None - - -def reject_protocol_version_mismatch( - request: JsonRpcRequest, - request_headers: dict[str, str], -) -> Optional[tuple[int, dict[str, Any]]]: - """Reject when header and ``_meta.mcp.protocolVersion`` both exist and differ.""" - header_version = get_header(request_headers, HEADER_MCP_PROTOCOL_VERSION) - meta_version = envelope_protocol_version(request.meta) - try: - validate_protocol_version_header_meta(header_version, meta_version) - except HeaderBodyMismatchError as exc: - return 400, exc.to_response(request.id) - return None - - -def reject_unsupported_protocol_version( - request: JsonRpcRequest, - request_headers: dict[str, str], - era: ProtocolEra, -) -> Optional[tuple[int, dict[str, Any]]]: - """ - Reject a present protocol version that the era does not support. - - Absent header and envelope versions are allowed (legacy default). The - header is not required on ``modern`` in this sidecar. - """ - header_version = get_header(request_headers, HEADER_MCP_PROTOCOL_VERSION) - meta_version = envelope_protocol_version(request.meta) - version = request_protocol_version(header_version, meta_version) - if not is_unsupported_protocol_version(version, era): - return None - try: - validate_supported_protocol_version(version, supported_protocol_versions_for_era(era)) - except UnsupportedProtocolVersionError as exc: - return 400, exc.to_response(request.id) - return None - - -class StatelessIngressPipeline: - """ - Deterministic JSON-RPC pre-dispatch for stateless POST /mcp. - - Handles ping and deprecated-method rejection inline. - ``server/discover`` is forwarded to the HTTP engine handler when provided. - Task and tool methods always return None so ``TaskManager`` plus the - low-level MCP server remain the only production task path. - """ - - def __init__( - self, - context: IngressContext, - *, - task_handler: Optional[TaskDispatchHandler] = None, - registry_handler: Optional[RegistryDispatchHandler] = None, - discover_handler: Optional[DiscoverHandler] = None, - initialize_handler: Optional[InitializeHandler] = None, - ) -> None: - self._context = context - self._task_handler = task_handler - self._registry_handler = registry_handler - self._discover_handler = discover_handler - self._initialize_handler = initialize_handler - - def reject_tools_call_mcp_name( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Replay-path ``Mcp-Name`` check for name-scoped methods.""" - try: - request = parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - return None - return reject_required_mcp_name( - request, request_headers, self._context.wire_mode - ) - - def reject_jsonrpc_mcp_method( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Replay-path ``Mcp-Method`` check for JSON-RPC POST.""" - try: - request = parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - return None - return reject_required_mcp_method(request, request_headers, self._context.wire_mode) - - def reject_protocol_version_cross_check( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Replay-path header vs envelope protocol version check.""" - try: - request = parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - return None - return reject_protocol_version_mismatch(request, request_headers) - - def reject_unsupported_protocol_version_header( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Replay-path unsupported protocol version check.""" - try: - request = parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - return None - return reject_unsupported_protocol_version( - request, request_headers, self._context.resolved_era() - ) - - def reject_method_policy( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Handshake and retired-method policy for POST, GET, and replay.""" - era = self._context.resolved_era() - try: - request = parse_jsonrpc_request(raw_body) - except (JsonRpcParseError, JsonRpcWireError): - header_method = get_header(request_headers, HEADER_MCP_METHOD) - if header_method is None: - return None - return reject_modern_method_policy(header_method.strip(), None, era) - return reject_modern_method_policy(request.method, request.id, era) - - def response_protocol_version(self) -> str: - """Advertised version used when the request does not name a supported one.""" - return self._context.protocol_version - - def response_supported_versions(self) -> frozenset[str]: - return supported_protocol_versions_for_era(self._context.resolved_era()) - - async def handle_post( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """ - Run ingress steps 2–5. Returns None to delegate to the underlying MCP app. - Step 1 (CORS) is handled by transport middleware. - """ - request_headers = prepare_sessionless_request_headers( - request_headers, self._context.wire_mode - ) - - if is_header_only_ping(raw_body, request_headers): - return 200, build_ping_response(None) - - try: - request = parse_jsonrpc_request(raw_body) - except JsonRpcParseError as exc: - return 400, exc.to_response(None) - except JsonRpcWireError as exc: - return 400, exc.to_response(None) - - rejected_handshake = reject_legacy_handshake( - request, self._context.resolved_era() - ) - if rejected_handshake is not None: - return rejected_handshake - - required_method = reject_required_mcp_method( - request, request_headers, self._context.wire_mode - ) - if required_method is not None: - return required_method - - required_name = reject_required_mcp_name( - request, request_headers, self._context.wire_mode - ) - if required_name is not None: - return required_name - - oversized = first_oversized_mcp_param(request_headers) - if oversized is not None: - return 400, InvalidRequestError( - "Mcp-Param header exceeds the maximum size" - ).to_response(request.id) - - header_name = get_header(request_headers, HEADER_MCP_NAME) - name_field = mcp_name_field(request.method) - body_name = request.params.get(name_field) if name_field else ( - request.params.get("name") or request.params.get("uri") - ) - if not mcp_name_is_required(request.method) and isinstance(body_name, str): - try: - validate_header_body_name(header_name, body_name) - except HeaderBodyMismatchError as exc: - return 400, exc.to_response(request.id) - - version_mismatch = reject_protocol_version_mismatch(request, request_headers) - if version_mismatch is not None: - return version_mismatch - - unsupported = reject_unsupported_protocol_version( - request, request_headers, self._context.resolved_era() - ) - if unsupported is not None: - return unsupported - - rejected = reject_legacy_wire(request, request_headers, self._context.resolved_era()) - if rejected is not None: - return rejected - - deprecated = reject_deprecated_method(request, self._context.resolved_era()) - if deprecated is not None: - return deprecated - - if request.method == PING_METHOD: - return 200, build_ping_response(request.id) - - if self._context.accepts_sessionless_initialize(): - if request.method == INITIALIZE_METHOD: - if self._initialize_handler is not None: - result = self._initialize_handler(request) - if isinstance(result, Awaitable): - result = await result - else: - requested = request.params.get("protocolVersion") - result = build_sessionless_initialize_result( - server_name=self._context.server_name, - server_version=self._context.server_version, - requested_version=requested if isinstance(requested, str) else None, - protocol_version=self._context.protocol_version, - advertise_tasks=self._context.advertise_tasks, - advertise_app=self._context.advertise_app, - custom_extensions=self._context.custom_extensions, - ) - return 200, jsonrpc_success(request.id, result) - if request.method == INITIALIZED_NOTIFICATION: - return 202, {} - - if request.method == SERVER_DISCOVER_METHOD: - if self._discover_handler is None: - return None - result = self._discover_handler(request) - if isinstance(result, Awaitable): - result = await result - return 200, jsonrpc_success(request.id, result) - - if is_task_wire_interception(request.method, request.params): - if self._task_handler is not None: - response = await self._task_handler(request) - if response is not None: - return 200, response - return None - - if self._registry_handler is not None: - response = await self._registry_handler(request) - if response is not None: - return 200, response - - return None - - @staticmethod - def serialize_response(jsonrpc_response: dict[str, Any]) -> bytes: - """Step 6 — JSON-RPC response serialization.""" - return json.dumps(jsonrpc_response).encode("utf-8") diff --git a/nitrostack/transports/http.py b/nitrostack/transports/http.py index a343747..17f9d2d 100644 --- a/nitrostack/transports/http.py +++ b/nitrostack/transports/http.py @@ -42,6 +42,7 @@ 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 @@ -311,7 +312,8 @@ class HeaderCompatMiddleware: version checks see what the client sent. Stack order on the HTTP app: CORS → this middleware (preserve) → handler. - Sidecar version checks run on the combined app outside this mount. + This middleware only normalizes transport headers; JSON-RPC dispatch remains + owned by the official MCP Streamable HTTP manager. """ def __init__(self, app: ASGIApp, *, drop_session_headers: bool = False) -> None: @@ -522,6 +524,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: @@ -693,7 +696,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", @@ -775,11 +778,15 @@ async def lifespan(app): 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( CORSMiddleware, - allow_origins=["*"], + allow_origins=allow_origins, allow_methods=["GET", "POST", "DELETE", "OPTIONS"], allow_headers=CORS_ALLOW_HEADERS, expose_headers=expose_headers, diff --git a/nitrostack/transports/middleware.py b/nitrostack/transports/middleware.py deleted file mode 100644 index 314f097..0000000 --- a/nitrostack/transports/middleware.py +++ /dev/null @@ -1,524 +0,0 @@ -"""Stateless HTTP ASGI middleware.""" - -from __future__ import annotations - -from typing import Any, Callable, Optional - -from nitrostack.protocol.errors import JsonRpcErrorCode -from nitrostack.protocol.jsonrpc import jsonrpc_error, jsonrpc_method_from_body -from nitrostack.protocol.version import MODERN_PROTOCOL_VERSION, ProtocolEra, WireMode -from nitrostack.runtime.stateless import assert_stateless_headers -from nitrostack.transports.cors import build_cors_headers, cors_preflight_response_headers -from nitrostack.transports.dispatch import ( - DiscoverHandler, - IngressContext, - InitializeHandler, - StatelessIngressPipeline, -) -from nitrostack.transports.headers import ( - MCP_HTTP_PATH, - build_mcp_echo_headers, - decode_asgi_headers, - get_header, - scope_with_header_snapshot, - scope_without_session_headers, - snapshot_validated_asgi_headers, - strip_legacy_session_headers, -) - -ASGIApp = Callable[..., Any] - -MCP_POST_PATHS = (MCP_HTTP_PATH, f"{MCP_HTTP_PATH}/") - - -class StatelessTransportMiddleware: - """ - ASGI wrapper implementing stateless HTTP transport invariants: - - OPTIONS 204 CORS preflight on MCP paths only - - Legacy session header stripping - - MCP response headers on all responses - - Pre-dispatch for POST /mcp (ping, server/discover) - """ - - def __init__( - self, - app: ASGIApp, - *, - pipeline: Optional[StatelessIngressPipeline] = None, - mcp_paths: tuple[str, ...] = MCP_POST_PATHS, - enable_cors: bool = True, - ) -> None: - self.app = app - self.pipeline = pipeline - self.mcp_paths = mcp_paths - self.enable_cors = enable_cors - - async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: - if scope.get("type") != "http": - await self.app(scope, receive, send) - return - - method = scope.get("method", "GET").upper() - path = scope.get("path", "") - - if method == "OPTIONS" and path in self.mcp_paths and self.enable_cors: - await self._send_options(scope, receive, send) - return - - buffered_body: Optional[bytes] = None - header_snapshot: Optional[tuple[tuple[bytes, bytes], ...]] = None - if method == "POST" and path in self.mcp_paths and self.pipeline is not None: - scope = self._strip_sessionless_scope(scope) - buffered_body = await self._read_body(receive) - handled = await self._try_pre_dispatch(scope, buffered_body, send) - if handled: - return - live_headers = decode_asgi_headers(list(scope.get("headers") or [])) - rejected = self._reject_replay_headers(buffered_body, live_headers) - if rejected is not None: - await self._send_pipeline_response( - scope, send, live_headers, rejected, body=buffered_body - ) - return - header_snapshot = snapshot_validated_asgi_headers(list(scope.get("headers") or [])) - snapshot_headers = decode_asgi_headers(list(header_snapshot)) - rejected = self._reject_replay_headers(buffered_body, snapshot_headers) - if rejected is not None: - await self._send_pipeline_response( - scope, send, snapshot_headers, rejected, body=buffered_body - ) - return - receive = self._replay_receive(buffered_body, receive) - scope = scope_with_header_snapshot(scope, header_snapshot) - - if path in self.mcp_paths and self.pipeline is not None: - scope = self._strip_sessionless_scope(scope) - raw_headers = decode_asgi_headers(list(scope.get("headers") or [])) - if method == "GET": - rejected = self.pipeline.reject_method_policy(b"", raw_headers) - if rejected is not None: - await self._send_pipeline_response( - scope, send, raw_headers, rejected, body=b"" - ) - return - - await self._forward_with_stateless_headers( - scope, receive, send, body=buffered_body - ) - - def _reject_replay_headers( - self, - raw_body: bytes, - request_headers: dict[str, str], - ) -> Optional[tuple[int, dict[str, Any]]]: - """Re-apply method policy, ``-32020``, and ``-32022`` on replay.""" - assert self.pipeline is not None - policy_rejected = self.pipeline.reject_method_policy(raw_body, request_headers) - if policy_rejected is not None: - return policy_rejected - method_rejected = self.pipeline.reject_jsonrpc_mcp_method(raw_body, request_headers) - if method_rejected is not None: - return method_rejected - name_rejected = self.pipeline.reject_tools_call_mcp_name(raw_body, request_headers) - if name_rejected is not None: - return name_rejected - version_rejected = self.pipeline.reject_protocol_version_cross_check( - raw_body, request_headers - ) - if version_rejected is not None: - return version_rejected - return self.pipeline.reject_unsupported_protocol_version_header( - raw_body, request_headers - ) - - async def _send_options(self, scope: dict[str, Any], receive: Any, send: Any) -> None: - headers_list = scope.get("headers") or [] - req_headers = { - k.decode("latin-1"): v.decode("latin-1") for k, v in headers_list - } - cors = cors_preflight_response_headers(req_headers) - response_headers = self._echo_headers(req_headers, extra=cors) - assert_stateless_headers(response_headers) - - await send( - { - "type": "http.response.start", - "status": 204, - "headers": self._encode_headers(response_headers), - } - ) - await send({"type": "http.response.body", "body": b""}) - - async def _try_pre_dispatch( - self, - scope: dict[str, Any], - body: bytes, - send: Any, - ) -> bool: - headers_list = scope.get("headers") or [] - raw_headers = {k.decode("latin-1"): v.decode("latin-1") for k, v in headers_list} - - assert self.pipeline is not None - result = await self.pipeline.handle_post(body, raw_headers) - if result is None: - return False - - await self._send_pipeline_response(scope, send, raw_headers, result, body=body) - return True - - async def _send_pipeline_response( - self, - scope: dict[str, Any], - send: Any, - raw_headers: dict[str, str], - result: tuple[int, dict[str, Any]], - *, - body: Optional[bytes] = None, - ) -> None: - status, jsonrpc_response = result - origin = get_header(raw_headers, "Origin") - cors = build_cors_headers(origin=origin) - response_headers = self._echo_headers(raw_headers, extra=cors, body=body) - assert_stateless_headers(response_headers) - assert self.pipeline is not None - if status == 202 and jsonrpc_response == {}: - payload = b"" - else: - payload = self.pipeline.serialize_response(jsonrpc_response) - await send( - { - "type": "http.response.start", - "status": status, - "headers": self._encode_headers(response_headers), - } - ) - await send({"type": "http.response.body", "body": payload}) - - async def _send_session_id_rejected( - self, - scope: dict[str, Any], - send: Any, - raw_headers: dict[str, str], - ) -> None: - origin = get_header(raw_headers, "Origin") - cors = build_cors_headers(origin=origin) - response_headers = self._echo_headers(raw_headers, extra=cors) - assert_stateless_headers(response_headers) - payload = StatelessIngressPipeline.serialize_response( - jsonrpc_error( - None, - int(JsonRpcErrorCode.INVALID_REQUEST), - "Invalid Request: Mcp-Session-Id is not supported", - ) - ) - await send( - { - "type": "http.response.start", - "status": 400, - "headers": self._encode_headers(response_headers), - } - ) - await send({"type": "http.response.body", "body": payload}) - - async def _forward_with_stateless_headers( - self, - scope: dict[str, Any], - receive: Any, - send: Any, - *, - body: Optional[bytes] = None, - ) -> None: - async def send_wrapper(message: dict[str, Any]) -> None: - if message["type"] == "http.response.start": - raw_headers = { - k.decode("latin-1"): v.decode("latin-1") - for k, v in message.get("headers", []) - } - req_headers = { - k.decode("latin-1"): v.decode("latin-1") - for k, v in (scope.get("headers") or []) - } - inner_headers = { - k: v - for k, v in strip_legacy_session_headers(raw_headers).items() - if k.lower() != "content-type" - } - merged = self._echo_headers( - req_headers, - content_type=raw_headers.get("content-type", "application/json"), - extra={ - **inner_headers, - **build_cors_headers(origin=get_header(req_headers, "Origin")), - }, - body=body, - ) - assert_stateless_headers(merged) - message = { - **message, - "headers": self._encode_headers(merged), - } - await send(message) - - await self.app( - scope_with_header_snapshot( - scope, - snapshot_validated_asgi_headers(list(scope.get("headers") or [])), - ), - receive, - send_wrapper, - ) - - def _strip_sessionless_scope(self, scope: dict[str, Any]) -> dict[str, Any]: - if self.pipeline is None: - return scope - from nitrostack.runtime.stateless import sessionless_strips_incoming_session_id - - if sessionless_strips_incoming_session_id(self.pipeline._context.wire_mode): - return scope_without_session_headers(scope) - return scope - - def _echo_headers( - self, - request_headers: dict[str, str], - *, - extra: Optional[dict[str, str]] = None, - content_type: str = "application/json", - body: Optional[bytes] = None, - ) -> dict[str, str]: - fallback = MODERN_PROTOCOL_VERSION - supported = None - if self.pipeline is not None: - fallback = self.pipeline.response_protocol_version() - supported = self.pipeline.response_supported_versions() - return build_mcp_echo_headers( - request_headers, - protocol_version=fallback, - method=jsonrpc_method_from_body(body), - supported_versions=supported, - content_type=content_type, - extra=extra, - ) - - @staticmethod - async def _read_body(receive: Any) -> bytes: - body = b"" - while True: - message = await receive() - if message["type"] == "http.request": - body += message.get("body", b"") - if not message.get("more_body", False): - break - return body - - @staticmethod - def _replay_receive(body: bytes, original_receive: Any) -> Any: - sent = False - - async def replay() -> dict[str, Any]: - nonlocal sent - if not sent: - sent = True - return {"type": "http.request", "body": body, "more_body": False} - # Body was already buffered. Wait for a real client disconnect - # instead of synthesizing one — Streamable HTTP treats disconnect - # as an abort of the in-flight request. - while True: - message = await original_receive() - if message.get("type") == "http.disconnect": - return message - - return replay - - @staticmethod - def _encode_headers(headers: dict[str, str]) -> list[tuple[bytes, bytes]]: - return [(k.lower().encode("latin-1"), v.encode("latin-1")) for k, v in headers.items()] - - -class SessionlessHttpGuard: - """ - Transport invariants official mcp 2.x does not own on sessionless ``/mcp``. - - Does not parse or answer ``tools/call``. Forwards those to the v2 app. - """ - - def __init__( - self, - app: ASGIApp, - *, - protocol_era: ProtocolEra = "auto", - enable_cors: bool = True, - discover_handler: Optional[DiscoverHandler] = None, - ) -> None: - self.app = app - self.protocol_era = protocol_era - self.enable_cors = enable_cors - self.discover_handler = discover_handler - wire_mode: WireMode = "reject" if protocol_era == "modern" else "stateless" - self._sender = StatelessTransportMiddleware( - app, - pipeline=StatelessIngressPipeline( - IngressContext( - server_name="", - server_version="", - protocol_version=MODERN_PROTOCOL_VERSION, - wire_mode=wire_mode, - protocol_era=protocol_era, - ), - discover_handler=discover_handler, - ), - enable_cors=enable_cors, - ) - - async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: - if scope.get("type") != "http": - await self.app(scope, receive, send) - return - method = scope.get("method", "").upper() - path = scope.get("path", "") - if method == "OPTIONS" and path in MCP_POST_PATHS and self.enable_cors: - await self._sender._send_options(scope, receive, send) - return - scope = scope_without_session_headers(scope) - raw_headers = decode_asgi_headers(list(scope.get("headers") or [])) - from nitrostack.transports.dispatch import ( - is_header_only_ping, - reject_modern_method_policy, - ) - from nitrostack.transports.headers import HEADER_MCP_METHOD, get_header - if path in MCP_POST_PATHS and method == "GET": - header_method = get_header(raw_headers, HEADER_MCP_METHOD) - if header_method is not None: - rejected = reject_modern_method_policy( - header_method.strip(), None, self.protocol_era - ) - if rejected is not None: - await self._sender._send_pipeline_response( - scope, send, raw_headers, rejected, body=b"" - ) - return - if method != "POST" or path not in MCP_POST_PATHS: - await self.app(scope, receive, send) - return - - body = await StatelessTransportMiddleware._read_body(receive) - if is_header_only_ping(body, raw_headers): - from nitrostack.protocol.jsonrpc import build_ping_response - - await self._sender._send_pipeline_response( - scope, send, raw_headers, (200, build_ping_response(None)), body=body - ) - return - - from nitrostack.protocol.jsonrpc import ( - JsonRpcParseError, - JsonRpcWireError, - parse_jsonrpc_request, - validate_header_body_method, - ) - try: - request = parse_jsonrpc_request(body) - except (JsonRpcParseError, JsonRpcWireError): - await self.app( - scope, - StatelessTransportMiddleware._replay_receive(body, receive), - send, - ) - return - - rejected = reject_modern_method_policy( - request.method, request.id, self.protocol_era - ) - if rejected is not None: - await self._sender._send_pipeline_response( - scope, send, raw_headers, rejected, body=body - ) - return - - header_method = get_header(raw_headers, HEADER_MCP_METHOD) - if header_method is not None: - try: - validate_header_body_method(header_method, request.method) - except Exception as exc: - from nitrostack.protocol.jsonrpc import map_exception_to_jsonrpc - - await self._sender._send_pipeline_response( - scope, - send, - raw_headers, - (400, map_exception_to_jsonrpc(exc, request.id)), - body=body, - ) - return - - if request.method == "server/discover" and self.discover_handler is not None: - from nitrostack.protocol.jsonrpc import jsonrpc_success - - await self._sender._send_pipeline_response( - scope, - send, - raw_headers, - (200, jsonrpc_success(request.id, self.discover_handler())), - body=body, - ) - return - - await self._sender._forward_with_stateless_headers( - scope, - StatelessTransportMiddleware._replay_receive(body, receive), - send, - body=body, - ) - - -def wrap_sessionless_http( - app: ASGIApp, - *, - protocol_era: ProtocolEra = "auto", - enable_cors: bool = True, - discover_handler: Optional[DiscoverHandler] = None, -) -> ASGIApp: - """Sessionless transport guard; official mcp 2.x still owns ``tools/call``.""" - return SessionlessHttpGuard( - app, - protocol_era=protocol_era, - enable_cors=enable_cors, - discover_handler=discover_handler, - ) - - -def wrap_modern_handshake_reject(app: ASGIApp, *, enable_cors: bool = True) -> ASGIApp: - """Reject 2025 ``initialize`` on era ``modern``; leave the v2 app otherwise.""" - return wrap_sessionless_http(app, protocol_era="modern", enable_cors=enable_cors) - - -def wrap_stateless_transport( - app: ASGIApp, - *, - server_name: str, - server_version: str, - protocol_version: str, - advertise_tasks: bool = True, - advertise_app: bool = False, - custom_extensions: Optional[dict[str, str]] = None, - wire_mode: WireMode = "stateless", - protocol_era: Optional[ProtocolEra] = None, - enable_cors: bool = True, - discover_handler: Optional[DiscoverHandler] = None, - initialize_handler: Optional[InitializeHandler] = None, -) -> ASGIApp: - """Wrap an ASGI app with stateless HTTP middleware.""" - pipeline = StatelessIngressPipeline( - IngressContext( - server_name=server_name, - server_version=server_version, - protocol_version=protocol_version, - advertise_tasks=advertise_tasks, - advertise_app=advertise_app, - custom_extensions=custom_extensions, - wire_mode=wire_mode, - protocol_era=protocol_era, - ), - discover_handler=discover_handler, - initialize_handler=initialize_handler, - ) - return StatelessTransportMiddleware(app, pipeline=pipeline, enable_cors=enable_cors) diff --git a/nitrostack/transports/stdio.py b/nitrostack/transports/stdio.py index d76e580..f6fe5cf 100644 --- a/nitrostack/transports/stdio.py +++ b/nitrostack/transports/stdio.py @@ -64,15 +64,37 @@ async def serve_stdio_streams( 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, + # A per-request envelope loop (no `initialize` handshake) built from + # documented, public `mcp.server.runner` building blocks, so this does + # not depend on a leading-underscore internal that mcp is free to + # remove without notice. + from mcp.server.runner import ( + Connection, + JSONRPCDispatcher, + LATEST_MODERN_VERSION, + NotifyOnlyOutbound, + ServerRunner, + aclose_shielded, + modern_on_request, ) + + dispatcher: JSONRPCDispatcher = JSONRPCDispatcher(read_stream, write_stream) + outbound = NotifyOnlyOutbound(dispatcher) + + async def _on_notify(dctx, method, params): + # Fresh per-notification `Connection`, mirroring the request path: + # notifications carry no envelope of their own at this era. + connection = Connection.from_envelope( + LATEST_MODERN_VERSION, None, None, outbound=outbound + ) + try: + await ServerRunner(server, connection, lifespan_state).on_notify( + dctx, method, params + ) + finally: + await aclose_shielded(connection) + + await dispatcher.run(modern_on_request(server, lifespan_state), _on_notify) else: await serve_loop( server, @@ -87,5 +109,6 @@ async def serve_stdio_streams( 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/test_mcp20_stdio.py b/tests/test_mcp20_stdio.py index 94a4b23..68705a4 100644 --- a/tests/test_mcp20_stdio.py +++ b/tests/test_mcp20_stdio.py @@ -135,8 +135,11 @@ def test_modern_stdio_rejects_initialize(monkeypatch): }, ) ) + # Modern has no `initialize` handshake handler at all (rejected via the + # public `modern_on_request` dispatch path, not a bespoke error message), + # so this surfaces as a standard JSON-RPC method-not-found. 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 new file mode 100644 index 0000000..7d33053 --- /dev/null +++ b/tests/test_mcp22_official_http_owner.py @@ -0,0 +1,98 @@ +"""Regression tests for Issue #22: one official owner for ``/mcp``.""" + +import asyncio +import json +from pathlib import Path + +import httpx + +from nitrostack import module +from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app + + +@module(name="issue22_probe", controllers=[]) +class Issue22ProbeModule: + pass + + +@mcp_app( + module=Issue22ProbeModule, + server=ServerConfig(name="issue22-probe", protocol_era="auto"), +) +class Issue22ProbeApp: + pass + + +def _request(path: str, *, method: str = "GET", payload: dict | None = None): + async def run(): + application = await McpApplicationFactory.create(Issue22ProbeApp) + http_app = application.get_combined_app(json_response=True) + 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={ + "content-type": "application/json", + "accept": "application/json, text/event-stream", + }, + content=json.dumps(payload) if payload is not None else None, + ) + + return asyncio.run(run()) + + +def test_combined_app_has_no_jsonrpc_sidecar_call(): + source = Path("nitrostack/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() + production = "\n".join( + path.read_text(encoding="utf-8") + for path in Path("nitrostack").rglob("*.py") + ) + for symbol in ( + "wrap_stateless_transport", + "wrap_sessionless_http", + "wrap_modern_handshake_reject", + "StatelessIngressPipeline", + "build_ping_response", + "is_header_only_ping", + ): + assert symbol not in production + + +def test_health_endpoint_remains_available(): + response = _request("/mcp/health") + assert response.status_code == 200 + assert response.json()["status"] == "ok" + + +def test_official_mcp_handles_ping_without_a_default_session_id(): + response = _request( + "/mcp", + method="POST", + payload={"jsonrpc": "2.0", "id": 1, "method": "ping"}, + ) + assert response.status_code == 200 + assert response.headers.get("mcp-session-id") is None + assert response.json() == {"jsonrpc": "2.0", "id": 1, "result": {}} + + +def test_official_mcp_returns_jsonrpc_error_for_unknown_method(): + response = _request( + "/mcp", + method="POST", + payload={"jsonrpc": "2.0", "id": 2, "method": "issue22/unknown"}, + ) + assert response.status_code == 200 + body = response.json() + assert body["jsonrpc"] == "2.0" + assert body["id"] == 2 + assert body["error"]["code"] == -32601 + assert "Method not found" in body["error"]["message"] diff --git a/tests/test_mcp23_http_factory.py b/tests/test_mcp23_http_factory.py new file mode 100644 index 0000000..82e6b6a --- /dev/null +++ b/tests/test_mcp23_http_factory.py @@ -0,0 +1,222 @@ +"""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 json +import os +import sys +from pathlib import Path + +import anyio +import httpx +import pytest +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 + +CLIENT_META = { + "io.modelcontextprotocol/clientInfo": {"name": "issue23-test", "version": "1.0"}, + "io.modelcontextprotocol/clientCapabilities": {}, +} + + +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}") + app.get_combined_app(json_response=True) # wires protocol_era onto mcp_server + discover = app.handle_server_discover() + caps = app.mcp_server.get_capabilities() + assert discover["capabilities"]["resources"]["subscribe"] == caps.resources.subscribe + # Only `modern` rejects `resources/subscribe` (see rejects_deprecated_method). + assert caps.resources.subscribe == (era != "modern") + + +# 3. No private `mcp.server.runner` symbol left in stdio.py. +def test_stdio_module_has_no_private_mcp_runner_import(): + source = Path("nitrostack/transports/stdio.py").read_text(encoding="utf-8") + assert "_serve_modern_stream" not in source + + +# 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 and serves real requests without the removed +# private import, via the public JSONRPCDispatcher/modern_on_request path. +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. The hand-built on_notify path (no public one-liner exists for it) must +# not crash or hang the dispatcher — 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"