diff --git a/README.md b/README.md index a43918464..e7a6e955b 100644 --- a/README.md +++ b/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(): diff --git a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py b/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py index 38dc5a916..a88c4ea6b 100644 --- a/examples/clients/simple-auth-client/mcp_simple_auth_client/main.py +++ b/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}") diff --git a/examples/snippets/clients/oauth_client.py b/examples/snippets/clients/oauth_client.py index 45026590a..140b38aed 100644 --- a/examples/snippets/clients/oauth_client.py +++ b/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(): diff --git a/examples/snippets/clients/streamable_basic.py b/examples/snippets/clients/streamable_basic.py index 108439613..071ea8155 100644 --- a/examples/snippets/clients/streamable_basic.py +++ b/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, _, diff --git a/src/mcp/client/session_group.py b/src/mcp/client/session_group.py index da45923e2..f82677d27 100644 --- a/src/mcp/client/session_group.py +++ b/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) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index fa0524e6e..ed28fcc27 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -6,19 +6,26 @@ and session management. """ +import contextlib import logging from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager from dataclasses import dataclass from datetime import timedelta +from typing import Any, overload +from warnings import warn import anyio import httpx from anyio.abc import TaskGroup from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream 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 ( + McpHttpClientFactory, + create_mcp_http_client, +) from mcp.shared.message import ClientMessageMetadata, SessionMessage from mcp.types import ( ErrorData, @@ -53,6 +60,9 @@ JSON = "application/json" SSE = "text/event-stream" +# Sentinel value for detecting unset optional parameters +_UNSET = object() + class StreamableHTTPError(Exception): """Base exception for StreamableHTTP transport errors.""" @@ -67,17 +77,25 @@ 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 + headers: dict[str, str] | None = None # Deprecated - no longer used + sse_read_timeout: float | None = None # Deprecated - no longer used class StreamableHTTPTransport: """StreamableHTTP client transport implementation.""" + @overload + def __init__(self, url: str) -> None: ... + + @overload + @deprecated( + "Parameters headers, timeout, sse_read_timeout, and auth are deprecated. " + "Configure these on the httpx.AsyncClient instead." + ) def __init__( self, url: str, @@ -85,6 +103,15 @@ def __init__( timeout: float | timedelta = 30, sse_read_timeout: float | timedelta = 60 * 5, auth: httpx.Auth | None = None, + ) -> None: ... + + def __init__( + self, + url: str, + headers: Any = _UNSET, + timeout: Any = _UNSET, + sse_read_timeout: Any = _UNSET, + auth: Any = _UNSET, ) -> None: """Initialize the StreamableHTTP transport. @@ -95,24 +122,40 @@ def __init__( sse_read_timeout: Timeout for SSE read operations. auth: Optional HTTPX authentication handler. """ + # Check for deprecated parameters and issue runtime warning + deprecated_params: list[str] = [] + if headers is not _UNSET: + deprecated_params.append("headers") + if timeout is not _UNSET: + deprecated_params.append("timeout") + if sse_read_timeout is not _UNSET: + deprecated_params.append("sse_read_timeout") + if auth is not _UNSET: + deprecated_params.append("auth") + + if deprecated_params: + warn( + f"Parameters {', '.join(deprecated_params)} are deprecated and will be ignored. " + "Configure these on the httpx.AsyncClient instead.", + DeprecationWarning, + stacklevel=2, + ) + 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 = { - ACCEPT: f"{JSON}, {SSE}", - CONTENT_TYPE: JSON, - **self.headers, - } - - 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) -> dict[str, str]: + """Build MCP-specific request headers. + + These headers will be merged with the httpx.AsyncClient's default headers, + with these MCP-specific headers taking precedence. + """ + headers: dict[str, str] = {} + # 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: @@ -216,7 +259,7 @@ async def handle_get_stream( if not self.session_id: return - headers = self._prepare_request_headers(self.request_headers) + headers = self._prepare_headers() if last_event_id: headers[LAST_EVENT_ID] = last_event_id # pragma: no cover @@ -225,7 +268,6 @@ async def handle_get_stream( "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") @@ -258,7 +300,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() if ctx.metadata and ctx.metadata.resumption_token: headers[LAST_EVENT_ID] = ctx.metadata.resumption_token else: @@ -274,7 +316,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") @@ -292,7 +333,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() message = ctx.session_message.message is_initialization = self._is_initialization_request(message) @@ -410,7 +451,7 @@ async def _handle_reconnection( delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS await anyio.sleep(delay_ms / 1000.0) - headers = self._prepare_request_headers(ctx.headers) + headers = self._prepare_headers() headers[LAST_EVENT_ID] = last_event_id # Extract original request ID to map responses @@ -424,7 +465,6 @@ async def _handle_reconnection( "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.info("Reconnected to SSE stream") @@ -512,12 +552,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(): @@ -544,7 +582,7 @@ async def terminate_session(self, client: httpx.AsyncClient) -> None: # pragma: return try: - headers = self._prepare_request_headers(self.request_headers) + headers = self._prepare_headers() response = await client.delete(self.url, headers=headers) if response.status_code == 405: @@ -560,14 +598,11 @@ def get_session_id(self) -> str | None: @asynccontextmanager -async def streamablehttp_client( +async def streamable_http_client( url: str, - headers: dict[str, str] | None = None, - timeout: float | timedelta = 30, - sse_read_timeout: float | timedelta = 60 * 5, + *, + http_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], @@ -579,30 +614,45 @@ async def streamablehttp_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. + http_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 = http_client is not None + client = http_client + + if client is None: + # Create default client with recommended MCP timeouts + client = create_mcp_http_client() + + transport = StreamableHTTPTransport(url) + 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) @@ -629,3 +679,44 @@ def start_get_stream() -> None: finally: await read_stream_writer.aclose() await write_stream.aclose() + + +@asynccontextmanager +@deprecated("Use `streamable_http_client` instead.") +async def streamablehttp_client( + url: str, + headers: dict[str, str] | None = None, + timeout: float | timedelta = 30, + sse_read_timeout: float | timedelta = 60 * 5, + terminate_on_close: bool = True, + httpx_client_factory: McpHttpClientFactory = create_mcp_http_client, + auth: httpx.Auth | None = None, +) -> AsyncGenerator[ + tuple[ + MemoryObjectReceiveStream[SessionMessage | Exception], + MemoryObjectSendStream[SessionMessage], + GetSessionIdCallback, + ], + None, +]: + # 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, + http_client=client, + terminate_on_close=terminate_on_close, + ) as streams: + yield streams diff --git a/src/mcp/shared/_httpx_utils.py b/src/mcp/shared/_httpx_utils.py index 3af64ad1e..945ef8095 100644 --- a/src/mcp/shared/_httpx_utils.py +++ b/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 diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py index 95e01ce57..ec38f3583 100644 --- a/tests/client/test_http_unicode.py +++ b/tests/client/test_http_unicode.py @@ -12,7 +12,7 @@ import pytest from mcp.client.session import ClientSession -from mcp.client.streamable_http import streamablehttp_client +from mcp.client.streamable_http import streamable_http_client from tests.test_helpers import wait_for_server # Test constants with various Unicode characters @@ -178,7 +178,7 @@ async def test_streamable_http_client_unicode_tool_call(running_unicode_server: base_url = running_unicode_server endpoint_url = f"{base_url}/mcp" - async with streamablehttp_client(endpoint_url) as (read_stream, write_stream, _get_session_id): + async with streamable_http_client(endpoint_url) as (read_stream, write_stream, _get_session_id): async with ClientSession(read_stream, write_stream) as session: await session.initialize() @@ -210,7 +210,7 @@ async def test_streamable_http_client_unicode_prompts(running_unicode_server: st base_url = running_unicode_server endpoint_url = f"{base_url}/mcp" - async with streamablehttp_client(endpoint_url) as (read_stream, write_stream, _get_session_id): + async with streamable_http_client(endpoint_url) as (read_stream, write_stream, _get_session_id): async with ClientSession(read_stream, write_stream) as session: await session.initialize() diff --git a/tests/client/test_notification_response.py b/tests/client/test_notification_response.py index 19d537400..7500abee7 100644 --- a/tests/client/test_notification_response.py +++ b/tests/client/test_notification_response.py @@ -18,7 +18,7 @@ from starlette.routing import Route from mcp import ClientSession, types -from mcp.client.streamable_http import streamablehttp_client +from mcp.client.streamable_http import streamable_http_client from mcp.shared.session import RequestResponder from mcp.types import ClientNotification, RootsListChangedNotification from tests.test_helpers import wait_for_server @@ -127,7 +127,7 @@ async def message_handler( # pragma: no cover if isinstance(message, Exception): returned_exception = message - async with streamablehttp_client(server_url) as (read_stream, write_stream, _): + async with streamable_http_client(server_url) as (read_stream, write_stream, _): async with ClientSession( read_stream, write_stream, diff --git a/tests/client/test_session_group.py b/tests/client/test_session_group.py index e61ea572b..755c613d9 100644 --- a/tests/client/test_session_group.py +++ b/tests/client/test_session_group.py @@ -280,7 +280,7 @@ async def test_disconnect_non_existent_server(self): ( StreamableHttpParameters(url="http://test.com/stream", terminate_on_close=False), "streamablehttp", - "mcp.client.session_group.streamablehttp_client", + "mcp.client.session_group.streamable_http_client", ), # url, headers, timeout, sse_read_timeout, terminate_on_close ], ) @@ -296,7 +296,7 @@ async def test_establish_session_parameterized( mock_read_stream = mock.AsyncMock(name=f"{client_type_name}Read") mock_write_stream = mock.AsyncMock(name=f"{client_type_name}Write") - # streamablehttp_client's __aenter__ returns three values + # streamable_http_client's __aenter__ returns three values if client_type_name == "streamablehttp": mock_extra_stream_val = mock.AsyncMock(name="StreamableExtra") mock_client_cm_instance.__aenter__.return_value = ( @@ -354,13 +354,14 @@ async def test_establish_session_parameterized( ) elif client_type_name == "streamablehttp": # pragma: no branch assert isinstance(server_params_instance, StreamableHttpParameters) - mock_specific_client_func.assert_called_once_with( - url=server_params_instance.url, - headers=server_params_instance.headers, - timeout=server_params_instance.timeout, - sse_read_timeout=server_params_instance.sse_read_timeout, - terminate_on_close=server_params_instance.terminate_on_close, - ) + # Verify streamable_http_client was called with url, httpx_client, and terminate_on_close + # The http_client is created by the real create_mcp_http_client + import httpx + + call_args = mock_specific_client_func.call_args + assert call_args.kwargs["url"] == server_params_instance.url + assert call_args.kwargs["terminate_on_close"] == server_params_instance.terminate_on_close + assert isinstance(call_args.kwargs["http_client"], httpx.AsyncClient) mock_client_cm_instance.__aenter__.assert_awaited_once() diff --git a/tests/server/fastmcp/test_integration.py b/tests/server/fastmcp/test_integration.py index b1cefca29..70948bd7e 100644 --- a/tests/server/fastmcp/test_integration.py +++ b/tests/server/fastmcp/test_integration.py @@ -34,7 +34,7 @@ ) from mcp.client.session import ClientSession from mcp.client.sse import sse_client -from mcp.client.streamable_http import GetSessionIdCallback, streamablehttp_client +from mcp.client.streamable_http import GetSessionIdCallback, streamable_http_client from mcp.shared.context import RequestContext from mcp.shared.message import SessionMessage from mcp.shared.session import RequestResponder @@ -179,7 +179,7 @@ def create_client_for_transport(transport: str, server_url: str): return sse_client(endpoint) elif transport == "streamable-http": endpoint = f"{server_url}/mcp" - return streamablehttp_client(endpoint) + return streamable_http_client(endpoint) else: # pragma: no cover raise ValueError(f"Invalid transport: {transport}") diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index a626e7385..731dd20dd 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -9,6 +9,7 @@ import socket import time from collections.abc import Generator +from datetime import timedelta from typing import Any from unittest.mock import MagicMock @@ -25,7 +26,11 @@ import mcp.types as types from mcp.client.session import ClientSession -from mcp.client.streamable_http import StreamableHTTPTransport, streamablehttp_client +from mcp.client.streamable_http import ( + StreamableHTTPTransport, + streamable_http_client, + streamablehttp_client, # pyright: ignore[reportDeprecated] +) from mcp.server import Server from mcp.server.streamable_http import ( MCP_PROTOCOL_VERSION_HEADER, @@ -40,6 +45,7 @@ ) from mcp.server.streamable_http_manager import StreamableHTTPSessionManager from mcp.server.transport_security import TransportSecuritySettings +from mcp.shared._httpx_utils import create_mcp_http_client from mcp.shared.context import RequestContext from mcp.shared.exceptions import McpError from mcp.shared.message import ClientMessageMetadata, ServerMessageMetadata, SessionMessage @@ -972,7 +978,7 @@ async def http_client(basic_server: None, basic_server_url: str): # pragma: no @pytest.fixture async def initialized_client_session(basic_server: None, basic_server_url: str): """Create initialized StreamableHTTP client session.""" - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -986,9 +992,9 @@ async def initialized_client_session(basic_server: None, basic_server_url: str): @pytest.mark.anyio -async def test_streamablehttp_client_basic_connection(basic_server: None, basic_server_url: str): +async def test_streamable_http_client_basic_connection(basic_server: None, basic_server_url: str): """Test basic client connection with initialization.""" - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1004,7 +1010,7 @@ async def test_streamablehttp_client_basic_connection(basic_server: None, basic_ @pytest.mark.anyio -async def test_streamablehttp_client_resource_read(initialized_client_session: ClientSession): +async def test_streamable_http_client_resource_read(initialized_client_session: ClientSession): """Test client resource read functionality.""" response = await initialized_client_session.read_resource(uri=AnyUrl("foobar://test-resource")) assert len(response.contents) == 1 @@ -1014,7 +1020,7 @@ async def test_streamablehttp_client_resource_read(initialized_client_session: C @pytest.mark.anyio -async def test_streamablehttp_client_tool_invocation(initialized_client_session: ClientSession): +async def test_streamable_http_client_tool_invocation(initialized_client_session: ClientSession): """Test client tool invocation.""" # First list tools tools = await initialized_client_session.list_tools() @@ -1029,7 +1035,7 @@ async def test_streamablehttp_client_tool_invocation(initialized_client_session: @pytest.mark.anyio -async def test_streamablehttp_client_error_handling(initialized_client_session: ClientSession): +async def test_streamable_http_client_error_handling(initialized_client_session: ClientSession): """Test error handling in client.""" with pytest.raises(McpError) as exc_info: await initialized_client_session.read_resource(uri=AnyUrl("unknown://test-error")) @@ -1038,9 +1044,9 @@ async def test_streamablehttp_client_error_handling(initialized_client_session: @pytest.mark.anyio -async def test_streamablehttp_client_session_persistence(basic_server: None, basic_server_url: str): +async def test_streamable_http_client_session_persistence(basic_server: None, basic_server_url: str): """Test that session ID persists across requests.""" - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1066,9 +1072,9 @@ async def test_streamablehttp_client_session_persistence(basic_server: None, bas @pytest.mark.anyio -async def test_streamablehttp_client_json_response(json_response_server: None, json_server_url: str): +async def test_streamable_http_client_json_response(json_response_server: None, json_server_url: str): """Test client with JSON response mode.""" - async with streamablehttp_client(f"{json_server_url}/mcp") as ( + async with streamable_http_client(f"{json_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1094,7 +1100,7 @@ async def test_streamablehttp_client_json_response(json_response_server: None, j @pytest.mark.anyio -async def test_streamablehttp_client_get_stream(basic_server: None, basic_server_url: str): +async def test_streamable_http_client_get_stream(basic_server: None, basic_server_url: str): """Test GET stream functionality for server-initiated messages.""" import mcp.types as types @@ -1107,7 +1113,7 @@ async def message_handler( # pragma: no branch if isinstance(message, types.ServerNotification): # pragma: no branch notifications_received.append(message) - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1134,13 +1140,13 @@ async def message_handler( # pragma: no branch @pytest.mark.anyio -async def test_streamablehttp_client_session_termination(basic_server: None, basic_server_url: str): +async def test_streamable_http_client_session_termination(basic_server: None, basic_server_url: str): """Test client session termination functionality.""" captured_session_id = None - # Create the streamablehttp_client with a custom httpx client to capture headers - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + # Create the streamable_http_client with a custom httpx client to capture headers + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, get_session_id, @@ -1160,22 +1166,23 @@ async def test_streamablehttp_client_session_termination(basic_server: None, bas if captured_session_id: # pragma: no cover headers[MCP_SESSION_ID_HEADER] = captured_session_id - async with streamablehttp_client(f"{basic_server_url}/mcp", headers=headers) as ( - read_stream, - write_stream, - _, - ): - async with ClientSession(read_stream, write_stream) as session: - # Attempt to make a request after termination - with pytest.raises( # pragma: no branch - McpError, - match="Session terminated", - ): - await session.list_tools() + async with create_mcp_http_client(headers=headers) as httpx_client: + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + # Attempt to make a request after termination + with pytest.raises( # pragma: no branch + McpError, + match="Session terminated", + ): + await session.list_tools() @pytest.mark.anyio -async def test_streamablehttp_client_session_termination_204( +async def test_streamable_http_client_session_termination_204( basic_server: None, basic_server_url: str, monkeypatch: pytest.MonkeyPatch ): """Test client session termination functionality with a 204 response. @@ -1205,8 +1212,8 @@ async def mock_delete(self: httpx.AsyncClient, *args: Any, **kwargs: Any) -> htt captured_session_id = None - # Create the streamablehttp_client with a custom httpx client to capture headers - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + # Create the streamable_http_client with a custom httpx client to capture headers + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, get_session_id, @@ -1226,22 +1233,23 @@ async def mock_delete(self: httpx.AsyncClient, *args: Any, **kwargs: Any) -> htt if captured_session_id: # pragma: no cover headers[MCP_SESSION_ID_HEADER] = captured_session_id - async with streamablehttp_client(f"{basic_server_url}/mcp", headers=headers) as ( - read_stream, - write_stream, - _, - ): - async with ClientSession(read_stream, write_stream) as session: - # Attempt to make a request after termination - with pytest.raises( # pragma: no branch - McpError, - match="Session terminated", - ): - await session.list_tools() + async with create_mcp_http_client(headers=headers) as httpx_client: + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + # Attempt to make a request after termination + with pytest.raises( # pragma: no branch + McpError, + match="Session terminated", + ): + await session.list_tools() @pytest.mark.anyio -async def test_streamablehttp_client_resumption(event_server: tuple[SimpleEventStore, str]): +async def test_streamable_http_client_resumption(event_server: tuple[SimpleEventStore, str]): """Test client session resumption using sync primitives for reliable coordination.""" _, server_url = event_server @@ -1268,7 +1276,7 @@ async def on_resumption_token_update(token: str) -> None: captured_resumption_token = token # First, start the client session and begin the tool that waits on lock - async with streamablehttp_client(f"{server_url}/mcp", terminate_on_close=False) as ( + async with streamable_http_client(f"{server_url}/mcp", terminate_on_close=False) as ( read_stream, write_stream, get_session_id, @@ -1324,42 +1332,44 @@ async def run_tool(): headers[MCP_SESSION_ID_HEADER] = captured_session_id if captured_protocol_version: # pragma: no cover headers[MCP_PROTOCOL_VERSION_HEADER] = captured_protocol_version - async with streamablehttp_client(f"{server_url}/mcp", headers=headers) as ( - read_stream, - write_stream, - _, - ): - async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session: - result = await session.send_request( - types.ClientRequest( - types.CallToolRequest( - params=types.CallToolRequestParams(name="release_lock", arguments={}), - ) - ), - types.CallToolResult, - ) - metadata = ClientMessageMetadata( - resumption_token=captured_resumption_token, - ) - result = await session.send_request( - types.ClientRequest( - types.CallToolRequest( - params=types.CallToolRequestParams(name="wait_for_lock_with_notification", arguments={}), - ) - ), - types.CallToolResult, - metadata=metadata, - ) - assert len(result.content) == 1 - assert result.content[0].type == "text" - assert result.content[0].text == "Completed" + async with create_mcp_http_client(headers=headers) as httpx_client: + async with streamable_http_client(f"{server_url}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session: + result = await session.send_request( + types.ClientRequest( + types.CallToolRequest( + params=types.CallToolRequestParams(name="release_lock", arguments={}), + ) + ), + types.CallToolResult, + ) + metadata = ClientMessageMetadata( + resumption_token=captured_resumption_token, + ) - # We should have received the remaining notifications - assert len(captured_notifications) == 1 + result = await session.send_request( + types.ClientRequest( + types.CallToolRequest( + params=types.CallToolRequestParams(name="wait_for_lock_with_notification", arguments={}), + ) + ), + types.CallToolResult, + metadata=metadata, + ) + assert len(result.content) == 1 + assert result.content[0].type == "text" + assert result.content[0].text == "Completed" + + # We should have received the remaining notifications + assert len(captured_notifications) == 1 - assert isinstance(captured_notifications[0].root, types.LoggingMessageNotification) - assert captured_notifications[0].root.params.data == "Second notification after lock" + assert isinstance(captured_notifications[0].root, types.LoggingMessageNotification) # pragma: no cover + assert captured_notifications[0].root.params.data == "Second notification after lock" # pragma: no cover @pytest.mark.anyio @@ -1391,7 +1401,7 @@ async def sampling_callback( ) # Create client with sampling callback - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1536,28 +1546,29 @@ async def test_streamablehttp_request_context_propagation(context_aware_server: "X-Trace-Id": "trace-123", } - async with streamablehttp_client(f"{basic_server_url}/mcp", headers=custom_headers) as ( - read_stream, - write_stream, - _, - ): - async with ClientSession(read_stream, write_stream) as session: - result = await session.initialize() - assert isinstance(result, InitializeResult) - assert result.serverInfo.name == "ContextAwareServer" + async with create_mcp_http_client(headers=custom_headers) as httpx_client: + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + result = await session.initialize() + assert isinstance(result, InitializeResult) + assert result.serverInfo.name == "ContextAwareServer" - # Call the tool that echoes headers back - tool_result = await session.call_tool("echo_headers", {}) + # Call the tool that echoes headers back + tool_result = await session.call_tool("echo_headers", {}) - # Parse the JSON response - assert len(tool_result.content) == 1 - assert isinstance(tool_result.content[0], TextContent) - headers_data = json.loads(tool_result.content[0].text) + # Parse the JSON response + assert len(tool_result.content) == 1 + assert isinstance(tool_result.content[0], TextContent) + headers_data = json.loads(tool_result.content[0].text) - # Verify headers were propagated - assert headers_data.get("authorization") == "Bearer test-token" - assert headers_data.get("x-custom-header") == "test-value" - assert headers_data.get("x-trace-id") == "trace-123" + # Verify headers were propagated + assert headers_data.get("authorization") == "Bearer test-token" + assert headers_data.get("x-custom-header") == "test-value" + assert headers_data.get("x-trace-id") == "trace-123" @pytest.mark.anyio @@ -1573,17 +1584,22 @@ async def test_streamablehttp_request_context_isolation(context_aware_server: No "Authorization": f"Bearer token-{i}", } - async with streamablehttp_client(f"{basic_server_url}/mcp", headers=headers) as (read_stream, write_stream, _): - async with ClientSession(read_stream, write_stream) as session: - await session.initialize() + async with create_mcp_http_client(headers=headers) as httpx_client: + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=httpx_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + await session.initialize() - # Call the tool that echoes context - tool_result = await session.call_tool("echo_context", {"request_id": f"request-{i}"}) + # Call the tool that echoes context + tool_result = await session.call_tool("echo_context", {"request_id": f"request-{i}"}) - assert len(tool_result.content) == 1 - assert isinstance(tool_result.content[0], TextContent) - context_data = json.loads(tool_result.content[0].text) - contexts.append(context_data) + assert len(tool_result.content) == 1 + assert isinstance(tool_result.content[0], TextContent) + context_data = json.loads(tool_result.content[0].text) + contexts.append(context_data) # Verify each request had its own context assert len(contexts) == 3 # pragma: no cover @@ -1597,7 +1613,7 @@ async def test_streamablehttp_request_context_isolation(context_aware_server: No @pytest.mark.anyio async def test_client_includes_protocol_version_header_after_init(context_aware_server: None, basic_server_url: str): """Test that client includes mcp-protocol-version header after initialization.""" - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1713,7 +1729,7 @@ async def test_client_crash_handled(basic_server: None, basic_server_url: str): # Simulate bad client that crashes after init async def bad_client(): """Client that triggers ClosedResourceError""" - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1731,7 +1747,7 @@ async def bad_client(): await anyio.sleep(0.1) # Try a good client, it should still be able to connect and list tools - async with streamablehttp_client(f"{basic_server_url}/mcp") as ( + async with streamable_http_client(f"{basic_server_url}/mcp") as ( read_stream, write_stream, _, @@ -1880,7 +1896,7 @@ async def test_close_sse_stream_callback_not_provided_for_old_protocol_version() @pytest.mark.anyio -async def test_streamablehttp_client_receives_priming_event( +async def test_streamable_http_client_receives_priming_event( event_server: tuple[SimpleEventStore, str], ) -> None: """Client should receive priming event (resumption token update) on POST SSE stream.""" @@ -1891,7 +1907,7 @@ async def test_streamablehttp_client_receives_priming_event( async def on_resumption_token_update(token: str) -> None: captured_resumption_tokens.append(token) - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -1932,7 +1948,7 @@ async def test_server_close_sse_stream_via_context( """Server tool can call ctx.close_sse_stream() to close connection.""" _, server_url = event_server - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -1953,7 +1969,7 @@ async def test_server_close_sse_stream_via_context( @pytest.mark.anyio -async def test_streamablehttp_client_auto_reconnects( +async def test_streamable_http_client_auto_reconnects( event_server: tuple[SimpleEventStore, str], ) -> None: """Client should auto-reconnect with Last-Event-ID when server closes after priming event.""" @@ -1969,7 +1985,7 @@ async def message_handler( if isinstance(message.root, types.LoggingMessageNotification): # pragma: no branch captured_notifications.append(str(message.root.params.data)) - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -1998,13 +2014,13 @@ async def message_handler( @pytest.mark.anyio -async def test_streamablehttp_client_respects_retry_interval( +async def test_streamable_http_client_respects_retry_interval( event_server: tuple[SimpleEventStore, str], ) -> None: """Client MUST respect retry field, waiting specified ms before reconnecting.""" _, server_url = event_server - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -2029,7 +2045,7 @@ async def test_streamablehttp_client_respects_retry_interval( @pytest.mark.anyio -async def test_streamablehttp_sse_polling_full_cycle( +async def test_streamable_http_sse_polling_full_cycle( event_server: tuple[SimpleEventStore, str], ) -> None: """End-to-end test: server closes stream, client reconnects, receives all events.""" @@ -2045,7 +2061,7 @@ async def message_handler( if isinstance(message.root, types.LoggingMessageNotification): # pragma: no branch all_notifications.append(str(message.root.params.data)) - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -2077,7 +2093,7 @@ async def message_handler( @pytest.mark.anyio -async def test_streamablehttp_events_replayed_after_disconnect( +async def test_streamable_http_events_replayed_after_disconnect( event_server: tuple[SimpleEventStore, str], ) -> None: """Events sent while client is disconnected should be replayed on reconnect.""" @@ -2093,7 +2109,7 @@ async def message_handler( if isinstance(message.root, types.LoggingMessageNotification): # pragma: no branch notification_data.append(str(message.root.params.data)) - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -2125,7 +2141,7 @@ async def message_handler( @pytest.mark.anyio -async def test_streamablehttp_multiple_reconnections( +async def test_streamable_http_multiple_reconnections( event_server: tuple[SimpleEventStore, str], ): """Verify multiple close_sse_stream() calls each trigger a client reconnect. @@ -2145,7 +2161,7 @@ async def test_streamablehttp_multiple_reconnections( async def on_resumption_token(token: str) -> None: resumption_tokens.append(token) - async with streamablehttp_client(f"{server_url}/mcp") as (read_stream, write_stream, _): + async with streamable_http_client(f"{server_url}/mcp") as (read_stream, write_stream, _): async with ClientSession(read_stream, write_stream) as session: await session.initialize() @@ -2205,7 +2221,7 @@ async def message_handler( if isinstance(message.root, types.ResourceUpdatedNotification): # pragma: no branch received_notifications.append(str(message.root.params.uri)) - async with streamablehttp_client(f"{server_url}/mcp") as ( + async with streamable_http_client(f"{server_url}/mcp") as ( read_stream, write_stream, _, @@ -2236,3 +2252,145 @@ async def message_handler( assert "http://notification_2/" in received_notifications, ( f"Should receive notification 2 after reconnect, got: {received_notifications}" ) + + +@pytest.mark.anyio +async def test_streamable_http_client_does_not_mutate_provided_client( + basic_server: None, basic_server_url: str +) -> None: + """Test that streamable_http_client does not mutate the provided httpx client's headers.""" + # Create a client with custom headers + original_headers = { + "X-Custom-Header": "custom-value", + "Authorization": "Bearer test-token", + } + + async with httpx.AsyncClient(headers=original_headers, follow_redirects=True) as custom_client: + # Use the client with streamable_http_client + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=custom_client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: + result = await session.initialize() + assert isinstance(result, InitializeResult) + + # Verify client headers were not mutated with MCP protocol headers + # If accept header exists, it should still be httpx default, not MCP's + if "accept" in custom_client.headers: # pragma: no branch + assert custom_client.headers.get("accept") == "*/*" + # MCP content-type should not have been added + assert custom_client.headers.get("content-type") != "application/json" + + # Verify custom headers are still present and unchanged + assert custom_client.headers.get("X-Custom-Header") == "custom-value" + assert custom_client.headers.get("Authorization") == "Bearer test-token" + + +@pytest.mark.anyio +async def test_streamable_http_client_mcp_headers_override_defaults( + context_aware_server: None, basic_server_url: str +) -> None: + """Test that MCP protocol headers override httpx.AsyncClient default headers.""" + # httpx.AsyncClient has default "accept: */*" header + # We need to verify that our MCP accept header overrides it in actual requests + + async with httpx.AsyncClient(follow_redirects=True) as client: + # Verify client has default accept header + assert client.headers.get("accept") == "*/*" + + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + await session.initialize() + + # Use echo_headers tool to see what headers the server actually received + tool_result = await session.call_tool("echo_headers", {}) + assert len(tool_result.content) == 1 + assert isinstance(tool_result.content[0], TextContent) + headers_data = json.loads(tool_result.content[0].text) + + # Verify MCP protocol headers were sent (not httpx defaults) + assert "accept" in headers_data + assert "application/json" in headers_data["accept"] + assert "text/event-stream" in headers_data["accept"] + + assert "content-type" in headers_data + assert headers_data["content-type"] == "application/json" + + +@pytest.mark.anyio +async def test_streamable_http_client_preserves_custom_with_mcp_headers( + context_aware_server: None, basic_server_url: str +) -> None: + """Test that both custom headers and MCP protocol headers are sent in requests.""" + custom_headers = { + "X-Custom-Header": "custom-value", + "X-Request-Id": "req-123", + "Authorization": "Bearer test-token", + } + + async with httpx.AsyncClient(headers=custom_headers, follow_redirects=True) as client: + async with streamable_http_client(f"{basic_server_url}/mcp", http_client=client) as ( + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + await session.initialize() + + # Use echo_headers tool to verify both custom and MCP headers are present + tool_result = await session.call_tool("echo_headers", {}) + assert len(tool_result.content) == 1 + assert isinstance(tool_result.content[0], TextContent) + headers_data = json.loads(tool_result.content[0].text) + + # Verify custom headers are present + assert headers_data.get("x-custom-header") == "custom-value" + assert headers_data.get("x-request-id") == "req-123" + assert headers_data.get("authorization") == "Bearer test-token" + + # Verify MCP protocol headers are also present + assert "accept" in headers_data + assert "application/json" in headers_data["accept"] + assert "text/event-stream" in headers_data["accept"] + + assert "content-type" in headers_data + assert headers_data["content-type"] == "application/json" + + +@pytest.mark.anyio +async def test_streamable_http_transport_deprecated_params_ignored(basic_server: None, basic_server_url: str) -> None: + """Test that deprecated parameters passed to StreamableHTTPTransport are properly ignored.""" + with pytest.warns(DeprecationWarning): + transport = StreamableHTTPTransport( # pyright: ignore[reportDeprecated] + url=f"{basic_server_url}/mcp", + headers={"X-Should-Be-Ignored": "ignored"}, + timeout=999, + sse_read_timeout=timedelta(seconds=999), + auth=None, + ) + + headers = transport._prepare_headers() + assert "X-Should-Be-Ignored" not in headers + assert headers["accept"] == "application/json, text/event-stream" + assert headers["content-type"] == "application/json" + + +@pytest.mark.anyio +async def test_streamablehttp_client_deprecation_warning(basic_server: None, basic_server_url: str) -> None: + """Test that the old streamablehttp_client() function issues a deprecation warning.""" + with pytest.warns(DeprecationWarning, match="Use `streamable_http_client` instead"): + async with streamablehttp_client(f"{basic_server_url}/mcp") as ( # pyright: ignore[reportDeprecated] + read_stream, + write_stream, + _, + ): + async with ClientSession(read_stream, write_stream) as session: # pragma: no branch + await session.initialize() + tools = await session.list_tools() + assert len(tools.tools) > 0