refactor: Simplify StreamableHTTPTransport to respect client configuration

Implements "principle of least surprise" by making the httpx client the
single source of truth for HTTP configuration (headers, timeout, auth).

Changes:
- StreamableHTTPTransport constructor now only takes url parameter
- Transport reads configuration from client when making requests
- Removed redundant config extraction and storage
- Removed headers and sse_read_timeout from RequestContext
- Removed MCP_DEFAULT_TIMEOUT and MCP_DEFAULT_SSE_READ_TIMEOUT from
  _httpx_utils public API (__all__)

This addresses PR feedback about awkward config extraction when client
is provided. The transport now only adds protocol requirements (MCP
headers, session headers) on top of the client's configuration rather
than extracting and overriding it.

All tests pass, no type errors.

Author: felixweinberger

src/mcp/client/streamable_http.py

@@ -20,12 +20,7 @@
 from httpx_sse import EventSource, ServerSentEvent, aconnect_sse
 from typing_extensions import deprecated
 
-from mcp.shared._httpx_utils import (
-    MCP_DEFAULT_SSE_READ_TIMEOUT,
-    MCP_DEFAULT_TIMEOUT,
-    McpHttpClientFactory,
-    create_mcp_http_client,
-)
+from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
     ErrorData,
@@ -70,52 +65,41 @@ class RequestContext:
     """Context for a request operation."""
 
     client: httpx.AsyncClient
-    headers: dict[str, str]
     session_id: str | None
     session_message: SessionMessage
     metadata: ClientMessageMetadata | None
     read_stream_writer: StreamWriter
-    sse_read_timeout: float
 
 
 class StreamableHTTPTransport:
     """StreamableHTTP client transport implementation."""
 
-    def __init__(
-        self,
-        url: str,
-        headers: dict[str, str] | None = None,
-        timeout: float | timedelta = 30,
-        sse_read_timeout: float | timedelta = 60 * 5,
-        auth: httpx.Auth | None = None,
-    ) -> None:
+    def __init__(self, url: str) -> None:
         """Initialize the StreamableHTTP transport.
 
         Args:
             url: The endpoint URL.
-            headers: Optional headers to include in requests.
-            timeout: HTTP timeout for regular operations.
-            sse_read_timeout: Timeout for SSE read operations.
-            auth: Optional HTTPX authentication handler.
         """
         self.url = url
-        self.headers = headers or {}
-        self.timeout = timeout.total_seconds() if isinstance(timeout, timedelta) else timeout
-        self.sse_read_timeout = (
-            sse_read_timeout.total_seconds() if isinstance(sse_read_timeout, timedelta) else sse_read_timeout
-        )
-        self.auth = auth
         self.session_id = None
         self.protocol_version = None
-        self.request_headers = {
-            **self.headers,
-            ACCEPT: f"{JSON}, {SSE}",
-            CONTENT_TYPE: JSON,
-        }
-
-    def _prepare_request_headers(self, base_headers: dict[str, str]) -> dict[str, str]:
-        """Update headers with session ID and protocol version if available."""
-        headers = base_headers.copy()
+
+    def _prepare_headers(self, client: httpx.AsyncClient) -> dict[str, str]:
+        """Build request headers from client config and protocol requirements.
+
+        Merges the client's default headers with MCP protocol headers and session headers.
+
+        Args:
+            client: The httpx client whose headers to use as base.
+
+        Returns:
+            Complete headers dict with client headers, protocol headers, and session headers.
+        """
+        headers = dict(client.headers) if client.headers else {}
+        # Add MCP protocol headers
+        headers[ACCEPT] = f"{JSON}, {SSE}"
+        headers[CONTENT_TYPE] = JSON
+        # Add session headers if available
         if self.session_id:
             headers[MCP_SESSION_ID] = self.session_id
         if self.protocol_version:
@@ -206,14 +190,13 @@ async def handle_get_stream(
             if not self.session_id:
                 return
 
-            headers = self._prepare_request_headers(self.request_headers)
+            headers = self._prepare_headers(client)
 
             async with aconnect_sse(
                 client,
                 "GET",
                 self.url,
                 headers=headers,
-                timeout=httpx.Timeout(self.timeout, read=self.sse_read_timeout),
             ) as event_source:
                 event_source.response.raise_for_status()
                 logger.debug("GET SSE connection established")
@@ -226,7 +209,7 @@ async def handle_get_stream(
 
     async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         """Handle a resumption request using GET with SSE."""
-        headers = self._prepare_request_headers(ctx.headers)
+        headers = self._prepare_headers(ctx.client)
         if ctx.metadata and ctx.metadata.resumption_token:
             headers[LAST_EVENT_ID] = ctx.metadata.resumption_token
         else:
@@ -242,7 +225,6 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
             "GET",
             self.url,
             headers=headers,
-            timeout=httpx.Timeout(self.timeout, read=self.sse_read_timeout),
         ) as event_source:
             event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
@@ -260,7 +242,7 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
 
     async def _handle_post_request(self, ctx: RequestContext) -> None:
         """Handle a POST request with response processing."""
-        headers = self._prepare_request_headers(ctx.headers)
+        headers = self._prepare_headers(ctx.client)
         message = ctx.session_message.message
         is_initialization = self._is_initialization_request(message)
 
@@ -401,12 +383,10 @@ async def post_writer(
 
                     ctx = RequestContext(
                         client=client,
-                        headers=self.request_headers,
                         session_id=self.session_id,
                         session_message=session_message,
                         metadata=metadata,
                         read_stream_writer=read_stream_writer,
-                        sse_read_timeout=self.sse_read_timeout,
                     )
 
                     async def handle_request_async():
@@ -433,7 +413,7 @@ async def terminate_session(self, client: httpx.AsyncClient) -> None:
             return
 
         try:
-            headers = self._prepare_request_headers(self.request_headers)
+            headers = self._prepare_headers(client)
             response = await client.delete(self.url, headers=headers)
 
             if response.status_code == 405:
@@ -493,16 +473,8 @@ async def streamable_http_client(
         # Create default client with recommended MCP timeouts
         client = create_mcp_http_client()
 
-    # Extract configuration from the client to pass to transport
-    headers_dict = dict(client.headers) if client.headers else None
-    timeout = client.timeout.connect if (client.timeout and client.timeout.connect is not None) else MCP_DEFAULT_TIMEOUT
-    sse_read_timeout = (
-        client.timeout.read if (client.timeout and client.timeout.read is not None) else MCP_DEFAULT_SSE_READ_TIMEOUT
-    )
-    auth = client.auth
-
-    # Create transport with extracted configuration
-    transport = StreamableHTTPTransport(url, headers_dict, timeout, sse_read_timeout, auth)
+    # Create transport
+    transport = StreamableHTTPTransport(url)
 
     async with anyio.create_task_group() as tg:
         try:

src/mcp/shared/_httpx_utils.py

@@ -4,7 +4,7 @@
 
 import httpx
 
-__all__ = ["create_mcp_http_client", "MCP_DEFAULT_TIMEOUT", "MCP_DEFAULT_SSE_READ_TIMEOUT"]
+__all__ = ["create_mcp_http_client"]
 
 # Default MCP timeout configuration
 MCP_DEFAULT_TIMEOUT = 30.0  # General operations (seconds)