Merge branch 'main' into client_notif_support

Author: kylestratis

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,
         _,
@@ -2420,11 +2420,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
 
 
@@ -2478,15 +2479,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/auth/oauth2.py

@@ -19,7 +19,7 @@
 import httpx
 from pydantic import BaseModel, Field, ValidationError
 
-from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
+from mcp.client.auth.exceptions import OAuthFlowError, OAuthTokenError
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -299,44 +299,6 @@ async def _handle_protected_resource_response(self, response: httpx.Response) ->
                 f"Protected Resource Metadata request failed: {response.status_code}"
             )  # pragma: no cover
 
-    async def _register_client(self) -> httpx.Request | None:
-        """Build registration request or skip if already registered."""
-        if self.context.client_info:
-            return None
-
-        if self.context.oauth_metadata and self.context.oauth_metadata.registration_endpoint:
-            registration_url = str(self.context.oauth_metadata.registration_endpoint)  # pragma: no cover
-        else:
-            auth_base_url = self.context.get_authorization_base_url(self.context.server_url)
-            registration_url = urljoin(auth_base_url, "/register")
-
-        registration_data = self.context.client_metadata.model_dump(by_alias=True, mode="json", exclude_none=True)
-
-        # If token_endpoint_auth_method is None, auto-select based on server support
-        if self.context.client_metadata.token_endpoint_auth_method is None:
-            preference_order = ["client_secret_basic", "client_secret_post", "none"]
-
-            if self.context.oauth_metadata and self.context.oauth_metadata.token_endpoint_auth_methods_supported:
-                supported = self.context.oauth_metadata.token_endpoint_auth_methods_supported
-                for method in preference_order:
-                    if method in supported:
-                        registration_data["token_endpoint_auth_method"] = method
-                        break
-                else:
-                    # No compatible methods between client and server
-                    raise OAuthRegistrationError(
-                        f"No compatible authentication methods. "
-                        f"Server supports: {supported}, "
-                        f"Client supports: {preference_order}"
-                    )
-            else:
-                # No server metadata available, use our default preference
-                registration_data["token_endpoint_auth_method"] = preference_order[0]
-
-        return httpx.Request(
-            "POST", registration_url, json=registration_data, headers={"Content-Type": "application/json"}
-        )
-
     async def _perform_authorization(self) -> httpx.Request:
         """Perform the authorization flow."""
         auth_code, code_verifier = await self._perform_authorization_code_grant()

src/mcp/client/session.py

@@ -25,7 +25,7 @@ async def __call__(
         self,
         context: RequestContext["ClientSession", Any],
         params: types.CreateMessageRequestParams,
-    ) -> types.CreateMessageResult | types.ErrorData: ...  # pragma: no branch
+    ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData: ...  # pragma: no branch
 
 
 class ElicitationFnT(Protocol):
@@ -104,7 +104,7 @@ async def _default_message_handler(
 async def _default_sampling_callback(
     context: RequestContext["ClientSession", Any],
     params: types.CreateMessageRequestParams,
-) -> types.CreateMessageResult | types.ErrorData:
+) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
         message="Sampling not supported",
@@ -196,6 +196,7 @@ def __init__(
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | None = None,
         *,
+        sampling_capabilities: types.SamplingCapability | None = None,
         experimental_task_handlers: ExperimentalTaskHandlers | None = None,
     ) -> None:
         super().__init__(
@@ -207,6 +208,7 @@ def __init__(
         )
         self._client_info = client_info or DEFAULT_CLIENT_INFO
         self._sampling_callback = sampling_callback or _default_sampling_callback
+        self._sampling_capabilities = sampling_capabilities
         self._elicitation_callback = elicitation_callback or _default_elicitation_callback
         self._elicit_complete_callback = elicit_complete_callback or _default_elicit_complete_callback
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
@@ -225,7 +227,11 @@ def __init__(
         self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers()
 
     async def initialize(self) -> types.InitializeResult:
-        sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
+        sampling = (
+            (self._sampling_capabilities or types.SamplingCapability())
+            if self._sampling_callback is not _default_sampling_callback
+            else None
+        )
         elicitation = (
             types.ElicitationCapability(
                 form=types.FormElicitationCapability(),

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)

src/mcp/client/sse.py

@@ -105,6 +105,9 @@ async def sse_reader(
                                         task_status.started(endpoint_url)
 
                                     case "message":
+                                        # Skip empty data (keep-alive pings)
+                                        if not sse.data:
+                                            continue
                                         try:
                                             message = types.JSONRPCMessage.model_validate_json(  # noqa: E501
                                                 sse.data