Merge branch 'main' into fix-leaked-stream

Author: aabmass

.github/actions/conformance/client.py

@@ -275,6 +275,27 @@ async def run_client_credentials_basic(server_url: str) -> None:
 async def run_auth_code_client(server_url: str) -> None:
     """Authorization code flow (default for auth/* scenarios)."""
     callback_handler = ConformanceOAuthCallbackHandler()
+    storage = InMemoryTokenStorage()
+
+    # Check for pre-registered client credentials from context
+    context_json = os.environ.get("MCP_CONFORMANCE_CONTEXT")
+    if context_json:
+        try:
+            context = json.loads(context_json)
+            client_id = context.get("client_id")
+            client_secret = context.get("client_secret")
+            if client_id:
+                await storage.set_client_info(
+                    OAuthClientInformationFull(
+                        client_id=client_id,
+                        client_secret=client_secret,
+                        redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+                        token_endpoint_auth_method="client_secret_basic" if client_secret else "none",
+                    )
+                )
+                logger.debug(f"Pre-loaded client credentials: client_id={client_id}")
+        except json.JSONDecodeError:
+            logger.exception("Failed to parse MCP_CONFORMANCE_CONTEXT")
 
     oauth_auth = OAuthClientProvider(
         server_url=server_url,
@@ -284,7 +305,7 @@ async def run_auth_code_client(server_url: str) -> None:
             grant_types=["authorization_code", "refresh_token"],
             response_types=["code"],
         ),
-        storage=InMemoryTokenStorage(),
+        storage=storage,
         redirect_handler=callback_handler.handle_redirect,
         callback_handler=callback_handler.handle_callback,
         client_metadata_url="https://conformance-test.local/client-metadata.json",

.github/workflows/conformance.yml

@@ -42,4 +42,4 @@ jobs:
         with:
           node-version: 24
       - run: uv sync --frozen --all-extras --package mcp
-      - run: npx @modelcontextprotocol/conformance@0.1.10 client --command 'uv run --frozen python .github/actions/conformance/client.py' --suite all
+      - run: npx @modelcontextprotocol/conformance@0.1.13 client --command 'uv run --frozen python .github/actions/conformance/client.py' --suite all

CLAUDE.md

@@ -29,6 +29,9 @@ This document contains critical information about working with this codebase. Fo
    - IMPORTANT: The `tests/client/test_client.py` is the most well designed test file. Follow its patterns.
    - IMPORTANT: Be minimal, and focus on E2E tests: Use the `mcp.client.Client` whenever possible.
 
+Test files mirror the source tree: `src/mcp/client/streamable_http.py` → `tests/client/test_streamable_http.py`
+Add tests to the existing file for that module.
+
 - For commits fixing bugs or adding features based on user reports add:
 
   ```bash

src/mcp/client/auth/oauth2.py

@@ -229,6 +229,7 @@ def __init__(
         callback_handler: Callable[[], Awaitable[tuple[str, str | None]]] | None = None,
         timeout: float = 300.0,
         client_metadata_url: str | None = None,
+        validate_resource_url: Callable[[str, str | None], Awaitable[None]] | None = None,
     ):
         """Initialize OAuth2 authentication.
 
@@ -243,6 +244,10 @@ def __init__(
                 advertises client_id_metadata_document_supported=true, this URL will be
                 used as the client_id instead of performing dynamic client registration.
                 Must be a valid HTTPS URL with a non-root pathname.
+            validate_resource_url: Optional callback to override resource URL validation.
+                Called with (server_url, prm_resource) where prm_resource is the resource
+                from Protected Resource Metadata (or None if not present). If not provided,
+                default validation rejects mismatched resources per RFC 8707.
 
         Raises:
             ValueError: If client_metadata_url is provided but not a valid HTTPS URL
@@ -263,6 +268,7 @@ def __init__(
             timeout=timeout,
             client_metadata_url=client_metadata_url,
         )
+        self._validate_resource_url_callback = validate_resource_url
         self._initialized = False
 
     async def _handle_protected_resource_response(self, response: httpx.Response) -> bool:
@@ -476,6 +482,26 @@ async def _handle_oauth_metadata_response(self, response: httpx.Response) -> Non
         metadata = OAuthMetadata.model_validate_json(content)
         self.context.oauth_metadata = metadata
 
+    async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None:
+        """Validate that PRM resource matches the server URL per RFC 8707."""
+        prm_resource = str(prm.resource) if prm.resource else None
+
+        if self._validate_resource_url_callback is not None:
+            await self._validate_resource_url_callback(self.context.server_url, prm_resource)
+            return
+
+        if not prm_resource:
+            return  # pragma: no cover
+        default_resource = resource_url_from_server_url(self.context.server_url)
+        # Normalize: Pydantic AnyHttpUrl adds trailing slash to root URLs
+        # (e.g. "https://example.com/") while resource_url_from_server_url may not.
+        if not default_resource.endswith("/"):
+            default_resource += "/"
+        if not prm_resource.endswith("/"):
+            prm_resource += "/"
+        if not check_resource_allowed(requested_resource=default_resource, configured_resource=prm_resource):
+            raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}")
+
     async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.Request, httpx.Response]:
         """HTTPX auth flow integration."""
         async with self.context.lock:
@@ -517,6 +543,8 @@ async def async_auth_flow(self, request: httpx.Request) -> AsyncGenerator[httpx.
 
                         prm = await handle_protected_resource_response(discovery_response)
                         if prm:
+                            # Validate PRM resource matches server URL (RFC 8707)
+                            await self._validate_resource_match(prm)
                             self.context.protected_resource_metadata = prm
 
                             # todo: try all authorization_servers to find the OASM

src/mcp/client/streamable_http.py

@@ -13,11 +13,14 @@
 from anyio.abc import TaskGroup
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from httpx_sse import EventSource, ServerSentEvent, aconnect_sse
+from pydantic import ValidationError
 
 from mcp.client._transport import TransportStreams
 from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
+    INVALID_REQUEST,
+    PARSE_ERROR,
     ErrorData,
     InitializeResult,
     JSONRPCError,
@@ -163,6 +166,11 @@ async def _handle_sse_event(
 
             except Exception as exc:  # pragma: no cover
                 logger.exception("Error parsing SSE message")
+                if original_request_id is not None:
+                    error_data = ErrorData(code=PARSE_ERROR, message=f"Failed to parse SSE message: {exc}")
+                    error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
+                    await read_stream_writer.send(error_msg)
+                    return True
                 await read_stream_writer.send(exc)
                 return False
         else:  # pragma: no cover
@@ -260,7 +268,9 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
 
             if response.status_code == 404:  # pragma: no branch
                 if isinstance(message, JSONRPCRequest):  # pragma: no branch
-                    await self._send_session_terminated_error(ctx.read_stream_writer, message.id)
+                    error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
+                    session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
+                    await ctx.read_stream_writer.send(session_message)
                 return
 
             response.raise_for_status()
@@ -272,20 +282,24 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
             if isinstance(message, JSONRPCRequest):
                 content_type = response.headers.get("content-type", "").lower()
                 if content_type.startswith("application/json"):
-                    await self._handle_json_response(response, ctx.read_stream_writer, is_initialization)
+                    await self._handle_json_response(
+                        response, ctx.read_stream_writer, is_initialization, request_id=message.id
+                    )
                 elif content_type.startswith("text/event-stream"):
                     await self._handle_sse_response(response, ctx, is_initialization)
                 else:
-                    await self._handle_unexpected_content_type(  # pragma: no cover
-                        content_type,  # pragma: no cover
-                        ctx.read_stream_writer,  # pragma: no cover
-                    )  # pragma: no cover
+                    logger.error(f"Unexpected content type: {content_type}")
+                    error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}")
+                    error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
+                    await ctx.read_stream_writer.send(error_msg)
 
     async def _handle_json_response(
         self,
         response: httpx.Response,
         read_stream_writer: StreamWriter,
         is_initialization: bool = False,
+        *,
+        request_id: RequestId,
     ) -> None:
         """Handle JSON response from the server."""
         try:
@@ -298,9 +312,11 @@ async def _handle_json_response(
 
             session_message = SessionMessage(message)
             await read_stream_writer.send(session_message)
-        except Exception as exc:  # pragma: no cover
+        except (httpx.StreamError, ValidationError) as exc:
             logger.exception("Error parsing JSON response")
-            await read_stream_writer.send(exc)
+            error_data = ErrorData(code=PARSE_ERROR, message=f"Failed to parse JSON response: {exc}")
+            error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
+            await read_stream_writer.send(error_msg)
 
     async def _handle_sse_response(
         self,
@@ -312,6 +328,11 @@ async def _handle_sse_response(
         last_event_id: str | None = None
         retry_interval_ms: int | None = None
 
+        # The caller (_handle_post_request) only reaches here inside
+        # isinstance(message, JSONRPCRequest), so this is always a JSONRPCRequest.
+        assert isinstance(ctx.session_message.message, JSONRPCRequest)
+        original_request_id = ctx.session_message.message.id
+
         try:
             event_source = EventSource(response)
             async for sse in event_source.aiter_sse():  # pragma: no branch
@@ -326,6 +347,7 @@ async def _handle_sse_response(
                 is_complete = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,
+                    original_request_id=original_request_id,
                     resumption_callback=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
                     is_initialization=is_initialization,
                 )
@@ -334,8 +356,8 @@ async def _handle_sse_response(
                 if is_complete:
                     await response.aclose()
                     return  # Normal completion, no reconnect needed
-        except Exception as e:
-            logger.debug(f"SSE stream ended: {e}")  # pragma: no cover
+        except Exception:
+            logger.debug("SSE stream ended", exc_info=True)  # pragma: no cover
 
         # Stream ended without response - reconnect if we received an event with ID
         if last_event_id is not None:  # pragma: no branch
@@ -400,24 +422,6 @@ async def _handle_reconnection(
             # Try to reconnect again if we still have an event ID
             await self._handle_reconnection(ctx, last_event_id, retry_interval_ms, attempt + 1)
 
-    async def _handle_unexpected_content_type(
-        self, content_type: str, read_stream_writer: StreamWriter
-    ) -> None:  # pragma: no cover
-        """Handle unexpected content type in response."""
-        error_msg = f"Unexpected content type: {content_type}"  # pragma: no cover
-        logger.error(error_msg)  # pragma: no cover
-        await read_stream_writer.send(ValueError(error_msg))  # pragma: no cover
-
-    async def _send_session_terminated_error(self, read_stream_writer: StreamWriter, request_id: RequestId) -> None:
-        """Send a session terminated error response."""
-        jsonrpc_error = JSONRPCError(
-            jsonrpc="2.0",
-            id=request_id,
-            error=ErrorData(code=32600, message="Session terminated"),
-        )
-        session_message = SessionMessage(jsonrpc_error)
-        await read_stream_writer.send(session_message)
-
     async def post_writer(
         self,
         client: httpx.AsyncClient,
@@ -467,8 +471,8 @@ async def handle_request_async():
                     else:
                         await handle_request_async()
 
-        except Exception:
-            logger.exception("Error in post_writer")  # pragma: no cover
+        except Exception:  # pragma: lax no cover
+            logger.exception("Error in post_writer")
         finally:
             await read_stream_writer.aclose()
             await write_stream.aclose()

src/mcp/types/jsonrpc.py

@@ -75,7 +75,7 @@ class JSONRPCError(BaseModel):
     """A response to a request that indicates an error occurred."""
 
     jsonrpc: Literal["2.0"]
-    id: str | int
+    id: RequestId
     error: ErrorData