Support SSL certificate verification option when creating httpx AsyncClient

src/mcp/client/sse.py

@@ -55,7 +55,9 @@ async def sse_client(
         try:
             logger.debug(f"Connecting to SSE endpoint: {remove_request_params(url)}")
             async with httpx_client_factory(
-                headers=headers, auth=auth, timeout=httpx.Timeout(timeout, read=sse_read_timeout)
+                headers=headers,
+                timeout=httpx.Timeout(timeout, read=sse_read_timeout),
+                auth=auth,
             ) as client:
                 async with aconnect_sse(
                     client,

src/mcp/shared/_httpx_utils.py

@@ -8,30 +8,20 @@
 
 
 class McpHttpClientFactory(Protocol):
-    def __call__(
-        self,
-        headers: dict[str, str] | None = None,
-        timeout: httpx.Timeout | None = None,
-        auth: httpx.Auth | None = None,
-    ) -> httpx.AsyncClient: ...
-
-
-def create_mcp_http_client(
-    headers: dict[str, str] | None = None,
-    timeout: httpx.Timeout | None = None,
-    auth: httpx.Auth | None = None,
-) -> httpx.AsyncClient:
+    def __call__(self, **kwargs: Any) -> httpx.AsyncClient: ...
+
+
+def create_mcp_http_client(**kwargs: Any) -> httpx.AsyncClient:
     """Create a standardized httpx AsyncClient with MCP defaults.
 
     This function provides common defaults used throughout the MCP codebase:
     - follow_redirects=True (always enabled)
     - Default timeout of 30 seconds if not specified
+    - You can pass any keyword argument accepted by httpx.AsyncClient
 
     Args:
-        headers: Optional headers to include with all requests.
-        timeout: Request timeout as httpx.Timeout object.
-            Defaults to 30 seconds if not specified.
-        auth: Optional authentication handler.
+        Any keyword argument supported by httpx.AsyncClient (e.g. headers, timeout, auth, verify, proxies, etc).
+        MCP defaults are applied unless overridden.
 
     Returns:
         Configured httpx.AsyncClient instance with MCP defaults.
@@ -47,37 +37,38 @@ def create_mcp_http_client(
 
         # With custom headers
         headers = {"Authorization": "Bearer token"}
-        async with create_mcp_http_client(headers) as client:
+        async with create_mcp_http_client(headers=headers) as client:
             response = await client.get("/endpoint")
 
         # With both custom headers and timeout
         timeout = httpx.Timeout(60.0, read=300.0)
-        async with create_mcp_http_client(headers, timeout) as client:
+        async with create_mcp_http_client(headers=headers, timeout=timeout) as client:
             response = await client.get("/long-request")
 
         # With authentication
         from httpx import BasicAuth
         auth = BasicAuth(username="user", password="pass")
-        async with create_mcp_http_client(headers, timeout, auth) as client:
+        async with create_mcp_http_client(headers=headers, timeout=timeout, auth=auth) as client:
             response = await client.get("/protected-endpoint")
+
+        # With SSL verification disabled
+        async with create_mcp_http_client(verify=False) as client:
+            response = await client.get("/insecure-endpoint")
+
+        # With custom SSL context
+        import ssl
+        ssl_ctx = ssl.create_default_context()
+        async with create_mcp_http_client(verify=ssl_ctx) as client:
+            response = await client.get("/custom-endpoint")
+
+        # With proxies and base_url
+        async with create_mcp_http_client(proxies="http://proxy:8080", base_url="https://api.example.com") as client:
+            response = await client.get("/resource")
     """
     # Set MCP defaults
-    kwargs: dict[str, Any] = {
+    default_kwargs: dict[str, Any] = {
         "follow_redirects": True,
+        "timeout": httpx.Timeout(30.0),
     }
-
-    # Handle timeout
-    if timeout is None:
-        kwargs["timeout"] = httpx.Timeout(30.0)
-    else:
-        kwargs["timeout"] = timeout
-
-    # Handle headers
-    if headers is not None:
-        kwargs["headers"] = headers
-
-    # Handle authentication
-    if auth is not None:
-        kwargs["auth"] = auth
-
-    return httpx.AsyncClient(**kwargs)
+    default_kwargs.update(kwargs)
+    return httpx.AsyncClient(**default_kwargs)

tests/shared/test_httpx_utils.py

@@ -18,7 +18,7 @@ def test_custom_parameters():
     headers = {"Authorization": "Bearer token"}
     timeout = httpx.Timeout(60.0)
 
-    client = create_mcp_http_client(headers, timeout)
+    client = create_mcp_http_client(headers=headers, timeout=timeout)
 
     assert client.headers["Authorization"] == "Bearer token"
     assert client.timeout.connect == 60.0