From 46f91eec2ec8ad4d4ad74d8426424cf814d02c77 Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Thu, 16 Jul 2026 15:59:42 +0200 Subject: [PATCH 1/7] feat(mcp): add Streamable HTTP client to extra/mcp Adds MCPClientStreamableHTTP + StreamableHTTPServerParams alongside the existing SSE and stdio MCP clients, for servers that speak the Streamable HTTP transport (--transport http). Credentials are set on the httpx client so they ride every request including initialize; the transport's own headers arg is deprecated. --- src/mistralai/extra/mcp/streamable_http.py | 64 ++++++++++++++++++++++ 1 file changed, 64 insertions(+) create mode 100644 src/mistralai/extra/mcp/streamable_http.py diff --git a/src/mistralai/extra/mcp/streamable_http.py b/src/mistralai/extra/mcp/streamable_http.py new file mode 100644 index 00000000..94cd6cbd --- /dev/null +++ b/src/mistralai/extra/mcp/streamable_http.py @@ -0,0 +1,64 @@ +import logging +from contextlib import AsyncExitStack +from typing import Any + +import httpx +from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream +from mcp.client.streamable_http import streamable_http_client # pyright: ignore[reportMissingImports] +from mcp.shared.message import SessionMessage # pyright: ignore[reportMissingImports] + +from mistralai.extra.mcp.base import ( + MCPClientBase, +) + +from mistralai.client.types import BaseModel + +logger = logging.getLogger(__name__) + + +class StreamableHTTPServerParams(BaseModel): + """Parameters required for a MCPClient with Streamable HTTP transport""" + + url: str + headers: dict[str, Any] | None = None + timeout: float = 30 + sse_read_timeout: float = 60 * 5 + + +class MCPClientStreamableHTTP(MCPClientBase): + """MCP client that uses the Streamable HTTP transport for communication. + + Credentials (for example a bearer token, or a per-request integration token) + are provided as ``headers`` and set as the default headers of the underlying + ``httpx.AsyncClient``, so they are sent on every request including the + initialize call. Recent ``mcp`` releases deprecate and ignore the transport's + own ``headers`` argument, so configuring them on the client is required. + """ + + _params: StreamableHTTPServerParams + + def __init__( + self, + params: StreamableHTTPServerParams, + name: str | None = None, + ): + super().__init__(name=name) + self._params = params + + async def _get_transport( + self, exit_stack: AsyncExitStack + ) -> tuple[ + MemoryObjectReceiveStream[SessionMessage | Exception], + MemoryObjectSendStream[SessionMessage], + ]: + http_client = await exit_stack.enter_async_context( + httpx.AsyncClient( + headers=self._params.headers, + timeout=self._params.timeout, + follow_redirects=True, + ) + ) + read_stream, write_stream, _ = await exit_stack.enter_async_context( + streamable_http_client(url=self._params.url, http_client=http_client) + ) + return read_stream, write_stream From f3d1efd22786d7881443dd811a584745acedb5ed Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Thu, 16 Jul 2026 18:31:34 +0200 Subject: [PATCH 2/7] refactor(mcp): drop unused sse_read_timeout from StreamableHTTPServerParams --- src/mistralai/extra/mcp/streamable_http.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/mistralai/extra/mcp/streamable_http.py b/src/mistralai/extra/mcp/streamable_http.py index 94cd6cbd..e69953ba 100644 --- a/src/mistralai/extra/mcp/streamable_http.py +++ b/src/mistralai/extra/mcp/streamable_http.py @@ -22,7 +22,6 @@ class StreamableHTTPServerParams(BaseModel): url: str headers: dict[str, Any] | None = None timeout: float = 30 - sse_read_timeout: float = 60 * 5 class MCPClientStreamableHTTP(MCPClientBase): From bc06284d4f8ee7a62930118520b3be7683259905 Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Thu, 16 Jul 2026 18:54:33 +0200 Subject: [PATCH 3/7] test(mcp): cover Streamable HTTP client header wiring --- .../extra/tests/test_mcp_streamable_http.py | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 src/mistralai/extra/tests/test_mcp_streamable_http.py diff --git a/src/mistralai/extra/tests/test_mcp_streamable_http.py b/src/mistralai/extra/tests/test_mcp_streamable_http.py new file mode 100644 index 00000000..3485f34b --- /dev/null +++ b/src/mistralai/extra/tests/test_mcp_streamable_http.py @@ -0,0 +1,50 @@ +"""Tests for the Streamable HTTP MCP client.""" + +import unittest +from contextlib import AsyncExitStack, asynccontextmanager +from typing import Any +from unittest import mock + +from mistralai.extra.mcp.streamable_http import ( + MCPClientStreamableHTTP, + StreamableHTTPServerParams, +) + + +class TestMCPClientStreamableHTTP(unittest.IsolatedAsyncioTestCase): + async def test_headers_are_set_on_the_http_client(self) -> None: + """Credentials must live on the httpx client (the transport's own headers + argument is deprecated/ignored), so they are sent on every request.""" + captured: dict[str, Any] = {} + + @asynccontextmanager + async def fake_streamable_http_client(url: str, http_client: Any) -> Any: + captured["url"] = url + captured["headers"] = dict(http_client.headers) + yield object(), object(), lambda: None + + client = MCPClientStreamableHTTP( + params=StreamableHTTPServerParams( + url="http://mcp.example/mcp", + headers={"Authorization": "Bearer gate", "Notion-Token": "ntn_x"}, + ), + name="test", + ) + + with mock.patch( + "mistralai.extra.mcp.streamable_http.streamable_http_client", + fake_streamable_http_client, + ): + async with AsyncExitStack() as stack: + read_stream, write_stream = await client._get_transport(stack) + + self.assertIsNotNone(read_stream) + self.assertIsNotNone(write_stream) + self.assertEqual(captured["url"], "http://mcp.example/mcp") + # httpx normalizes header names to lower case + self.assertEqual(captured["headers"]["authorization"], "Bearer gate") + self.assertEqual(captured["headers"]["notion-token"], "ntn_x") + + +if __name__ == "__main__": + unittest.main() From 8ba57bc9c471a125bd9239d3d032a126b70664c1 Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Thu, 16 Jul 2026 19:07:00 +0200 Subject: [PATCH 4/7] test(mcp): fix asynccontextmanager return type for pyright --- src/mistralai/extra/tests/test_mcp_streamable_http.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/mistralai/extra/tests/test_mcp_streamable_http.py b/src/mistralai/extra/tests/test_mcp_streamable_http.py index 3485f34b..c58ac5d8 100644 --- a/src/mistralai/extra/tests/test_mcp_streamable_http.py +++ b/src/mistralai/extra/tests/test_mcp_streamable_http.py @@ -1,6 +1,7 @@ """Tests for the Streamable HTTP MCP client.""" import unittest +from collections.abc import AsyncIterator from contextlib import AsyncExitStack, asynccontextmanager from typing import Any from unittest import mock @@ -18,7 +19,7 @@ async def test_headers_are_set_on_the_http_client(self) -> None: captured: dict[str, Any] = {} @asynccontextmanager - async def fake_streamable_http_client(url: str, http_client: Any) -> Any: + async def fake_streamable_http_client(url: str, http_client: Any) -> AsyncIterator[Any]: captured["url"] = url captured["headers"] = dict(http_client.headers) yield object(), object(), lambda: None From 96c20e6e87f90f1e37cfea75b4957df306dfc2ea Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Fri, 17 Jul 2026 13:42:38 +0200 Subject: [PATCH 5/7] fix(mcp): do not inherit ambient HTTP proxy in Streamable HTTP client --- src/mistralai/extra/mcp/streamable_http.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/mistralai/extra/mcp/streamable_http.py b/src/mistralai/extra/mcp/streamable_http.py index e69953ba..83f98fde 100644 --- a/src/mistralai/extra/mcp/streamable_http.py +++ b/src/mistralai/extra/mcp/streamable_http.py @@ -50,11 +50,15 @@ async def _get_transport( MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage], ]: + # trust_env=False so the client does not inherit an ambient HTTP(S)_PROXY: + # the MCP endpoint is reached directly, and routing it through an egress + # proxy meant for external traffic can make the connection hang. http_client = await exit_stack.enter_async_context( httpx.AsyncClient( headers=self._params.headers, timeout=self._params.timeout, follow_redirects=True, + trust_env=False, ) ) read_stream, write_stream, _ = await exit_stack.enter_async_context( From 82a043790c42520c934db2e5b4ace15809394fa1 Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Fri, 17 Jul 2026 13:55:55 +0200 Subject: [PATCH 6/7] refactor(mcp): make trust_env configurable (default True) instead of hardcoded --- src/mistralai/extra/mcp/streamable_http.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/mistralai/extra/mcp/streamable_http.py b/src/mistralai/extra/mcp/streamable_http.py index 83f98fde..4da520ef 100644 --- a/src/mistralai/extra/mcp/streamable_http.py +++ b/src/mistralai/extra/mcp/streamable_http.py @@ -22,6 +22,11 @@ class StreamableHTTPServerParams(BaseModel): url: str headers: dict[str, Any] | None = None timeout: float = 30 + # Whether the httpx client trusts the ambient environment (HTTP(S)_PROXY, + # SSL_CERT_FILE/DIR, .netrc). Defaults to httpx's default (True). Set False to + # reach the endpoint directly without an ambient egress proxy, e.g. for an + # in-cluster URL on a host whose external egress is proxied. + trust_env: bool = True class MCPClientStreamableHTTP(MCPClientBase): @@ -50,15 +55,16 @@ async def _get_transport( MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage], ]: - # trust_env=False so the client does not inherit an ambient HTTP(S)_PROXY: - # the MCP endpoint is reached directly, and routing it through an egress - # proxy meant for external traffic can make the connection hang. + # trust_env controls whether the client inherits the ambient + # HTTP(S)_PROXY / cert / netrc env. Set it False (see params) to reach an + # in-cluster endpoint directly, bypassing a proxy meant for external + # traffic that would make the connection hang. http_client = await exit_stack.enter_async_context( httpx.AsyncClient( headers=self._params.headers, timeout=self._params.timeout, follow_redirects=True, - trust_env=False, + trust_env=self._params.trust_env, ) ) read_stream, write_stream, _ = await exit_stack.enter_async_context( From 6560fd27255a26229fda70b78b35a18127e250c0 Mon Sep 17 00:00:00 2001 From: Adrien Duchemin Date: Fri, 24 Jul 2026 00:59:50 +0200 Subject: [PATCH 7/7] feat(mcp): add follow_redirects (default False) to Streamable HTTP client On a redirect, httpx only strips Authorization on cross-origin hops, so the client's default headers (a per-request integration token passed via headers) would be resent verbatim to the redirect target, letting a compromised or open-redirecting server exfiltrate them. Default follow_redirects to False (also httpx's own default) and let callers opt in via StreamableHTTPServerParams when the server relies on redirects and every redirect host is trusted. --- src/mistralai/extra/mcp/streamable_http.py | 12 +++++++- .../extra/tests/test_mcp_streamable_http.py | 30 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/src/mistralai/extra/mcp/streamable_http.py b/src/mistralai/extra/mcp/streamable_http.py index 4da520ef..14bf0d2a 100644 --- a/src/mistralai/extra/mcp/streamable_http.py +++ b/src/mistralai/extra/mcp/streamable_http.py @@ -27,6 +27,13 @@ class StreamableHTTPServerParams(BaseModel): # reach the endpoint directly without an ambient egress proxy, e.g. for an # in-cluster URL on a host whose external egress is proxied. trust_env: bool = True + # Whether httpx follows HTTP redirects (3xx). Defaults to False (also httpx's + # default). On a redirect httpx only strips Authorization on cross-origin hops, + # so the ``headers`` above (e.g. a per-request integration token) would be resent + # verbatim to the redirect target, letting a compromised or open-redirecting + # server exfiltrate them. Set True only if the server relies on redirects and + # every host it can redirect to is trusted. + follow_redirects: bool = False class MCPClientStreamableHTTP(MCPClientBase): @@ -59,11 +66,14 @@ async def _get_transport( # HTTP(S)_PROXY / cert / netrc env. Set it False (see params) to reach an # in-cluster endpoint directly, bypassing a proxy meant for external # traffic that would make the connection hang. + # + # follow_redirects defaults False (see params): the secret default headers + # would otherwise be resent to a redirect target on same-origin hops. http_client = await exit_stack.enter_async_context( httpx.AsyncClient( headers=self._params.headers, timeout=self._params.timeout, - follow_redirects=True, + follow_redirects=self._params.follow_redirects, trust_env=self._params.trust_env, ) ) diff --git a/src/mistralai/extra/tests/test_mcp_streamable_http.py b/src/mistralai/extra/tests/test_mcp_streamable_http.py index c58ac5d8..eea80325 100644 --- a/src/mistralai/extra/tests/test_mcp_streamable_http.py +++ b/src/mistralai/extra/tests/test_mcp_streamable_http.py @@ -46,6 +46,36 @@ async def fake_streamable_http_client(url: str, http_client: Any) -> AsyncIterat self.assertEqual(captured["headers"]["authorization"], "Bearer gate") self.assertEqual(captured["headers"]["notion-token"], "ntn_x") + async def test_follow_redirects_defaults_false_and_is_configurable(self) -> None: + """Redirects are not followed by default (secret headers would be resent to + the redirect target); the flag is forwarded to the httpx client when set.""" + captured: dict[str, Any] = {} + + @asynccontextmanager + async def fake_streamable_http_client(url: str, http_client: Any) -> AsyncIterator[Any]: + captured["follow_redirects"] = http_client.follow_redirects + yield object(), object(), lambda: None + + with mock.patch( + "mistralai.extra.mcp.streamable_http.streamable_http_client", + fake_streamable_http_client, + ): + default_client = MCPClientStreamableHTTP( + params=StreamableHTTPServerParams(url="http://mcp.example/mcp"), + name="test", + ) + async with AsyncExitStack() as stack: + await default_client._get_transport(stack) + self.assertFalse(captured["follow_redirects"]) + + opted_in_client = MCPClientStreamableHTTP( + params=StreamableHTTPServerParams(url="http://mcp.example/mcp", follow_redirects=True), + name="test", + ) + async with AsyncExitStack() as stack: + await opted_in_client._get_transport(stack) + self.assertTrue(captured["follow_redirects"]) + if __name__ == "__main__": unittest.main()