Disable redirects in streamable HTTP transport

🤖 Generated with GPT-5.2-Codex

src/fastmcp/client/transports/http.py

@@ -10,7 +10,7 @@ from typing import Literal, cast
 import httpx
 from mcp import ClientSession
 from mcp.client.streamable_http import streamable_http_client
-from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
+from mcp.shared._httpx_utils import McpHttpClientFactory
 from pydantic import AnyUrl
 from typing_extensions import Unpack
 
@@ -106,21 +106,20 @@ class StreamableHttpTransport(ClientTransport):
             timeout = httpx.Timeout(30.0, read=read_timeout_seconds.total_seconds())
 
         # Create httpx client from factory or use default with MCP-appropriate timeouts
-        # create_mcp_http_client uses 30s connect/5min read timeout by default,
-        # and always enables follow_redirects
         if self.httpx_client_factory is not None:
             # Factory clients get the full kwargs for backwards compatibility
             http_client = self.httpx_client_factory(
                 headers=headers,
                 auth=self.auth,
-                follow_redirects=True,  # type: ignore[call-arg]
+                follow_redirects=False,  # type: ignore[call-arg]
                 **({"timeout": timeout} if timeout else {}),
             )
         else:
-            http_client = create_mcp_http_client(
+            http_client = httpx.AsyncClient(
                 headers=headers,
-                timeout=timeout,
                 auth=self.auth,
+                follow_redirects=False,
+                timeout=timeout or httpx.Timeout(30.0, read=300.0),
             )
 
         # Ensure httpx client is closed after use

tests/client/transports/test_transports.py

@@ -1,6 +1,8 @@
 from ssl import VerifyMode
+from unittest.mock import patch
 
 import httpx
+import pytest
 
 from fastmcp.client.auth.oauth import OAuth
 from fastmcp.client.transports import SSETransport, StreamableHttpTransport
@@ -40,3 +42,49 @@ async def test_oauth_uses_same_client_as_transport_sse():
             httpx_client._transport._pool._ssl_context.verify_mode  # type: ignore[attr-defined]
             == VerifyMode.CERT_NONE
         )
+
+
+@pytest.mark.anyio
+async def test_streamable_http_factory_disables_redirects():
+    captured_kwargs: dict[str, object] = {}
+
+    def factory(**kwargs):
+        captured_kwargs.update(kwargs)
+        return httpx.AsyncClient(**kwargs)
+
+    transport = StreamableHttpTransport(
+        "https://some.fake.url/",
+        httpx_client_factory=factory,
+    )
+
+    with patch(
+        "fastmcp.client.transports.http.streamable_http_client",
+        side_effect=RuntimeError("stop"),
+    ):
+        with pytest.raises(RuntimeError, match="stop"):
+            async with transport.connect_session():
+                pass
+
+    assert captured_kwargs["follow_redirects"] is False
+
+
+@pytest.mark.anyio
+async def test_streamable_http_default_client_disables_redirects():
+    transport = StreamableHttpTransport("https://some.fake.url/")
+
+    with (
+        patch(
+            "fastmcp.client.transports.http.httpx.AsyncClient",
+            wraps=httpx.AsyncClient,
+        ) as async_client,
+        patch(
+            "fastmcp.client.transports.http.streamable_http_client",
+            side_effect=RuntimeError("stop"),
+        ),
+    ):
+        with pytest.raises(RuntimeError, match="stop"):
+            async with transport.connect_session():
+                pass
+
+    assert async_client.call_args is not None
+    assert async_client.call_args.kwargs["follow_redirects"] is False