From 2955d849df9bdcdf87022971dd0877be27d6a50c Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 14:08:14 +0800 Subject: [PATCH 1/7] fix: handle ClosedResourceError in StreamableHTTP message router --- src/mcp/server/streamable_http.py | 5 + ...est_1363_race_condition_streamable_http.py | 234 ++++++++++++++++++ 2 files changed, 239 insertions(+) create mode 100644 tests/issues/test_1363_race_condition_streamable_http.py diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index b45d742b0..4f5944b8a 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -875,6 +875,11 @@ async def message_router(): for message. Still processing message as the client might reconnect and replay.""" ) + except anyio.ClosedResourceError: + if self._terminated: + logging.debug("Read stream closed by client") + else: + logging.exception("Unexpected closure of read stream in message router") except Exception: logger.exception("Error in message router") diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py new file mode 100644 index 000000000..6dbcf6699 --- /dev/null +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -0,0 +1,234 @@ +"""Test for issue #1363 - Race condition in StreamableHTTP transport causes ClosedResourceError. + +This test reproduces the race condition described in issue #1363 where MCP servers +in HTTP Streamable mode experience ClosedResourceError exceptions when requests +fail validation early (e.g., due to incorrect Accept headers). + +The race condition occurs because: +1. Transport setup creates a message_router task +2. Message router enters async for write_stream_reader loop +3. write_stream_reader calls checkpoint() in receive(), yielding control +4. Request handling processes HTTP request +5. If validation fails early, request returns immediately +6. Transport termination closes all streams including write_stream_reader +7. Message router may still be in checkpoint() yield and hasn't returned to check stream state +8. When message router resumes, it encounters a closed stream, raising ClosedResourceError +""" + +import socket +import subprocess +import sys +import time +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + +import httpx +import pytest +import uvicorn +from starlette.applications import Starlette +from starlette.routing import Mount +from starlette.types import Receive, Scope, Send + +from mcp.server import Server +from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.types import Tool + +SERVER_NAME = "test_race_condition_server" + + +def check_server_logs_for_errors(process, test_name: str): + """ + Check server logs for ClosedResourceError and other race condition errors. + + Args: + process: The server process + test_name: Name of the test for better error messages + """ + # Get logs from the process + try: + stdout, stderr = process.communicate(timeout=10) + server_logs = stderr + stdout + except Exception: + server_logs = "" + + # Check for specific race condition errors + errors_found = [] + + if "ClosedResourceError" in server_logs: + errors_found.append("ClosedResourceError") + + if "Error in message router" in server_logs: + errors_found.append("Error in message router") + + if "anyio.ClosedResourceError" in server_logs: + errors_found.append("anyio.ClosedResourceError") + + # Assert no race condition errors occurred + if errors_found: + error_msg = f"Test '{test_name}' found race condition errors: {', '.join(errors_found)}\n" + error_msg += f"Server logs:\n{server_logs}" + pytest.fail(error_msg) + + # If we get here, no race condition errors were found + print(f"✓ Test '{test_name}' passed: No race condition errors detected") + + +@pytest.fixture +def server_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def server_url(server_port: int) -> str: + return f"http://127.0.0.1:{server_port}" + + +class RaceConditionTestServer(Server): + def __init__(self): + super().__init__(SERVER_NAME) + + async def on_list_tools(self) -> list[Tool]: + return [] + + +def run_server_with_logging(port: int): + """Run the StreamableHTTP server with logging to capture race condition errors.""" + app = RaceConditionTestServer() + + # Create session manager + session_manager = StreamableHTTPSessionManager( + app=app, + json_response=False, + stateless=True, # Use stateless mode to trigger the race condition + ) + + # Create the ASGI handler + async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None: + await session_manager.handle_request(scope, receive, send) + + # Create Starlette app with lifespan + @asynccontextmanager + async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: + async with session_manager.run(): + yield + + routes = [ + Mount("/", app=handle_streamable_http), + ] + + starlette_app = Starlette(routes=routes, lifespan=lifespan) + uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="debug") + + +def start_server_process(port: int): + """Start server in a separate process.""" + # Create a temporary script to run the server + import os + import tempfile + + script_content = f""" +import sys +sys.path.insert(0, '{os.getcwd()}') +from tests.issues.test_1363_race_condition_streamable_http import run_server_with_logging +run_server_with_logging({port}) +""" + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write(script_content) + script_path = f.name + + process = subprocess.Popen([sys.executable, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + + # Give server time to start + time.sleep(1) + return process + + +@pytest.mark.anyio +async def test_race_condition_invalid_accept_headers(server_port: int): + """ + Test the race condition with invalid Accept headers. + + This test reproduces the exact scenario described in issue #1363: + - Send POST request with incorrect Accept headers (missing either application/json or text/event-stream) + - Request fails validation early and returns quickly + - This should trigger the race condition where message_router encounters ClosedResourceError + """ + process = start_server_process(server_port) + + try: + # Test with missing text/event-stream in Accept header + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers={ + "Accept": "application/json", # Missing text/event-stream + "Content-Type": "application/json", + }, + ) + # Should get 406 Not Acceptable due to missing text/event-stream + assert response.status_code == 406 + + # Test with missing application/json in Accept header + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers={ + "Accept": "text/event-stream", # Missing application/json + "Content-Type": "application/json", + }, + ) + # Should get 406 Not Acceptable due to missing application/json + assert response.status_code == 406 + + # Test with completely invalid Accept header + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers={ + "Accept": "text/plain", # Invalid Accept header + "Content-Type": "application/json", + }, + ) + # Should get 406 Not Acceptable + assert response.status_code == 406 + + finally: + process.terminate() + process.wait() + # Check server logs for race condition errors + check_server_logs_for_errors(process, "test_race_condition_invalid_accept_headers") + + +@pytest.mark.anyio +async def test_race_condition_invalid_content_type(server_port: int): + """ + Test the race condition with invalid Content-Type headers. + + This test reproduces the race condition scenario with Content-Type validation failure. + """ + process = start_server_process(server_port) + + try: + # Test with invalid Content-Type + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.post( + f"http://127.0.0.1:{server_port}/", + json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, + headers={ + "Accept": "application/json, text/event-stream", + "Content-Type": "text/plain", # Invalid Content-Type + }, + ) + assert response.status_code == 400 + + finally: + process.terminate() + process.wait() + # Check server logs for race condition errors + check_server_logs_for_errors(process, "test_race_condition_invalid_content_type") From f8072c90f0932d6de3b5904d2853671a69e7d3b0 Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 14:14:41 +0800 Subject: [PATCH 2/7] Add type annotations to race condition test functions --- tests/issues/test_1363_race_condition_streamable_http.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py index 6dbcf6699..7146a8385 100644 --- a/tests/issues/test_1363_race_condition_streamable_http.py +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -36,7 +36,7 @@ SERVER_NAME = "test_race_condition_server" -def check_server_logs_for_errors(process, test_name: str): +def check_server_logs_for_errors(process: subprocess.Popen[str], test_name: str) -> None: """ Check server logs for ClosedResourceError and other race condition errors. @@ -52,7 +52,7 @@ def check_server_logs_for_errors(process, test_name: str): server_logs = "" # Check for specific race condition errors - errors_found = [] + errors_found: list[str] = [] if "ClosedResourceError" in server_logs: errors_found.append("ClosedResourceError") @@ -93,7 +93,7 @@ async def on_list_tools(self) -> list[Tool]: return [] -def run_server_with_logging(port: int): +def run_server_with_logging(port: int) -> None: """Run the StreamableHTTP server with logging to capture race condition errors.""" app = RaceConditionTestServer() @@ -122,7 +122,7 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="debug") -def start_server_process(port: int): +def start_server_process(port: int) -> subprocess.Popen[str]: """Start server in a separate process.""" # Create a temporary script to run the server import os From 6a7ecd5e5eb49a8347e12390ead610cf8b93236d Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 14:24:27 +0800 Subject: [PATCH 3/7] fix: add connection waiting mechanism for Windows compatibility --- ...est_1363_race_condition_streamable_http.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py index 7146a8385..ddf288faf 100644 --- a/tests/issues/test_1363_race_condition_streamable_http.py +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -141,8 +141,23 @@ def start_server_process(port: int) -> subprocess.Popen[str]: process = subprocess.Popen([sys.executable, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - # Give server time to start - time.sleep(1) + # Wait for server to be running with connection testing (like other tests) + max_attempts = 20 + attempt = 0 + while attempt < max_attempts: + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.connect(("127.0.0.1", port)) + break + except ConnectionRefusedError: + time.sleep(0.1) + attempt += 1 + else: + # If server failed to start, terminate the process and raise an error + process.terminate() + process.wait() + raise RuntimeError(f"Server failed to start after {max_attempts} attempts") + return process From 3f3c8ae5560c2be74f391371b0af660a81ed4fee Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 14:54:14 +0800 Subject: [PATCH 4/7] fix: escape path in test script content using repr() --- tests/issues/test_1363_race_condition_streamable_http.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py index ddf288faf..e9e7e603d 100644 --- a/tests/issues/test_1363_race_condition_streamable_http.py +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -130,7 +130,7 @@ def start_server_process(port: int) -> subprocess.Popen[str]: script_content = f""" import sys -sys.path.insert(0, '{os.getcwd()}') +sys.path.insert(0, {repr(os.getcwd())}) from tests.issues.test_1363_race_condition_streamable_http import run_server_with_logging run_server_with_logging({port}) """ From c76b28d3ab0735adb65514e699e5e3fc2971adce Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 15:04:08 +0800 Subject: [PATCH 5/7] refactor(test_1363): eliminate temporary files by using python -c --- .../test_1363_race_condition_streamable_http.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py index e9e7e603d..088db0aca 100644 --- a/tests/issues/test_1363_race_condition_streamable_http.py +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -126,20 +126,17 @@ def start_server_process(port: int) -> subprocess.Popen[str]: """Start server in a separate process.""" # Create a temporary script to run the server import os - import tempfile - script_content = f""" + server_code = f""" import sys sys.path.insert(0, {repr(os.getcwd())}) from tests.issues.test_1363_race_condition_streamable_http import run_server_with_logging run_server_with_logging({port}) """ - with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: - f.write(script_content) - script_path = f.name - - process = subprocess.Popen([sys.executable, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + process = subprocess.Popen( + [sys.executable, "-c", server_code], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) # Wait for server to be running with connection testing (like other tests) max_attempts = 20 From 1b8b779312e8b5807d5eb9f7a2625dcb13057c5b Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Sun, 21 Sep 2025 17:14:42 +0800 Subject: [PATCH 6/7] fix: use logger instance instead of logging module in streamable_http --- src/mcp/server/streamable_http.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 4f5944b8a..8d3b4104c 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -870,16 +870,16 @@ async def message_router(): # Stream might be closed, remove from registry self._request_streams.pop(request_stream_id, None) else: - logging.debug( + logger.debug( f"""Request stream {request_stream_id} not found for message. Still processing message as the client might reconnect and replay.""" ) except anyio.ClosedResourceError: if self._terminated: - logging.debug("Read stream closed by client") + logger.debug("Read stream closed by client") else: - logging.exception("Unexpected closure of read stream in message router") + logger.exception("Unexpected closure of read stream in message router") except Exception: logger.exception("Error in message router") From 2c02f06ae8c80f9a579bd71a0de544e62b8b164c Mon Sep 17 00:00:00 2001 From: Edison-A-N Date: Tue, 23 Sep 2025 13:48:02 +0800 Subject: [PATCH 7/7] refactor: inline server code in test_1363 to avoid external imports --- ...est_1363_race_condition_streamable_http.py | 48 ++++++++++--------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py index 088db0aca..09cbd8db8 100644 --- a/tests/issues/test_1363_race_condition_streamable_http.py +++ b/tests/issues/test_1363_race_condition_streamable_http.py @@ -19,19 +19,9 @@ import subprocess import sys import time -from collections.abc import AsyncGenerator -from contextlib import asynccontextmanager import httpx import pytest -import uvicorn -from starlette.applications import Starlette -from starlette.routing import Mount -from starlette.types import Receive, Scope, Send - -from mcp.server import Server -from mcp.server.streamable_http_manager import StreamableHTTPSessionManager -from mcp.types import Tool SERVER_NAME = "test_race_condition_server" @@ -85,6 +75,32 @@ def server_url(server_port: int) -> str: return f"http://127.0.0.1:{server_port}" +def start_server_process(port: int) -> subprocess.Popen[str]: + """Start server in a separate process.""" + # Create a temporary script to run the server + import os + + server_code = f""" +import sys +import os +sys.path.insert(0, {repr(os.getcwd())}) + +import socket +import time +from collections.abc import AsyncGenerator +from contextlib import asynccontextmanager + +import uvicorn +from starlette.applications import Starlette +from starlette.routing import Mount +from starlette.types import Receive, Scope, Send + +from mcp.server import Server +from mcp.server.streamable_http_manager import StreamableHTTPSessionManager +from mcp.types import Tool + +SERVER_NAME = "test_race_condition_server" + class RaceConditionTestServer(Server): def __init__(self): super().__init__(SERVER_NAME) @@ -92,9 +108,7 @@ def __init__(self): async def on_list_tools(self) -> list[Tool]: return [] - def run_server_with_logging(port: int) -> None: - """Run the StreamableHTTP server with logging to capture race condition errors.""" app = RaceConditionTestServer() # Create session manager @@ -121,16 +135,6 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: starlette_app = Starlette(routes=routes, lifespan=lifespan) uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="debug") - -def start_server_process(port: int) -> subprocess.Popen[str]: - """Start server in a separate process.""" - # Create a temporary script to run the server - import os - - server_code = f""" -import sys -sys.path.insert(0, {repr(os.getcwd())}) -from tests.issues.test_1363_race_condition_streamable_http import run_server_with_logging run_server_with_logging({port}) """