Replace httpx_client_factory with httpx_client parameter

Modernize streamable_http_client API by accepting httpx.AsyncClient instances
directly instead of factory functions, following industry standards.

- New API: httpx_client: httpx.AsyncClient | None parameter
- Default client created with recommended timeouts if None
- Deprecated wrapper provides backward compatibility
- Updated examples to show custom client usage
- Add MCP_DEFAULT_TIMEOUT constants to _httpx_utils

Author: felixweinberger

README.md

@@ -2375,6 +2375,7 @@ from pydantic import AnyUrl
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
 from mcp.client.streamable_http import streamable_http_client
+from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -2428,15 +2429,16 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with streamable_http_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
+    async with create_mcp_http_client(auth=oauth_auth) as custom_client:
+        async with streamable_http_client("http://localhost:8001/mcp", httpx_client=custom_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
 
-            tools = await session.list_tools()
-            print(f"Available tools: {[tool.name for tool in tools.tools]}")
+                tools = await session.list_tools()
+                print(f"Available tools: {[tool.name for tool in tools.tools]}")
 
-            resources = await session.list_resources()
-            print(f"Available resources: {[r.uri for r in resources.resources]}")
+                resources = await session.list_resources()
+                print(f"Available resources: {[r.uri for r in resources.resources]}")
 
 
 def run():

examples/clients/simple-auth-client/mcp_simple_auth_client/main.py

@@ -11,15 +11,17 @@
 import threading
 import time
 import webbrowser
-from datetime import timedelta
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any
 from urllib.parse import parse_qs, urlparse
 
+import httpx
+
 from mcp.client.auth import OAuthClientProvider, TokenStorage
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
+from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -194,9 +196,7 @@ async def _default_redirect_handler(authorization_url: str) -> None:
             # Use client_metadata_url to enable CIMD when the server supports it
             oauth_auth = OAuthClientProvider(
                 server_url=self.server_url.replace("/mcp", ""),
-                client_metadata=OAuthClientMetadata.model_validate(
-                    client_metadata_dict
-                ),
+                client_metadata=OAuthClientMetadata.model_validate(client_metadata_dict),
                 storage=InMemoryTokenStorage(),
                 redirect_handler=_default_redirect_handler,
                 callback_handler=callback_handler,
@@ -214,12 +214,12 @@ 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 streamable_http_client(
-                    url=self.server_url,
-                    auth=oauth_auth,
-                    timeout=timedelta(seconds=60),
-                ) as (read_stream, write_stream, get_session_id):
-                    await self._run_session(read_stream, write_stream, get_session_id)
+                async with create_mcp_http_client(auth=oauth_auth) as custom_client:
+                    async with streamable_http_client(
+                        url=self.server_url,
+                        httpx_client=custom_client,
+                    ) as (read_stream, write_stream, get_session_id):
+                        await self._run_session(read_stream, write_stream, get_session_id)
 
         except Exception as e:
             print(f"❌ Failed to connect: {e}")
@@ -329,9 +329,7 @@ async def interactive_loop(self):
                     await self.call_tool(tool_name, arguments)
 
                 else:
-                    print(
-                        "❌ Unknown command. Try 'list', 'call <tool_name>', or 'quit'"
-                    )
+                    print("❌ Unknown command. Try 'list', 'call <tool_name>', or 'quit'")
 
             except KeyboardInterrupt:
                 print("\n\n👋 Goodbye!")

examples/snippets/clients/oauth_client.py

@@ -15,6 +15,7 @@
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
 from mcp.client.streamable_http import streamable_http_client
+from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -68,15 +69,16 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with streamable_http_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
+    async with create_mcp_http_client(auth=oauth_auth) as custom_client:
+        async with streamable_http_client("http://localhost:8001/mcp", httpx_client=custom_client) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
 
-            tools = await session.list_tools()
-            print(f"Available tools: {[tool.name for tool in tools.tools]}")
+                tools = await session.list_tools()
+                print(f"Available tools: {[tool.name for tool in tools.tools]}")
 
-            resources = await session.list_resources()
-            print(f"Available resources: {[r.uri for r in resources.resources]}")
+                resources = await session.list_resources()
+                print(f"Available resources: {[r.uri for r in resources.resources]}")
 
 
 def run():

src/mcp/client/session_group.py

@@ -17,6 +17,7 @@
 from typing import Any, TypeAlias, overload
 
 import anyio
+import httpx
 from pydantic import BaseModel
 from typing_extensions import Self, deprecated
 
@@ -26,6 +27,7 @@
 from mcp.client.sse import sse_client
 from mcp.client.stdio import StdioServerParameters
 from mcp.client.streamable_http import streamable_http_client
+from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.exceptions import McpError
 from mcp.shared.session import ProgressFnT
 
@@ -309,11 +311,18 @@ async def _establish_session(
                 )
                 read, write = await session_stack.enter_async_context(client)
             else:
+                httpx_client = create_mcp_http_client(
+                    headers=server_params.headers,
+                    timeout=httpx.Timeout(
+                        server_params.timeout.total_seconds(),
+                        read=server_params.sse_read_timeout.total_seconds(),
+                    ),
+                )
+                await session_stack.enter_async_context(httpx_client)
+
                 client = streamable_http_client(
                     url=server_params.url,
-                    headers=server_params.headers,
-                    timeout=server_params.timeout,
-                    sse_read_timeout=server_params.sse_read_timeout,
+                    httpx_client=httpx_client,
                     terminate_on_close=server_params.terminate_on_close,
                 )
                 read, write, _ = await session_stack.enter_async_context(client)

src/mcp/client/streamable_http.py

@@ -6,6 +6,7 @@
 and session management.
 """
 
+import contextlib
 import logging
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
@@ -19,7 +20,12 @@
 from httpx_sse import EventSource, ServerSentEvent, aconnect_sse
 from typing_extensions import deprecated
 
-from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
+from mcp.shared._httpx_utils import (
+    MCP_DEFAULT_SSE_READ_TIMEOUT,
+    MCP_DEFAULT_TIMEOUT,
+    McpHttpClientFactory,
+    create_mcp_http_client,
+)
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
     ErrorData,
@@ -106,9 +112,9 @@ def __init__(
         self.session_id = None
         self.protocol_version = None
         self.request_headers = {
+            **self.headers,
             ACCEPT: f"{JSON}, {SSE}",
             CONTENT_TYPE: JSON,
-            **self.headers,
         }
 
     def _prepare_request_headers(self, base_headers: dict[str, str]) -> dict[str, str]:
@@ -563,12 +569,9 @@ def get_session_id(self) -> str | None:
 @asynccontextmanager
 async def streamable_http_client(
     url: str,
-    headers: dict[str, str] | None = None,
-    timeout: float | timedelta = 30,
-    sse_read_timeout: float | timedelta = 60 * 5,
+    *,
+    httpx_client: httpx.AsyncClient | None = None,
     terminate_on_close: bool = True,
-    httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
-    auth: httpx.Auth | None = None,
 ) -> AsyncGenerator[
     tuple[
         MemoryObjectReceiveStream[SessionMessage | Exception],
@@ -580,30 +583,57 @@ async def streamable_http_client(
     """
     Client transport for StreamableHTTP.
 
-    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
-    event before disconnecting. All other HTTP operations are controlled by `timeout`.
+    Args:
+        url: The MCP server endpoint URL.
+        httpx_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.
+        terminate_on_close: If True, send a DELETE request to terminate the session
+            when the context exits.
 
     Yields:
         Tuple containing:
             - read_stream: Stream for reading messages from the server
             - write_stream: Stream for sending messages to the server
             - get_session_id_callback: Function to retrieve the current session ID
-    """
-    transport = StreamableHTTPTransport(url, headers, timeout, sse_read_timeout, auth)
 
+    Example:
+        See examples/snippets/clients/ for usage patterns.
+    """
     read_stream_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0)
     write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)
 
+    # Determine if we need to create and manage the client
+    client_provided = httpx_client is not None
+    client = httpx_client
+
+    if client is None:
+        # 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)
+
+    # Sync client headers with transport's merged headers (includes MCP protocol requirements)
+    client.headers.update(transport.request_headers)
+
     async with anyio.create_task_group() as tg:
         try:
             logger.debug(f"Connecting to StreamableHTTP endpoint: {url}")
 
-            async with httpx_client_factory(
-                headers=transport.request_headers,
-                timeout=httpx.Timeout(transport.timeout, read=transport.sse_read_timeout),
-                auth=transport.auth,
-            ) as client:
-                # Define callbacks that need access to tg
+            async with contextlib.AsyncExitStack() as stack:
+                # Only manage client lifecycle if we created it
+                if not client_provided:
+                    await stack.enter_async_context(client)
+
                 def start_get_stream() -> None:
                     tg.start_soon(transport.handle_get_stream, client, read_stream_writer)
 
@@ -650,7 +680,24 @@ async def streamablehttp_client(
     ],
     None,
 ]:
-    async with streamable_http_client(
-        url, headers, timeout, sse_read_timeout, terminate_on_close, httpx_client_factory, auth
-    ) as streams:
-        yield streams
+    # Convert timeout parameters
+    timeout_seconds = timeout.total_seconds() if isinstance(timeout, timedelta) else timeout
+    sse_read_timeout_seconds = (
+        sse_read_timeout.total_seconds() if isinstance(sse_read_timeout, timedelta) else sse_read_timeout
+    )
+
+    # Create httpx client using the factory with old-style parameters
+    client = httpx_client_factory(
+        headers=headers,
+        timeout=httpx.Timeout(timeout_seconds, read=sse_read_timeout_seconds),
+        auth=auth,
+    )
+
+    # Manage client lifecycle since we created it
+    async with client:
+        async with streamable_http_client(
+            url,
+            httpx_client=client,
+            terminate_on_close=terminate_on_close,
+        ) as streams:
+            yield streams

src/mcp/shared/_httpx_utils.py

@@ -4,7 +4,11 @@
 
 import httpx
 
-__all__ = ["create_mcp_http_client"]
+__all__ = ["create_mcp_http_client", "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)
 
 
 class McpHttpClientFactory(Protocol):  # pragma: no branch
@@ -68,7 +72,7 @@ def create_mcp_http_client(
 
     # Handle timeout
     if timeout is None:
-        kwargs["timeout"] = httpx.Timeout(30.0)
+        kwargs["timeout"] = httpx.Timeout(MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT)
     else:
         kwargs["timeout"] = timeout