refactor: use typed models and add test for session 404 response

- Replace raw dict with JSONRPCError and ErrorData types for consistency
  with the rest of the codebase
- Use INVALID_REQUEST (-32600) error code instead of -32001, which is in
  the reserved JSON-RPC implementation range (see spec issue #509)
- Remove pragma: no cover and add unit test for unknown session ID
- Test verifies HTTP 404 status and proper JSON-RPC error format

Github-Issue:#1727

Author: maxisbey

src/mcp/server/streamable_http_manager.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import contextlib
-import json
 import logging
 from collections.abc import AsyncIterator
 from http import HTTPStatus
@@ -23,6 +22,7 @@
     StreamableHTTPServerTransport,
 )
 from mcp.server.transport_security import TransportSecuritySettings
+from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError
 
 logger = logging.getLogger(__name__)
 
@@ -277,21 +277,18 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
 
                 # Handle the HTTP request and return the response
                 await http_transport.handle_request(scope, receive, send)
-        else:  # pragma: no cover
+        else:
             # Unknown or expired session ID - return 404 per MCP spec
-            # Match TypeScript SDK exactly: jsonrpc, error, id order
-            error_body = json.dumps(
-                {
-                    "jsonrpc": "2.0",
-                    "error": {
-                        "code": -32001,
-                        "message": "Session not found",
-                    },
-                    "id": None,
-                }
+            error_response = JSONRPCError(
+                jsonrpc="2.0",
+                id="server-error",
+                error=ErrorData(
+                    code=INVALID_REQUEST,
+                    message="Session not found",
+                ),
             )
             response = Response(
-                content=error_body,
+                content=error_response.model_dump_json(by_alias=True, exclude_none=True),
                 status_code=HTTPStatus.NOT_FOUND,
                 media_type="application/json",
             )

tests/server/test_streamable_http_manager.py

@@ -1,5 +1,6 @@
 """Tests for StreamableHTTPSessionManager."""
 
+import json
 from typing import Any
 from unittest.mock import AsyncMock, patch
 
@@ -11,6 +12,7 @@
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.types import INVALID_REQUEST
 
 
 @pytest.mark.anyio
@@ -262,3 +264,52 @@ async def mock_receive():
 
             # Verify internal state is cleaned up
             assert len(transport._request_streams) == 0, "Transport should have no active request streams"
+
+
+@pytest.mark.anyio
+async def test_unknown_session_id_returns_404():
+    """Test that requests with unknown session IDs return HTTP 404 per MCP spec."""
+    app = Server("test-unknown-session")
+    manager = StreamableHTTPSessionManager(app=app)
+
+    async with manager.run():
+        sent_messages: list[Message] = []
+        response_body = b""
+
+        async def mock_send(message: Message):
+            nonlocal response_body
+            sent_messages.append(message)
+            if message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+
+        # Request with a non-existent session ID
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"accept", b"application/json, text/event-stream"),
+                (b"mcp-session-id", b"non-existent-session-id"),
+            ],
+        }
+
+        async def mock_receive():
+            return {"type": "http.request", "body": b"{}", "more_body": False}
+
+        await manager.handle_request(scope, mock_receive, mock_send)
+
+        # Find the response start message
+        response_start = next(
+            (msg for msg in sent_messages if msg["type"] == "http.response.start"),
+            None,
+        )
+        assert response_start is not None, "Should have sent a response"
+        assert response_start["status"] == 404, "Should return HTTP 404 for unknown session ID"
+
+        # Verify JSON-RPC error format
+        error_data = json.loads(response_body)
+        assert error_data["jsonrpc"] == "2.0"
+        assert error_data["id"] == "server-error"
+        assert error_data["error"]["code"] == INVALID_REQUEST
+        assert error_data["error"]["message"] == "Session not found"