Add streamable_http_client which accepts httpx.AsyncClient instead of httpx_client_factory (#1177)

Co-authored-by: Felix Weinberger <[email protected]>

Author: Kludex

README.md

@@ -2241,12 +2241,12 @@ Run from the repository root:
 import asyncio
 
 from mcp import ClientSession
-from mcp.client.streamable_http import streamablehttp_client
+from mcp.client.streamable_http import streamable_http_client
 
 
 async def main():
     # Connect to a streamable HTTP server
-    async with streamablehttp_client("http://localhost:8000/mcp") as (
+    async with streamable_http_client("http://localhost:8000/mcp") as (
         read_stream,
         write_stream,
         _,
@@ -2370,11 +2370,12 @@ cd to the `examples/snippets` directory and run:
 import asyncio
 from urllib.parse import parse_qs, urlparse
 
+import httpx
 from pydantic import AnyUrl
 
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
-from mcp.client.streamable_http import streamablehttp_client
+from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -2428,15 +2429,16 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with streamablehttp_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
+    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) 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()
 
-            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,15 @@
 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 streamablehttp_client
+from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -193,7 +193,7 @@ async def _default_redirect_handler(authorization_url: str) -> None:
             # Create OAuth authentication handler using the new interface
             # Use client_metadata_url to enable CIMD when the server supports it
             oauth_auth = OAuthClientProvider(
-                server_url=self.server_url,
+                server_url=self.server_url.replace("/mcp", ""),
                 client_metadata=OAuthClientMetadata.model_validate(client_metadata_dict),
                 storage=InMemoryTokenStorage(),
                 redirect_handler=_default_redirect_handler,
@@ -212,12 +212,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 streamablehttp_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 httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client:
+                    async with streamable_http_client(
+                        url=self.server_url,
+                        http_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}")

examples/snippets/clients/oauth_client.py

@@ -10,11 +10,12 @@
 import asyncio
 from urllib.parse import parse_qs, urlparse
 
+import httpx
 from pydantic import AnyUrl
 
 from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
-from mcp.client.streamable_http import streamablehttp_client
+from mcp.client.streamable_http import streamable_http_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
@@ -68,15 +69,16 @@ async def main():
         callback_handler=handle_callback,
     )
 
-    async with streamablehttp_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
-        async with ClientSession(read, write) as session:
-            await session.initialize()
+    async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) 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()
 
-            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/snippets/clients/streamable_basic.py

@@ -6,12 +6,12 @@
 import asyncio
 
 from mcp import ClientSession
-from mcp.client.streamable_http import streamablehttp_client
+from mcp.client.streamable_http import streamable_http_client
 
 
 async def main():
     # Connect to a streamable HTTP server
-    async with streamablehttp_client("http://localhost:8000/mcp") as (
+    async with streamable_http_client("http://localhost:8000/mcp") as (
         read_stream,
         write_stream,
         _,

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
 
@@ -25,7 +26,8 @@
 from mcp.client.session import ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
 from mcp.client.sse import sse_client
 from mcp.client.stdio import StdioServerParameters
-from mcp.client.streamable_http import streamablehttp_client
+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
 
@@ -47,7 +49,7 @@ class SseServerParameters(BaseModel):
 
 
 class StreamableHttpParameters(BaseModel):
-    """Parameters for intializing a streamablehttp_client."""
+    """Parameters for intializing a streamable_http_client."""
 
     # The endpoint URL.
     url: str
@@ -309,11 +311,18 @@ async def _establish_session(
                 )
                 read, write = await session_stack.enter_async_context(client)
             else:
-                client = streamablehttp_client(
-                    url=server_params.url,
+                httpx_client = create_mcp_http_client(
                     headers=server_params.headers,
-                    timeout=server_params.timeout,
-                    sse_read_timeout=server_params.sse_read_timeout,
+                    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,
+                    http_client=httpx_client,
                     terminate_on_close=server_params.terminate_on_close,
                 )
                 read, write, _ = await session_stack.enter_async_context(client)