From b8a107d16c8543391af58e2e06acdf296bb34cda Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Tue, 7 Jul 2026 19:23:56 +0000 Subject: [PATCH] Scope HTTP client redirect following to the request's origin create_mcp_http_client followed every redirect, so everything configured on a client (headers, auth, request bodies) was re-sent to whatever host a Location header named. Clients built by the factory now follow redirects within the same origin (scheme, host, and port), plus http-to-https upgrades of the same host on default ports, and raise the new RedirectError for anything else - before the next request is sent. - transports resolve a refused redirect in-band: requests get a JSON-RPC error naming the target and the remedy, notifications are delivered to the session's message handler; the standalone GET stream stops retrying an endpoint that keeps redirecting - caller-supplied clients that follow no redirects get the same clear error on POST, GET stream, and SSE connect instead of an opaque content-type error - OAuth discovery, registration, token, refresh, and the identity-assertion token exchange fail loudly on redirect responses instead of silently trying the next URL or abandoning the discovery chain - RedirectError and create_mcp_http_client are exported from the top-level mcp package; migration.md documents the behavior change; docs and examples configure clients through the factory, and the general-purpose fetch example uses a browser-like client of its own --- docs/client/transports.md | 2 +- docs/migration.md | 35 ++- docs/run/asgi.md | 2 +- docs_src/client_transports/tutorial003.py | 5 +- docs_src/identity_assertion/tutorial001.py | 5 +- docs_src/oauth_clients/tutorial001.py | 5 +- docs_src/oauth_clients/tutorial002.py | 6 +- .../mcp_simple_auth_client/main.py | 4 +- .../simple-tool/mcp_simple_tool/server.py | 7 +- .../clients/identity_assertion_client.py | 6 +- examples/snippets/clients/oauth_client.py | 5 +- src/mcp/__init__.py | 3 + .../auth/extensions/identity_assertion.py | 3 + src/mcp/client/auth/oauth2.py | 5 + src/mcp/client/auth/utils.py | 11 + src/mcp/client/sse.py | 46 +++- src/mcp/client/streamable_http.py | 60 ++++- src/mcp/shared/_httpx_utils.py | 134 ++++++++++- .../extensions/test_identity_assertion.py | 24 ++ tests/client/test_auth.py | 47 ++++ tests/shared/conftest.py | 12 + tests/shared/test_httpx_utils.py | 209 +++++++++++++++++- tests/shared/test_sse.py | 125 ++++++++++- tests/shared/test_streamable_http.py | 174 ++++++++++++++- 24 files changed, 878 insertions(+), 57 deletions(-) diff --git a/docs/client/transports.md b/docs/client/transports.md index 1587a1a7d7..e12e4c892e 100644 --- a/docs/client/transports.md +++ b/docs/client/transports.md @@ -29,7 +29,7 @@ Pass a URL string and you get **Streamable HTTP**, the transport you deploy behi --8<-- "docs_src/client_transports/tutorial002.py" ``` -That is the whole production client. `Client` wraps the URL in `streamable_http_client(...)` for you, on top of an `httpx.AsyncClient` configured the way MCP needs: `follow_redirects=True`, a 30-second timeout for connect/write/pool, and a 300-second read timeout because the server may hold a response stream open. +That is the whole production client. `Client` wraps the URL in `streamable_http_client(...)` for you, on top of an `httpx.AsyncClient` configured the way MCP needs: same-origin redirect handling, a 30-second timeout for connect/write/pool, and a 300-second read timeout because the server may hold a response stream open. !!! check A `Client` you have constructed is **not** connected. Construction only picks the transport; diff --git a/docs/migration.md b/docs/migration.md index 8822d449d0..e69bb0d674 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -1595,14 +1595,15 @@ async with streamablehttp_client( ```python import httpx +from mcp import create_mcp_http_client from mcp.client.streamable_http import streamable_http_client -# Configure headers, timeout, and auth on the httpx.AsyncClient -http_client = httpx.AsyncClient( +# Configure headers, timeout, and auth on the client. create_mcp_http_client +# applies the SDK defaults (timeouts, same-origin redirect handling). +http_client = create_mcp_http_client( headers={"Authorization": "Bearer token"}, timeout=httpx.Timeout(30, read=300), auth=my_auth, - follow_redirects=True, ) async with http_client: @@ -1613,7 +1614,26 @@ async with http_client: ... ``` -v1's internal client set `follow_redirects=True`; set it explicitly when supplying your own `httpx.AsyncClient` to preserve that behavior. +Prefer `create_mcp_http_client` when supplying your own client: a plain `httpx.AsyncClient` does not follow redirects at all, and configuring `follow_redirects=True` yourself sends everything on the client (headers, auth, bodies) to whichever server a redirect names. + +### HTTP clients only follow same-origin redirects + +Clients created by the SDK (including `Client("http://...")`, `sse_client`, and the +default client inside `streamable_http_client`) now vet redirect targets before +following. Redirects that stay on the same origin (scheme, host, and port) and +http-to-https upgrades of the same host still work; any other redirect raises +`mcp.RedirectError` naming the target instead of being followed. Everything +configured on a client - headers, auth, request bodies - is sent to whichever +server it talks to, so requests now stay with the server you configured, and a +misconfigured URL fails with a clear error instead of silently bouncing. + +To migrate: connect to the final URL the error names. If your deployment +genuinely requires following a redirect to a different origin, supply your own +`httpx.AsyncClient(follow_redirects=True)` - after verifying you trust every +destination in the chain. + +`create_mcp_http_client` (the SDK's client factory) and `RedirectError` are now +exported from the top-level `mcp` package. ### `get_session_id` callback removed from `streamable_http_client` @@ -1638,6 +1658,7 @@ async with streamable_http_client(url) as (read_stream, write_stream, get_sessio ```python import httpx +from mcp import create_mcp_http_client from mcp.client.streamable_http import streamable_http_client # Option 1: Simply ignore if you don't need the session ID @@ -1653,10 +1674,8 @@ async def capture_session_id(response: httpx.Response) -> None: if session_id: captured_session_ids.append(session_id) -http_client = httpx.AsyncClient( - event_hooks={"response": [capture_session_id]}, - follow_redirects=True, -) +http_client = create_mcp_http_client() +http_client.event_hooks["response"].append(capture_session_id) async with http_client: async with streamable_http_client(url, http_client=http_client) as (read_stream, write_stream): diff --git a/docs/run/asgi.md b/docs/run/asgi.md index 2eca9273cd..29cfbbdd09 100644 --- a/docs/run/asgi.md +++ b/docs/run/asgi.md @@ -94,7 +94,7 @@ That trailing `/mcp` is `streamable_http_path`. Set it to `"/"` and the mount pr --8<-- "docs_src/asgi/tutorial004.py" ``` -Now clients connect to `/notes`, not `/notes/mcp`. +Now clients connect to `/notes/`, not `/notes/mcp`. (The trailing slash is the canonical path for this mount shape; the bare `/notes` answers with a same-origin redirect to it.) ## CORS for browser clients diff --git a/docs_src/client_transports/tutorial003.py b/docs_src/client_transports/tutorial003.py index 0134a72561..b000a761f0 100644 --- a/docs_src/client_transports/tutorial003.py +++ b/docs_src/client_transports/tutorial003.py @@ -1,14 +1,13 @@ import httpx -from mcp import Client +from mcp import Client, create_mcp_http_client from mcp.client.streamable_http import streamable_http_client async def main() -> None: - async with httpx.AsyncClient( + async with create_mcp_http_client( headers={"Authorization": "Bearer ..."}, timeout=httpx.Timeout(30.0, read=300.0), - follow_redirects=True, ) as http_client: transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client) async with Client(transport) as client: diff --git a/docs_src/identity_assertion/tutorial001.py b/docs_src/identity_assertion/tutorial001.py index 8a7e9a050b..8ba44e26cc 100644 --- a/docs_src/identity_assertion/tutorial001.py +++ b/docs_src/identity_assertion/tutorial001.py @@ -1,10 +1,9 @@ import time import uuid -import httpx import jwt -from mcp import Client +from mcp import Client, create_mcp_http_client from mcp.client.auth.extensions.identity_assertion import IdentityAssertionOAuthProvider from mcp.client.streamable_http import streamable_http_client from mcp.shared.auth import OAuthClientInformationFull, OAuthToken @@ -62,7 +61,7 @@ async def fetch_id_jag(audience: str, resource: str) -> str: async def main() -> None: - async with httpx.AsyncClient(auth=oauth, follow_redirects=True) as http_client: + async with create_mcp_http_client(auth=oauth) as http_client: transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client) async with Client(transport) as client: result = await client.list_tools() diff --git a/docs_src/oauth_clients/tutorial001.py b/docs_src/oauth_clients/tutorial001.py index 4360379a29..001835673c 100644 --- a/docs_src/oauth_clients/tutorial001.py +++ b/docs_src/oauth_clients/tutorial001.py @@ -1,9 +1,8 @@ from urllib.parse import parse_qs, urlparse -import httpx from pydantic import AnyUrl -from mcp import Client +from mcp import Client, create_mcp_http_client from mcp.client.auth import AuthorizationCodeResult, OAuthClientProvider from mcp.client.streamable_http import streamable_http_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken @@ -55,7 +54,7 @@ async def wait_for_callback() -> AuthorizationCodeResult: async def main() -> None: - async with httpx.AsyncClient(auth=oauth, follow_redirects=True) as http_client: + async with create_mcp_http_client(auth=oauth) as http_client: transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client) async with Client(transport) as client: result = await client.list_tools() diff --git a/docs_src/oauth_clients/tutorial002.py b/docs_src/oauth_clients/tutorial002.py index b5b052c962..72a0d35771 100644 --- a/docs_src/oauth_clients/tutorial002.py +++ b/docs_src/oauth_clients/tutorial002.py @@ -1,6 +1,4 @@ -import httpx - -from mcp import Client +from mcp import Client, create_mcp_http_client from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider from mcp.client.streamable_http import streamable_http_client from mcp.shared.auth import OAuthClientInformationFull, OAuthToken @@ -34,7 +32,7 @@ async def set_client_info(self, client_info: OAuthClientInformationFull) -> None async def main() -> None: - async with httpx.AsyncClient(auth=oauth, follow_redirects=True) as http_client: + async with create_mcp_http_client(auth=oauth) as http_client: transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client) async with Client(transport) as client: result = await client.list_tools() diff --git a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py b/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py index 0d461d5d11..4171b8827d 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 @@ -17,7 +17,7 @@ from typing import Any from urllib.parse import parse_qs, urlparse -import httpx +from mcp import create_mcp_http_client from mcp.client._transport import ReadStream, WriteStream from mcp.client.auth import AuthorizationCodeResult, OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession @@ -233,7 +233,7 @@ async def _default_redirect_handler(authorization_url: str) -> None: await self._run_session(read_stream, write_stream) else: print("📡 Opening StreamableHTTP transport connection with auth...") - async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client: + async with create_mcp_http_client(auth=oauth_auth) as custom_client: async with streamable_http_client(url=self.server_url, http_client=custom_client) as ( read_stream, write_stream, diff --git a/examples/servers/simple-tool/mcp_simple_tool/server.py b/examples/servers/simple-tool/mcp_simple_tool/server.py index b16249e068..0c3a136551 100644 --- a/examples/servers/simple-tool/mcp_simple_tool/server.py +++ b/examples/servers/simple-tool/mcp_simple_tool/server.py @@ -1,15 +1,18 @@ import anyio import click +import httpx import mcp_types as types from mcp.server import Server, ServerRequestContext -from mcp.shared._httpx_utils import create_mcp_http_client async def fetch_website( url: str, ) -> list[types.ContentBlock]: headers = {"User-Agent": "MCP Test Server (github.com/modelcontextprotocol/python-sdk)"} - async with create_mcp_http_client(headers=headers) as client: + # A general-purpose web fetch wants ordinary browser-like redirect + # behavior; create_mcp_http_client is for talking to a configured MCP + # endpoint and only follows same-origin redirects. + async with httpx.AsyncClient(headers=headers, follow_redirects=True, timeout=30.0) as client: response = await client.get(url) response.raise_for_status() return [types.TextContent(type="text", text=response.text)] diff --git a/examples/snippets/clients/identity_assertion_client.py b/examples/snippets/clients/identity_assertion_client.py index 218df4bcfc..f74f6d3c88 100644 --- a/examples/snippets/clients/identity_assertion_client.py +++ b/examples/snippets/clients/identity_assertion_client.py @@ -16,9 +16,7 @@ import asyncio -import httpx - -from mcp import ClientSession +from mcp import ClientSession, create_mcp_http_client from mcp.client.auth.extensions.identity_assertion import IdentityAssertionOAuthProvider from mcp.client.streamable_http import streamable_http_client from mcp.shared.auth import OAuthClientInformationFull, OAuthToken @@ -66,7 +64,7 @@ async def main() -> None: scope="user", ) - async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as http_client: + async with create_mcp_http_client(auth=oauth_auth) as http_client: async with streamable_http_client("http://localhost:8001/mcp", http_client=http_client) as (read, write): async with ClientSession(read, write) as session: await session.initialize() diff --git a/examples/snippets/clients/oauth_client.py b/examples/snippets/clients/oauth_client.py index 2085b9a1db..e35aaba669 100644 --- a/examples/snippets/clients/oauth_client.py +++ b/examples/snippets/clients/oauth_client.py @@ -9,10 +9,9 @@ import asyncio from urllib.parse import parse_qs, urlparse -import httpx from pydantic import AnyUrl -from mcp import ClientSession +from mcp import ClientSession, create_mcp_http_client from mcp.client.auth import AuthorizationCodeResult, OAuthClientProvider, TokenStorage from mcp.client.streamable_http import streamable_http_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken @@ -72,7 +71,7 @@ async def main(): callback_handler=handle_callback, ) - async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client: + async with create_mcp_http_client(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/__init__.py b/src/mcp/__init__.py index 085e445d4a..31315fd6cd 100644 --- a/src/mcp/__init__.py +++ b/src/mcp/__init__.py @@ -65,6 +65,7 @@ from .client.stdio import StdioServerParameters, stdio_client from .server.session import ServerSession from .server.stdio import stdio_server +from .shared._httpx_utils import RedirectError, create_mcp_http_client from .shared.exceptions import MCPDeprecationWarning, MCPError, UrlElicitationRequiredError from .shared.uri_template import InvalidUriTemplate, UriTemplate @@ -108,6 +109,7 @@ "PromptsCapability", "ReadResourceRequest", "ReadResourceResult", + "RedirectError", "Resource", "ResourcesCapability", "ResourceUpdatedNotification", @@ -137,6 +139,7 @@ "UriTemplate", "UrlElicitationRequiredError", "InvalidUriTemplate", + "create_mcp_http_client", "stdio_client", "stdio_server", ] diff --git a/src/mcp/client/auth/extensions/identity_assertion.py b/src/mcp/client/auth/extensions/identity_assertion.py index 2d97e5eff2..3e8e49d9f7 100644 --- a/src/mcp/client/auth/extensions/identity_assertion.py +++ b/src/mcp/client/auth/extensions/identity_assertion.py @@ -39,6 +39,7 @@ union_scopes, validate_metadata_issuer, ) +from mcp.shared._httpx_utils import redirect_error from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthClientInformationFull, OAuthToken from mcp.shared.auth_utils import calculate_token_expiry, resource_url_from_server_url @@ -199,6 +200,8 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx. assertion = await self._assertion_provider(self._issuer, self._resource) token_response = yield self._build_token_request(scope_to_request, assertion) + if token_response.has_redirect_location: + raise redirect_error(token_response, context="OAuth token exchange request") if token_response.status_code != 200: body = (await token_response.aread()).decode(errors="replace") raise OAuthTokenError(f"Token exchange failed ({token_response.status_code}): {body}") diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py index 711848d724..f72b60e9da 100644 --- a/src/mcp/client/auth/oauth2.py +++ b/src/mcp/client/auth/oauth2.py @@ -41,6 +41,7 @@ validate_authorization_response_iss, validate_metadata_issuer, ) +from mcp.shared._httpx_utils import redirect_error from mcp.shared.auth import ( AuthorizationCodeResult, OAuthClientInformationFull, @@ -416,6 +417,8 @@ async def _exchange_token_authorization_code( async def _handle_token_response(self, response: httpx.Response) -> None: """Handle token exchange response.""" + if response.has_redirect_location: + raise redirect_error(response, context="OAuth token request") if response.status_code not in {200, 201}: body = await response.aread() body_text = body.decode("utf-8") @@ -468,6 +471,8 @@ async def _refresh_token(self) -> httpx.Request: async def _handle_refresh_response(self, response: httpx.Response) -> bool: """Handle token refresh response. Returns True if successful.""" + if response.has_redirect_location: + raise redirect_error(response, context="OAuth token refresh request") if response.status_code != 200: logger.warning(f"Token refresh failed: {response.status_code}") self.context.clear_tokens() diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py index d6b05e0667..aff5606dd5 100644 --- a/src/mcp/client/auth/utils.py +++ b/src/mcp/client/auth/utils.py @@ -6,6 +6,7 @@ from pydantic import AnyUrl, ValidationError from mcp.client.auth import OAuthFlowError, OAuthRegistrationError, OAuthTokenError +from mcp.shared._httpx_utils import redirect_error from mcp.shared.auth import ( OAuthClientInformationFull, OAuthClientMetadata, @@ -206,6 +207,10 @@ async def handle_protected_resource_response( Returns: ProtectedResourceMetadata if successfully discovered, None if we should try next URL """ + if response.has_redirect_location: + # A redirect here would silently move discovery to another URL; + # fail loudly instead of treating it as a miss. + raise redirect_error(response, context="OAuth protected resource metadata request") if response.status_code == 200: try: content = await response.aread() @@ -221,6 +226,10 @@ async def handle_protected_resource_response( async def handle_auth_metadata_response(response: Response) -> tuple[bool, OAuthMetadata | None]: + if response.has_redirect_location: + # A redirect here would silently abandon discovery (any non-4xx stops + # the fallback chain); fail loudly instead. + raise redirect_error(response, context="OAuth authorization server metadata request") if response.status_code == 200: try: content = await response.aread() @@ -293,6 +302,8 @@ def create_client_registration_request( async def handle_registration_response(response: Response) -> OAuthClientInformationFull: """Handle registration response.""" + if response.has_redirect_location: + raise redirect_error(response, context="OAuth client registration request") if response.status_code not in (200, 201): await response.aread() raise OAuthRegistrationError(f"Registration failed: {response.status_code} {response.text}") diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 8b482932aa..b83fb177b9 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -1,6 +1,6 @@ import logging from collections.abc import Callable -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress from typing import Any from urllib.parse import parse_qs, urljoin, urlparse @@ -12,7 +12,7 @@ from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import create_context_streams -from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client +from mcp.shared._httpx_utils import McpHttpClientFactory, RedirectError, create_mcp_http_client, redirect_error from mcp.shared.message import SessionMessage logger = logging.getLogger(__name__) @@ -48,6 +48,7 @@ async def sse_client( timeout: HTTP timeout for regular operations (in seconds). sse_read_timeout: Timeout for SSE read operations (in seconds). httpx_client_factory: Factory function for creating the HTTPX client. + The default follows only same-origin redirects; see create_mcp_http_client. auth: Optional HTTPX authentication handler. on_session_created: Optional callback invoked with the session ID when received. """ @@ -56,6 +57,10 @@ async def sse_client( headers=headers, auth=auth, timeout=httpx.Timeout(timeout, read=sse_read_timeout) ) as client: async with aconnect_sse(client, "GET", url) as event_source: + if event_source.response.has_redirect_location: + # Only reachable with a client that does not follow redirects; + # see redirect_error. + raise redirect_error(event_source.response) event_source.response.raise_for_status() logger.debug("SSE connection established") @@ -121,14 +126,35 @@ async def post_writer(endpoint_url: str): async def _send_message(session_message: SessionMessage) -> None: logger.debug(f"Sending client message: {session_message}") - response = await client.post( - endpoint_url, - json=session_message.message.model_dump( - by_alias=True, - mode="json", - exclude_unset=True, - ), - ) + try: + response = await client.post( + endpoint_url, + json=session_message.message.model_dump( + by_alias=True, + mode="json", + exclude_unset=True, + ), + ) + if response.has_redirect_location: + # Only reachable with a client that does not follow + # redirects; see redirect_error. + raise redirect_error(response) + except RedirectError as exc: + # Handled here, before the task group wraps it: answer + # a request in-band so its caller is not parked forever; + # deliver anything else to the session's message handler. + logger.exception("Error sending client message") + message = session_message.message + if isinstance(message, types.JSONRPCRequest): + error_data = types.ErrorData(code=types.INVALID_REQUEST, message=str(exc)) + payload: SessionMessage | Exception = SessionMessage( + types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data) + ) + else: + payload = exc + with suppress(anyio.BrokenResourceError, anyio.ClosedResourceError): + await read_stream_writer.send(payload) + return response.raise_for_status() logger.debug(f"Client message sent successfully: {response.status_code}") diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 5c3501d8c4..21aeffa255 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -33,7 +33,7 @@ from mcp.client._transport import TransportStreams from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams -from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared._httpx_utils import RedirectError, create_mcp_http_client, redirect_error from mcp.shared.inbound import MCP_PROTOCOL_VERSION_HEADER from mcp.shared.jsonrpc_dispatcher import cancelled_request_id_from_params from mcp.shared.message import ClientMessageMetadata, SessionMessage @@ -211,6 +211,10 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer: headers[LAST_EVENT_ID] = last_event_id async with aconnect_sse(client, "GET", self.url, headers=headers) as event_source: + if event_source.response.has_redirect_location: + # Only reachable with a client that does not follow + # redirects; see redirect_error. + raise redirect_error(event_source.response) event_source.response.raise_for_status() logger.debug("GET SSE connection established") @@ -227,6 +231,13 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer: # Stream ended normally (server closed) - reset attempt counter attempt = 0 + except RedirectError as exc: + # A redirected stream endpoint will redirect on every retry; + # deliver once so the session can see it, then stop. + logger.exception("GET stream closed: the server redirected the stream endpoint") + with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError): + await read_stream_writer.send(exc) + return except Exception: logger.debug("GET stream error", exc_info=True) attempt += 1 @@ -313,6 +324,25 @@ async def _run_request_post( if self._in_flight_posts.get(request_id) is post: del self._in_flight_posts[request_id] + async def _resolve_redirected_message(self, ctx: RequestContext, exc: RedirectError) -> None: + """Resolve a message whose HTTP exchange ended in a refused redirect. + + Answers a request in-band so its caller is not parked forever; delivers + anything else to the session's message handler. The error text names + the redirect target and the remedy. + """ + message = ctx.session_message.message + if isinstance(message, JSONRPCRequest): + error_data = ErrorData(code=INVALID_REQUEST, message=str(exc)) + payload: SessionMessageOrError = SessionMessage( + JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data) + ) + else: + logger.exception("Error sending client message") + payload = exc + with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError): + await ctx.read_stream_writer.send(payload) + async def _handle_post_request(self, ctx: RequestContext) -> None: """Handle a POST request with response processing.""" message = ctx.session_message.message @@ -339,6 +369,11 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: ) return + if response.has_redirect_location: + # Only reachable with a client that does not follow redirects; + # see redirect_error. + raise redirect_error(response) + if response.status_code >= 400: if isinstance(message, JSONRPCRequest): # A spec-correct server may return the JSON-RPC error in the @@ -528,6 +563,11 @@ async def _handle_reconnection( # Stream ended again without response - reconnect again (reset attempt counter) logger.info("SSE stream disconnected, reconnecting...") await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0) + except RedirectError: + # A redirected stream endpoint will redirect on every retry; + # propagate so the request resolves with the guidance instead + # of burning the retry budget. + raise except Exception as e: # pragma: no cover logger.debug(f"Reconnection failed: {e}") # Try to reconnect again if we still have an event ID @@ -583,10 +623,14 @@ async def _handle_message(session_message: SessionMessage) -> None: ) async def handle_request_async(): - if is_resumption: - await self._handle_resumption_request(ctx) - else: - await self._handle_post_request(ctx) + try: + if is_resumption: + await self._handle_resumption_request(ctx) + else: + await self._handle_post_request(ctx) + except RedirectError as exc: + # Handled here, before any task group wraps it. + await self._resolve_redirected_message(ctx, exc) # If this is a request, start a new task to handle it if isinstance(message, JSONRPCRequest): @@ -657,7 +701,11 @@ async def streamable_http_client( url: The MCP server endpoint URL. 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. + authentication, or other HTTP settings, create a client with + mcp.create_mcp_http_client and pass it here; it applies the same defaults, + including following only same-origin redirects (a plain httpx.AsyncClient + follows none, and a redirect response then resolves the affected request + with an error naming the target). terminate_on_close: If True, send a DELETE request to terminate the session when the context exits. Yields: diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py index 6a121aff6d..c01728a81d 100644 --- a/src/mcp/shared/_httpx_utils.py +++ b/src/mcp/shared/_httpx_utils.py @@ -4,12 +4,30 @@ import httpx -__all__ = ["create_mcp_http_client", "MCP_DEFAULT_TIMEOUT", "MCP_DEFAULT_SSE_READ_TIMEOUT"] +__all__ = [ + "create_mcp_http_client", + "RedirectError", + "MCP_DEFAULT_TIMEOUT", + "MCP_DEFAULT_SSE_READ_TIMEOUT", +] # Default MCP timeout configuration MCP_DEFAULT_TIMEOUT = 30.0 # General operations (seconds) MCP_DEFAULT_SSE_READ_TIMEOUT = 300.0 # SSE streams - 5 minutes (seconds) +_DEFAULT_PORTS = {"http": 80, "https": 443} + + +class RedirectError(httpx.HTTPStatusError): + """Raised when a server redirects a request somewhere the client will not go automatically. + + Clients created by `create_mcp_http_client` follow redirects that stay on the + same origin (scheme, host, and port), including http-to-https upgrades of the + same host on default ports. Any other redirect raises this error instead of + being followed, because everything configured on the client (headers, auth, + request bodies) is sent to whichever server it talks to. + """ + class McpHttpClientFactory(Protocol): # pragma: no branch def __call__( # pragma: no branch @@ -20,6 +38,105 @@ def __call__( # pragma: no branch ) -> httpx.AsyncClient: ... +def _port_or_default(url: httpx.URL) -> int | None: + if url.port is not None: + return url.port + return _DEFAULT_PORTS.get(url.scheme) + + +def _same_origin(url: httpx.URL, other: httpx.URL) -> bool: + return url.scheme == other.scheme and url.host == other.host and _port_or_default(url) == _port_or_default(other) + + +def _is_https_upgrade(old_url: httpx.URL, new_url: httpx.URL) -> bool: + # The one scheme change treated as staying on the same origin: the same + # host moving from http on its default port to https on its default port. + if old_url.host != new_url.host: + return False + return ( + old_url.scheme == "http" + and new_url.scheme == "https" + and _port_or_default(old_url) == 80 + and _port_or_default(new_url) == 443 + ) + + +def _resolve_redirect_target(request_url: httpx.URL, location: str) -> httpx.URL: + """Resolve a Location header value the way httpx builds its redirect URL.""" + url = httpx.URL(location) + + # An "absolute" Location with a scheme but no host keeps the request's + # host. Only the host is copied - not the port - matching httpx. + if url.scheme and not url.host: + url = url.copy_with(host=request_url.host) + + # Location may be relative (RFC 9110 section 10.2.2). + if url.is_relative_url: + url = request_url.join(url) + + return url + + +def _redirect_allowed(request_url: httpx.URL, target: httpx.URL) -> bool: + return _same_origin(target, request_url) or _is_https_upgrade(request_url, target) + + +def redirect_error( + response: httpx.Response, *, context: str | None = None, target: httpx.URL | None = None +) -> RedirectError: + """Build the error for a redirect response that will not be followed. + + Used both by the client factory's response hook (which only sees redirects + leaving the origin, and passes the target it already resolved) and by + transports that receive a redirect response from a client configured not + to follow redirects at all. + """ + request = response.request + location = response.headers.get("location", "") + prefix = f"{context}: " if context else "" + if target is None: + try: + target = _resolve_redirect_target(request.url, location) + except httpx.InvalidURL: + return RedirectError( + f"{prefix}server responded with {response.status_code} and an unparsable Location header {location!r}.", + request=request, + response=response, + ) + if _redirect_allowed(request.url, target): + # Only reachable with a client that does not follow redirects at all + # (the factory's clients follow these): point at the final URL. + message = ( + f"{prefix}server redirected this request to {str(target)!r} and the configured " + "HTTP client does not follow redirects. Connect to that URL directly instead." + ) + else: + message = ( + f"{prefix}server at {request.url} responded with {response.status_code} redirecting to " + f"{str(target)!r}, which is on a different origin. Redirects are only followed within " + "the same origin (scheme, host, and port) or for an http-to-https upgrade of the same " + "host. If the new location is correct, connect to it directly - note that headers and " + "request bodies configured for this client are sent to whichever server it talks to. " + "To follow other redirects, supply your own pre-configured httpx.AsyncClient." + ) + return RedirectError(message, request=request, response=response) + + +async def _raise_on_disallowed_redirect(response: httpx.Response) -> None: + """Response hook: stop redirects that leave the origin before they are followed.""" + if not response.has_redirect_location: + return + request = response.request + try: + target = _resolve_redirect_target(request.url, response.headers["location"]) + except httpx.InvalidURL: + # Let httpx surface its own error for an unparsable Location. + return + if _redirect_allowed(request.url, target): + return + raise redirect_error(response, target=target) + + def create_mcp_http_client( headers: dict[str, str] | None = None, timeout: httpx.Timeout | None = None, @@ -27,7 +144,10 @@ def create_mcp_http_client( ) -> httpx.AsyncClient: """Create a standardized httpx AsyncClient with MCP defaults. - Always enables follow_redirects and applies an SSE-friendly default timeout. + Follows redirects that stay on the same origin (scheme, host, and port), + including http-to-https upgrades of the same host on default ports; any + other redirect raises `RedirectError` instead of being followed. Applies + an SSE-friendly default timeout. Args: headers: Optional headers to include with all requests. @@ -42,6 +162,11 @@ def create_mcp_http_client( The returned AsyncClient must be used as a context manager to ensure proper cleanup of connections. + Raises: + RedirectError: When a response redirects to a different origin. Connect + to the final URL directly, or supply your own pre-configured + httpx.AsyncClient where following such redirects is intended. + Example: Basic usage with MCP defaults: @@ -76,7 +201,10 @@ def create_mcp_http_client( ``` """ # Set MCP defaults - kwargs: dict[str, Any] = {"follow_redirects": True} + kwargs: dict[str, Any] = { + "follow_redirects": True, + "event_hooks": {"response": [_raise_on_disallowed_redirect]}, + } # Handle timeout if timeout is None: diff --git a/tests/client/auth/extensions/test_identity_assertion.py b/tests/client/auth/extensions/test_identity_assertion.py index 1bc63a1173..28a3d6ac4c 100644 --- a/tests/client/auth/extensions/test_identity_assertion.py +++ b/tests/client/auth/extensions/test_identity_assertion.py @@ -14,6 +14,7 @@ from mcp.client.auth import OAuthFlowError, OAuthTokenError from mcp.client.auth.extensions.identity_assertion import IdentityAssertionOAuthProvider, _origin +from mcp.shared._httpx_utils import RedirectError from mcp.shared.auth import JWT_BEARER_GRANT_TYPE, OAuthClientInformationFull, OAuthToken ISSUER = "https://auth.example.com" @@ -409,3 +410,26 @@ def test_origin_normalizes_default_ports() -> None: assert _origin("http://host") == _origin("http://host:80") assert _origin("https://host") != _origin("https://host:8443") assert _origin("https://host") != _origin("https://other") + + +@pytest.mark.anyio +async def test_token_exchange_redirect_raises() -> None: + """SDK-defined policy: token exchange responses must come from the requested URL.""" + requests: list[httpx.Request] = [] + storage = InMemoryStorage() + auth = make_provider(storage) + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + host, path = request.url.host, request.url.path + if host == "mcp.example.com": + return httpx.Response(401) + if host == "auth.example.com" and path in (ASM_PATH, OIDC_PATH): + return httpx.Response(200, content=asm_body(), headers={"content-type": "application/json"}) + if host == "auth.example.com" and path == "/token": + return httpx.Response(307, headers={"location": "https://other.example.com/token"}) + raise AssertionError(f"unexpected request: {request.method} {request.url}") # pragma: no cover + + async with httpx.AsyncClient(transport=httpx.MockTransport(handle), auth=auth) as http: + with pytest.raises(RedirectError, match="OAuth token exchange request"): + await http.get("https://mcp.example.com/mcp") diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py index 1ec38ccf6f..8420b6f517 100644 --- a/tests/client/test_auth.py +++ b/tests/client/test_auth.py @@ -24,6 +24,8 @@ extract_resource_metadata_from_www_auth, extract_scope_from_www_auth, get_client_metadata_scopes, + handle_auth_metadata_response, + handle_protected_resource_response, handle_registration_response, is_valid_client_metadata_url, should_use_client_metadata_url, @@ -33,6 +35,7 @@ ) from mcp.server.auth.routes import build_metadata from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions +from mcp.shared._httpx_utils import RedirectError from mcp.shared.auth import ( AuthorizationCodeResult, OAuthClientInformationFull, @@ -3169,3 +3172,47 @@ async def echo_callback() -> AuthorizationCodeResult: await auth_flow.asend(httpx.Response(200, request=final_req)) except StopAsyncIteration: pass + + +def _redirect_response(url: str) -> httpx.Response: + request = httpx.Request("POST", url) + return httpx.Response(307, headers={"location": "https://other.example.com/moved"}, request=request) + + +@pytest.mark.anyio +async def test_protected_resource_metadata_redirect_raises(): + """Discovery must not silently move to another URL via a redirect.""" + with pytest.raises(RedirectError, match="OAuth protected resource metadata request"): + await handle_protected_resource_response( + _redirect_response("https://api.example.com/.well-known/oauth-protected-resource") + ) + + +@pytest.mark.anyio +async def test_auth_server_metadata_redirect_raises(): + """A redirect must not silently abandon the metadata fallback chain.""" + with pytest.raises(RedirectError, match="OAuth authorization server metadata request"): + await handle_auth_metadata_response( + _redirect_response("https://auth.example.com/.well-known/oauth-authorization-server") + ) + + +@pytest.mark.anyio +async def test_registration_redirect_raises(): + """SDK-defined policy: registration responses must come from the requested URL.""" + with pytest.raises(RedirectError, match="OAuth client registration request"): + await handle_registration_response(_redirect_response("https://auth.example.com/register")) + + +@pytest.mark.anyio +async def test_token_response_redirect_raises(oauth_provider: OAuthClientProvider): + """SDK-defined policy: token responses must come from the requested URL.""" + with pytest.raises(RedirectError, match="OAuth token request"): + await oauth_provider._handle_token_response(_redirect_response("https://auth.example.com/token")) + + +@pytest.mark.anyio +async def test_refresh_response_redirect_raises(oauth_provider: OAuthClientProvider): + """SDK-defined policy: refresh responses must come from the requested URL.""" + with pytest.raises(RedirectError, match="OAuth token refresh request"): + await oauth_provider._handle_refresh_response(_redirect_response("https://auth.example.com/token")) diff --git a/tests/shared/conftest.py b/tests/shared/conftest.py index 7b53b42654..428799c40e 100644 --- a/tests/shared/conftest.py +++ b/tests/shared/conftest.py @@ -59,3 +59,15 @@ def pair_factory(request: pytest.FixtureRequest) -> PairFactory: __all__ = ["PairFactory", "direct_pair", "jsonrpc_pair"] + + +@pytest.fixture +def no_proxy_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Clear proxy environment variables for tests that swap in a mock transport. + + Proxy variables make httpx build proxy mounts at client construction, which + would take precedence over a transport assigned after construction. + """ + for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY"): + monkeypatch.delenv(name, raising=False) + monkeypatch.delenv(name.lower(), raising=False) diff --git a/tests/shared/test_httpx_utils.py b/tests/shared/test_httpx_utils.py index dcc6fd003c..ffdc3a5ce3 100644 --- a/tests/shared/test_httpx_utils.py +++ b/tests/shared/test_httpx_utils.py @@ -1,8 +1,14 @@ """Tests for httpx utility functions.""" import httpx +import pytest -from mcp.shared._httpx_utils import create_mcp_http_client +from mcp.shared._httpx_utils import RedirectError, _resolve_redirect_target, create_mcp_http_client, redirect_error + + +@pytest.fixture(autouse=True) +def _no_proxy_env(no_proxy_env: None) -> None: + """Every test here swaps a mock transport into a factory-built client.""" def test_default_settings(): @@ -22,3 +28,204 @@ def test_custom_parameters(): assert client.headers["Authorization"] == "Bearer token" assert client.timeout.connect == 60.0 + + +def _redirecting_transport(location: str, requests: list[httpx.Request]) -> httpx.MockTransport: + """First request gets a 307 to `location`; every request is recorded.""" + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + return httpx.Response(307, headers={"location": location}) + return httpx.Response(200, json={"ok": True}) + + return httpx.MockTransport(handle) + + +@pytest.mark.anyio +async def test_follows_same_origin_redirect(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("/canonical", requests) + client = create_mcp_http_client(headers={"X-Custom": "value"}) + client._transport = transport # swap in the mock transport + + response = await client.post("http://example.com/endpoint", content=b"payload") + + assert response.status_code == 200 + assert len(requests) == 2 + assert requests[1].url == httpx.URL("http://example.com/canonical") + assert requests[1].method == "POST" + assert requests[1].headers["X-Custom"] == "value" + assert requests[1].content == b"payload" + await client.aclose() + + +@pytest.mark.anyio +async def test_follows_https_upgrade_redirect(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("https://example.com/endpoint", requests) + client = create_mcp_http_client() + client._transport = transport + + response = await client.get("http://example.com/endpoint") + + assert response.status_code == 200 + assert len(requests) == 2 + assert requests[1].url == httpx.URL("https://example.com/endpoint") + await client.aclose() + + +@pytest.mark.anyio +async def test_rejects_redirect_to_other_host(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("https://other.example.com/collect", requests) + client = create_mcp_http_client(headers={"X-Custom": "value"}) + client._transport = transport + + with pytest.raises(RedirectError) as excinfo: + await client.post("https://example.com/endpoint", content=b"payload") + + # The redirect was not followed: exactly one request went out. + assert len(requests) == 1 + assert "https://other.example.com/collect" in str(excinfo.value) + assert "different origin" in str(excinfo.value) + await client.aclose() + + +@pytest.mark.anyio +async def test_rejects_redirect_to_other_port(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("http://example.com:9000/endpoint", requests) + client = create_mcp_http_client() + client._transport = transport + + with pytest.raises(RedirectError): + await client.get("http://example.com:8000/endpoint") + + assert len(requests) == 1 + await client.aclose() + + +@pytest.mark.anyio +async def test_rejects_https_to_http_redirect(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("http://example.com/endpoint", requests) + client = create_mcp_http_client() + client._transport = transport + + with pytest.raises(RedirectError): + await client.get("https://example.com/endpoint") + + assert len(requests) == 1 + await client.aclose() + + +@pytest.mark.anyio +async def test_rejects_absolute_form_location_without_host(): + # An "absolute" Location with no host keeps the request's host but not its + # port (matching httpx's own resolution), so from a non-default port this + # resolves to a different origin and is refused. + requests: list[httpx.Request] = [] + transport = _redirecting_transport("http:///moved", requests) + client = create_mcp_http_client() + client._transport = transport + + with pytest.raises(RedirectError) as excinfo: + await client.get("http://example.com:8000/endpoint") + + assert len(requests) == 1 + assert "http://example.com/moved" in str(excinfo.value) + await client.aclose() + + +@pytest.mark.anyio +async def test_unparsable_location_defers_to_httpx(): + requests: list[httpx.Request] = [] + transport = _redirecting_transport("http://\x00bad/", requests) + client = create_mcp_http_client() + client._transport = transport + + with pytest.raises(httpx.RemoteProtocolError): + await client.get("https://example.com/endpoint") + + await client.aclose() + + +@pytest.mark.anyio +async def test_user_supplied_client_still_follows_everything(): + """A caller's own follow_redirects=True client is not policed by the SDK.""" + requests: list[httpx.Request] = [] + transport = _redirecting_transport("https://other.example.com/moved", requests) + + async with httpx.AsyncClient(transport=transport, follow_redirects=True) as client: + response = await client.get("https://example.com/endpoint") + + assert response.status_code == 200 + assert len(requests) == 2 + assert requests[1].url == httpx.URL("https://other.example.com/moved") + + +def test_redirect_error_builder_same_origin_names_target(): + request = httpx.Request("POST", "http://example.com/endpoint") + response = httpx.Response(307, headers={"location": "/canonical"}, request=request) + + error = redirect_error(response) + + assert "http://example.com/canonical" in str(error) + assert "Connect to that URL directly" in str(error) + + +def test_redirect_error_builder_includes_context(): + request = httpx.Request("POST", "https://auth.example.com/token") + response = httpx.Response(307, headers={"location": "https://other.example.com/token"}, request=request) + + error = redirect_error(response, context="OAuth token request") + + assert str(error).startswith("OAuth token request: ") + assert "different origin" in str(error) + + +def test_redirect_error_builder_unparsable_location(): + request = httpx.Request("GET", "https://example.com/endpoint") + response = httpx.Response(307, headers={"location": "http://\x00bad/"}, request=request) + + error = redirect_error(response) + + assert "unparsable Location" in str(error) + + +def test_redirect_error_is_httpx_status_error(): + """Existing handlers that catch httpx.HTTPStatusError keep working.""" + request = httpx.Request("GET", "https://example.com/endpoint") + response = httpx.Response(307, headers={"location": "https://other.example.com/"}, request=request) + + assert isinstance(redirect_error(response), httpx.HTTPStatusError) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "location", + [ + "/relative/path?q=1", + "relative-no-slash", + "http:///absolute-form-no-host", + "https://other.example.com/absolute", + "//protocol-relative.example.com/x", + "http://example.com:8000/same-origin", + ], +) +async def test_redirect_target_resolution_matches_httpx(location: str) -> None: + """The policy must judge exactly the URL httpx itself would follow. + + Guards against drift between _resolve_redirect_target and httpx's own + redirect URL construction across httpx upgrades. + """ + + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(307, headers={"location": location}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client: + response = await client.get("http://example.com:8000/base/path") + + assert response.next_request is not None + assert _resolve_redirect_target(response.request.url, location) == response.next_request.url diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index 6427fb21a6..e9908ee76c 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -39,8 +39,9 @@ from mcp.server import Server, ServerRequestContext from mcp.server.sse import SseServerTransport from mcp.server.transport_security import TransportSecuritySettings -from mcp.shared._httpx_utils import McpHttpClientFactory +from mcp.shared._httpx_utils import McpHttpClientFactory, RedirectError, create_mcp_http_client from mcp.shared.exceptions import MCPError +from mcp.shared.message import SessionMessage from tests.interaction.transports import StreamingASGITransport SERVER_NAME = "test_server_for_SSE" @@ -428,6 +429,7 @@ async def mock_aiter_sse() -> AsyncGenerator[ServerSentEvent, None]: mock_event_source = MagicMock() mock_event_source.aiter_sse.return_value = mock_aiter_sse() mock_event_source.response = MagicMock() + mock_event_source.response.has_redirect_location = False mock_event_source.response.raise_for_status = MagicMock() mock_aconnect_sse = MagicMock() @@ -437,7 +439,9 @@ async def mock_aiter_sse() -> AsyncGenerator[ServerSentEvent, 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())) + mock_client.post = AsyncMock( + return_value=MagicMock(status_code=200, has_redirect_location=False, raise_for_status=MagicMock()) + ) with ( patch("mcp.client.sse.create_mcp_http_client", return_value=mock_client), @@ -480,3 +484,120 @@ async def test_sse_session_cleanup_on_disconnect() -> None: headers={"Content-Type": "application/json"}, ) assert response.status_code == 404 + + +class _EndpointThenOpen(httpx.AsyncByteStream): + """SSE body: announce the message endpoint, then hold the stream open.""" + + def __init__(self, endpoint: str) -> None: + self._endpoint = endpoint + + async def __aiter__(self) -> AsyncGenerator[bytes, None]: + yield f"event: endpoint\r\ndata: {self._endpoint}\r\n\r\n".encode() + await anyio.sleep_forever() + + +def _sse_redirecting_setup(post_location: str, seen: list[httpx.Request]) -> httpx.MockTransport: + """GET serves an SSE stream announcing /messages/; POST answers 307 to `post_location`.""" + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.method == "GET": + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=_EndpointThenOpen("/messages/") + ) + return httpx.Response(307, headers={"location": post_location}) + + return httpx.MockTransport(handle) + + +@pytest.mark.anyio +async def test_sse_post_redirect_to_other_origin_is_delivered(no_proxy_env: None) -> None: + """A redirected message POST surfaces on the read stream instead of dying in a log.""" + seen: list[httpx.Request] = [] + transport = _sse_redirecting_setup("https://other.example.com/collect", seen) + + def factory(headers: Any = None, timeout: Any = None, auth: Any = None) -> httpx.AsyncClient: + client = create_mcp_http_client(headers=headers, timeout=timeout, auth=auth) + client._transport = transport # swap in the mock transport + return client + + with anyio.fail_after(5): + async with sse_client("https://example.com/sse", httpx_client_factory=factory) as (read_stream, write_stream): + await write_stream.send(SessionMessage(types.JSONRPCRequest(jsonrpc="2.0", id=1, method="ping"))) + item = await read_stream.receive() + + assert isinstance(item, SessionMessage) + assert isinstance(item.message, types.JSONRPCError) + assert item.message.id == 1 + assert "different origin" in item.message.error.message + # The POST was not re-sent to the other origin. + assert all(request.url.host == "example.com" for request in seen) + + +@pytest.mark.anyio +async def test_sse_post_same_origin_redirect_is_followed(no_proxy_env: None) -> None: + """A same-origin redirect on the message POST is followed transparently.""" + seen: list[httpx.Request] = [] + followed = anyio.Event() + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + if request.method == "GET": + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=_EndpointThenOpen("/messages/") + ) + if str(request.url.path) == "/canonical/": + followed.set() + return httpx.Response(202) + return httpx.Response(307, headers={"location": "/canonical/"}) + + transport = httpx.MockTransport(handle) + + def factory(headers: Any = None, timeout: Any = None, auth: Any = None) -> httpx.AsyncClient: + client = create_mcp_http_client(headers=headers, timeout=timeout, auth=auth) + client._transport = transport # swap in the mock transport + return client + + with anyio.fail_after(5): + async with sse_client("https://example.com/sse", httpx_client_factory=factory) as (_read_stream, write_stream): + notification = types.JSONRPCNotification(jsonrpc="2.0", method="notifications/roots/list_changed") + await write_stream.send(SessionMessage(notification)) + await followed.wait() + + posts = [request for request in seen if request.method == "POST"] + assert [str(request.url.path) for request in posts] == ["/messages/", "/canonical/"] + + +@pytest.mark.anyio +async def test_sse_post_redirect_with_non_following_client_is_delivered() -> None: + """A caller-supplied client that follows no redirects gets the guidance error delivered.""" + seen: list[httpx.Request] = [] + transport = _sse_redirecting_setup("/canonical/", seen) + + def factory(headers: Any = None, timeout: Any = None, auth: Any = None) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=transport) + + with anyio.fail_after(5): + async with sse_client("https://example.com/sse", httpx_client_factory=factory) as (read_stream, write_stream): + notification = types.JSONRPCNotification(jsonrpc="2.0", method="notifications/roots/list_changed") + await write_stream.send(SessionMessage(notification)) + item = await read_stream.receive() + + assert isinstance(item, RedirectError) + assert "Connect to that URL directly" in str(item) + + +@pytest.mark.anyio +async def test_sse_connect_redirect_with_non_following_client_raises() -> None: + """A redirect on the SSE connect GET raises the guidance error to the caller.""" + + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(307, headers={"location": "/sse/"}) + + def factory(headers: Any = None, timeout: Any = None, auth: Any = None) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handle)) + + with anyio.fail_after(5), pytest.raises(RedirectError, match="Connect to that URL directly"): + async with sse_client("https://example.com/sse", httpx_client_factory=factory): + pass # pragma: no cover - connect fails before the body runs diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index cbce222eca..7ce555f56f 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -45,7 +45,7 @@ from mcp import MCPError from mcp.client import ClientRequestContext from mcp.client.session import ClientSession -from mcp.client.streamable_http import StreamableHTTPTransport, streamable_http_client +from mcp.client.streamable_http import RequestContext, StreamableHTTPTransport, streamable_http_client from mcp.server import Server, ServerRequestContext from mcp.server.streamable_http import ( GET_STREAM_KEY, @@ -63,6 +63,7 @@ from mcp.server.transport_security import TransportSecuritySettings from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import create_context_streams +from mcp.shared._httpx_utils import RedirectError, create_mcp_http_client from mcp.shared.message import ClientMessageMetadata, ServerMessageMetadata, SessionMessage from mcp.shared.session import RequestResponder from tests.interaction.transports import StreamingASGITransport @@ -2283,3 +2284,174 @@ async def asgi_receive() -> Message: assert body_chunks[-1] == {"type": "http.response.body", "body": b"", "more_body": False} assert "Error in standalone SSE writer" not in caplog.text assert "Error in standalone SSE response" not in caplog.text + + +@pytest.mark.anyio +async def test_post_redirect_to_other_origin_resolves_request_in_band(no_proxy_env: None) -> None: + """A redirected request POST resolves with a JSON-RPC error naming the target.""" + seen: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(307, headers={"location": "https://other.example.com/mcp"}) + + client = create_mcp_http_client() + client._transport = httpx.MockTransport(handle) # swap in the mock transport + with anyio.fail_after(5): + async with client: + # coverage misreports the ->exit arc for nested async with (see AGENTS.md) + async with streamable_http_client( # pragma: no branch + "https://example.com/mcp", http_client=client, terminate_on_close=False + ) as (read_stream, write_stream): + await write_stream.send(SessionMessage(types.JSONRPCRequest(jsonrpc="2.0", id=1, method="ping"))) + item = await read_stream.receive() + + assert isinstance(item, SessionMessage) + assert isinstance(item.message, types.JSONRPCError) + assert item.message.id == 1 + assert "different origin" in item.message.error.message + # The redirect was not followed. + assert len(seen) == 1 + + +@pytest.mark.anyio +async def test_post_redirect_with_non_following_client_resolves_request_in_band() -> None: + """A caller-supplied client that follows no redirects gets the guidance error, not a parse error.""" + + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(307, headers={"location": "/mcp/"}) + + with anyio.fail_after(5): + async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client: + # coverage misreports the ->exit arc for nested async with (see AGENTS.md) + async with streamable_http_client( # pragma: no branch + "https://example.com/mcp", http_client=client, terminate_on_close=False + ) as (read_stream, write_stream): + await write_stream.send(SessionMessage(types.JSONRPCRequest(jsonrpc="2.0", id=1, method="ping"))) + item = await read_stream.receive() + + assert isinstance(item, SessionMessage) + assert isinstance(item.message, types.JSONRPCError) + assert "https://example.com/mcp/" in item.message.error.message + assert "Connect to that URL directly" in item.message.error.message + + +@pytest.mark.anyio +async def test_notification_redirect_is_delivered_to_session(no_proxy_env: None) -> None: + """A redirected notification POST surfaces on the read stream instead of dying in a log.""" + + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(307, headers={"location": "https://other.example.com/mcp"}) + + client = create_mcp_http_client() + client._transport = httpx.MockTransport(handle) # swap in the mock transport + with anyio.fail_after(5): + async with client: + # coverage misreports the ->exit arc for nested async with (see AGENTS.md) + async with streamable_http_client( # pragma: no branch + "https://example.com/mcp", http_client=client, terminate_on_close=False + ) as (read_stream, write_stream): + notification = types.JSONRPCNotification(jsonrpc="2.0", method="notifications/roots/list_changed") + await write_stream.send(SessionMessage(notification)) + item = await read_stream.receive() + + assert isinstance(item, RedirectError) + assert "different origin" in str(item) + + +@pytest.mark.anyio +async def test_get_stream_stops_on_redirect(no_proxy_env: None) -> None: + """A redirected standalone GET stream gives up instead of burning the retry budget.""" + seen: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(307, headers={"location": "https://other.example.com/mcp"}) + + transport = StreamableHTTPTransport("https://example.com/mcp") + transport.session_id = "session-1" + client = create_mcp_http_client() + client._transport = httpx.MockTransport(handle) # swap in the mock transport + writer, reader = create_context_streams[SessionMessage | Exception](1) + async with client, writer, reader: + with anyio.fail_after(5): + await transport.handle_get_stream(client, writer) + delivered = await reader.receive() + + assert len(seen) == 1 + assert isinstance(delivered, RedirectError) + + +@pytest.mark.anyio +async def test_reconnection_propagates_redirect_error(no_proxy_env: None) -> None: + """A redirected reconnection GET propagates instead of burning the retry budget.""" + seen: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(307, headers={"location": "https://other.example.com/mcp"}) + + transport = StreamableHTTPTransport("https://example.com/mcp") + client = create_mcp_http_client() + client._transport = httpx.MockTransport(handle) # swap in the mock transport + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0) + ctx = RequestContext( + client=client, + session_id="session-1", + session_message=SessionMessage(types.JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")), + metadata=None, + read_stream_writer=read_stream_writer, + ) + async with client, read_stream_writer, read_stream: + with anyio.fail_after(5), pytest.raises(RedirectError): + await transport._handle_reconnection(ctx, "event-1", None, 0) + + assert len(seen) == 1 + + +@pytest.mark.anyio +async def test_get_stream_stops_on_redirect_with_non_following_client() -> None: + """A caller-supplied non-following client gets the same guidance on the GET stream.""" + seen: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(307, headers={"location": "/mcp/"}) + + transport = StreamableHTTPTransport("https://example.com/mcp") + transport.session_id = "session-1" + writer, reader = create_context_streams[SessionMessage | Exception](1) + async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client, writer, reader: + with anyio.fail_after(5): + await transport.handle_get_stream(client, writer) + delivered = await reader.receive() + + assert len(seen) == 1 + assert isinstance(delivered, RedirectError) + assert "Connect to that URL directly" in str(delivered) + + +@pytest.mark.anyio +async def test_resumption_redirect_resolves_request_in_band(no_proxy_env: None) -> None: + """A redirected resumption GET resolves the request in-band instead of killing the session.""" + + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(307, headers={"location": "https://other.example.com/mcp"}) + + client = create_mcp_http_client() + client._transport = httpx.MockTransport(handle) # swap in the mock transport + with anyio.fail_after(5): + async with client: + # coverage misreports the ->exit arc for nested async with (see AGENTS.md) + async with streamable_http_client( # pragma: no branch + "https://example.com/mcp", http_client=client, terminate_on_close=False + ) as (read_stream, write_stream): + request = types.JSONRPCRequest(jsonrpc="2.0", id=7, method="tools/list") + metadata = ClientMessageMetadata(resumption_token="event-1") + await write_stream.send(SessionMessage(request, metadata=metadata)) + item = await read_stream.receive() + + assert isinstance(item, SessionMessage) + assert isinstance(item.message, types.JSONRPCError) + assert item.message.id == 7 + assert "different origin" in item.message.error.message