diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md
index cd7de35626..3766134518 100644
--- a/docs/client/oauth-clients.md
+++ b/docs/client/oauth-clients.md
@@ -83,6 +83,8 @@ The first time `Client` sends a request, the server answers `401`. The provider
 
 After that it is quiet. Tokens come out of storage, an expired access token is refreshed with the refresh token, and only when none of that works does it run the flow again.
 
+One transport rule applies to all of these requests: they are made while an MCP request is in flight, and like it they do not follow redirects to other addresses (they follow none at all), so the metadata, registration and token URLs must answer directly.
+
 You wrote none of it. Two keyword arguments remain (`client_metadata_url` and `validate_resource_url`), and this file needs neither. `client_metadata_url` is the one worth knowing about; it gets its own section below.
 
 ### Try it
diff --git a/docs/client/transports.md b/docs/client/transports.md
index afb33caf38..be72a8bbb7 100644
--- a/docs/client/transports.md
+++ b/docs/client/transports.md
@@ -29,7 +29,7 @@ Pass a URL string and you get **Streamable HTTP**, the transport you deploy behi
 --8<-- "docs_src/client_transports/tutorial002.py"
 ```
 
-That is the whole production client. `Client` wraps the URL in `streamable_http_client(...)` for you, on top of an `httpx2.AsyncClient` configured the way MCP needs: `follow_redirects=True`, a 30-second timeout for connect/write/pool, and a 300-second read timeout because the server may hold a response stream open.
+That is the whole production client. `Client` wraps the URL in `streamable_http_client(...)` for you, on top of an `httpx2.AsyncClient` configured the way MCP needs: a 30-second timeout for connect/write/pool, and a 300-second read timeout because the server may hold a response stream open. Whichever client is underneath, the transport follows a redirect only when it stays on the endpoint's origin (same scheme, host and port, or `http` to `https` on the same host with the default ports) and keeps the request method, which covers a 307/308 trailing-slash redirect. Any other redirect is not followed, and the call it answered fails with an `MCPError` naming the location; if that address is the server you meant, use it as the URL.
 
 !!! check
     A `Client` you have constructed is **not** connected. Construction only picks the transport;
@@ -45,7 +45,7 @@ That is the whole production client. `Client` wraps the URL in `streamable_http_
 
 The moment you need an `Authorization` header, a cookie, a proxy, mTLS, or a different timeout, build the `httpx2.AsyncClient` yourself and hand it to `streamable_http_client`:
 
-```python title="client.py" hl_lines="8-14"
+```python title="client.py" hl_lines="8-13"
 --8<-- "docs_src/client_transports/tutorial003.py"
 ```
 
@@ -75,7 +75,10 @@ environment variables or pass an explicit `verify=ssl_context` to your `httpx2.A
 !!! info
     `httpx2` keeps the familiar `httpx` API, so if you know `httpx` you already know how to do auth,
     proxies, event hooks, retries and connection limits here. The SDK adds nothing on top and takes
-    nothing away. It is also where OAuth plugs in:
+    nothing away, with one exception: redirects. MCP requests follow the same-origin rule above rather
+    than the client's `follow_redirects`, and requests an `auth=` handler makes while one is in flight
+    (OAuth discovery, registration, token) do not follow redirects, so those URLs must answer directly.
+    It is also where OAuth plugs in:
     `httpx2.AsyncClient(auth=OAuthClientProvider(...))`. That whole flow is **[OAuth clients](oauth-clients.md)**.
 
 ## stdio
diff --git a/docs/migration.md b/docs/migration.md
index 7927c60611..1b51cae4a8 100644
--- a/docs/migration.md
+++ b/docs/migration.md
@@ -119,7 +119,7 @@ them:
 ```python
 import httpx
 
-http_client = httpx.AsyncClient(follow_redirects=True)
+http_client = httpx.AsyncClient(timeout=httpx.Timeout(30, read=300))
 ```
 
 **After (v2):**
@@ -127,7 +127,7 @@ http_client = httpx.AsyncClient(follow_redirects=True)
 ```python
 import httpx2
 
-http_client = httpx2.AsyncClient(follow_redirects=True)
+http_client = httpx2.AsyncClient(timeout=httpx2.Timeout(30, read=300))
 ```
 
 `httpx2` is API-compatible with `httpx`, so usually only the import name
@@ -2092,7 +2092,6 @@ http_client = httpx2.AsyncClient(
     headers={"Authorization": "Bearer token"},
     timeout=httpx2.Timeout(30, read=300),
     auth=my_auth,
-    follow_redirects=True,
 )
 
 async with http_client:
@@ -2103,11 +2102,11 @@ async with http_client:
         ...
 ```
 
-v1's internal client set `follow_redirects=True`; set it explicitly when supplying your own `httpx2.AsyncClient` to preserve that behavior.
+v1's internal client set `follow_redirects=True`. You don't need it on your own client: the transport follows a redirect within the endpoint's origin (a trailing-slash redirect, say) itself, and does not follow one anywhere else, whatever the client is configured to do.
 
 `streamable_http_client` itself keeps a small signature — `streamable_http_client(url, *, http_client=None, terminate_on_close=True)` — and now yields a 2-tuple (next section). The removed function's other parameters map onto the client you build:
 
-- `headers`, `timeout`, `sse_read_timeout`, `auth`: set them on the `httpx2.AsyncClient` as above. `streamablehttp_client` defaulted to `httpx.Timeout(30, read=300)`; a bare `httpx2.AsyncClient()` falls back to httpx2's flat 5-second timeout, too short for the long-lived GET stream, so set `timeout=httpx2.Timeout(30, read=300)` (as shown) to keep v1's values. Omitting `http_client` still gives you a default client with those timeouts and `follow_redirects=True`.
+- `headers`, `timeout`, `sse_read_timeout`, `auth`: set them on the `httpx2.AsyncClient` as above. `streamablehttp_client` defaulted to `httpx.Timeout(30, read=300)`; a bare `httpx2.AsyncClient()` falls back to httpx2's flat 5-second timeout, too short for the long-lived GET stream, so set `timeout=httpx2.Timeout(30, read=300)` (as shown) to keep v1's values. Omitting `http_client` still gives you a default client with those timeouts.
 - `httpx_client_factory`: gone with no replacement — call your factory yourself and pass the result as `http_client`.
 - `terminate_on_close`: unchanged (default `True`).
 
@@ -2151,10 +2150,7 @@ async def capture_session_id(response: httpx2.Response) -> None:
     if session_id:
         captured_session_ids.append(session_id)
 
-http_client = httpx2.AsyncClient(
-    event_hooks={"response": [capture_session_id]},
-    follow_redirects=True,
-)
+http_client = httpx2.AsyncClient(event_hooks={"response": [capture_session_id]})
 
 async with http_client:
     async with streamable_http_client(url, http_client=http_client) as (read_stream, write_stream):
diff --git a/docs_src/client_transports/tutorial003.py b/docs_src/client_transports/tutorial003.py
index 4df055e229..488737dc89 100644
--- a/docs_src/client_transports/tutorial003.py
+++ b/docs_src/client_transports/tutorial003.py
@@ -8,7 +8,6 @@ async def main() -> None:
     async with httpx2.AsyncClient(
         headers={"Authorization": "Bearer ..."},
         timeout=httpx2.Timeout(30.0, read=300.0),
-        follow_redirects=True,
     ) as http_client:
         transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client)
         async with Client(transport) as client:
diff --git a/docs_src/identity_assertion/tutorial001.py b/docs_src/identity_assertion/tutorial001.py
index afcd537896..24bc26572f 100644
--- a/docs_src/identity_assertion/tutorial001.py
+++ b/docs_src/identity_assertion/tutorial001.py
@@ -62,7 +62,7 @@ async def fetch_id_jag(audience: str, resource: str) -> str:
 
 
 async def main() -> None:
-    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
+    async with httpx2.AsyncClient(auth=oauth) as http_client:
         transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
         async with Client(transport) as client:
             result = await client.list_tools()
diff --git a/docs_src/oauth_clients/tutorial001.py b/docs_src/oauth_clients/tutorial001.py
index d150dc5da6..6e01553dc5 100644
--- a/docs_src/oauth_clients/tutorial001.py
+++ b/docs_src/oauth_clients/tutorial001.py
@@ -55,7 +55,7 @@ async def wait_for_callback() -> AuthorizationCodeResult:
 
 
 async def main() -> None:
-    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
+    async with httpx2.AsyncClient(auth=oauth) as http_client:
         transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
         async with Client(transport) as client:
             result = await client.list_tools()
diff --git a/docs_src/oauth_clients/tutorial002.py b/docs_src/oauth_clients/tutorial002.py
index dd4105f937..eac562318f 100644
--- a/docs_src/oauth_clients/tutorial002.py
+++ b/docs_src/oauth_clients/tutorial002.py
@@ -34,7 +34,7 @@ async def set_client_info(self, client_info: OAuthClientInformationFull) -> None
 
 
 async def main() -> None:
-    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
+    async with httpx2.AsyncClient(auth=oauth) as http_client:
         transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
         async with Client(transport) as client:
             result = await client.list_tools()
diff --git a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py b/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py
index a190b89970..b04e6fb546 100644
--- a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py
+++ b/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py
@@ -233,7 +233,7 @@ async def _default_redirect_handler(authorization_url: str) -> None:
                     await self._run_session(read_stream, write_stream)
             else:
                 print("📡 Opening StreamableHTTP transport connection with auth...")
-                async with httpx2.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+                async with httpx2.AsyncClient(auth=oauth_auth) as custom_client:
                     async with streamable_http_client(url=self.server_url, http_client=custom_client) as (
                         read_stream,
                         write_stream,
diff --git a/examples/servers/simple-tool/mcp_simple_tool/server.py b/examples/servers/simple-tool/mcp_simple_tool/server.py
index a43dd0f7b4..ac99110b66 100644
--- a/examples/servers/simple-tool/mcp_simple_tool/server.py
+++ b/examples/servers/simple-tool/mcp_simple_tool/server.py
@@ -1,15 +1,16 @@
 import anyio
 import click
+import httpx2
 import mcp.types as types
 from mcp.server import Server, ServerRequestContext
-from mcp.shared._httpx_utils import create_mcp_http_client
 
 
 async def fetch_website(
     url: str,
 ) -> list[types.ContentBlock]:
     headers = {"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"}
-    async with create_mcp_http_client(headers=headers) as client:
+    timeout = httpx2.Timeout(30, read=300)
+    async with httpx2.AsyncClient(headers=headers, timeout=timeout, follow_redirects=True) as client:
         response = await client.get(url)
         response.raise_for_status()
         return [types.TextContent(type="text", text=response.text)]
diff --git a/examples/snippets/clients/identity_assertion_client.py b/examples/snippets/clients/identity_assertion_client.py
index 19cde274c5..8c80f28997 100644
--- a/examples/snippets/clients/identity_assertion_client.py
+++ b/examples/snippets/clients/identity_assertion_client.py
@@ -66,7 +66,7 @@ async def main() -> None:
         scope="user",
     )
 
-    async with httpx2.AsyncClient(auth=oauth_auth, follow_redirects=True) as http_client:
+    async with httpx2.AsyncClient(auth=oauth_auth) as http_client:
         async with streamable_http_client("http://localhost:8001/mcp", http_client=http_client) as (read, write):
             async with ClientSession(read, write) as session:
                 await session.initialize()
diff --git a/examples/snippets/clients/oauth_client.py b/examples/snippets/clients/oauth_client.py
index 58c542ea43..11e0f5f912 100644
--- a/examples/snippets/clients/oauth_client.py
+++ b/examples/snippets/clients/oauth_client.py
@@ -72,7 +72,7 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with httpx2.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+    async with httpx2.AsyncClient(auth=oauth_auth) as custom_client:
         async with streamable_http_client("http://localhost:8001/mcp", http_client=custom_client) as (read, write):
             async with ClientSession(read, write) as session:
                 await session.initialize()
diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py
index 31e2e5cade..59ce05d7ca 100644
--- a/src/mcp/client/auth/utils.py
+++ b/src/mcp/client/auth/utils.py
@@ -230,9 +230,9 @@ async def handle_auth_metadata_response(response: Response) -> tuple[bool, OAuth
             return True, asm
         except ValidationError:  # pragma: no cover
             return True, None
-    elif response.status_code < 400 or response.status_code >= 500:
-        return False, None  # Non-4XX error, stop trying
-    return True, None
+    elif 300 <= response.status_code < 500:
+        return True, None  # Not served at this URL (redirects are not followed) - try the next candidate
+    return False, None  # Server error or unexpected status, stop trying
 
 
 def validate_authorization_response_iss(iss: str | None, oauth_metadata: OAuthMetadata | None) -> None:
diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py
index 31d0f35391..a4011545af 100644
--- a/src/mcp/client/sse.py
+++ b/src/mcp/client/sse.py
@@ -12,7 +12,12 @@
 
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import create_context_streams
-from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
+from mcp.shared._httpx_utils import (
+    McpHttpClientFactory,
+    create_mcp_http_client,
+    request_within_origin,
+    sse_within_origin,
+)
 from mcp.shared.message import SessionMessage
 
 logger = logging.getLogger(__name__)
@@ -47,7 +52,13 @@ async def sse_client(
         headers: Optional headers to include in requests.
         timeout: HTTP timeout for regular operations (in seconds).
         sse_read_timeout: Timeout for SSE read operations (in seconds).
-        httpx_client_factory: Factory function for creating the httpx2 client.
+        httpx_client_factory: Factory function for creating the httpx2 client. Whichever client it
+            returns, MCP requests follow a redirect only when it stays on the endpoint's origin
+            (same scheme, host and port, or http to https on the same host with default ports) and
+            keeps the request method; any other redirect is not followed, so connecting fails with
+            `httpx2.HTTPStatusError` for the redirect response. The client's `follow_redirects`
+            setting is not consulted, and requests `auth` makes during an MCP request do not follow
+            redirects.
         auth: Optional httpx2 authentication handler.
         on_session_created: Optional callback invoked with the session ID when received.
     """
@@ -55,7 +66,7 @@ async def sse_client(
     async with httpx_client_factory(
         headers=headers, auth=auth, timeout=httpx2.Timeout(timeout, read=sse_read_timeout)
     ) as client:
-        async with client.sse(url) as event_source:
+        async with sse_within_origin(client, url) as event_source:
             event_source.response.raise_for_status()
             logger.debug("SSE connection established")
 
@@ -121,13 +132,11 @@ async def post_writer(endpoint_url: str):
 
                         async def _send_message(session_message: SessionMessage) -> None:
                             logger.debug(f"Sending client message: {session_message}")
-                            response = await client.post(
+                            response = await request_within_origin(
+                                client,
+                                "POST",
                                 endpoint_url,
-                                json=session_message.message.model_dump(
-                                    by_alias=True,
-                                    mode="json",
-                                    exclude_unset=True,
-                                ),
+                                json=session_message.message.model_dump(by_alias=True, mode="json", exclude_unset=True),
                             )
                             response.raise_for_status()
                             logger.debug(f"Client message sent successfully: {response.status_code}")
diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py
index 226b0fecf9..1b1532504e 100644
--- a/src/mcp/client/streamable_http.py
+++ b/src/mcp/client/streamable_http.py
@@ -33,7 +33,12 @@
 from mcp.client._transport import TransportStreams
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
-from mcp.shared._httpx_utils import create_mcp_http_client
+from mcp.shared._httpx_utils import (
+    create_mcp_http_client,
+    request_within_origin,
+    sse_within_origin,
+    stream_within_origin,
+)
 from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER
 from mcp.shared.jsonrpc_dispatcher import cancelled_request_id_from_params
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
@@ -62,6 +67,14 @@ class ResumptionError(StreamableHTTPError):
     """Raised when resumption request is invalid."""
 
 
+def _unfollowed_redirect(response: httpx2.Response) -> str | None:
+    """Describe a redirect `stream_within_origin` left unfollowed, or None if `response` is not one."""
+    if response.next_request is None:
+        return None
+    location = response.next_request.url
+    return f"Redirect to {location} not followed; use that URL as the endpoint if it is the intended server"
+
+
 @dataclass
 class RequestContext:
     """Context for a request operation."""
@@ -210,7 +223,11 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
                 if last_event_id:
                     headers[LAST_EVENT_ID] = last_event_id
 
-                async with client.sse(self.url, headers=headers) as event_source:
+                async with sse_within_origin(client, self.url, headers=headers) as event_source:
+                    if (redirect := _unfollowed_redirect(event_source.response)) is not None:
+                        # The same GET would be redirected again, so retrying cannot help.
+                        logger.warning(f"GET stream not opened: {redirect}")
+                        return
                     event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
@@ -253,7 +270,14 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
             original_request_id = ctx.session_message.message.id
 
-        async with ctx.client.sse(self.url, headers=headers) as event_source:
+        async with sse_within_origin(ctx.client, self.url, headers=headers) as event_source:
+            if (redirect := _unfollowed_redirect(event_source.response)) is not None:
+                logger.warning(redirect)
+                assert original_request_id is not None
+                await self._resolve_abandoned_request(
+                    ctx.read_stream_writer, original_request_id, redirect, code=INVALID_REQUEST
+                )
+                return
             event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
 
@@ -320,7 +344,8 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
         if ctx.metadata is not None and ctx.metadata.headers is not None:
             headers.update(ctx.metadata.headers)
 
-        async with ctx.client.stream(
+        async with stream_within_origin(
+            ctx.client,
             "POST",
             self.url,
             json=message.model_dump(by_alias=True, mode="json", exclude_unset=True),
@@ -339,6 +364,14 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                     )
                 return
 
+            if (redirect := _unfollowed_redirect(response)) is not None:
+                logger.warning(redirect)
+                if isinstance(message, JSONRPCRequest):
+                    await self._resolve_abandoned_request(
+                        ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST
+                    )
+                return
+
             if response.status_code >= 400:
                 if isinstance(message, JSONRPCRequest):
                     # A spec-correct server may return the JSON-RPC error in the
@@ -501,7 +534,7 @@ async def _handle_reconnection(
         headers[LAST_EVENT_ID] = last_event_id
 
         try:
-            async with ctx.client.sse(self.url, headers=headers) as event_source:
+            async with sse_within_origin(ctx.client, self.url, headers=headers) as event_source:
                 event_source.response.raise_for_status()
                 logger.info("Reconnected to SSE stream")
 
@@ -626,7 +659,7 @@ async def terminate_session(self, client: httpx2.AsyncClient) -> None:
 
         try:
             headers = self._prepare_headers()
-            response = await client.delete(self.url, headers=headers)
+            response = await request_within_origin(client, "DELETE", self.url, headers=headers)
 
             if response.status_code == 405:
                 logger.debug("Server does not allow session termination")
@@ -650,6 +683,12 @@ async def streamable_http_client(
         http_client: Optional pre-configured httpx2.AsyncClient. If None, a default
             client with recommended MCP timeouts will be created. To configure headers,
             authentication, or other HTTP settings, create an httpx2.AsyncClient and pass it here.
+            Whichever client is used, MCP requests follow a redirect only when it stays on the
+            endpoint's origin (same scheme, host and port, or http to https on the same host with
+            default ports) and keeps the request method (307/308); any other redirect is not
+            followed and the message it answered fails with an error naming the location. The
+            client's `follow_redirects` setting is not consulted, and requests its `auth` handler
+            makes during an MCP request do not follow redirects.
         terminate_on_close: If True, send a DELETE request to terminate the session when the context exits.
 
     Yields:
diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py
index 6bb638886a..4abfbcab19 100644
--- a/src/mcp/shared/_httpx_utils.py
+++ b/src/mcp/shared/_httpx_utils.py
@@ -1,5 +1,7 @@
-"""Utilities for creating standardized httpx2 AsyncClient instances."""
+"""Utilities for creating and using httpx2 AsyncClient instances in the MCP transports."""
 
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 from typing import Any, Protocol
 
 import httpx2
@@ -10,6 +12,9 @@
 MCP_DEFAULT_TIMEOUT = 30.0  # General operations (seconds)
 MCP_DEFAULT_SSE_READ_TIMEOUT = 300.0  # SSE streams - 5 minutes (seconds)
 
+# The headers httpx2.AsyncClient.sse() adds to an event-stream request.
+_SSE_HEADERS = {"Accept": "text/event-stream", "Cache-Control": "no-store"}
+
 
 class McpHttpClientFactory(Protocol):  # pragma: no branch
     def __call__(  # pragma: no branch
@@ -25,9 +30,12 @@ def create_mcp_http_client(
     timeout: httpx2.Timeout | None = None,
     auth: httpx2.Auth | None = None,
 ) -> httpx2.AsyncClient:
-    """Create a standardized httpx2 AsyncClient with MCP defaults.
+    """Create an httpx2 AsyncClient with the MCP transports' default timeouts.
 
-    Always enables follow_redirects and applies an SSE-friendly default timeout.
+    The client uses a 30-second timeout for connect/write/pool and a 300-second
+    read timeout, because a server may hold a response stream open. Redirect
+    following is left at the httpx2 default (off): the MCP transports follow
+    redirects within the endpoint's origin themselves, see `stream_within_origin`.
 
     Args:
         headers: Optional headers to include with all requests.
@@ -36,60 +44,102 @@ def create_mcp_http_client(
         auth: Optional authentication handler.
 
     Returns:
-        Configured httpx2.AsyncClient instance with MCP defaults.
+        Configured httpx2.AsyncClient instance.
 
     Note:
         The returned AsyncClient must be used as a context manager to ensure
         proper cleanup of connections.
-
-    Example:
-        Basic usage with MCP defaults:
-
-        ```python
-        async with create_mcp_http_client() as client:
-            response = await client.get("https://api.example.com")
-        ```
-
-        With custom headers:
-
-        ```python
-        headers = {"Authorization": "Bearer token"}
-        async with create_mcp_http_client(headers) as client:
-            response = await client.get("/endpoint")
-        ```
-
-        With both custom headers and timeout:
-
-        ```python
-        timeout = httpx2.Timeout(60.0, read=300.0)
-        async with create_mcp_http_client(headers, timeout) as client:
-            response = await client.get("/long-request")
-        ```
-
-        With authentication:
-
-        ```python
-        from httpx2 import BasicAuth
-        auth = BasicAuth(username="user", password="pass")
-        async with create_mcp_http_client(headers, timeout, auth) as client:
-            response = await client.get("/protected-endpoint")
-        ```
     """
-    # Set MCP defaults
-    kwargs: dict[str, Any] = {"follow_redirects": True}
-
-    # Handle timeout
     if timeout is None:
-        kwargs["timeout"] = httpx2.Timeout(MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT)
-    else:
-        kwargs["timeout"] = timeout
-
-    # Handle headers
+        timeout = httpx2.Timeout(MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT)
+    kwargs: dict[str, Any] = {"timeout": timeout}
     if headers is not None:
         kwargs["headers"] = headers
-
-    # Handle authentication
     if auth is not None:  # pragma: no cover
         kwargs["auth"] = auth
-
     return httpx2.AsyncClient(**kwargs)
+
+
+def _within_origin(url: httpx2.URL, location: httpx2.URL) -> bool:
+    """Whether `location` is on `url`'s origin, or is its https upgrade on the default ports.
+
+    httpx2 normalises a scheme's default port to None and lower-cases hosts, so
+    plain tuple comparison is exact. The upgrade rule is the one httpx2 itself
+    uses to decide a redirect has not left the origin (`_is_https_redirect`).
+    """
+    if (url.scheme, url.host, url.port) == (location.scheme, location.host, location.port):
+        return True
+    return (
+        url.host == location.host
+        and url.scheme == "http"
+        and url.port is None
+        and location.scheme == "https"
+        and location.port is None
+    )
+
+
+@asynccontextmanager
+async def stream_within_origin(
+    client: httpx2.AsyncClient, method: str, url: httpx2.URL | str, **kwargs: Any
+) -> AsyncIterator[httpx2.Response]:
+    """`client.stream(...)`, following redirects only while they stay within the request's origin.
+
+    An MCP transport talks to one configured endpoint, and everything on a request
+    (headers, auth, body) was configured for that endpoint. A redirect that stays
+    on the origin of the request just sent (same scheme, host and port, or http to
+    https on the same host with default ports) and keeps the request's method,
+    such as a 307/308 trailing-slash normalisation, is followed using httpx2's
+    own next-request rules. Any other redirect is not followed: the redirect
+    response itself is yielded, the way httpx2 hands one back when
+    `follow_redirects` is off, and the caller treats it as the non-success it
+    is. (httpx2 rewrites a POST into a body-less GET for 301/302/303, which
+    would drop the message, so those count as not followed for anything but a
+    GET.) The client's own `follow_redirects` setting is not consulted, and
+    requests an `httpx2.Auth` flow makes during the call are sent the same way,
+    so they do not follow redirects either.
+
+    Raises:
+        httpx2.TooManyRedirects: More than `client.max_redirects` redirects were followed.
+    """
+    request = client.build_request(method, url, **kwargs)
+    for _ in range(client.max_redirects + 1):
+        response = await client.send(request, stream=True, follow_redirects=False)
+        # Set by httpx2, with its own method/body/header rules, only when the response is a redirect.
+        next_request = response.next_request
+        if (
+            next_request is None
+            or next_request.method != response.request.method
+            or not _within_origin(response.request.url, next_request.url)
+        ):
+            try:
+                yield response
+            finally:
+                await response.aclose()
+            return
+        try:
+            # Drain the redirect body so the connection returns to the pool, as httpx2 does when it follows.
+            await response.aread()
+        finally:
+            await response.aclose()
+        request = next_request
+    raise httpx2.TooManyRedirects("Exceeded maximum allowed redirects.", request=request)
+
+
+async def request_within_origin(
+    client: httpx2.AsyncClient, method: str, url: httpx2.URL | str, **kwargs: Any
+) -> httpx2.Response:
+    """`client.request(...)` with the redirect handling of `stream_within_origin`."""
+    async with stream_within_origin(client, method, url, **kwargs) as response:
+        await response.aread()
+    return response
+
+
+@asynccontextmanager
+async def sse_within_origin(
+    client: httpx2.AsyncClient, url: httpx2.URL | str, *, headers: dict[str, str] | None = None
+) -> AsyncIterator[httpx2.EventSource]:
+    """`client.sse(url)` with the redirect handling of `stream_within_origin`."""
+    merged = httpx2.Headers(_SSE_HEADERS)
+    merged.update(headers or {})
+    async with stream_within_origin(client, "GET", url, headers=merged) as response:
+        yield httpx2.EventSource(response)
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index be96cc8eec..4bba5b19cd 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -24,6 +24,7 @@
     extract_resource_metadata_from_www_auth,
     extract_scope_from_www_auth,
     get_client_metadata_scopes,
+    handle_auth_metadata_response,
     handle_registration_response,
     is_valid_client_metadata_url,
     should_use_client_metadata_url,
@@ -824,6 +825,17 @@ async def test_resource_param_included_with_protected_resource_metadata(self, oa
         assert "resource=" in content
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(("status", "keep_trying"), [(404, True), (307, True), (500, False)])
+async def test_auth_metadata_response_says_whether_to_try_the_next_discovery_url(
+    status: int, keep_trying: bool
+) -> None:
+    """SDK-defined: a 4xx or a 3xx (redirects are not followed on these requests) from a discovery
+    candidate means the metadata is not served there and the next well-known URL is tried; a 5xx
+    stops discovery."""
+    assert await handle_auth_metadata_response(httpx2.Response(status)) == (keep_trying, None)
+
+
 @pytest.mark.parametrize(
     ("protocol_version", "expected"),
     [
diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py
index ef9511fbe0..9996c19228 100644
--- a/tests/client/test_http_unicode.py
+++ b/tests/client/test_http_unicode.py
@@ -112,11 +112,9 @@ async def unicode_session() -> AsyncIterator[ClientSession]:
 
     async with (
         session_manager.run(),
-        # follow_redirects matches the SDK's own client factory; Starlette's Mount 307-redirects
-        # the bare /mcp path to /mcp/.
-        httpx2.AsyncClient(
-            transport=StreamingASGITransport(app), base_url=BASE_URL, follow_redirects=True
-        ) as http_client,
+        # Starlette's Mount 307-redirects the bare /mcp path to /mcp/; the transport follows that
+        # same-origin redirect itself.
+        httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client,
         streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client) as (read_stream, write_stream),
         ClientSession(read_stream, write_stream) as session,
     ):
diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py
index d21f520daf..6445889828 100644
--- a/tests/client/test_streamable_http.py
+++ b/tests/client/test_streamable_http.py
@@ -26,18 +26,24 @@
     JSONRPCNotification,
     JSONRPCRequest,
     JSONRPCResponse,
+    ListToolsResult,
+    PaginatedRequestParams,
 )
 from mcp_types.version import LATEST_MODERN_VERSION
+from starlette.applications import Starlette
+from starlette.routing import Mount
 from starlette.types import Receive, Scope, Send
 
+from mcp import Client, MCPError
 from mcp.client.streamable_http import (
     MAX_RECONNECTION_ATTEMPTS,
     RequestContext,
     StreamableHTTPTransport,
     streamable_http_client,
 )
-from mcp.server import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.server._streamable_http_modern import handle_modern_request
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, ServerEvent
 from mcp.shared._context_streams import ContextSendStream, create_context_streams
 from mcp.shared.dispatcher import CallOptions, DispatchContext
@@ -748,3 +754,164 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
                 _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
             )
     send.close()
+
+
+@pytest.mark.anyio
+async def test_trailing_slash_redirect_within_origin_is_followed_by_the_transport() -> None:
+    """SDK-defined: a redirect that stays on the endpoint's origin (here Starlette's Mount sending
+    /mcp to /mcp/) is followed by the transport itself, so a caller-supplied client left at
+    httpx2's no-follow default still connects."""
+    session_manager = StreamableHTTPSessionManager(app=Server("redirect-test"))
+    app = Starlette(routes=[Mount("/mcp", app=session_manager.handle_request)])
+    urls: list[str] = []
+
+    async def record(request: httpx2.Request) -> None:
+        urls.append(str(request.url))
+
+    with anyio.fail_after(5):
+        async with (
+            session_manager.run(),
+            httpx2.AsyncClient(transport=StreamingASGITransport(app), event_hooks={"request": [record]}) as http,
+            Client(streamable_http_client("http://mcp.example/mcp", http_client=http)) as client,
+        ):
+            assert client.server_info is not None
+            assert client.server_info.name == "redirect-test"
+
+    assert urls[:2] == ["http://mcp.example/mcp", "http://mcp.example/mcp/"]
+
+
+class _RedirectPromptsListElsewhere(httpx2.AsyncBaseTransport):
+    """Serves `app` in process, except that a prompts/list POST is answered with a redirect to
+    another origin."""
+
+    def __init__(self, app: Starlette) -> None:
+        self.inner = StreamingASGITransport(app)
+
+    async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response:
+        await request.aread()
+        if request.method == "POST" and json.loads(request.content).get("method") == "prompts/list":
+            return httpx2.Response(307, headers={"location": "http://other.example/mcp/"})
+        return await self.inner.handle_async_request(request)
+
+    async def __aenter__(self) -> "_RedirectPromptsListElsewhere":
+        await self.inner.__aenter__()
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        await self.inner.__aexit__(*args)
+
+
+@pytest.mark.anyio
+async def test_redirect_to_another_origin_fails_that_call_and_keeps_the_session() -> None:
+    """SDK-defined: a redirect pointing outside the endpoint's origin is not followed, whatever the
+    caller's client is configured to do: the call it answered fails with MCPError naming the
+    location, nothing is sent to the other origin, and the session stays usable."""
+
+    async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+        return ListToolsResult(tools=[])
+
+    session_manager = StreamableHTTPSessionManager(app=Server("redirect-test", on_list_tools=list_tools))
+    app = Starlette(routes=[Mount("/mcp", app=session_manager.handle_request)])
+    urls: list[str] = []
+
+    async def record(request: httpx2.Request) -> None:
+        urls.append(str(request.url))
+
+    with anyio.fail_after(5):
+        async with (
+            session_manager.run(),
+            httpx2.AsyncClient(
+                transport=_RedirectPromptsListElsewhere(app), event_hooks={"request": [record]}, follow_redirects=True
+            ) as http,
+            Client(streamable_http_client("http://mcp.example/mcp/", http_client=http)) as client,
+        ):
+            with pytest.raises(MCPError) as exc_info:
+                await client.list_prompts()
+            assert (await client.list_tools()).tools == []
+
+    assert exc_info.value.error.code == INVALID_REQUEST
+    assert exc_info.value.error.message == snapshot(
+        "Redirect to http://other.example/mcp/ not followed; use that URL as the endpoint if it is the intended server"
+    )
+    assert [url for url in urls if "other.example" in url] == []
+
+
+@pytest.mark.anyio
+async def test_redirected_notification_is_dropped_and_the_next_message_still_goes_out() -> None:
+    """SDK-defined: a notification whose POST is redirected outside the origin has no waiter to
+    resolve, so it is logged and dropped; the transport keeps serving the write stream."""
+    urls: list[str] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        urls.append(str(request.url))
+        return httpx2.Response(307, headers={"location": "http://other.example/mcp"})
+
+    with anyio.fail_after(5):
+        async with (
+            httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/roots/list_changed"))
+            )
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=7, method="tools/list", params={})))
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == 7
+    assert reply.message.error.code == INVALID_REQUEST
+    assert urls == ["http://test/mcp", "http://test/mcp"]
+
+
+@pytest.mark.anyio
+async def test_get_stream_gives_up_without_retrying_when_the_endpoint_redirects_elsewhere() -> None:
+    """SDK-defined: the standalone GET stream is not opened through a redirect to another origin,
+    and since the same GET would be redirected again the transport logs it and stops instead of
+    spending its reconnection attempts."""
+    gets: list[str] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        gets.append(str(request.url))
+        return httpx2.Response(307, headers={"location": "http://other.example/mcp"})
+
+    transport = StreamableHTTPTransport("http://test/mcp")
+    transport.session_id = "session-1"
+    send, receive = create_context_streams[SessionMessage | Exception](1)
+    with anyio.fail_after(5):
+        async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
+            await transport.handle_get_stream(http, send)
+    assert gets == ["http://test/mcp"]
+    send.close()
+    receive.close()
+
+
+@pytest.mark.anyio
+async def test_resumption_redirected_elsewhere_resolves_that_request_with_an_error() -> None:
+    """SDK-defined: a resumption GET answered with a redirect to another origin is not followed;
+    the resumed request is resolved with an error naming the location rather than left waiting."""
+    seen: list[tuple[str, str | None]] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen.append((f"{request.method} {request.url}", request.headers.get("last-event-id")))
+        return httpx2.Response(307, headers={"location": "http://other.example/mcp"})
+
+    with anyio.fail_after(5):
+        async with (
+            httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(
+                    JSONRPCRequest(jsonrpc="2.0", id="resume-1", method="tools/call", params={}),
+                    metadata=ClientMessageMetadata(resumption_token="evt-41"),
+                )
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == "resume-1"
+    assert reply.message.error.code == INVALID_REQUEST
+    assert reply.message.error.message == snapshot(
+        "Redirect to http://other.example/mcp not followed; use that URL as the endpoint if it is the intended server"
+    )
+    assert seen == [("GET http://test/mcp", "evt-41")]
diff --git a/tests/shared/test_httpx_utils.py b/tests/shared/test_httpx_utils.py
index a94d7c9299..158f2e43e9 100644
--- a/tests/shared/test_httpx_utils.py
+++ b/tests/shared/test_httpx_utils.py
@@ -1,16 +1,27 @@
-"""Tests for httpx2 utility functions."""
+"""Tests for the httpx2 helpers the client transports are built on."""
+
+from collections.abc import AsyncGenerator, AsyncIterator
+from typing import Any
 
 import httpx2
+import pytest
+
+from mcp.shared._httpx_utils import (
+    create_mcp_http_client,
+    request_within_origin,
+    sse_within_origin,
+    stream_within_origin,
+)
 
-from mcp.shared._httpx_utils import create_mcp_http_client
+pytestmark = pytest.mark.anyio
 
 
-def test_default_settings():
-    """Test that default settings are applied correctly."""
+def test_default_client_uses_mcp_timeouts_and_httpx_redirect_default():
+    """The factory applies the transports' timeouts and leaves redirect following to the transports."""
     client = create_mcp_http_client()
 
-    assert client.follow_redirects is True
-    assert client.timeout.connect == 30.0
+    assert client.follow_redirects is False
+    assert client.timeout == httpx2.Timeout(30.0, read=300.0)
 
 
 def test_custom_parameters():
@@ -22,3 +33,222 @@ def test_custom_parameters():
 
     assert client.headers["Authorization"] == "Bearer token"
     assert client.timeout.connect == 60.0
+
+
+class _Body(httpx2.AsyncByteStream):
+    """A response body served as a real stream, recording whether the client closed it."""
+
+    def __init__(self, data: bytes, closed: list[bool]) -> None:
+        self._data = data
+        self._closed = closed
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        yield self._data
+
+    async def aclose(self) -> None:
+        self._closed.append(True)
+
+
+def _recording_client(
+    redirects: dict[str, tuple[int, str]], **client_kwargs: Any
+) -> tuple[httpx2.AsyncClient, list[str], list[bool]]:
+    """A client whose server redirects each URL in `redirects` (status, Location) and answers 200
+    to anything else; plus the `METHOD url` lines the server received and one entry per redirect
+    response body the client closed."""
+    received: list[str] = []
+    closed: list[bool] = []
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        received.append(f"{request.method} {request.url}")
+        if str(request.url) in redirects:
+            status, location = redirects[str(request.url)]
+            return httpx2.Response(status, headers={"location": location}, stream=_Body(b"moved", closed))
+        return httpx2.Response(200, text=request.content.decode() or "ok")
+
+    return httpx2.AsyncClient(transport=httpx2.MockTransport(serve), **client_kwargs), received, closed
+
+
+@pytest.mark.parametrize(
+    ("url", "location"),
+    [
+        ("http://mcp.example/mcp", "http://mcp.example/mcp/"),
+        ("http://mcp.example/mcp", "/other/path"),
+        ("http://mcp.example:8080/mcp", "http://mcp.example:8080/v2/mcp"),
+        ("http://mcp.example/mcp", "http://MCP.EXAMPLE:80/mcp/"),
+        ("http://mcp.example/mcp", "https://mcp.example:443/mcp"),
+    ],
+)
+async def test_redirect_within_origin_is_followed_with_method_and_body(url: str, location: str):
+    """A redirect that stays on the request's origin (or upgrades it to https) is followed, and a
+    307 keeps the method and body (SDK-defined policy; the re-send itself is httpx2's)."""
+    client, received, closed = _recording_client({url: (307, location)})
+
+    async with client, stream_within_origin(client, "POST", url, content=b"payload") as response:
+        await response.aread()
+
+    assert response.status_code == 200
+    assert response.text == "payload"
+    assert received == [f"POST {url}", f"POST {httpx2.URL(url).join(location)}"]
+    assert closed == [True]
+
+
+@pytest.mark.parametrize(
+    "location",
+    [
+        "http://other.example/mcp",
+        "http://mcp.example:8080/mcp",
+        "http://sub.mcp.example/mcp",
+        "https://mcp.example:8443/mcp",
+        "ftp://mcp.example/mcp",
+    ],
+)
+async def test_redirect_outside_origin_is_not_followed(location: str):
+    """A redirect to another origin is handed back unfollowed, the way httpx2 hands back a redirect
+    with following off, and the location is never requested (SDK-defined policy)."""
+    url = "http://mcp.example/mcp"
+    client, received, closed = _recording_client({url: (307, location)})
+
+    async with client, stream_within_origin(client, "POST", url, content=b"payload") as response:
+        pass
+
+    assert response.status_code == 307
+    assert response.next_request is not None
+    assert response.next_request.url == location
+    assert received == [f"POST {url}"]
+    assert closed == [True]
+
+
+@pytest.mark.parametrize("status", [301, 302, 303])
+async def test_method_changing_redirect_of_a_post_is_not_followed(status: int):
+    """httpx2 turns a POST into a body-less GET for 301/302/303, which would drop the message, so a
+    same-origin redirect with one of those codes is handed back unfollowed (SDK-defined)."""
+    url = "http://mcp.example/mcp"
+    client, received, _ = _recording_client({url: (status, "/mcp/")})
+
+    async with client, stream_within_origin(client, "POST", url, content=b"payload") as response:
+        pass
+
+    assert response.status_code == status
+    assert received == [f"POST {url}"]
+
+
+@pytest.mark.parametrize("status", [301, 302, 303, 307, 308])
+async def test_same_origin_redirect_of_a_get_is_followed_for_every_redirect_status(status: int):
+    """A GET keeps its method under every redirect status, so the SSE GET follows all of them
+    within the origin (SDK-defined policy over httpx2's method rules)."""
+    url = "http://mcp.example/sse"
+    client, received, _ = _recording_client({url: (status, "/sse/")})
+
+    async with client, stream_within_origin(client, "GET", url) as response:
+        await response.aread()
+
+    assert response.status_code == 200
+    assert received == [f"GET {url}", "GET http://mcp.example/sse/"]
+
+
+async def test_https_to_http_on_same_host_is_outside_origin():
+    """Only the upgrade direction counts as staying on the origin; a downgrade is not followed."""
+    url = "https://mcp.example/mcp"
+    client, received, _ = _recording_client({url: (302, "http://mcp.example/mcp")})
+
+    async with client, stream_within_origin(client, "GET", url) as response:
+        pass
+
+    assert response.status_code == 302
+    assert received == [f"GET {url}"]
+
+
+async def test_client_configured_to_follow_redirects_is_still_scoped_to_origin():
+    """The client's own follow_redirects=True does not widen the policy: the transport helper
+    decides per request (SDK-defined)."""
+    url = "http://mcp.example/mcp"
+    client, received, _ = _recording_client({url: (307, "http://other.example/mcp")}, follow_redirects=True)
+
+    async with client, stream_within_origin(client, "POST", url) as response:
+        pass
+
+    assert response.status_code == 307
+    assert received == [f"POST {url}"]
+
+
+async def test_redirect_chain_longer_than_client_max_redirects_raises_too_many_redirects():
+    """Same-origin hops are bounded by the client's max_redirects, as httpx2 bounds its own."""
+    url = "http://mcp.example/a"
+    client, received, closed = _recording_client(
+        {
+            "http://mcp.example/a": (307, "/b"),
+            "http://mcp.example/b": (307, "/c"),
+            "http://mcp.example/c": (307, "/d"),
+        },
+        max_redirects=2,
+    )
+
+    async with client:
+        with pytest.raises(httpx2.TooManyRedirects):
+            await request_within_origin(client, "GET", url)
+
+    assert received == ["GET http://mcp.example/a", "GET http://mcp.example/b", "GET http://mcp.example/c"]
+    assert closed == [True, True, True]
+
+
+async def test_request_within_origin_returns_a_read_response():
+    """The non-streaming form hands back a response whose body is already read."""
+    url = "http://mcp.example/mcp"
+    client, received, _ = _recording_client({url: (307, "/mcp/")})
+
+    async with client:
+        response = await request_within_origin(client, "DELETE", url)
+
+    assert response.status_code == 200
+    assert response.text == "ok"
+    assert received == [f"DELETE {url}", "DELETE http://mcp.example/mcp/"]
+
+
+async def test_sse_within_origin_sends_event_stream_headers_and_caller_headers():
+    """The SSE form asks for an event stream exactly as client.sse() does, merged case-insensitively
+    with the caller's headers, and yields an EventSource over the final response."""
+    seen: list[httpx2.Headers] = []
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        seen.append(request.headers)
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, text="data: hello\n\n")
+
+    client = httpx2.AsyncClient(transport=httpx2.MockTransport(serve))
+    async with client:
+        async with sse_within_origin(client, "http://mcp.example/sse") as source:
+            events = [event.data async for event in source]
+        async with sse_within_origin(client, "http://mcp.example/sse", headers={"accept": "x/y", "k": "v"}):
+            pass
+
+    assert events == ["hello"]
+    assert seen[0]["accept"] == "text/event-stream"
+    assert seen[0]["cache-control"] == "no-store"
+    assert seen[1].get_list("accept") == ["x/y"]
+    assert seen[1]["cache-control"] == "no-store"
+    assert seen[1]["k"] == "v"
+
+
+async def test_auth_flow_requests_are_not_redirected():
+    """Requests an httpx2 Auth flow issues while a transport request is in flight (a token refresh,
+    say) inherit the per-request no-follow setting, so a redirect on them is handed back to the
+    auth flow rather than followed (httpx2 behaviour the transports rely on)."""
+    received: list[str] = []
+
+    class TokenThenRequest(httpx2.Auth):
+        async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+            token_response = yield httpx2.Request("POST", "http://mcp.example/token", content=b"grant")
+            request.headers["x-token-status"] = str(token_response.status_code)
+            yield request
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        received.append(f"{request.method} {request.url}")
+        if request.url.path == "/token":
+            return httpx2.Response(307, headers={"location": "http://other.example/token"})
+        return httpx2.Response(200, text=request.headers["x-token-status"])
+
+    client = httpx2.AsyncClient(transport=httpx2.MockTransport(serve), auth=TokenThenRequest(), follow_redirects=True)
+    async with client:
+        response = await request_within_origin(client, "POST", "http://mcp.example/mcp")
+
+    assert response.text == "307"
+    assert received == ["POST http://mcp.example/token", "POST http://mcp.example/mcp"]
diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py
index c27dd69db3..77d1b28a0a 100644
--- a/tests/shared/test_sse.py
+++ b/tests/shared/test_sse.py
@@ -3,14 +3,13 @@
 import json
 from collections.abc import AsyncGenerator
 from typing import Any
-from unittest.mock import AsyncMock, MagicMock, Mock
+from unittest.mock import Mock
 from urllib.parse import urlparse
 
 import anyio
 import httpx2
 import mcp_types as types
 import pytest
-from httpx2 import ServerSentEvent
 from inline_snapshot import snapshot
 from mcp_types import (
     CallToolRequestParams,
@@ -41,6 +40,7 @@
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._httpx_utils import McpHttpClientFactory
 from mcp.shared.exceptions import MCPError
+from mcp.shared.message import SessionMessage
 from tests.interaction.transports import StreamingASGITransport
 
 SERVER_NAME = "test_server_for_SSE"
@@ -58,15 +58,13 @@ def factory(
         auth: httpx2.Auth | None = None,
     ) -> httpx2.AsyncClient:
         # The SSE GET runs until it observes a disconnect, so the bridge must let the
-        # application drain on close rather than cancelling it. follow_redirects matches
-        # create_mcp_http_client, the factory this one stands in for.
+        # application drain on close rather than cancelling it.
         return httpx2.AsyncClient(
             transport=StreamingASGITransport(app, cancel_on_close=False),
             base_url=BASE_URL,
             headers=headers,
             timeout=timeout,
             auth=auth,
-            follow_redirects=True,
         )
 
     return factory
@@ -400,10 +398,10 @@ async def test_sse_client_handles_empty_keepalive_pings() -> None:
     send an SSE event consisting of an event ID and an empty data field in order
     to prime the client to reconnect."
 
-    This test mocks the SSE event stream to include empty "message" events and
-    verifies the client skips them without crashing.
+    The event stream served here carries an endpoint event, an empty "message"
+    event (the case under test), then a real response; the client must skip the
+    empty one and deliver the response.
     """
-    # Build a proper JSON-RPC response using types (not hardcoded strings)
     init_result = InitializeResult(
         protocol_version="2024-11-05",
         capabilities=ServerCapabilities(),
@@ -415,41 +413,86 @@ async def test_sse_client_handles_empty_keepalive_pings() -> None:
         result=init_result.model_dump(by_alias=True, exclude_none=True),
     )
     response_json = response.model_dump_json(by_alias=True, exclude_none=True)
+    event_stream = (
+        "event: endpoint\ndata: /messages/?session_id=abc123\n\n"
+        "event: message\ndata: \n\n"
+        f"event: message\ndata: {response_json}\n\n"
+    )
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        assert request.url.path == "/sse"
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, text=event_stream)
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx2.Timeout | None = None,
+        auth: httpx2.Auth | None = None,
+    ) -> httpx2.AsyncClient:
+        return httpx2.AsyncClient(transport=httpx2.MockTransport(serve))
+
+    with anyio.fail_after(5):
+        async with sse_client("http://test/sse", httpx_client_factory=factory) as (read_stream, _):
+            msg = await read_stream.receive()
+
+    assert isinstance(msg, SessionMessage)
+    assert isinstance(msg.message, types.JSONRPCResponse)
+    assert msg.message.id == 1
+
+
+@pytest.mark.anyio
+async def test_sse_client_follows_redirect_within_origin_on_connect() -> None:
+    """SDK-defined: a redirect of the SSE GET that stays on the endpoint's origin is followed by
+    the transport itself, with a client left at httpx2's no-follow default."""
+    received: list[str] = []
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        received.append(str(request.url))
+        if request.url.path == "/sse":
+            return httpx2.Response(307, headers={"location": "/sse/"})
+        assert request.url.path == "/sse/"
+        return httpx2.Response(
+            200, headers={"content-type": "text/event-stream"}, text="event: endpoint\ndata: /messages/\n\n"
+        )
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx2.Timeout | None = None,
+        auth: httpx2.Auth | None = None,
+    ) -> httpx2.AsyncClient:
+        return httpx2.AsyncClient(transport=httpx2.MockTransport(serve))
+
+    with anyio.fail_after(5):
+        async with sse_client("http://test/sse", httpx_client_factory=factory):
+            pass
+
+    assert received == ["http://test/sse", "http://test/sse/"]
 
-    # Mock SSE events using httpx2's ServerSentEvent: an endpoint event, an
-    # empty keep-alive ping (the case under test), then a real response.
-    mock_event_source = MagicMock()
-    mock_event_source.__aiter__.return_value = [
-        ServerSentEvent(event="endpoint", data="/messages/?session_id=abc123"),
-        ServerSentEvent(event="message", data=""),
-        ServerSentEvent(event="message", data=response_json),
-    ]
-    mock_event_source.response.raise_for_status = MagicMock()
-
-    mock_sse = MagicMock()
-    mock_sse.__aenter__ = AsyncMock(return_value=mock_event_source)
-    mock_sse.__aexit__ = AsyncMock(return_value=None)
-
-    mock_client = MagicMock()
-    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
-    mock_client.__aexit__ = AsyncMock(return_value=None)
-    mock_client.sse = MagicMock(return_value=mock_sse)
-    mock_client.post = AsyncMock(return_value=MagicMock(status_code=200, raise_for_status=MagicMock()))
-
-    def mock_factory(
+
+@pytest.mark.anyio
+async def test_sse_client_does_not_follow_redirect_to_another_origin_on_connect() -> None:
+    """SDK-defined: a redirect of the SSE GET to another origin is not followed, even with a client
+    configured to follow redirects: connecting fails with HTTPStatusError for the redirect response
+    and that origin is never contacted."""
+    received: list[str] = []
+
+    def serve(request: httpx2.Request) -> httpx2.Response:
+        received.append(str(request.url))
+        return httpx2.Response(307, headers={"location": "http://other.example/sse"})
+
+    def factory(
         headers: dict[str, str] | None = None,
         timeout: httpx2.Timeout | None = None,
         auth: httpx2.Auth | None = None,
     ) -> httpx2.AsyncClient:
-        return mock_client
-
-    async with sse_client("http://test/sse", httpx_client_factory=mock_factory) as (read_stream, _):
-        # Read the message - should skip the empty one and get the real response
-        msg = await read_stream.receive()
-        # If we get here without error, the empty message was skipped successfully
-        assert not isinstance(msg, Exception)
-        assert isinstance(msg.message, types.JSONRPCResponse)
-        assert msg.message.id == 1
+        return httpx2.AsyncClient(transport=httpx2.MockTransport(serve), follow_redirects=True)
+
+    with anyio.fail_after(5):
+        with pytest.raises(httpx2.HTTPStatusError) as exc_info:
+            async with sse_client("http://test/sse", httpx_client_factory=factory):
+                pytest.fail("should not connect")  # pragma: no cover
+
+    assert exc_info.value.response.status_code == 307
+    assert received == ["http://test/sse"]
 
 
 @pytest.mark.anyio
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index aeef25a278..bdc14e3507 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -356,10 +356,11 @@ async def running_app(
 
 
 def make_client(app: Starlette, headers: dict[str, str] | None = None) -> httpx2.AsyncClient:
-    """An httpx2 client served in process by `app`, with create_mcp_http_client's redirect default.
+    """An httpx2 client served in process by `app`.
 
-    (Starlette's Mount 307-redirects the bare /mcp path to /mcp/, which the SDK's own client
-    factory follows.)
+    Starlette's Mount 307-redirects the bare /mcp path to /mcp/. The MCP transport follows that
+    same-origin redirect itself; `follow_redirects=True` is here for the tests in this file that
+    POST to /mcp with this client directly.
     """
     return httpx2.AsyncClient(
         transport=StreamingASGITransport(app), base_url=BASE_URL, headers=headers, follow_redirects=True
@@ -996,7 +997,9 @@ async def message_handler(message: IncomingMessage) -> None:  # pragma: no branc
         assert resource_update_found, "ResourceUpdatedNotification not received via GET stream"
 
 
-def create_session_id_capturing_client(app: Starlette) -> tuple[httpx2.AsyncClient, list[str]]:
+def create_session_id_capturing_client(
+    app: Starlette, transport: httpx2.AsyncBaseTransport | None = None
+) -> tuple[httpx2.AsyncClient, list[str]]:
     """Create an in-process httpx2 client that captures the session ID from responses."""
     captured_ids: list[str] = []
 
@@ -1006,9 +1009,8 @@ async def capture_session_id(response: httpx2.Response) -> None:
             captured_ids.append(session_id)
 
     client = httpx2.AsyncClient(
-        transport=StreamingASGITransport(app),
+        transport=transport or StreamingASGITransport(app),
         base_url=BASE_URL,
-        follow_redirects=True,
         event_hooks={"response": [capture_session_id]},
     )
     return client, captured_ids
@@ -1052,36 +1054,34 @@ async def test_streamable_http_client_session_termination(basic_app: Starlette)
 
 
 @pytest.mark.anyio
-async def test_streamable_http_client_session_termination_204(
-    basic_app: Starlette, monkeypatch: pytest.MonkeyPatch
-) -> None:
+async def test_streamable_http_client_session_termination_204(basic_app: Starlette) -> None:
     """Session termination also succeeds when the server answers the DELETE with 204.
 
-    This test patches the httpx2 client to return a 204 response for DELETEs.
+    The in-process server answers the DELETE with 200; a wrapping HTTP transport rewrites that to
+    204 on the way back, which is what some servers send.
     """
 
-    # Save the original delete method to restore later
-    original_delete = httpx2.AsyncClient.delete
+    class AnswerDeleteWith204(httpx2.AsyncBaseTransport):
+        def __init__(self, inner: StreamingASGITransport) -> None:
+            self.inner = inner
 
-    # Mock the client's delete method to return a 204
-    async def mock_delete(self: httpx2.AsyncClient, *args: Any, **kwargs: Any) -> httpx2.Response:
-        # Call the original method to get the real response
-        response = await original_delete(self, *args, **kwargs)
+        async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response:
+            response = await self.inner.handle_async_request(request)
+            if request.method != "DELETE" or response.status_code != 200:
+                return response
+            await response.aread()
+            return httpx2.Response(204, headers=response.headers, request=request)
 
-        # Create a new response with 204 status code but same headers
-        mocked_response = httpx2.Response(
-            204,
-            headers=response.headers,
-            content=response.content,
-            request=response.request,
-        )
-        return mocked_response
+        async def __aenter__(self) -> AnswerDeleteWith204:
+            await self.inner.__aenter__()
+            return self
 
-    # Apply the patch to the httpx2 client
-    monkeypatch.setattr(httpx2.AsyncClient, "delete", mock_delete)
+        async def __aexit__(self, *args: Any) -> None:
+            await self.inner.__aexit__(*args)
 
-    # Use httpx2 client with event hooks to capture session ID
-    httpx_client, captured_ids = create_session_id_capturing_client(basic_app)
+    httpx_client, captured_ids = create_session_id_capturing_client(
+        basic_app, transport=AnswerDeleteWith204(StreamingASGITransport(basic_app))
+    )
 
     async with httpx_client:
         async with streamable_http_client(f"{BASE_URL}/mcp", http_client=httpx_client) as (