From e63f6a6e95fe66540cabddda4873589094a6b1f6 Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 4 May 2026 22:02:09 +0000 Subject: [PATCH 1/3] Python: Add regression tests for GET stream resilience in streamable HTTP transport (#5317) The issue was already resolved by the MCP library upgrade to v1.27.0 (commit 094f9903b) which catches GET stream exceptions in handle_get_stream() and retries gracefully instead of propagating failures through the TaskGroup. This commit adds regression tests covering: - 405 Method Not Allowed on GET SSE (Learn MCP server behavior) - Connection reset/error on GET SSE (D365 F&O MCP server behavior) Both scenarios verify the session remains usable after the background notification stream fails. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/tests/core/test_mcp.py | 184 ++++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 01cf1717bdd..4ede31d0cd3 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -4178,3 +4178,187 @@ def provider(kwargs): # endregion + + +# region: MCPStreamableHTTPTool GET stream resilience + + +async def test_streamable_http_get_stream_405_does_not_crash_session(): + """Test that a 405 response on the GET SSE notification stream is handled gracefully. + + Some MCP servers (e.g. Learn MCP) reject GET requests with 405 because they only + support the modern Streamable HTTP transport. The background GET notification task + must not propagate that failure to the main session. + """ + import asyncio + + import httpx + from mcp.client.streamable_http import streamable_http_client + + session_id = "test-session-resilience" + + async def handle_request(request: httpx.Request) -> httpx.Response: + if request.method == "GET": + return httpx.Response( + status_code=405, + headers={"content-type": "text/plain"}, + content=b"This endpoint does not support SSE transport.", + ) + + if request.method == "POST": + body = json.loads(request.content) + method = body.get("method") + + if method == "initialize": + return httpx.Response( + status_code=200, + headers={ + "content-type": "application/json", + "mcp-session-id": session_id, + }, + content=json.dumps({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "protocolVersion": "2025-03-26", + "capabilities": {"tools": {"listChanged": True}}, + "serverInfo": {"name": "learn-mcp", "version": "1.0.0"}, + }, + }).encode(), + ) + + if method == "notifications/initialized": + return httpx.Response(status_code=202) + + if method == "tools/list": + return httpx.Response( + status_code=200, + headers={"content-type": "application/json"}, + content=json.dumps({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "tools": [{ + "name": "search_docs", + "description": "Search documentation", + "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}, + }] + }, + }).encode(), + ) + + return httpx.Response(status_code=202) + + if request.method == "DELETE": + return httpx.Response(status_code=200) + + return httpx.Response(status_code=404) + + transport = httpx.MockTransport(handle_request) + http_client = httpx.AsyncClient(transport=transport) + + async with http_client: + async with streamable_http_client( + url="http://test-server/mcp", + http_client=http_client, + terminate_on_close=False, + ) as (read_stream, write_stream, get_session_id): + async with ClientSession( + read_stream=read_stream, + write_stream=write_stream, + ) as session: + result = await session.initialize() + assert result.serverInfo.name == "learn-mcp" + + # Allow time for the background GET stream attempt to fail + await asyncio.sleep(0.5) + + # Session must remain usable after the GET stream failure + tools = await session.list_tools() + assert len(tools.tools) == 1 + assert tools.tools[0].name == "search_docs" + + +async def test_streamable_http_get_stream_connection_error_does_not_crash_session(): + """Test that a connection error on GET SSE notification stream is handled gracefully. + + Some MCP servers only accept POST and the background GET either resets or refuses + the connection. This must not cancel the main session task. + """ + import asyncio + + import httpx + from mcp.client.streamable_http import streamable_http_client + + async def handle_request(request: httpx.Request) -> httpx.Response: + if request.method == "GET": + raise httpx.ConnectError("Connection reset by peer") + + if request.method == "POST": + body = json.loads(request.content) + method = body.get("method") + + if method == "initialize": + return httpx.Response( + status_code=200, + headers={ + "content-type": "application/json", + "mcp-session-id": "d365-session", + }, + content=json.dumps({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "protocolVersion": "2025-03-26", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "d365-server", "version": "1.0.0"}, + }, + }).encode(), + ) + + if method == "notifications/initialized": + return httpx.Response(status_code=202) + + if method == "tools/list": + return httpx.Response( + status_code=200, + headers={"content-type": "application/json"}, + content=json.dumps({ + "jsonrpc": "2.0", + "id": body["id"], + "result": {"tools": []}, + }).encode(), + ) + + return httpx.Response(status_code=202) + + if request.method == "DELETE": + return httpx.Response(status_code=200) + + return httpx.Response(status_code=404) + + transport = httpx.MockTransport(handle_request) + http_client = httpx.AsyncClient(transport=transport) + + async with http_client: + async with streamable_http_client( + url="http://test-server/mcp", + http_client=http_client, + terminate_on_close=False, + ) as (read_stream, write_stream, get_session_id): + async with ClientSession( + read_stream=read_stream, + write_stream=write_stream, + ) as session: + result = await session.initialize() + assert result.serverInfo.name == "d365-server" + + # Allow time for the background GET stream reconnection attempts + await asyncio.sleep(1.0) + + # Session must remain usable after the GET stream failure + tools = await session.list_tools() + assert tools.tools is not None + + +# endregion From c6026739919a6bdd2f52e2a0a518e58a87389995 Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 4 May 2026 22:20:47 +0000 Subject: [PATCH 2/3] Python: Add regression tests for MCPStreamableHTTPTool GET stream resilience (#5317) Fixes #5317 --- python/packages/core/tests/core/test_mcp.py | 78 +++++++++++---------- 1 file changed, 42 insertions(+), 36 deletions(-) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index 4ede31d0cd3..aa9bc6f5178 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -4238,11 +4238,13 @@ async def handle_request(request: httpx.Request) -> httpx.Response: "jsonrpc": "2.0", "id": body["id"], "result": { - "tools": [{ - "name": "search_docs", - "description": "Search documentation", - "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}, - }] + "tools": [ + { + "name": "search_docs", + "description": "Search documentation", + "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}, + } + ] }, }).encode(), ) @@ -4257,26 +4259,28 @@ async def handle_request(request: httpx.Request) -> httpx.Response: transport = httpx.MockTransport(handle_request) http_client = httpx.AsyncClient(transport=transport) - async with http_client: - async with streamable_http_client( + async with ( + http_client, + streamable_http_client( url="http://test-server/mcp", http_client=http_client, terminate_on_close=False, - ) as (read_stream, write_stream, get_session_id): - async with ClientSession( - read_stream=read_stream, - write_stream=write_stream, - ) as session: - result = await session.initialize() - assert result.serverInfo.name == "learn-mcp" + ) as (read_stream, write_stream, get_session_id), + ClientSession( + read_stream=read_stream, + write_stream=write_stream, + ) as session, + ): + result = await session.initialize() + assert result.serverInfo.name == "learn-mcp" - # Allow time for the background GET stream attempt to fail - await asyncio.sleep(0.5) + # Allow time for the background GET stream attempt to fail + await asyncio.sleep(0.5) - # Session must remain usable after the GET stream failure - tools = await session.list_tools() - assert len(tools.tools) == 1 - assert tools.tools[0].name == "search_docs" + # Session must remain usable after the GET stream failure + tools = await session.list_tools() + assert len(tools.tools) == 1 + assert tools.tools[0].name == "search_docs" async def test_streamable_http_get_stream_connection_error_does_not_crash_session(): @@ -4340,25 +4344,27 @@ async def handle_request(request: httpx.Request) -> httpx.Response: transport = httpx.MockTransport(handle_request) http_client = httpx.AsyncClient(transport=transport) - async with http_client: - async with streamable_http_client( + async with ( + http_client, + streamable_http_client( url="http://test-server/mcp", http_client=http_client, terminate_on_close=False, - ) as (read_stream, write_stream, get_session_id): - async with ClientSession( - read_stream=read_stream, - write_stream=write_stream, - ) as session: - result = await session.initialize() - assert result.serverInfo.name == "d365-server" - - # Allow time for the background GET stream reconnection attempts - await asyncio.sleep(1.0) - - # Session must remain usable after the GET stream failure - tools = await session.list_tools() - assert tools.tools is not None + ) as (read_stream, write_stream, get_session_id), + ClientSession( + read_stream=read_stream, + write_stream=write_stream, + ) as session, + ): + result = await session.initialize() + assert result.serverInfo.name == "d365-server" + + # Allow time for the background GET stream reconnection attempts + await asyncio.sleep(1.0) + + # Session must remain usable after the GET stream failure + tools = await session.list_tools() + assert tools.tools is not None # endregion From c8cf0df20d32a2d4cdb009c03af2b4877ad9b331 Mon Sep 17 00:00:00 2001 From: Copilot Date: Mon, 4 May 2026 22:28:28 +0000 Subject: [PATCH 3/3] fix(tests): replace sleep-based waits with event synchronization in GET stream tests Replace asyncio.sleep() with asyncio.Event tracking in the GET stream resilience regression tests. This ensures the tests assert that the background GET request was actually attempted, preventing false positives when the GET path is never exercised. Fixes #5317 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- python/packages/core/tests/core/test_mcp.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py index aa9bc6f5178..7aefcad654a 100644 --- a/python/packages/core/tests/core/test_mcp.py +++ b/python/packages/core/tests/core/test_mcp.py @@ -4196,9 +4196,11 @@ async def test_streamable_http_get_stream_405_does_not_crash_session(): from mcp.client.streamable_http import streamable_http_client session_id = "test-session-resilience" + get_attempted = asyncio.Event() async def handle_request(request: httpx.Request) -> httpx.Response: if request.method == "GET": + get_attempted.set() return httpx.Response( status_code=405, headers={"content-type": "text/plain"}, @@ -4274,8 +4276,8 @@ async def handle_request(request: httpx.Request) -> httpx.Response: result = await session.initialize() assert result.serverInfo.name == "learn-mcp" - # Allow time for the background GET stream attempt to fail - await asyncio.sleep(0.5) + # Wait for the background GET stream to be attempted + await asyncio.wait_for(get_attempted.wait(), timeout=5.0) # Session must remain usable after the GET stream failure tools = await session.list_tools() @@ -4294,8 +4296,11 @@ async def test_streamable_http_get_stream_connection_error_does_not_crash_sessio import httpx from mcp.client.streamable_http import streamable_http_client + get_attempted = asyncio.Event() + async def handle_request(request: httpx.Request) -> httpx.Response: if request.method == "GET": + get_attempted.set() raise httpx.ConnectError("Connection reset by peer") if request.method == "POST": @@ -4359,8 +4364,8 @@ async def handle_request(request: httpx.Request) -> httpx.Response: result = await session.initialize() assert result.serverInfo.name == "d365-server" - # Allow time for the background GET stream reconnection attempts - await asyncio.sleep(1.0) + # Wait for the background GET stream to be attempted + await asyncio.wait_for(get_attempted.wait(), timeout=5.0) # Session must remain usable after the GET stream failure tools = await session.list_tools()