diff --git a/docs/clients/transports.mdx b/docs/clients/transports.mdx index 7e3eb03db..efcda3366 100644 --- a/docs/clients/transports.mdx +++ b/docs/clients/transports.mdx @@ -124,6 +124,38 @@ client = Client( ) ``` +### SSL Verification + +By default, HTTPS connections verify the server's SSL certificate. You can customize this behavior with the `verify` parameter, which accepts the same values as [httpx](https://www.python-httpx.org/advanced/ssl/): + +```python +from fastmcp import Client + +# Disable SSL verification (e.g., for self-signed certs in development) +client = Client("https://dev-server.internal/mcp", verify=False) + +# Use a custom CA bundle +client = Client("https://corp-server.internal/mcp", verify="/path/to/ca-bundle.pem") + +# Use a custom SSL context for full control +import ssl +ctx = ssl.create_default_context() +ctx.load_verify_locations("/path/to/internal-ca.pem") +client = Client("https://corp-server.internal/mcp", verify=ctx) +``` + +The `verify` parameter is also available directly on `StreamableHttpTransport` and `SSETransport`: + +```python +from fastmcp.client.transports import StreamableHttpTransport + +transport = StreamableHttpTransport( + url="https://dev-server.internal/mcp", + verify=False, +) +client = Client(transport) +``` + ### SSE Transport Server-Sent Events transport is maintained for backward compatibility. Use Streamable HTTP for new deployments unless you have specific infrastructure requirements. diff --git a/src/fastmcp/client/client.py b/src/fastmcp/client/client.py index efacf671e..94cbbe169 100644 --- a/src/fastmcp/client/client.py +++ b/src/fastmcp/client/client.py @@ -4,6 +4,7 @@ import asyncio import copy import datetime import secrets +import ssl import weakref from collections.abc import Coroutine from contextlib import AsyncExitStack, asynccontextmanager, suppress @@ -20,6 +21,7 @@ from mcp.types import GetTaskResult, TaskStatusNotification from pydantic import AnyUrl import fastmcp +from fastmcp.client.auth.oauth import OAuth from fastmcp.client.elicitation import ElicitationHandler, create_elicitation_callback from fastmcp.client.logging import ( LogHandler, @@ -258,10 +260,34 @@ class Client( init_timeout: datetime.timedelta | float | int | None = None, client_info: mcp.types.Implementation | None = None, auth: httpx.Auth | Literal["oauth"] | str | None = None, + verify: ssl.SSLContext | bool | str | None = None, ) -> None: self.name = name or self.generate_name() self.transport = cast(ClientTransportT, infer_transport(transport)) + + if verify is not None: + from fastmcp.client.transports.http import StreamableHttpTransport + from fastmcp.client.transports.sse import SSETransport + + if isinstance(self.transport, StreamableHttpTransport | SSETransport): + self.transport.verify = verify + # Re-sync existing OAuth auth with the new verify setting, + # but only if the transport doesn't have a custom factory + # (which takes precedence and was already applied to OAuth). + if ( + isinstance(self.transport.auth, OAuth) + and auth is None + and self.transport.httpx_client_factory is None + ): + verify_factory = self.transport._make_verify_factory() + if verify_factory is not None: + self.transport.auth.httpx_client_factory = verify_factory + else: + raise ValueError( + "The 'verify' parameter is only supported for HTTP transports." + ) + if auth is not None: self.transport._set_auth(auth) diff --git a/src/fastmcp/client/transports/http.py b/src/fastmcp/client/transports/http.py index f7e74ca60..e4532c203 100644 --- a/src/fastmcp/client/transports/http.py +++ b/src/fastmcp/client/transports/http.py @@ -4,8 +4,9 @@ from __future__ import annotations import contextlib import datetime +import ssl from collections.abc import AsyncIterator, Callable -from typing import Literal, cast +from typing import Any, Literal, cast import httpx from mcp import ClientSession @@ -32,6 +33,7 @@ class StreamableHttpTransport(ClientTransport): auth: httpx.Auth | Literal["oauth"] | str | None = None, sse_read_timeout: datetime.timedelta | float | int | None = None, httpx_client_factory: McpHttpClientFactory | None = None, + verify: ssl.SSLContext | bool | str | None = None, ): """Initialize a Streamable HTTP transport. @@ -45,6 +47,10 @@ class StreamableHttpTransport(ClientTransport): If provided, must accept keyword arguments: headers, auth, follow_redirects, and optionally timeout. Using **kwargs is recommended to ensure forward compatibility. + verify: SSL certificate verification. Accepts False to disable + verification, a path to a CA bundle, or an ssl.SSLContext + for full control. None (default) uses httpx defaults (verification + enabled). Ignored when httpx_client_factory is provided. """ if isinstance(url, AnyUrl): url = str(url) @@ -57,6 +63,20 @@ class StreamableHttpTransport(ClientTransport): self.url: str = url self.headers = headers or {} self.httpx_client_factory = httpx_client_factory + self.verify: ssl.SSLContext | bool | str | None = verify + + if httpx_client_factory is not None and verify is not None: + import warnings + + warnings.warn( + "Both 'httpx_client_factory' and 'verify' were provided. " + "The 'verify' parameter will be ignored because " + "'httpx_client_factory' takes precedence. Configure SSL " + "verification directly in your httpx_client_factory instead.", + UserWarning, + stacklevel=2, + ) + self._set_auth(auth) if sse_read_timeout is not None: @@ -78,9 +98,19 @@ class StreamableHttpTransport(ClientTransport): def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): resolved: httpx.Auth | None if auth == "oauth": - resolved = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) + resolved = OAuth( + self.url, + httpx_client_factory=self.httpx_client_factory + or self._make_verify_factory(), + ) elif isinstance(auth, OAuth): auth._bind(self.url) + # Only inject the transport's factory into OAuth if OAuth still + # has the bare default — preserve any factory the caller attached + if auth.httpx_client_factory is httpx.AsyncClient: + factory = self.httpx_client_factory or self._make_verify_factory() + if factory is not None: + auth.httpx_client_factory = factory resolved = auth elif isinstance(auth, str): resolved = BearerAuth(auth) @@ -88,6 +118,31 @@ class StreamableHttpTransport(ClientTransport): resolved = auth self.auth: httpx.Auth | None = resolved + def _make_verify_factory(self) -> McpHttpClientFactory | None: + if self.verify is None: + return None + verify = self.verify + + def factory( + headers: dict[str, str] | None = None, + timeout: httpx.Timeout | None = None, + auth: httpx.Auth | None = None, + ) -> httpx.AsyncClient: + if timeout is None: + timeout = httpx.Timeout(30.0, read=300.0) + kwargs: dict[str, Any] = { + "follow_redirects": True, + "timeout": timeout, + "verify": verify, + } + if headers is not None: + kwargs["headers"] = headers + if auth is not None: + kwargs["auth"] = auth + return httpx.AsyncClient(**kwargs) + + return cast(McpHttpClientFactory, factory) + @contextlib.asynccontextmanager async def connect_session( self, **session_kwargs: Unpack[SessionKwargs] @@ -108,6 +163,7 @@ class StreamableHttpTransport(ClientTransport): # Create httpx client from factory or use default with MCP-appropriate timeouts # create_mcp_http_client uses 30s connect/5min read timeout by default, # and always enables follow_redirects + verify_factory = self._make_verify_factory() if self.httpx_client_factory is not None: # Factory clients get the full kwargs for backwards compatibility http_client = self.httpx_client_factory( @@ -116,6 +172,12 @@ class StreamableHttpTransport(ClientTransport): follow_redirects=True, # type: ignore[call-arg] **({"timeout": timeout} if timeout else {}), ) + elif verify_factory is not None: + http_client = verify_factory( + headers=headers, + timeout=timeout, + auth=self.auth, + ) else: http_client = create_mcp_http_client( headers=headers, diff --git a/src/fastmcp/client/transports/sse.py b/src/fastmcp/client/transports/sse.py index 36fa0ebc0..fa900eb7b 100644 --- a/src/fastmcp/client/transports/sse.py +++ b/src/fastmcp/client/transports/sse.py @@ -4,6 +4,7 @@ from __future__ import annotations import contextlib import datetime +import ssl from collections.abc import AsyncIterator from typing import Any, Literal, cast @@ -31,6 +32,7 @@ class SSETransport(ClientTransport): auth: httpx.Auth | Literal["oauth"] | str | None = None, sse_read_timeout: datetime.timedelta | float | int | None = None, httpx_client_factory: McpHttpClientFactory | None = None, + verify: ssl.SSLContext | bool | str | None = None, ): if isinstance(url, AnyUrl): url = str(url) @@ -43,6 +45,20 @@ class SSETransport(ClientTransport): self.url: str = url self.headers = headers or {} self.httpx_client_factory = httpx_client_factory + self.verify: ssl.SSLContext | bool | str | None = verify + + if httpx_client_factory is not None and verify is not None: + import warnings + + warnings.warn( + "Both 'httpx_client_factory' and 'verify' were provided. " + "The 'verify' parameter will be ignored because " + "'httpx_client_factory' takes precedence. Configure SSL " + "verification directly in your httpx_client_factory instead.", + UserWarning, + stacklevel=2, + ) + self._set_auth(auth) self.sse_read_timeout = normalize_timeout_to_timedelta(sse_read_timeout) @@ -50,9 +66,19 @@ class SSETransport(ClientTransport): def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): resolved: httpx.Auth | None if auth == "oauth": - resolved = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) + resolved = OAuth( + self.url, + httpx_client_factory=self.httpx_client_factory + or self._make_verify_factory(), + ) elif isinstance(auth, OAuth): auth._bind(self.url) + # Only inject the transport's factory into OAuth if OAuth still + # has the bare default — preserve any factory the caller attached + if auth.httpx_client_factory is httpx.AsyncClient: + factory = self.httpx_client_factory or self._make_verify_factory() + if factory is not None: + auth.httpx_client_factory = factory resolved = auth elif isinstance(auth, str): resolved = BearerAuth(auth) @@ -60,6 +86,31 @@ class SSETransport(ClientTransport): resolved = auth self.auth: httpx.Auth | None = resolved + def _make_verify_factory(self) -> McpHttpClientFactory | None: + if self.verify is None: + return None + verify = self.verify + + def factory( + headers: dict[str, str] | None = None, + timeout: httpx.Timeout | None = None, + auth: httpx.Auth | None = None, + ) -> httpx.AsyncClient: + if timeout is None: + timeout = httpx.Timeout(30.0, read=300.0) + kwargs: dict[str, Any] = { + "follow_redirects": True, + "timeout": timeout, + "verify": verify, + } + if headers is not None: + kwargs["headers"] = headers + if auth is not None: + kwargs["auth"] = auth + return httpx.AsyncClient(**kwargs) + + return cast(McpHttpClientFactory, factory) + @contextlib.asynccontextmanager async def connect_session( self, **session_kwargs: Unpack[SessionKwargs] @@ -85,6 +136,10 @@ class SSETransport(ClientTransport): if self.httpx_client_factory is not None: client_kwargs["httpx_client_factory"] = self.httpx_client_factory + else: + verify_factory = self._make_verify_factory() + if verify_factory is not None: + client_kwargs["httpx_client_factory"] = verify_factory async with sse_client(self.url, auth=self.auth, **client_kwargs) as transport: read_stream, write_stream = transport diff --git a/tests/client/transports/test_transports.py b/tests/client/transports/test_transports.py index acf6001cb..1481aca25 100644 --- a/tests/client/transports/test_transports.py +++ b/tests/client/transports/test_transports.py @@ -1,7 +1,12 @@ +import ssl from ssl import VerifyMode +from typing import cast import httpx +import pytest +from mcp.shared._httpx_utils import McpHttpClientFactory +from fastmcp import Client from fastmcp.client.auth.oauth import OAuth from fastmcp.client.transports import SSETransport, StreamableHttpTransport @@ -40,3 +45,219 @@ async def test_oauth_uses_same_client_as_transport_sse(): httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] == VerifyMode.CERT_NONE ) + + +class TestSSLVerify: + def test_streamable_http_transport_stores_verify_false(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=False, + ) + assert transport.verify is False + + def test_streamable_http_transport_stores_verify_ssl_context(self): + ctx = ssl.create_default_context() + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=ctx, + ) + assert transport.verify is ctx + + def test_streamable_http_transport_stores_verify_cert_path(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify="/path/to/cert.pem", + ) + assert transport.verify == "/path/to/cert.pem" + + def test_streamable_http_transport_verify_default_is_none(self): + transport = StreamableHttpTransport("https://example.com/mcp") + assert transport.verify is None + + def test_sse_transport_stores_verify_false(self): + transport = SSETransport( + "https://example.com/sse", + verify=False, + ) + assert transport.verify is False + + def test_sse_transport_stores_verify_ssl_context(self): + ctx = ssl.create_default_context() + transport = SSETransport( + "https://example.com/sse", + verify=ctx, + ) + assert transport.verify is ctx + + def test_sse_transport_verify_default_is_none(self): + transport = SSETransport("https://example.com/sse") + assert transport.verify is None + + def test_client_passes_verify_to_streamable_http_transport(self): + client = Client("https://example.com/mcp", verify=False) + assert isinstance(client.transport, StreamableHttpTransport) + assert client.transport.verify is False + + def test_client_passes_verify_ssl_context_to_transport(self): + ctx = ssl.create_default_context() + client = Client("https://example.com/mcp", verify=ctx) + assert isinstance(client.transport, StreamableHttpTransport) + assert client.transport.verify is ctx + + def test_client_passes_verify_cert_path_to_transport(self): + client = Client( + "https://example.com/mcp", + verify="/path/to/cert.pem", + ) + assert isinstance(client.transport, StreamableHttpTransport) + assert client.transport.verify == "/path/to/cert.pem" + + def test_client_verify_none_leaves_transport_default(self): + client = Client("https://example.com/mcp") + assert isinstance(client.transport, StreamableHttpTransport) + assert client.transport.verify is None + + def test_client_verify_raises_for_non_http_transport(self): + from fastmcp import FastMCP + + server = FastMCP("test") + with pytest.raises( + ValueError, + match="only supported for HTTP transports", + ): + Client(server, verify=False) + + def test_client_passes_verify_to_sse_transport(self): + client = Client("https://example.com/sse", verify=False) + assert isinstance(client.transport, SSETransport) + assert client.transport.verify is False + + async def test_streamable_http_verify_propagates_to_oauth(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=False, + auth="oauth", + ) + assert isinstance(transport.auth, OAuth) + async with transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + == VerifyMode.CERT_NONE + ) + + async def test_sse_verify_propagates_to_oauth(self): + transport = SSETransport( + "https://example.com/sse", + verify=False, + auth="oauth", + ) + assert isinstance(transport.auth, OAuth) + async with transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + == VerifyMode.CERT_NONE + ) + + async def test_client_verify_propagates_to_oauth(self): + client = Client( + "https://example.com/mcp", + verify=False, + auth="oauth", + ) + assert isinstance(client.transport, StreamableHttpTransport) + assert isinstance(client.transport.auth, OAuth) + async with client.transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + == VerifyMode.CERT_NONE + ) + + async def test_verify_propagates_to_preconstructed_oauth_instance(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=False, + auth=OAuth(), + ) + assert isinstance(transport.auth, OAuth) + async with transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + == VerifyMode.CERT_NONE + ) + + async def test_client_verify_resyncs_existing_oauth_on_transport(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + auth="oauth", + ) + assert isinstance(transport.auth, OAuth) + # OAuth was created without verify — factory should be default + async with transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + != VerifyMode.CERT_NONE + ) + + # Now wrap in Client with verify=False — should resync OAuth + client = Client(transport, verify=False) + assert isinstance(client.transport.auth, OAuth) + async with client.transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode + == VerifyMode.CERT_NONE + ) + + async def test_client_verify_overrides_transport_verify_in_oauth(self): + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=False, + auth="oauth", + ) + assert isinstance(transport.auth, OAuth) + # OAuth should initially have verify=False + async with transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode # type: ignore[attr-defined] + == VerifyMode.CERT_NONE + ) + + # Client overrides verify to True — OAuth should update + client = Client(transport, verify=True) + assert isinstance(client.transport.auth, OAuth) + async with client.transport.auth.httpx_client_factory() as httpx_client: + assert ( + httpx_client._transport._pool._ssl_context.verify_mode + != VerifyMode.CERT_NONE + ) + + async def test_oauth_custom_factory_preserved_with_verify(self): + custom_factory = cast( + McpHttpClientFactory, + lambda **kwargs: httpx.AsyncClient(verify=False, **kwargs), + ) + auth = OAuth(httpx_client_factory=custom_factory) + transport = StreamableHttpTransport( + "https://example.com/mcp", + verify=True, + auth=auth, + ) + assert isinstance(transport.auth, OAuth) + assert transport.auth.httpx_client_factory is custom_factory + + def test_warns_when_both_factory_and_verify_provided_streamable(self): + factory = cast(McpHttpClientFactory, httpx.AsyncClient) + with pytest.warns(UserWarning, match="httpx_client_factory.*takes precedence"): + StreamableHttpTransport( + "https://example.com/mcp", + httpx_client_factory=factory, + verify=False, + ) + + def test_warns_when_both_factory_and_verify_provided_sse(self): + factory = cast(McpHttpClientFactory, httpx.AsyncClient) + with pytest.warns(UserWarning, match="httpx_client_factory.*takes precedence"): + SSETransport( + "https://example.com/sse", + httpx_client_factory=factory, + verify=False, + )