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..481ab6b 100644 --- a/nitrostack/core/app.py +++ b/nitrostack/core/app.py @@ -75,7 +75,7 @@ strip_tool_arguments, ) from nitrostack.protocol.observability import TraceContext, extract_trace_context -from nitrostack.runtime.correlation import InFlightRegistry, new_correlation_id +from nitrostack.runtime.correlation import new_correlation_id from nitrostack.transports.headers import ( extract_mcp_param_headers, extract_mcp_scope_headers, @@ -541,7 +541,6 @@ def __init__(self, app_class: Type): self._prompts: Dict[str, _PromptEntry] = {} self._initial_tools: List[Tuple[Any, Callable, ToolConfig]] = [] self.task_manager = TaskManager() - self._in_flight = InFlightRegistry() self._bootstrap() @@ -766,6 +765,11 @@ async def health_status_resource(context: ExecutionContext) -> str: # ------------------------------------------------------------------ def _advertise_tasks_extension(self) -> bool: + # Tasks remain available to in-process callers and the legacy wire only. + # MCP 2026 has no complete tasks surface (notably tasks/update), so + # advertising this partial implementation would make discovery lie. + if self.protocol_era in ("auto", "modern"): + return False return any( entry.config.task_support in ("optional", "required") for entry in self._tools.values() @@ -1275,6 +1279,11 @@ async def _call_tool(self, name: str, arguments: Dict[str, Any], task: Any = Non trace = _trace_context_from_request_ctx(rc) is_task = task_metadata is not None + if self.protocol_era in ("auto", "modern") and is_task: + raise MCPError( + types.INVALID_PARAMS, + "Server does not support task augmentation on the MCP 2026 wire", + ) if cfg.task_support == "forbidden" and task_metadata is not None: raise MCPError( types.METHOD_NOT_FOUND, @@ -1369,7 +1378,6 @@ async def background_execution(): trace=trace, ) _apply_request_envelope(ctx, rc) - ticket = self._in_flight.register(correlation_id, jsonrpc_id=jsonrpc_id) try: result = await run_pipeline( handler=entry.method, @@ -1391,13 +1399,6 @@ async def background_execution(): content=[types.TextContent(type="text", text=str(exc))], isError=True, ) - finally: - self._in_flight.discard(correlation_id) - if ticket.cancel_requested.is_set(): - return types.CallToolResult( - content=[types.TextContent(type="text", text="Request was cancelled.")], - isError=True, - ) return self._to_call_tool_result(result, entry.component, ctx) async def _read_resource(self, uri: str) -> List[ReadResourceContents]: @@ -1546,6 +1547,10 @@ def _build_get_task_result(self, task) -> types.GetTaskResult: return GetTaskResult(**payload) def _register_task_handlers(self, server: NitroStackMcpServer) -> None: + if self.protocol_era in ("auto", "modern"): + server.has_task_support = False + return + async def handle_list_tasks(req): if rejects_deprecated_method("tasks/list", self.protocol_era): message = deprecated_method_message("tasks/list") @@ -1769,37 +1774,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/context.py b/nitrostack/core/context.py index 42cf129..440e2ad 100644 --- a/nitrostack/core/context.py +++ b/nitrostack/core/context.py @@ -107,8 +107,9 @@ class TaskContext: whichever transport (STDIO or Streamable HTTP) initiated the task — transport-agnostic since both use the same ``mcp.server.session.ServerSession``. The client only receives these if it supplied a ``progressToken`` in the - original ``tools/call`` request's ``_meta``; ``TaskManager``-backed polling via - ``tasks/get`` always works regardless, so this is additive, not required. + original ``tools/call`` request's ``_meta``. TaskManager-backed polling is + retained for in-process callers and the legacy wire; the incomplete Tasks + surface is deliberately not advertised on MCP 2026. """ def __init__( 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/protocol/method_contract.py b/nitrostack/protocol/method_contract.py index b9142ab..2c3403d 100644 --- a/nitrostack/protocol/method_contract.py +++ b/nitrostack/protocol/method_contract.py @@ -1,8 +1,8 @@ """SEP-2243 method contracts for the 2026 method surface. -Official v2 owns this table once mounted. Until then the sidecar validator -reads one row per method: required ``Mcp-Method``, optional ``Mcp-Name`` field, -and whether ``auto`` requires the method header. +The official MCP SDK owns request dispatch. This table records the method/header +contract used by the 2026 protocol checks: required ``Mcp-Method``, optional +``Mcp-Name`` field, and whether ``auto`` requires the method header. """ from __future__ import annotations @@ -63,10 +63,6 @@ def _contract( _contract("prompts/get", name_field=NAME_FIELD_NAME), _contract("notifications/prompts/list_changed"), _contract("completion/complete"), - _contract("tasks/get"), - _contract("tasks/cancel"), - _contract("tasks/result"), - _contract("tasks/list"), _contract("logging/setLevel"), ) @@ -82,8 +78,6 @@ def _contract( # (``initialize``, ``notifications/initialized``) are not in this table: # ``modern`` rejects them as method-not-found; ``auto`` still answers them. DEPRECATED_MODERN_METHODS: dict[str, str] = { - "tasks/result": "Method 'tasks/result' is not supported in MCP 2026-07-28; use 'tasks/get'.", - "tasks/list": "Method 'tasks/list' is not supported in modern stateless MCP 2026-07-28.", "resources/subscribe": ( "Method 'resources/subscribe' is not supported in stateless MCP 2026-07-28; " "use SSE subscriptions/listen." diff --git a/nitrostack/runtime/acceptance.py b/nitrostack/runtime/acceptance.py index 17992ed..7183445 100644 --- a/nitrostack/runtime/acceptance.py +++ b/nitrostack/runtime/acceptance.py @@ -103,7 +103,7 @@ class AcceptanceCriterion: ), ProtocolDeliverable( ProtocolArea.ASYNC_TASKS, - "Task state machine and tasks/get embedded result", + "Legacy/in-process Task state machine and embedded result", "tests/test_mcp20_tasks.py", ), ProtocolDeliverable( @@ -146,7 +146,7 @@ class AcceptanceCriterion: ), AcceptanceCriterion( "task_cancellation", - "tasks/cancel marks cancelled; terminal cancel returns -32602", + "Legacy tasks/cancel marks cancelled; terminal cancel returns -32602", "tests/test_mcp20_tasks.py", ), AcceptanceCriterion( diff --git a/nitrostack/runtime/conformance.py b/nitrostack/runtime/conformance.py index dc352b0..177c169 100644 --- a/nitrostack/runtime/conformance.py +++ b/nitrostack/runtime/conformance.py @@ -32,7 +32,7 @@ class ConformanceArea(str, Enum): "Standard JSON-RPC error codes; deprecated methods rejected on modern wire" ), ConformanceArea.TASKS: ( - "Task-augmented tools/call, tasks/get lifecycle, cancel, terminal-only TTL eviction" + "MCP 2026 hides the incomplete Tasks surface; legacy/in-process task lifecycle remains" ), ConformanceArea.MULTI_TENANT: ( "Cross-tenant task access raises TaskNotFoundError without enumeration" diff --git a/nitrostack/runtime/stateless.py b/nitrostack/runtime/stateless.py index cb849db..778b90d 100644 --- a/nitrostack/runtime/stateless.py +++ b/nitrostack/runtime/stateless.py @@ -47,7 +47,8 @@ def request_protocol_version( The header wins when present. Otherwise ``_meta.mcp.protocolVersion`` is used. When neither is present the request proceeds (legacy default). The - header is not required on ``modern`` in this sidecar. + header is not required on ``modern`` because the official SDK owns the + HTTP request lifecycle. """ for raw in (header_version, envelope_version): if isinstance(raw, str) and raw.strip(): diff --git a/nitrostack/tasks/notify.py b/nitrostack/tasks/notify.py deleted file mode 100644 index fc81677..0000000 --- a/nitrostack/tasks/notify.py +++ /dev/null @@ -1,194 +0,0 @@ -"""Fan-out for ``notifications/tasks/status`` across live transports.""" - -from __future__ import annotations - -import logging -from typing import Any, Callable, Optional, Protocol - -from nitrostack.tasks.authorization import entry_matches_access_context -from nitrostack.tasks.types import TaskAccessContext, TaskEntry - -logger = logging.getLogger(__name__) - -TASK_STATUS_METHOD = "notifications/tasks/status" - - -def build_task_status_notification( - task_id: str, - status: str, - *, - status_message: Optional[str] = None, -) -> dict[str, Any]: - """Wire notification for a task status or progress change.""" - params: dict[str, Any] = {"taskId": task_id, "status": status} - if status_message is not None: - params["statusMessage"] = status_message - return {"method": TASK_STATUS_METHOD, "params": params} - - -class TaskStatusSink(Protocol): - """One connected transport that can receive task status.""" - - access: Optional[TaskAccessContext] - session_id: Optional[str] - task_id: Optional[str] - - async def send(self, notification: dict[str, Any]) -> None: - """Deliver one status notification. Must not raise to the router.""" - ... - - -class CallbackTaskSink: - """Test and in-process sink that records or forwards notifications.""" - - def __init__( - self, - callback: Callable[[dict[str, Any]], Any], - *, - access: Optional[TaskAccessContext] = None, - session_id: Optional[str] = None, - task_id: Optional[str] = None, - ) -> None: - self._callback = callback - self.access = access - self.session_id = session_id - self.task_id = task_id - - async def send(self, notification: dict[str, Any]) -> None: - result = self._callback(notification) - if hasattr(result, "__await__"): - await result - - -class QueueTaskSink: - """Push notifications onto an asyncio queue (listen / SSE attach).""" - - def __init__( - self, - put: Callable[[dict[str, Any]], None], - *, - access: Optional[TaskAccessContext] = None, - session_id: Optional[str] = None, - task_id: Optional[str] = None, - ) -> None: - self._put = put - self.access = access - self.session_id = session_id - self.task_id = task_id - - async def send(self, notification: dict[str, Any]) -> None: - self._put(notification) - - -class StreamTaskSink: - """Write a JSON-RPC notification onto an official stream write side.""" - - def __init__( - self, - write_stream: Any, - *, - access: Optional[TaskAccessContext] = None, - session_id: Optional[str] = None, - task_id: Optional[str] = None, - ) -> None: - self._write_stream = write_stream - self.access = access - self.session_id = session_id - self.task_id = task_id - - async def send(self, notification: dict[str, Any]) -> None: - from mcp.shared.message import SessionMessage - from mcp_types import jsonrpc_message_adapter - - message = jsonrpc_message_adapter.validate_python( - { - "jsonrpc": "2.0", - "method": notification["method"], - "params": notification.get("params") or {}, - } - ) - await self._write_stream.send(SessionMessage(message)) - - -class SessionTaskSink: - """Notify the originating MCP session (stdio or legacy SSE).""" - - def __init__( - self, - session: Any, - *, - access: Optional[TaskAccessContext] = None, - session_id: Optional[str] = None, - task_id: Optional[str] = None, - ) -> None: - self._session = session - self.access = access - self.session_id = session_id - self.task_id = task_id - - async def send(self, notification: dict[str, Any]) -> None: - session = self._session - if session is None: - return - method = notification["method"] - params = notification.get("params") or {} - sender = getattr(session, "send_notification", None) - if sender is not None: - await sender(method, params) - return - outbound = getattr(session, "outbound", None) - notify = getattr(outbound, "notify", None) if outbound is not None else None - if notify is not None: - await notify(method, params) - - -def _sink_accepts(sink: TaskStatusSink, entry: TaskEntry) -> bool: - task_id = getattr(sink, "task_id", None) - if task_id is not None and task_id != entry.task_id: - return False - session_id = getattr(sink, "session_id", None) - if session_id is not None and entry.session_id and session_id != entry.session_id: - return False - access = getattr(sink, "access", None) - if access is not None and not entry_matches_access_context(entry, access): - return False - return True - - -class TaskStatusRouter: - """Deliver task status to every live channel. A failed send is ignored.""" - - def __init__(self) -> None: - self._sinks: dict[object, TaskStatusSink] = {} - - def register(self, sink: TaskStatusSink) -> Callable[[], None]: - token = object() - self._sinks[token] = sink - - def unsubscribe() -> None: - self._sinks.pop(token, None) - - return unsubscribe - - async def notify_task_status( - self, - entry: TaskEntry, - *, - status: Optional[str] = None, - status_message: Optional[str] = None, - ) -> None: - """Fan out one status change. Never raises to the task lifecycle.""" - notification = build_task_status_notification( - entry.task_id, - status if status is not None else entry.status, - status_message=status_message - if status_message is not None - else entry.data.status_message, - ) - for sink in list(self._sinks.values()): - if not _sink_accepts(sink, entry): - continue - try: - await sink.send(notification) - except Exception: - logger.exception("task status notify failed; continuing") diff --git a/nitrostack/testing/__init__.py b/nitrostack/testing/__init__.py index 0db12b6..d10766d 100644 --- a/nitrostack/testing/__init__.py +++ b/nitrostack/testing/__init__.py @@ -15,9 +15,17 @@ class NitroTestingModule: `request_handlers` (the same dict the real stdio/HTTP transports use). """ @classmethod - async def create(cls, app_module: Type) -> "NitroTestingModule": + async def create( + cls, + app_module: Type, + *, + protocol_era: str = "auto", + ) -> "NitroTestingModule": # Construct a dummy App class decorated with @mcp_app - @mcp_app(module=app_module, server=ServerConfig(name="test-server")) + @mcp_app( + module=app_module, + server=ServerConfig(name="test-server", protocol_era=protocol_era), + ) class TestApp: pass 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_mcp20_task_authorization.py b/tests/test_mcp20_task_authorization.py index 3c6eb35..67e9617 100644 --- a/tests/test_mcp20_task_authorization.py +++ b/tests/test_mcp20_task_authorization.py @@ -167,7 +167,10 @@ async def owner_tool(self, input: EchoInput, context: ExecutionContext) -> str: class AuthModule: pass - @mcp_app(module=AuthModule, server=ServerConfig(name="task-auth")) + @mcp_app( + module=AuthModule, + server=ServerConfig(name="task-auth", protocol_era="legacy"), + ) class AuthApp: pass diff --git a/tests/test_mcp20_tasks.py b/tests/test_mcp20_tasks.py index da6bf1d..2e611c5 100644 --- a/tests/test_mcp20_tasks.py +++ b/tests/test_mcp20_tasks.py @@ -96,7 +96,10 @@ async def sync_only(self, input: EchoInput, context: ExecutionContext) -> str: class ForbiddenModule: pass - @mcp_app(module=ForbiddenModule, server=ServerConfig(name="tasks-forbidden")) + @mcp_app( + module=ForbiddenModule, + server=ServerConfig(name="tasks-forbidden", protocol_era="legacy"), + ) class ForbiddenApp: pass @@ -164,7 +167,10 @@ async def noop(self, input: EchoInput, context: ExecutionContext) -> str: class GetModule: pass - @mcp_app(module=GetModule, server=ServerConfig(name="tasks-get")) + @mcp_app( + module=GetModule, + server=ServerConfig(name="tasks-get", protocol_era="legacy"), + ) class GetApp: pass @@ -203,10 +209,7 @@ class ListApp: async def _run(): app = await McpApplicationFactory.create(ListApp) - handler = app.mcp_server.request_handlers[types.ListTasksRequest] - with pytest.raises(McpError) as exc: - await handler(types.ListTasksRequest(method="tasks/list", params={})) - assert exc.value.error.code == types.METHOD_NOT_FOUND + assert types.ListTasksRequest not in app.mcp_server.request_handlers asyncio.run(_run()) @@ -226,7 +229,10 @@ async def noop3(self, input: EchoInput, context: ExecutionContext) -> str: class CancelModule: pass - @mcp_app(module=CancelModule, server=ServerConfig(name="tasks-cancel")) + @mcp_app( + module=CancelModule, + server=ServerConfig(name="tasks-cancel", protocol_era="legacy"), + ) class CancelApp: pass @@ -256,7 +262,10 @@ async def slow_echo(self, input: EchoInput, context: ExecutionContext) -> str: class CreateModule: pass - @mcp_app(module=CreateModule, server=ServerConfig(name="tasks-create")) + @mcp_app( + module=CreateModule, + server=ServerConfig(name="tasks-create", protocol_era="legacy"), + ) class CreateApp: pass 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_mcp25_hide_tasks.py b/tests/test_mcp25_hide_tasks.py new file mode 100644 index 0000000..72a9a33 --- /dev/null +++ b/tests/test_mcp25_hide_tasks.py @@ -0,0 +1,173 @@ +"""Issue #25: the partial Tasks surface is hidden on the 2026 wire.""" + +import asyncio +import json + +import httpx +import mcp.types as types +import pytest +from mcp import MCPError +from pydantic import BaseModel + +from nitrostack import ExecutionContext, injectable, module, tool +from nitrostack.core.app import McpApplicationFactory, ServerConfig, mcp_app +from nitrostack.core.di import DIContainer +from nitrostack.core.errors import TaskNotFoundError + + +class TaskInput(BaseModel): + value: str = "" + + +@injectable() +class HideTasksController: + @tool( + name="taskable", + description="taskable", + input_schema=TaskInput, + task_support="optional", + ) + async def taskable(self, input: TaskInput, context: ExecutionContext) -> str: + return "ok" + + +@module(name="hide_tasks", controllers=[HideTasksController]) +class HideTasksModule: + pass + + +def _app(era: str): + @mcp_app(module=HideTasksModule, server=ServerConfig(name="hide-tasks", protocol_era=era)) + class App: + pass + + return asyncio.run(McpApplicationFactory.create(App)) + + +def setup_function() -> None: + DIContainer.reset() + + +def teardown_function() -> None: + DIContainer.reset() + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_2026_capabilities_and_handlers_hide_tasks(era): + app = _app(era) + server = app.mcp_server + assert server is not None + assert server.has_task_support is False + assert server.get_capabilities(protocol_version="2026-07-28").tasks is None + assert types.GetTaskRequest not in server.request_handlers + assert types.CancelTaskRequest not in server.request_handlers + assert types.GetTaskPayloadRequest not in server.request_handlers + + +def test_legacy_keeps_the_existing_in_process_task_handlers(): + app = _app("legacy") + server = app.mcp_server + assert server is not None + assert server.has_task_support is True + assert types.GetTaskRequest in server.request_handlers + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_task_augmented_tool_call_is_rejected_without_creating_a_task(era): + app = _app(era) + + async def run(): + with pytest.raises(MCPError) as exc_info: + await app._call_tool( + "taskable", + {}, + task=types.TaskMetadata(ttl=60_000), + ) + assert getattr(exc_info.value, "error", None).code == types.INVALID_PARAMS + with pytest.raises(TaskNotFoundError): + await app.task_manager.get_task("task_does_not_exist") + + asyncio.run(run()) + + +def _http_request(app, payload: dict, *, extra_headers: dict[str, str] | None = None): + async def run(): + http_app = app.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: + headers = { + "content-type": "application/json", + "accept": "application/json, text/event-stream", + } + headers.update(extra_headers or {}) + return await client.post( + "/mcp", + headers=headers, + content=json.dumps(payload), + ) + + return asyncio.run(run()) + + +@pytest.mark.parametrize("era", ["auto", "modern"]) +def test_2026_discover_does_not_advertise_tasks(era): + response = _http_request( + _app(era), + { + "jsonrpc": "2.0", + "id": 1, + "method": "server/discover", + "params": { + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientCapabilities": {}, + } + }, +}, + extra_headers={ + "Mcp-Method": "server/discover", + "MCP-Protocol-Version": "2026-07-28", + }, + ) + assert response.status_code == 200 + body = response.json() + if "error" in body: + # Modern discovery is optional in auto; when present it must still + # agree with the low-level capabilities and never advertise Tasks. + assert body["error"]["code"] == -32601 + return + capabilities = body["result"].get("capabilities", {}) + assert "tasks" not in capabilities + assert "io.modelcontextprotocol/tasks" not in capabilities.get("extensions", {}) + + +@pytest.mark.parametrize("method", ["tasks/get", "tasks/list", "tasks/result", "tasks/cancel"]) +def test_2026_task_methods_are_method_not_found(method): + response = _http_request( + _app("auto"), + {"jsonrpc": "2.0", "id": 1, "method": method, "params": {}}, + ) + assert response.status_code == 200 + body = response.json() + assert body["error"]["code"] == -32601 + + +def test_2026_task_augmented_http_call_is_not_a_task_result(): + response = _http_request( + _app("auto"), + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "taskable", + "arguments": {}, + "task": {"ttl": 60_000}, + }, + }, + ) + assert response.status_code == 200 + body = response.json() + assert body["error"]["code"] == -32602 + assert "resultType" not in body.get("result", {}) diff --git a/tests/test_tasks.py b/tests/test_tasks.py index b1dad3c..08bde04 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -471,7 +471,7 @@ class TestTasksModule: async def _mcp_task_flow(): - harness = await NitroTestingModule.create(TestTasksModule) + harness = await NitroTestingModule.create(TestTasksModule, protocol_era="legacy") req = types.CallToolRequest( method="tools/call",