From 50484d840797ff018b43402698ae10063d4e87b0 Mon Sep 17 00:00:00 2001 From: ROHIT-JR Date: Tue, 15 Sep 2026 12:37:22 +0530 Subject: [PATCH] fix: remove duplicate MCP JSON-RPC sidecar --- nitrostack/__init__.py | 3 - nitrostack/core/app.py | 35 +- nitrostack/protocol/__init__.py | 2 - nitrostack/protocol/jsonrpc.py | 5 - nitrostack/transports/__init__.py | 16 +- nitrostack/transports/dispatch.py | 558 ------------------------ nitrostack/transports/http.py | 3 +- nitrostack/transports/middleware.py | 524 ---------------------- tests/test_mcp22_official_http_owner.py | 98 +++++ 9 files changed, 105 insertions(+), 1139 deletions(-) delete mode 100644 nitrostack/transports/dispatch.py delete mode 100644 nitrostack/transports/middleware.py create mode 100644 tests/test_mcp22_official_http_owner.py 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..581ab88 100644 --- a/nitrostack/core/app.py +++ b/nitrostack/core/app.py @@ -1769,37 +1769,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/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/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..86f7546 100644 --- a/nitrostack/transports/http.py +++ b/nitrostack/transports/http.py @@ -311,7 +311,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: 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/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"]