From 0a2fe62cdee1b370af6dd93212802e513f6fc03a Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:21:14 +0000 Subject: [PATCH 1/3] [v1.x] Follow redirects only within the MCP endpoint's origin Backport of #3397 to the 1.x line (httpx / httpx-sse). The Streamable HTTP and SSE client transports now send every request with redirect following off and follow a redirect themselves only when it stays on the endpoint's origin (same scheme, host and port, or http -> https on the same host with default ports), keeps the request method, and carries no userinfo. Any other redirect is handed back unfollowed and fails the way a non-2xx response does on 1.x (httpx.HTTPStatusError), with a message naming the location. The caller's client `follow_redirects` setting is not consulted either way, so the usual `/mcp` -> `/mcp/` trailing-slash redirect keeps working with any client. OAuthClientProvider applies the same rule to the requests its flow makes (metadata discovery, registration, token) via a RedirectAwareAuth base; a 3xx from a discovery URL is treated like a 4xx (try the next candidate). create_mcp_http_client no longer sets follow_redirects=True; the simple-tool example that used it to fetch arbitrary pages gets its own plain client. sse_client now also closes its receive streams when connecting fails. --- docs/authorization.md | 2 +- docs/client.md | 23 ++ .../mcp_simple_auth_client/main.py | 2 +- .../simple-tool/mcp_simple_tool/server.py | 5 +- examples/snippets/clients/oauth_client.py | 2 +- src/mcp/client/auth/oauth2.py | 13 +- src/mcp/client/auth/utils.py | 11 +- src/mcp/client/sse.py | 30 +- src/mcp/client/streamable_http.py | 67 +++-- src/mcp/shared/_httpx_utils.py | 216 ++++++++++++--- tests/client/test_auth.py | 149 ++++++++-- tests/shared/test_httpx_utils.py | 259 +++++++++++++++++- tests/shared/test_sse.py | 127 ++++++--- tests/shared/test_streamable_http.py | 215 +++++++++++++-- 14 files changed, 927 insertions(+), 194 deletions(-) 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)_ +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..57dcb9efc9 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,183 @@ 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 carries no userinfo (which httpx would otherwise + send as Basic auth). 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 + 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..8dc2dd7d96 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,239 @@ 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_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..110203ba1c 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,83 @@ 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, write_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.message.root, types.JSONRPCResponse) - assert msg.message.root.id == 1 + + 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..b65cbb1e55 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,164 @@ 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 _redirected_post_error(url: str, location: str) -> httpx.HTTPStatusError: + """Send one request through streamable_http_client, with a client configured to follow redirects, + to a server answering `url` with a 307 to `location`; return the error that ends the connection, + having checked 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 urls == [url] + return error + + +@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.""" + error = await _redirected_post_error("http://mcp.example/mcp", "http://other.example/mcp/") + + assert str(error) == snapshot( + "Redirect to http://other.example/mcp/ 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.""" + error = await _redirected_post_error("https://mcp.example/mcp", "http://mcp.example/mcp/") + + assert str(error) == 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.""" + error = await _redirected_post_error("http://mcp.example/mcp", "https://sso.example/login?state=s3cr3t&nonce=n") + + assert str(error) == snapshot( + "Redirect to https://sso.example/login 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.""" + error = await _redirected_post_error("https://mcp.example/mcp", "http://backend.lan:8000/mcp/") + + assert str(error) == 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")] From 0cff7e3299accd5137f2bcab40ca9a7ff31b3513 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:41:16 +0000 Subject: [PATCH 2/3] tests: keep assertions inside the awaited helpers Python 3.11's tracer does not report the line after a coroutine that handled an exception group returns, so the snapshot comparisons move into the helper and the keep-alive assertions back inside the sse_client block; drop an unused stream name. --- tests/shared/test_sse.py | 9 +++--- tests/shared/test_streamable_http.py | 48 ++++++++++++++++------------ 2 files changed, 32 insertions(+), 25 deletions(-) diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index 110203ba1c..77d6cac65f 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -579,12 +579,11 @@ def factory( 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, write_stream): + 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.root, types.JSONRPCResponse) - assert msg.message.root.id == 1 + assert isinstance(msg, SessionMessage) + assert isinstance(msg.message.root, types.JSONRPCResponse) + assert msg.message.root.id == 1 @pytest.mark.anyio diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index b65cbb1e55..35376bc1c9 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -2455,10 +2455,10 @@ def _leaf_exception(exc: BaseException) -> BaseException: return exc -async def _redirected_post_error(url: str, location: str) -> httpx.HTTPStatusError: +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`; return the error that ends the connection, - having checked that nothing but `url` was requested.""" + 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: @@ -2480,8 +2480,8 @@ def handler(request: httpx.Request) -> httpx.Response: 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] - return error @pytest.mark.anyio @@ -2489,10 +2489,12 @@ async def test_redirect_to_another_origin_is_not_followed_and_fails_the_request( """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.""" - error = await _redirected_post_error("http://mcp.example/mcp", "http://other.example/mcp/") - - assert str(error) == snapshot( - "Redirect to http://other.example/mcp/ not followed; use that URL as the endpoint if it is the intended server" + 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" + ), ) @@ -2500,23 +2502,27 @@ async def test_redirect_to_another_origin_is_not_followed_and_fails_the_request( 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.""" - error = await _redirected_post_error("https://mcp.example/mcp", "http://mcp.example/mcp/") - - assert str(error) == snapshot("""\ + 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.""" - error = await _redirected_post_error("http://mcp.example/mcp", "https://sso.example/login?state=s3cr3t&nonce=n") - - assert str(error) == snapshot( - "Redirect to https://sso.example/login not followed; use that URL as the endpoint if it is the intended server" + 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" + ), ) @@ -2524,13 +2530,15 @@ async def test_unfollowed_redirect_location_is_named_without_its_query_string() 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.""" - error = await _redirected_post_error("https://mcp.example/mcp", "http://backend.lan:8000/mcp/") - - assert str(error) == snapshot("""\ + 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 From 113c83f97350fe7abc2124d5c682741ee212290e Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:56:54 +0000 Subject: [PATCH 3/3] Keep following a relative redirect when the endpoint URL carries userinfo A relative Location joined onto a URL with user:pass@ keeps that userinfo, which is the caller's own credential for the same origin; only refuse userinfo the redirect itself introduces. --- src/mcp/shared/_httpx_utils.py | 8 +++++--- tests/shared/test_httpx_utils.py | 14 ++++++++++++++ 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py index 57dcb9efc9..248e797efd 100644 --- a/src/mcp/shared/_httpx_utils.py +++ b/src/mcp/shared/_httpx_utils.py @@ -91,8 +91,10 @@ def next_request_within_origin(response: httpx.Response) -> httpx.Request | None 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 carries no userinfo (which httpx would otherwise - send as Basic auth). None for anything else, including a non-redirect. + 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: @@ -100,7 +102,7 @@ def next_request_within_origin(response: httpx.Response) -> httpx.Request | None sent = response.request if ( next_request.method != sent.method - or next_request.url.userinfo + or (next_request.url.userinfo and next_request.url.userinfo != sent.url.userinfo) or not _within_origin(sent.url, next_request.url) ): return None diff --git a/tests/shared/test_httpx_utils.py b/tests/shared/test_httpx_utils.py index 8dc2dd7d96..7709fc968c 100644 --- a/tests/shared/test_httpx_utils.py +++ b/tests/shared/test_httpx_utils.py @@ -208,6 +208,20 @@ async def test_redirect_location_with_userinfo_is_not_followed(): 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"