diff --git a/docs/authorization.md b/docs/authorization.md
index 171871ee58..70a3824231 100644
--- a/docs/authorization.md
+++ b/docs/authorization.md
@@ -151,7 +151,7 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+    async with httpx.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/docs/client.md b/docs/client.md
index 77c2729f27..2241cfd260 100644
--- a/docs/client.md
+++ b/docs/client.md
@@ -130,6 +130,29 @@ if __name__ == "__main__":
 _Full example: [examples/snippets/clients/streamable_basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/examples/snippets/clients/streamable_basic.py)_
 <!-- /snippet-source -->
 
+To configure headers, authentication or timeouts, create an `httpx.AsyncClient` and pass it as `http_client=`.
+
+## HTTP redirects
+
+The transport connects to the URL you gave it, and only that origin.
+
+* A `307`/`308` redirect that stays on the same scheme, host and port is followed, and so is `http://` → `https://` on the same host. That covers the usual `/mcp` → `/mcp/` trailing-slash redirect.
+* A redirect anywhere else is **not** followed. Connecting fails with:
+
+    ```text
+    httpx.HTTPStatusError: Redirect to https://other.example.com/mcp not followed; use that URL as the endpoint if it is the intended server
+    ```
+
+    If that URL is the server you meant, put it in your config. If it isn't, the server or a proxy in front of it is misconfigured.
+
+This holds for any `httpx.AsyncClient` you pass in: its `follow_redirects` setting is not consulted for MCP requests, in either direction. The SDK's OAuth providers apply the same rule to their own requests, and so does `sse_client()`.
+
+!!! tip
+    `Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP` means the
+    server sits behind a TLS-terminating proxy it doesn't know about and is issuing `http://` redirects.
+    That is fixed on the server (for uvicorn: `--proxy-headers` and `--forwarded-allow-ips`), or by
+    using the exact `https://…/` URL the message suggests.
+
 ## Client Display Utilities
 
 When building MCP clients, the SDK provides utilities to help display human-readable names for tools, resources, and prompts:
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 a88c4ea6b6..01d1ac709a 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
@@ -212,7 +212,7 @@ async def _default_redirect_handler(authorization_url: str) -> None:
                     await self._run_session(read_stream, write_stream, None)
             else:
                 print("📡 Opening StreamableHTTP transport connection with auth...")
-                async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+                async with httpx.AsyncClient(auth=oauth_auth) as custom_client:
                     async with streamable_http_client(
                         url=self.server_url,
                         http_client=custom_client,
diff --git a/examples/servers/simple-tool/mcp_simple_tool/server.py b/examples/servers/simple-tool/mcp_simple_tool/server.py
index 5b2b7d068d..b75e328e01 100644
--- a/examples/servers/simple-tool/mcp_simple_tool/server.py
+++ b/examples/servers/simple-tool/mcp_simple_tool/server.py
@@ -2,9 +2,9 @@
 
 import anyio
 import click
+import httpx
 import mcp.types as types
 from mcp.server.lowlevel import Server
-from mcp.shared._httpx_utils import create_mcp_http_client
 from starlette.requests import Request
 
 
@@ -12,7 +12,8 @@ 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 = httpx.Timeout(30, read=300)
+    async with httpx.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/oauth_client.py b/examples/snippets/clients/oauth_client.py
index 140b38aedb..523dfdf099 100644
--- a/examples/snippets/clients/oauth_client.py
+++ b/examples/snippets/clients/oauth_client.py
@@ -69,7 +69,7 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+    async with httpx.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/oauth2.py b/src/mcp/client/auth/oauth2.py
index aea037f1f0..680cbfd022 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -40,6 +40,7 @@
     validate_metadata_issuer,
 )
 from mcp.client.streamable_http import MCP_PROTOCOL_VERSION
+from mcp.shared._httpx_utils import RedirectAwareAuth, redirect_note
 from mcp.shared.auth import (
     OAuthClientInformationFull,
     OAuthClientMetadata,
@@ -224,7 +225,7 @@ def _origin_issuer(server_url: str) -> str:
     return str(AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}"))
 
 
-class OAuthClientProvider(httpx.Auth):
+class OAuthClientProvider(RedirectAwareAuth):
     """
     OAuth2 authentication for httpx.
     Handles OAuth flow with automatic client registration and token storage.
@@ -421,7 +422,9 @@ async def _handle_token_response(self, response: httpx.Response) -> None:
         if response.status_code != 200:
             body = await response.aread()  # pragma: no cover
             body_text = body.decode("utf-8")  # pragma: no cover
-            raise OAuthTokenError(f"Token exchange failed ({response.status_code}): {body_text}")  # pragma: no cover
+            raise OAuthTokenError(  # pragma: no cover
+                f"Token exchange failed ({response.status_code}){redirect_note(response)}: {body_text}"
+            )
 
         # Parse and validate response with scope validation
         token_response = await handle_token_response_scopes(response)
@@ -464,7 +467,7 @@ async def _refresh_token(self) -> httpx.Request:
     async def _handle_refresh_response(self, response: httpx.Response) -> bool:  # pragma: no cover
         """Handle token refresh response. Returns True if successful."""
         if response.status_code != 200:
-            logger.warning(f"Token refresh failed: {response.status_code}")
+            logger.warning(f"Token refresh failed: {response.status_code}{redirect_note(response)}")
             self.context.clear_tokens()
             return False
 
@@ -508,8 +511,8 @@ def _expected_issuer(self) -> str:
         the 2025-03-26 well-known URL is built from (RFC 8414 §3.3)."""
         return self.context.auth_server_url or _origin_issuer(self.context.server_url)
 
-    async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
-        """HTTPX auth flow integration."""
+    async def _auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
+        """The OAuth flow proper; `async_auth_flow` drives it (see `RedirectAwareAuth`)."""
         async with self.context.lock:
             if not self._initialized:
                 await self._initialize()  # pragma: no cover
diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py
index 807d1da984..413ac8405e 100644
--- a/src/mcp/client/auth/utils.py
+++ b/src/mcp/client/auth/utils.py
@@ -9,6 +9,7 @@
 
 from mcp.client.auth import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
 from mcp.client.streamable_http import MCP_PROTOCOL_VERSION
+from mcp.shared._httpx_utils import redirect_note
 from mcp.shared.auth import (
     OAuthClientInformationFull,
     OAuthClientMetadata,
@@ -205,9 +206,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_metadata_issuer(oauth_metadata: OAuthMetadata, expected_issuer: str) -> None:
@@ -262,7 +263,9 @@ async def handle_registration_response(response: Response) -> OAuthClientInforma
     """Handle registration response."""
     if response.status_code not in (200, 201):
         await response.aread()
-        raise OAuthRegistrationError(f"Registration failed: {response.status_code} {response.text}")
+        raise OAuthRegistrationError(
+            f"Registration failed: {response.status_code}{redirect_note(response)} {response.text}"
+        )
 
     try:
         content = await response.aread()
diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py
index 0d7fa0fb46..08e8927887 100644
--- a/src/mcp/client/sse.py
+++ b/src/mcp/client/sse.py
@@ -8,10 +8,15 @@
 import httpx
 from anyio.abc import TaskStatus
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
-from httpx_sse import SSEError, aconnect_sse
+from httpx_sse import SSEError
 
 import mcp.types as types
-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,6 +52,13 @@ async def sse_client(
         headers: Optional headers to include in requests.
         timeout: HTTP timeout for regular operations.
         sse_read_timeout: Timeout for SSE read operations.
+        httpx_client_factory: Factory function for creating the httpx 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 status for the SSE GET, 307/308 for a message POST); any
+            other redirect is not followed, so connecting fails with `httpx.HTTPStatusError` for
+            the redirect response. The client's `follow_redirects` setting is not consulted; the
+            SDK's OAuth providers apply the same rule to the requests they make.
         auth: Optional HTTPX authentication handler.
         on_session_created: Optional callback invoked with the session ID when received.
     """
@@ -65,11 +77,7 @@ async def sse_client(
             async with httpx_client_factory(
                 headers=headers, auth=auth, timeout=httpx.Timeout(timeout, read=sse_read_timeout)
             ) as client:
-                async with aconnect_sse(
-                    client,
-                    "GET",
-                    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")
 
@@ -135,7 +143,9 @@ async def post_writer(endpoint_url: str):
                             async with write_stream_reader:
                                 async for session_message in write_stream_reader:
                                     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,
@@ -161,3 +171,7 @@ async def post_writer(endpoint_url: str):
         finally:
             await read_stream_writer.aclose()
             await write_stream.aclose()
+            # The receive sides too, so that failing to connect (which raises before the
+            # streams are handed to the caller) does not leave them to the garbage collector.
+            await read_stream.aclose()
+            await write_stream_reader.aclose()
diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py
index ed28fcc275..4fd743b9c4 100644
--- a/src/mcp/client/streamable_http.py
+++ b/src/mcp/client/streamable_http.py
@@ -19,12 +19,16 @@
 import httpx
 from anyio.abc import TaskGroup
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
-from httpx_sse import EventSource, ServerSentEvent, aconnect_sse
+from httpx_sse import EventSource, ServerSentEvent
 from typing_extensions import deprecated
 
 from mcp.shared._httpx_utils import (
     McpHttpClientFactory,
     create_mcp_http_client,
+    redirect_location,
+    request_within_origin,
+    sse_within_origin,
+    stream_within_origin,
 )
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
@@ -72,6 +76,28 @@ class ResumptionError(StreamableHTTPError):
     """Raised when resumption request is invalid."""
 
 
+def _unfollowed_redirect(response: httpx.Response) -> str | None:
+    """Describe a redirect `stream_within_origin` left unfollowed, or None if `response` is not one."""
+    location = redirect_location(response)
+    if location is None:
+        return None
+    if response.request.url.scheme == "https" and location.scheme == "http":
+        return (
+            f"Redirect to {location} not followed: it would downgrade this HTTPS endpoint to plain HTTP.\n"
+            "The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust,\n"
+            f"often combined with a trailing-slash difference. Try {location.copy_with(scheme='https')} instead, "
+            "or fix the proxy settings."
+        )
+    return f"Redirect to {location} not followed; use that URL as the endpoint if it is the intended server"
+
+
+def _raise_for_unfollowed_redirect(response: httpx.Response) -> None:
+    """Raise `httpx.HTTPStatusError`, as `raise_for_status()` does for a redirect response, saying why
+    this one was not followed."""
+    if (redirect := _unfollowed_redirect(response)) is not None:
+        raise httpx.HTTPStatusError(redirect, request=response.request, response=response)
+
+
 @dataclass
 class RequestContext:
     """Context for a request operation."""
@@ -263,12 +289,11 @@ async def handle_get_stream(
                 if last_event_id:
                     headers[LAST_EVENT_ID] = last_event_id  # pragma: no cover
 
-                async with aconnect_sse(
-                    client,
-                    "GET",
-                    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")
 
@@ -311,12 +336,8 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         if isinstance(ctx.session_message.message.root, JSONRPCRequest):  # pragma: no branch
             original_request_id = ctx.session_message.message.root.id
 
-        async with aconnect_sse(
-            ctx.client,
-            "GET",
-            self.url,
-            headers=headers,
-        ) as event_source:
+        async with sse_within_origin(ctx.client, self.url, headers=headers) as event_source:
+            _raise_for_unfollowed_redirect(event_source.response)
             event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
 
@@ -337,7 +358,8 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
         message = ctx.session_message.message
         is_initialization = self._is_initialization_request(message)
 
-        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_none=True),
@@ -355,6 +377,7 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                     )  # pragma: no cover
                 return  # pragma: no cover
 
+            _raise_for_unfollowed_redirect(response)
             response.raise_for_status()
             if is_initialization:
                 self._maybe_extract_session_id_from_response(response)
@@ -460,12 +483,7 @@ async def _handle_reconnection(
             original_request_id = ctx.session_message.message.root.id
 
         try:
-            async with aconnect_sse(
-                ctx.client,
-                "GET",
-                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")
 
@@ -583,7 +601,7 @@ async def terminate_session(self, client: httpx.AsyncClient) -> None:  # pragma:
 
         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")
@@ -619,6 +637,13 @@ async def streamable_http_client(
         http_client: Optional pre-configured httpx.AsyncClient. If None, a default
             client with recommended MCP timeouts will be created. To configure headers,
             authentication, or other HTTP settings, create an httpx.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 for a POST; any status for the GET
+            stream); any other redirect is not followed and, like any non-2xx response, raises
+            `httpx.HTTPStatusError`, here naming the location. The client's `follow_redirects`
+            setting is not consulted; the SDK's OAuth providers apply the same rule to the
+            requests they make.
         terminate_on_close: If True, send a DELETE request to terminate the session
             when the context exits.
 
diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py
index 945ef80955..248e797efd 100644
--- a/src/mcp/shared/_httpx_utils.py
+++ b/src/mcp/shared/_httpx_utils.py
@@ -1,8 +1,12 @@
-"""Utilities for creating standardized httpx AsyncClient instances."""
+"""Utilities for creating and using httpx AsyncClient instances in the MCP transports."""
 
+from abc import ABC, abstractmethod
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
 from typing import Any, Protocol
 
 import httpx
+from httpx_sse import EventSource
 
 __all__ = ["create_mcp_http_client", "MCP_DEFAULT_TIMEOUT", "MCP_DEFAULT_SSE_READ_TIMEOUT"]
 
@@ -10,6 +14,12 @@
 MCP_DEFAULT_TIMEOUT = 30.0  # General operations (seconds)
 MCP_DEFAULT_SSE_READ_TIMEOUT = 300.0  # SSE streams - 5 minutes (seconds)
 
+# The headers httpx_sse.aconnect_sse() adds to an event-stream request.
+_SSE_HEADERS = {"Accept": "text/event-stream", "Cache-Control": "no-store"}
+
+# How many redirects one auth-flow request may follow within its origin (see RedirectAwareAuth).
+_AUTH_REDIRECT_LIMIT = 5
+
 
 class McpHttpClientFactory(Protocol):  # pragma: no branch
     def __call__(  # pragma: no branch
@@ -25,63 +35,185 @@ def create_mcp_http_client(
     timeout: httpx.Timeout | None = None,
     auth: httpx.Auth | None = None,
 ) -> httpx.AsyncClient:
-    """Create a standardized httpx AsyncClient with MCP defaults.
+    """Create an httpx AsyncClient with the MCP transports' default timeouts.
 
-    This function provides common defaults used throughout the MCP codebase:
-    - follow_redirects=True (always enabled)
-    - Default timeout of 30 seconds if not specified
+    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 httpx 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.
-        timeout: Request timeout as httpx.Timeout object.
-            Defaults to 30 seconds if not specified.
+        timeout: Request timeout as httpx.Timeout object. Defaults to 30s for
+            connect/write/pool and 300s for read (for long-lived SSE streams).
         auth: Optional authentication handler.
 
     Returns:
-        Configured httpx.AsyncClient instance with MCP defaults.
+        Configured httpx.AsyncClient instance.
 
     Note:
         The returned AsyncClient must be used as a context manager to ensure
         proper cleanup of connections.
-
-    Examples:
-        # Basic usage with MCP defaults
-        async with create_mcp_http_client() as client:
-            response = await client.get("https://api.example.com")
-
-        # With custom headers
-        headers = {"Authorization": "Bearer token"}
-        async with create_mcp_http_client(headers) as client:
-            response = await client.get("/endpoint")
-
-        # With both custom headers and timeout
-        timeout = httpx.Timeout(60.0, read=300.0)
-        async with create_mcp_http_client(headers, timeout) as client:
-            response = await client.get("/long-request")
-
-        # With authentication
-        from httpx 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"] = httpx.Timeout(MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT)
-    else:
-        kwargs["timeout"] = timeout
-
-    # Handle headers
+        timeout = httpx.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 httpx.AsyncClient(**kwargs)
+
+
+def _within_origin(url: httpx.URL, location: httpx.URL) -> bool:
+    """Whether `location` is on `url`'s origin, or is its https upgrade on the default ports.
+
+    httpx normalises a scheme's default port to None and lower-cases hosts, so
+    plain tuple comparison is exact. The upgrade rule is the one httpx 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
+    )
+
+
+def next_request_within_origin(response: httpx.Response) -> httpx.Request | None:
+    """The request that follows `response`'s redirect, if it is one the MCP transports follow.
+
+    That is when httpx built a next request for it (a redirect status with a
+    Location), the next request keeps the method (307/308, or any redirect of a
+    GET: httpx turns a POST into a body-less GET for 301/302/303, which would
+    drop the message), its URL stays within the origin of the request just sent
+    (same scheme, host and port, or http to https on the same host with default
+    ports), and the Location does not bring userinfo of its own (which httpx
+    would otherwise send as Basic auth; userinfo the configured URL already had
+    is kept by a relative Location and is fine). None for anything else,
+    including a non-redirect.
+    """
+    next_request = response.next_request
+    if next_request is None:
+        return None
+    sent = response.request
+    if (
+        next_request.method != sent.method
+        or (next_request.url.userinfo and next_request.url.userinfo != sent.url.userinfo)
+        or not _within_origin(sent.url, next_request.url)
+    ):
+        return None
+    return next_request
+
+
+@asynccontextmanager
+async def stream_within_origin(
+    client: httpx.AsyncClient, method: str, url: httpx.URL | str, **kwargs: Any
+) -> AsyncGenerator[httpx.Response, None]:
+    """`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
+    `next_request_within_origin` accepts, such as a 307/308 trailing-slash
+    normalisation, is followed, at most `client.max_redirects` times. Any other
+    redirect (or one past that budget) is not followed: the redirect response
+    itself is yielded, the way httpx hands one back when `follow_redirects` is
+    off, and the caller treats it as the non-success it is. The client's own
+    `follow_redirects` setting is not consulted. Requests an `httpx.Auth` flow
+    makes during the call are sent without following either; the SDK's OAuth
+    providers apply the same rule to their own requests.
+    """
+    request = client.build_request(method, url, **kwargs)
+    followed = 0
+    while True:
+        response = await client.send(request, stream=True, follow_redirects=False)
+        next_request = next_request_within_origin(response)
+        if next_request is None or followed == client.max_redirects:
+            break
+        try:
+            # Drain the redirect body so the connection returns to the pool, as httpx does when it follows.
+            await response.aread()
+        finally:
+            await response.aclose()
+        request = next_request
+        followed += 1
+    try:
+        yield response
+    finally:
+        await response.aclose()
+
+
+async def request_within_origin(
+    client: httpx.AsyncClient, method: str, url: httpx.URL | str, **kwargs: Any
+) -> httpx.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: httpx.AsyncClient, url: httpx.URL | str, *, headers: dict[str, str] | None = None
+) -> AsyncGenerator[EventSource, None]:
+    """`httpx_sse.aconnect_sse(client, "GET", url)` with the redirect handling of `stream_within_origin`."""
+    merged = httpx.Headers(_SSE_HEADERS)
+    merged.update(headers or {})
+    async with stream_within_origin(client, "GET", url, headers=merged) as response:
+        yield EventSource(response)
+
+
+def redirect_location(response: httpx.Response) -> httpx.URL | None:
+    """Where `response` redirects to, for use in a message: without userinfo, query or fragment,
+    which can carry state that does not belong in an error or a log line. None if not a redirect."""
+    if response.next_request is None:
+        return None
+    return response.next_request.url.copy_with(userinfo=b"", query=None, fragment=None)
+
+
+def redirect_note(response: httpx.Response) -> str:
+    """A suffix naming the location of a redirect response that was not followed, else empty."""
+    location = redirect_location(response)
+    if location is None:
+        return ""
+    return f" (redirected to {location}; not followed)"
+
+
+class RedirectAwareAuth(ABC, httpx.Auth):
+    """An `httpx.Auth` whose own requests follow redirects the way MCP transport requests do.
+
+    The transports send every request with redirect following off and follow a
+    redirect themselves only within the endpoint's origin (`stream_within_origin`).
+    httpx applies that per-request setting to the requests an auth flow makes
+    too (metadata discovery, registration, token), so on their own those would
+    follow nothing. Subclasses write their flow as `_auth_flow`; this class
+    drives it and, for each request the flow makes other than the one being
+    authenticated, follows a redirect that `next_request_within_origin` accepts,
+    up to `_AUTH_REDIRECT_LIMIT` times. Any other redirect response is handed
+    to the flow as it is.
+    """
+
+    @abstractmethod
+    def _auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
+        """The subclass's flow, written as `httpx.Auth.async_auth_flow` otherwise would be."""
+
+    async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
+        flow = self._auth_flow(request)
+        try:
+            outgoing = await flow.__anext__()
+            while True:
+                response = yield outgoing
+                if outgoing is not request:
+                    for _ in range(_AUTH_REDIRECT_LIMIT):
+                        follow = next_request_within_origin(response)
+                        if follow is None:
+                            break
+                        response = yield follow
+                outgoing = await flow.asend(response)
+        except StopAsyncIteration:
+            return
+        finally:
+            await flow.aclose()
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index 3ebd2e061a..7c3e5bd60c 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -5,6 +5,7 @@
 import base64
 import json
 import time
+from collections.abc import AsyncGenerator
 from unittest import mock
 from urllib.parse import parse_qs, unquote, urlparse
 
@@ -26,6 +27,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,40 +826,129 @@ async def test_resource_param_included_with_protected_resource_metadata(self, oa
         assert "resource=" in content
 
 
+async def _start_discovery(
+    provider: OAuthClientProvider,
+) -> tuple[AsyncGenerator[httpx.Request, httpx.Response], httpx.Request]:
+    """Drive `provider`'s auth flow to the point where it has sent the MCP request, seen a 401 and
+    issued its first protected-resource-metadata request; returns (flow, that request)."""
+    provider.context.current_tokens = None
+    provider.context.token_expiry_time = None
+    provider._initialized = True
+    mcp_request = httpx.Request("POST", "https://api.example.com/v1/mcp")
+    flow = provider.async_auth_flow(mcp_request)
+    sent = await flow.__anext__()
+    assert sent is mcp_request
+    # No resource_metadata hint, so discovery tries the path-based well-known URL, then the root one.
+    unauthorized = httpx.Response(401, request=mcp_request)
+    prm_request = await flow.asend(unauthorized)
+    assert (prm_request.method, str(prm_request.url)) == (
+        "GET",
+        "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp",
+    )
+    return flow, prm_request
+
+
+async def _redirect(request: httpx.Request, status: int, location: str) -> httpx.Response:
+    """A redirect answer to `request`, as httpx hands it back when it does not follow it."""
+    transport = httpx.MockTransport(lambda r: httpx.Response(status, headers={"location": location}))
+    async with httpx.AsyncClient(transport=transport) as client:
+        return await client.send(request)
+
+
+@pytest.mark.anyio
+async def test_auth_flow_follows_a_same_origin_redirect_of_its_own_request(oauth_provider: OAuthClientProvider):
+    """SDK-defined: a request the OAuth flow makes (here protected-resource metadata discovery)
+    follows a redirect that stays within its origin and keeps its method, like an MCP request."""
+    flow, prm_request = await _start_discovery(oauth_provider)
+
+    follow_up = await flow.asend(await _redirect(prm_request, 307, "/.well-known/oauth-protected-resource/v1/mcp/"))
+
+    assert (follow_up.method, str(follow_up.url)) == (
+        "GET",
+        "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp/",
+    )
+    await flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_auth_flow_does_not_follow_a_redirect_of_its_own_request_to_another_origin(
+    oauth_provider: OAuthClientProvider,
+):
+    """SDK-defined: a redirect of a flow request to another origin is handed to the flow unfollowed,
+    which treats it as "not served here" and moves to its next discovery URL."""
+    flow, prm_request = await _start_discovery(oauth_provider)
+
+    next_request = await flow.asend(await _redirect(prm_request, 307, "https://elsewhere.example/prm"))
+
+    assert (next_request.method, str(next_request.url)) == (
+        "GET",
+        "https://api.example.com/.well-known/oauth-protected-resource",
+    )
+    await flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_auth_flow_stops_following_a_redirecting_request_after_a_few_hops(
+    oauth_provider: OAuthClientProvider,
+):
+    """SDK-defined: a flow request that keeps redirecting within its origin is followed a bounded
+    number of times; the redirect after that is handed to the flow unfollowed."""
+    flow, request = await _start_discovery(oauth_provider)
+
+    hops = 0
+    while str(request.url) != "https://api.example.com/.well-known/oauth-protected-resource":
+        request = await flow.asend(
+            await _redirect(request, 307, f"/.well-known/oauth-protected-resource/v1/mcp/{hops}")
+        )
+        hops += 1
+
+    assert hops == 6  # five followed, the sixth handed back and taken as "try the next URL"
+    await flow.aclose()
+
+
+@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(httpx.Response(status)) == (keep_trying, None)
+
+
 class TestRegistrationResponse:
     """Test client registration response handling."""
 
     @pytest.mark.anyio
     async def test_handle_registration_response_reads_before_accessing_text(self):
-        """Test that response.aread() is called before accessing response.text."""
-
-        # Track if aread() was called
-        class MockResponse(httpx.Response):
-            def __init__(self):
-                self.status_code = 400
-                self._aread_called = False
-                self._text = "Registration failed with error"
-
-            async def aread(self):
-                self._aread_called = True
-                return b"test content"
-
-            @property
-            def text(self):
-                if not self._aread_called:
-                    raise RuntimeError("Response.text accessed before response.aread()")  # pragma: no cover
-                return self._text
-
-        mock_response = MockResponse()
-
-        # This should call aread() before accessing text
-        with pytest.raises(Exception) as exc_info:
-            await handle_registration_response(mock_response)
-
-        # Verify aread() was called
-        assert mock_response._aread_called
-        # Verify the error message includes the response text
-        assert "Registration failed: 400" in str(exc_info.value)
+        """The registration error carries the response text, which for a streamed response means
+        reading it first (a streamed httpx response raises ResponseNotRead otherwise)."""
+        response = httpx.Response(400, stream=httpx.ByteStream(b"Registration failed with error"))
+
+        with pytest.raises(OAuthRegistrationError) as exc_info:
+            await handle_registration_response(response)
+
+        assert str(exc_info.value) == snapshot("Registration failed: 400 Registration failed with error")
+
+    @pytest.mark.anyio
+    async def test_registration_error_names_an_unfollowed_redirect(self):
+        """SDK-defined: when the registration endpoint answered with a redirect that was not followed,
+        the error says where it pointed (without userinfo or query) instead of only the bare status."""
+        request = httpx.Request("POST", "https://as.example/register")
+        async with httpx.AsyncClient(
+            transport=httpx.MockTransport(
+                lambda r: httpx.Response(307, headers={"location": "https://u:p@elsewhere.example/register?state=x"})
+            )
+        ) as client:
+            response = await client.send(request)
+
+        with pytest.raises(OAuthRegistrationError) as exc_info:
+            await handle_registration_response(response)
+
+        assert str(exc_info.value) == snapshot(
+            "Registration failed: 307 (redirected to https://elsewhere.example/register; not followed) "
+        )
 
 
 class TestCreateClientRegistrationRequest:
diff --git a/tests/shared/test_httpx_utils.py b/tests/shared/test_httpx_utils.py
index dcc6fd003c..7709fc968c 100644
--- a/tests/shared/test_httpx_utils.py
+++ b/tests/shared/test_httpx_utils.py
@@ -1,16 +1,27 @@
-"""Tests for httpx utility functions."""
+"""Tests for the httpx helpers the client transports are built on."""
+
+from collections.abc import AsyncGenerator, AsyncIterator
+from typing import Any
 
 import httpx
+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 == httpx.Timeout(30.0, read=300.0)
 
 
 def test_custom_parameters():
@@ -22,3 +33,253 @@ def test_custom_parameters():
 
     assert client.headers["Authorization"] == "Bearer token"
     assert client.timeout.connect == 60.0
+
+
+class _Body(httpx.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[httpx.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: httpx.Request) -> httpx.Response:
+        received.append(f"{request.method} {request.url}")
+        if str(request.url) in redirects:
+            status, location = redirects[str(request.url)]
+            return httpx.Response(status, headers={"location": location}, stream=_Body(b"moved", closed))
+        return httpx.Response(200, text=request.content.decode() or "ok")
+
+    return httpx.AsyncClient(transport=httpx.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 httpx'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 {httpx.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 httpx 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):
+    """httpx 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 httpx'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_past_the_client_max_redirects_budget_is_handed_back_unfollowed():
+    """Same-origin hops are bounded by the client's max_redirects; the redirect after that is not
+    followed but handed back like any other, so a loop fails the one call rather than raising
+    (SDK-defined; max_redirects=0 therefore means "follow none")."""
+    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:
+        response = await request_within_origin(client, "GET", url)
+
+    assert response.status_code == 307
+    assert response.next_request is not None
+    assert response.next_request.url == "http://mcp.example/d"
+    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_redirect_location_with_userinfo_is_not_followed():
+    """A Location carrying user:password is handed back unfollowed even within the origin, since
+    httpx would otherwise send that userinfo as Basic auth (SDK-defined)."""
+    url = "http://mcp.example/mcp"
+    client, received, _ = _recording_client({url: (307, "http://user:secret@mcp.example/mcp/")})
+
+    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_userinfo_of_the_configured_url_kept_by_a_relative_location_is_followed():
+    """Userinfo the caller put in the endpoint URL is carried over by a relative Location (URL join
+    keeps the authority); that is the caller's own credential for the same origin, so the redirect
+    is followed as httpx itself would (SDK-defined)."""
+    url = "http://user:secret@mcp.example/mcp"
+    client, received, _ = _recording_client({url: (307, "/mcp/")})
+
+    async with client, stream_within_origin(client, "POST", url, content=b"payload") as response:
+        await response.aread()
+
+    assert response.status_code == 200
+    assert received == [f"POST {url}", "POST http://user:secret@mcp.example/mcp/"]
+
+
+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 httpx_sse.aconnect_sse() does, merged case-insensitively
+    with the caller's headers, and yields an EventSource over the final response."""
+    seen: list[httpx.Headers] = []
+
+    def serve(request: httpx.Request) -> httpx.Response:
+        seen.append(request.headers)
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, text="data: hello\n\n")
+
+    client = httpx.AsyncClient(transport=httpx.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.aiter_sse()]
+        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 httpx 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 (httpx behaviour the transports rely on)."""
+    received: list[str] = []
+
+    class TokenThenRequest(httpx.Auth):
+        async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
+            token_response = yield httpx.Request("POST", "http://mcp.example/token", content=b"grant")
+            request.headers["x-token-status"] = str(token_response.status_code)
+            yield request
+
+    def serve(request: httpx.Request) -> httpx.Response:
+        received.append(f"{request.method} {request.url}")
+        if request.url.path == "/token":
+            return httpx.Response(307, headers={"location": "http://other.example/token"})
+        return httpx.Response(200, text=request.headers["x-token-status"])
+
+    client = httpx.AsyncClient(transport=httpx.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 7604450f81..77d6cac65f 100644
--- a/tests/shared/test_sse.py
+++ b/tests/shared/test_sse.py
@@ -4,13 +4,12 @@
 import time
 from collections.abc import AsyncGenerator, Generator
 from typing import Any
-from unittest.mock import AsyncMock, MagicMock, Mock, patch
+from unittest.mock import Mock
 
 import anyio
 import httpx
 import pytest
 import uvicorn
-from httpx_sse import ServerSentEvent
 from inline_snapshot import snapshot
 from pydantic import AnyUrl
 from starlette.applications import Starlette
@@ -26,6 +25,7 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import McpError
+from mcp.shared.message import SessionMessage
 from mcp.types import (
     EmptyResult,
     ErrorData,
@@ -538,12 +538,6 @@ def test_sse_server_transport_endpoint_validation(endpoint: str, expected_result
         assert sse._endpoint.startswith("/")
 
 
-# ResourceWarning filter: When mocking aconnect_sse, the sse_client's internal task
-# group doesn't receive proper cancellation signals, so the sse_reader task's finally
-# block (which closes read_stream_writer) doesn't execute. This is a test artifact -
-# the actual code path (`if not sse.data: continue`) IS exercised and works correctly.
-# Production code with real SSE connections cleans up properly.
-@pytest.mark.filterwarnings("ignore::ResourceWarning")
 @pytest.mark.anyio
 async def test_sse_client_handles_empty_keepalive_pings() -> None:
     """Test that SSE client properly handles empty data lines (keep-alive pings).
@@ -552,10 +546,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(
         protocolVersion="2024-11-05",
         capabilities=ServerCapabilities(),
@@ -567,38 +561,82 @@ 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"
+    )
 
-    # Create mock SSE events using httpx_sse's ServerSentEvent
-    async def mock_aiter_sse() -> AsyncGenerator[ServerSentEvent, None]:
-        # First: endpoint event
-        yield ServerSentEvent(event="endpoint", data="/messages/?session_id=abc123")
-        # Empty data keep-alive ping - this is what we're testing
-        yield ServerSentEvent(event="message", data="")
-        # Real JSON-RPC response
-        yield ServerSentEvent(event="message", data=response_json)
-
-    mock_event_source = MagicMock()
-    mock_event_source.aiter_sse.return_value = mock_aiter_sse()
-    mock_event_source.response = MagicMock()
-    mock_event_source.response.raise_for_status = MagicMock()
-
-    mock_aconnect_sse = MagicMock()
-    mock_aconnect_sse.__aenter__ = AsyncMock(return_value=mock_event_source)
-    mock_aconnect_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.post = AsyncMock(return_value=MagicMock(status_code=200, raise_for_status=MagicMock()))
-
-    with (
-        patch("mcp.client.sse.create_mcp_http_client", return_value=mock_client),
-        patch("mcp.client.sse.aconnect_sse", return_value=mock_aconnect_sse),
-    ):
-        async with sse_client("http://test/sse") as (read_stream, _):
-            # Read the message - should skip the empty one and get the real response
+    def serve(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/sse"
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=event_stream)
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(transport=httpx.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()
-            # If we get here without error, the empty message was skipped successfully
-            assert not isinstance(msg, Exception)
+            assert isinstance(msg, SessionMessage)
             assert isinstance(msg.message.root, types.JSONRPCResponse)
             assert msg.message.root.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 httpx's no-follow default."""
+    received: list[str] = []
+
+    def serve(request: httpx.Request) -> httpx.Response:
+        received.append(str(request.url))
+        if request.url.path == "/sse":
+            return httpx.Response(307, headers={"location": "/sse/"})
+        assert request.url.path == "/sse/"
+        return httpx.Response(
+            200, headers={"content-type": "text/event-stream"}, text="event: endpoint\ndata: /messages/\n\n"
+        )
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(transport=httpx.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/"]
+
+
+@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
+    (raised inside sse_client's task group) and that origin is never contacted."""
+    received: list[str] = []
+
+    def serve(request: httpx.Request) -> httpx.Response:
+        received.append(str(request.url))
+        return httpx.Response(307, headers={"location": "http://other.example/sse"})
+
+    def factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(transport=httpx.MockTransport(serve), follow_redirects=True)
+
+    with anyio.fail_after(5):
+        with pytest.raises(Exception) 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.group_contains(httpx.HTTPStatusError, match="307 Temporary Redirect")
+    assert received == ["http://test/sse"]
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index db96706be5..35376bc1c9 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -20,6 +20,7 @@
 import requests
 import uvicorn
 from httpx_sse import ServerSentEvent
+from inline_snapshot import snapshot
 from pydantic import AnyUrl
 from starlette.applications import Starlette
 from starlette.requests import Request
@@ -1255,41 +1256,36 @@ async def test_streamable_http_client_session_termination(basic_server: None, ba
 
 
 @pytest.mark.anyio
-async def test_streamable_http_client_session_termination_204(
-    basic_server: None, basic_server_url: str, monkeypatch: pytest.MonkeyPatch
-):
+async def test_streamable_http_client_session_termination_204(basic_server: None, basic_server_url: str):
     """Test client session termination functionality with a 204 response.
 
-    This test patches the httpx client to return a 204 response for DELETEs.
+    The 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 = httpx.AsyncClient.delete
-
-    # Mock the client's delete method to return a 204
-    async def mock_delete(self: httpx.AsyncClient, *args: Any, **kwargs: Any) -> httpx.Response:
-        # Call the original method to get the real response
-        response = await original_delete(self, *args, **kwargs)
+    class AnswerDeleteWith204(httpx.AsyncBaseTransport):
+        def __init__(self) -> None:
+            self.inner = httpx.AsyncHTTPTransport()
 
-        # Create a new response with 204 status code but same headers
-        mocked_response = httpx.Response(
-            204,
-            headers=response.headers,
-            content=response.content,
-            request=response.request,
-        )
-        return mocked_response
+        async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+            response = await self.inner.handle_async_request(request)
+            if request.method != "DELETE" or response.status_code != 200:
+                return response
+            await response.aread()
+            return httpx.Response(204, headers=response.headers, request=request)
 
-    # Apply the patch to the httpx client
-    monkeypatch.setattr(httpx.AsyncClient, "delete", mock_delete)
+        async def aclose(self) -> None:
+            await self.inner.aclose()
 
     captured_session_id = None
 
-    # Create the streamable_http_client with a custom httpx client to capture headers
-    async with streamable_http_client(f"{basic_server_url}/mcp") as (
-        read_stream,
-        write_stream,
-        get_session_id,
+    async with (
+        httpx.AsyncClient(transport=AnswerDeleteWith204()) as terminating_client,
+        streamable_http_client(f"{basic_server_url}/mcp", http_client=terminating_client) as (
+            read_stream,
+            write_stream,
+            get_session_id,
+        ),
     ):
         async with ClientSession(read_stream, write_stream) as session:
             # Initialize the session
@@ -2297,7 +2293,7 @@ async def test_streamable_http_client_does_not_mutate_provided_client(
         "Authorization": "Bearer test-token",
     }
 
-    async with httpx.AsyncClient(headers=original_headers, follow_redirects=True) as custom_client:
+    async with httpx.AsyncClient(headers=original_headers) as custom_client:
         # Use the client with streamable_http_client
         async with streamable_http_client(f"{basic_server_url}/mcp", http_client=custom_client) as (
             read_stream,
@@ -2328,7 +2324,7 @@ async def test_streamable_http_client_mcp_headers_override_defaults(
     # httpx.AsyncClient has default "accept: */*" header
     # We need to verify that our MCP accept header overrides it in actual requests
 
-    async with httpx.AsyncClient(follow_redirects=True) as client:
+    async with httpx.AsyncClient() as client:
         # Verify client has default accept header
         assert client.headers.get("accept") == "*/*"
 
@@ -2366,7 +2362,7 @@ async def test_streamable_http_client_preserves_custom_with_mcp_headers(
         "Authorization": "Bearer test-token",
     }
 
-    async with httpx.AsyncClient(headers=custom_headers, follow_redirects=True) as client:
+    async with httpx.AsyncClient(headers=custom_headers) as client:
         async with streamable_http_client(f"{basic_server_url}/mcp", http_client=client) as (
             read_stream,
             write_stream,
@@ -2426,3 +2422,172 @@ async def test_streamablehttp_client_deprecation_warning(basic_server: None, bas
                 await session.initialize()
                 tools = await session.list_tools()
                 assert len(tools.tools) > 0
+
+
+@pytest.mark.anyio
+async def test_trailing_slash_redirect_within_origin_is_followed_by_the_transport(
+    basic_server: None, basic_server_url: str
+) -> 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
+    httpx's no-follow default still connects."""
+    urls: list[str] = []
+
+    async def record(request: httpx.Request) -> None:
+        urls.append(str(request.url))
+
+    with anyio.fail_after(10):
+        async with (
+            httpx.AsyncClient(event_hooks={"request": [record]}) as http,
+            streamable_http_client(f"{basic_server_url}/mcp", http_client=http) as (read_stream, write_stream, _),
+            ClientSession(read_stream, write_stream) as session,
+        ):
+            result = await session.initialize()
+
+    assert result.serverInfo.name == SERVER_NAME
+    assert urls[:2] == [f"{basic_server_url}/mcp", f"{basic_server_url}/mcp/"]
+
+
+def _leaf_exception(exc: BaseException) -> BaseException:
+    """The one exception inside the (possibly nested) exception group an anyio task group raises."""
+    while (inner := getattr(exc, "exceptions", None)) is not None:
+        (exc,) = inner
+    return exc
+
+
+async def _assert_redirected_post_fails(url: str, location: str, expected_message: str) -> None:
+    """Send one request through streamable_http_client, with a client configured to follow redirects,
+    to a server answering `url` with a 307 to `location`, and check that the connection ends with
+    HTTPStatusError carrying `expected_message` and that nothing but `url` was requested."""
+    urls: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        urls.append(str(request.url))
+        return httpx.Response(307, headers={"location": location})
+
+    with anyio.fail_after(5):
+        # The request's POST fails inside the transport's task group, which ends the connection.
+        with pytest.raises(Exception) as exc_info:
+            async with (  # pragma: no branch
+                httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True) as http,
+                streamable_http_client(url, http_client=http) as (read_stream, write_stream, _),
+                read_stream,
+                write_stream,
+            ):
+                request = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+                await write_stream.send(SessionMessage(JSONRPCMessage(request)))
+                await read_stream.receive()
+    error = _leaf_exception(exc_info.value)
+    assert isinstance(error, httpx.HTTPStatusError)
+    assert error.response.status_code == 307
+    assert str(error) == expected_message
+    assert urls == [url]
+
+
+@pytest.mark.anyio
+async def test_redirect_to_another_origin_is_not_followed_and_fails_the_request() -> None:
+    """SDK-defined: a redirect pointing outside the endpoint's origin is not followed, whatever the
+    caller's client is configured to do: nothing is sent to the other origin, and the request fails
+    the way any non-2xx response does, with HTTPStatusError naming the location."""
+    await _assert_redirected_post_fails(
+        "http://mcp.example/mcp",
+        "http://other.example/x",
+        snapshot(
+            "Redirect to http://other.example/x not followed; use that URL as the endpoint if it is the intended server"
+        ),
+    )
+
+
+@pytest.mark.anyio
+async def test_https_endpoint_redirected_to_plain_http_is_explained_and_the_https_form_suggested() -> None:
+    """SDK-authored text: a redirect of an HTTPS endpoint to plain HTTP on the same host (the usual
+    sign of a TLS-terminating proxy the server does not trust) never suggests the http:// URL."""
+    await _assert_redirected_post_fails(
+        "https://mcp.example/mcp",
+        "http://mcp.example/mcp/",
+        snapshot("""\
+Redirect to http://mcp.example/mcp/ not followed: it would downgrade this HTTPS endpoint to plain HTTP.
+The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust,
+often combined with a trailing-slash difference. Try https://mcp.example/mcp/ instead, or fix the proxy settings.\
+"""),
+    )
+
+
+@pytest.mark.anyio
+async def test_unfollowed_redirect_location_is_named_without_its_query_string() -> None:
+    """SDK-authored text: the location is reported without query or userinfo, which may carry state
+    that does not belong in an error message or a log line."""
+    await _assert_redirected_post_fails(
+        "http://mcp.example/mcp",
+        "https://idp.example/l?state=s3cr3t&nonce=n",
+        snapshot(
+            "Redirect to https://idp.example/l not followed; use that URL as the endpoint if it is the intended server"
+        ),
+    )
+
+
+@pytest.mark.anyio
+async def test_https_endpoint_redirected_to_plain_http_elsewhere_never_suggests_the_http_url() -> None:
+    """SDK-authored text: the downgrade explanation applies whatever host the http:// location names,
+    so the message never offers a plain-HTTP URL as the endpoint to configure."""
+    await _assert_redirected_post_fails(
+        "https://mcp.example/mcp",
+        "http://backend.lan:8000/mcp/",
+        snapshot("""\
+Redirect to http://backend.lan:8000/mcp/ not followed: it would downgrade this HTTPS endpoint to plain HTTP.
+The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust,
+often combined with a trailing-slash difference. Try https://backend.lan:8000/mcp/ instead, or fix the proxy settings.\
+"""),
+    )
+
+
+@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: httpx.Request) -> httpx.Response:
+        gets.append(str(request.url))
+        return httpx.Response(307, headers={"location": "http://other.example/mcp"})
+
+    transport = StreamableHTTPTransport("http://test/mcp")
+    transport.session_id = "session-1"
+    writer, reader = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    with anyio.fail_after(5):
+        async with httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True) as http:
+            await transport.handle_get_stream(http, writer)
+    writer.close()
+    reader.close()
+    assert gets == ["http://test/mcp"]
+
+
+@pytest.mark.anyio
+async def test_resumption_redirected_elsewhere_fails_the_resumed_request() -> None:
+    """SDK-defined: a resumption GET answered with a redirect to another origin is not followed;
+    the resumed request fails with HTTPStatusError naming the location, like a redirected POST."""
+    seen: list[tuple[str, str | None]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append((f"{request.method} {request.url}", request.headers.get("last-event-id")))
+        return httpx.Response(307, headers={"location": "http://other.example/mcp"})
+
+    with anyio.fail_after(5):
+        with pytest.raises(Exception) as exc_info:
+            async with (  # pragma: no branch
+                httpx.AsyncClient(transport=httpx.MockTransport(handler), follow_redirects=True) as http,
+                streamable_http_client("http://test/mcp", http_client=http) as (read_stream, write_stream, _),
+                read_stream,
+                write_stream,
+            ):
+                request = JSONRPCRequest(jsonrpc="2.0", id="resume-1", method="tools/call", params={})
+                metadata = ClientMessageMetadata(resumption_token="evt-41")
+                await write_stream.send(SessionMessage(JSONRPCMessage(request), metadata=metadata))
+                await read_stream.receive()
+    error = _leaf_exception(exc_info.value)
+    assert isinstance(error, httpx.HTTPStatusError)
+    assert str(error) == 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")]