From 98edf1a05d0467c905b980a35303829063ff5cf2 Mon Sep 17 00:00:00 2001 From: "marvin-context-protocol[bot]" <225465937+marvin-context-protocol[bot]@users.noreply.github.com> Date: Thu, 28 Aug 2025 22:37:37 +0000 Subject: [PATCH] Implement SSE fallback for 405 errors in MCP client MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add backwards compatibility support for MCP clients when connecting to SSE-only servers. When a StreamableHttpTransport receives a 405 Method Not Allowed error, the client now automatically falls back to SSETransport using the same URL and authentication settings. This implements the backwards compatibility requirement from the MCP specification: https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#backwards-compatibility Key changes: - Modified Client._context_manager() to detect 405 errors from StreamableHttpTransport - Automatically retry with SSETransport preserving auth, headers, and timeouts - Added comprehensive test coverage for fallback behavior - Non-405 HTTP errors are re-raised without fallback attempt Fixes #853 🤖 Generated with [Claude Code](https://claude.ai/code) Co-authored-by: William Easton --- src/fastmcp/client/client.py | 46 ++++++- tests/client/test_client.py | 239 +++++++++++++++++++++++++++++++++++ 2 files changed, 284 insertions(+), 1 deletion(-) diff --git a/src/fastmcp/client/client.py b/src/fastmcp/client/client.py index 2592873d7..a46baf56c 100644 --- a/src/fastmcp/client/client.py +++ b/src/fastmcp/client/client.py @@ -351,7 +351,51 @@ class Client(Generic[ClientTransportT]): @asynccontextmanager async def _context_manager(self): with catch(get_catch_handlers()): - async with self.transport.connect_session( + transport_to_use = self.transport + + # Check if we should try SSE fallback for 405 errors + # This implements the backwards compatibility requirement from the MCP specification + if isinstance(self.transport, StreamableHttpTransport): + try: + async with transport_to_use.connect_session( + **self._session_kwargs + ) as session: + self._session_state.session = session + # Initialize the session + try: + with anyio.fail_after(self._init_timeout): + self._session_state.initialize_result = ( + await self._session_state.session.initialize() + ) + yield + return # Success, no need to try fallback + except anyio.ClosedResourceError: + raise RuntimeError("Server session was closed unexpectedly") + except TimeoutError: + raise RuntimeError("Failed to initialize server session") + finally: + self._session_state.session = None + self._session_state.initialize_result = None + except httpx.HTTPStatusError as e: + # Handle 405 Method Not Allowed - try SSE fallback + if e.response.status_code == 405: + logger.debug( + f"[{self.name}] Received 405 error from StreamableHttpTransport, attempting SSE fallback" + ) + # Create SSE transport with same URL and auth settings + transport_to_use = SSETransport( + url=self.transport.url, + headers=self.transport.headers, + auth=self.transport.auth, + sse_read_timeout=self.transport.sse_read_timeout, + httpx_client_factory=self.transport.httpx_client_factory, + ) + else: + # Re-raise non-405 HTTP errors + raise + + # Use the (possibly fallback) transport + async with transport_to_use.connect_session( **self._session_kwargs ) as session: self._session_state.session = session diff --git a/tests/client/test_client.py b/tests/client/test_client.py index bae0ca81f..ef0c8a231 100644 --- a/tests/client/test_client.py +++ b/tests/client/test_client.py @@ -1032,3 +1032,242 @@ class TestAuth: assert isinstance(client.transport, SSETransport) assert isinstance(client.transport.auth, BearerAuth) assert client.transport.auth.token.get_secret_value() == "test_token" + + +class TestSSEFallback: + """Tests for the SSE fallback functionality when StreamableHttpTransport receives 405 errors.""" + + async def test_sse_fallback_on_405_error(self, fastmcp_server): + """Test that client falls back to SSE transport when receiving 405 error from StreamableHttp.""" + from contextlib import asynccontextmanager + from unittest.mock import Mock, patch + + import httpx + + # Create a client with StreamableHttpTransport + from fastmcp.client.transports import StreamableHttpTransport + + # Mock the StreamableHttpTransport to raise 405 error + sse_connect_called = False + + @asynccontextmanager + async def mock_streamable_connect(*args, **kwargs): + response = Mock() + response.status_code = 405 + raise httpx.HTTPStatusError( + "405 Method Not Allowed", request=Mock(), response=response + ) + yield # This won't be reached due to the exception + + @asynccontextmanager + async def mock_sse_connect(*args, **kwargs): + nonlocal sse_connect_called + sse_connect_called = True + # Use the in-memory FastMCP transport for successful connection + transport = FastMCPTransport(fastmcp_server) + async with transport.connect_session(**kwargs) as session: + yield session + + with ( + patch.object( + StreamableHttpTransport, "connect_session", mock_streamable_connect + ), + patch.object(SSETransport, "connect_session", mock_sse_connect), + ): + client = Client("http://localhost:8888/mcp") + assert isinstance(client.transport, StreamableHttpTransport) + + # The client should fallback to SSE and succeed + async with client: + tools = await client.list_tools() + assert len(tools) >= 1 # We have tools from fastmcp_server + assert sse_connect_called # Verify SSE fallback was used + + async def test_sse_fallback_preserves_auth_and_headers(self): + """Test that SSE fallback preserves auth and headers from original transport.""" + from contextlib import asynccontextmanager + from unittest.mock import Mock, patch + + import httpx + + from fastmcp.client.auth.bearer import BearerAuth + + # Create client with auth and headers + transport = StreamableHttpTransport( + "http://localhost:8888/mcp", + headers={"Custom-Header": "test-value"}, + auth="test-token", + ) + client = Client(transport) + + fallback_transport = None + + @asynccontextmanager + async def mock_streamable_connect(*args, **kwargs): + response = Mock() + response.status_code = 405 + raise httpx.HTTPStatusError( + "405 Method Not Allowed", request=Mock(), response=response + ) + yield # This won't be reached due to the exception + + @asynccontextmanager + async def mock_sse_connect(self, *args, **kwargs): + nonlocal fallback_transport + fallback_transport = self + # Just return without yielding to avoid actual connection + if False: + yield + + with ( + patch.object( + StreamableHttpTransport, "connect_session", mock_streamable_connect + ), + patch.object(SSETransport, "connect_session", mock_sse_connect), + ): + try: + async with client: + pass + except Exception: + pass # We expect this to fail since we're not yielding from mock_sse_connect + + # Verify fallback transport has correct configuration + assert fallback_transport is not None + assert fallback_transport.url == "http://localhost:8888/mcp" + assert fallback_transport.headers == {"Custom-Header": "test-value"} + assert isinstance(fallback_transport.auth, BearerAuth) + assert fallback_transport.auth.token.get_secret_value() == "test-token" + + async def test_non_405_http_errors_not_fallback(self): + """Test that non-405 HTTP errors are re-raised without SSE fallback.""" + from contextlib import asynccontextmanager + from unittest.mock import Mock, patch + + import httpx + + @asynccontextmanager + async def mock_streamable_connect(*args, **kwargs): + response = Mock() + response.status_code = 404 + raise httpx.HTTPStatusError( + "404 Not Found", request=Mock(), response=response + ) + yield # This won't be reached due to the exception + + with patch.object( + StreamableHttpTransport, "connect_session", mock_streamable_connect + ): + client = Client("http://localhost:8888/mcp") + + # Should re-raise the 404 error without attempting SSE fallback + with pytest.raises(httpx.HTTPStatusError, match="404 Not Found"): + async with client: + pass + + async def test_sse_transport_not_affected_by_fallback_logic(self, fastmcp_server): + """Test that SSE transport works normally and isn't affected by fallback logic.""" + from contextlib import asynccontextmanager + from unittest.mock import patch + + # Create client with SSE transport directly + client = Client(transport=SSETransport("http://localhost:8888/sse")) + + # Mock SSE transport to succeed + @asynccontextmanager + async def mock_sse_connect(*args, **kwargs): + transport = FastMCPTransport(fastmcp_server) + async with transport.connect_session(**kwargs) as session: + yield session + + with patch.object(SSETransport, "connect_session", mock_sse_connect): + async with client: + tools = await client.list_tools() + assert len(tools) >= 1 + + async def test_other_transport_types_not_affected(self, fastmcp_server): + """Test that other transport types (e.g., FastMCPTransport) work normally.""" + client = Client(transport=FastMCPTransport(fastmcp_server)) + + async with client: + tools = await client.list_tools() + assert len(tools) >= 1 + + async def test_sse_fallback_happens_on_405(self, fastmcp_server): + """Test that SSE fallback occurs when 405 error is received.""" + from contextlib import asynccontextmanager + from unittest.mock import Mock, patch + + import httpx + + # Track whether fallback happened + fallback_occurred = False + + @asynccontextmanager + async def mock_streamable_connect(*args, **kwargs): + response = Mock() + response.status_code = 405 + raise httpx.HTTPStatusError( + "405 Method Not Allowed", request=Mock(), response=response + ) + yield # This won't be reached due to the exception + + @asynccontextmanager + async def mock_sse_connect(*args, **kwargs): + nonlocal fallback_occurred + fallback_occurred = True + transport = FastMCPTransport(fastmcp_server) + async with transport.connect_session(**kwargs) as session: + yield session + + with ( + patch.object( + StreamableHttpTransport, "connect_session", mock_streamable_connect + ), + patch.object(SSETransport, "connect_session", mock_sse_connect), + ): + client = Client("http://localhost:8888/mcp") + + async with client: + await client.list_tools() + + # Verify that fallback occurred + assert fallback_occurred + + async def test_sse_fallback_with_successful_tool_call(self, fastmcp_server): + """Test that after SSE fallback, tool calls work correctly.""" + from contextlib import asynccontextmanager + from unittest.mock import Mock, patch + + import httpx + + @asynccontextmanager + async def mock_streamable_connect(*args, **kwargs): + response = Mock() + response.status_code = 405 + raise httpx.HTTPStatusError( + "405 Method Not Allowed", request=Mock(), response=response + ) + yield # This won't be reached due to the exception + + @asynccontextmanager + async def mock_sse_connect(*args, **kwargs): + transport = FastMCPTransport(fastmcp_server) + async with transport.connect_session(**kwargs) as session: + yield session + + with ( + patch.object( + StreamableHttpTransport, "connect_session", mock_streamable_connect + ), + patch.object(SSETransport, "connect_session", mock_sse_connect), + ): + client = Client("http://localhost:8888/mcp") + + async with client: + # Test tool call after fallback + result = await client.call_tool("greet", {"name": "World"}) + assert result.data == "Hello, World!" + + # Test another tool call + result = await client.call_tool("add", {"a": 5, "b": 3}) + assert result.data == 8