Add verify parameter for SSL certificate configuration (#3487)

* feat: add `verify` parameter for SSL certificate configuration

* Propagate verify to OAuth preflight clients

* Propagate verify to pre-constructed OAuth instances

* Fix verify override not propagating to existing OAuth factory

* Warn when both httpx_client_factory and verify are provided

* Preserve user-provided OAuth factory when transport has verify

* Skip OAuth re-sync when transport has custom httpx_client_factory

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.

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)
 

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,

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

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,
+            )