1  """
  2  SSL/TLS configuration support for MCP connections.
  3  
  4  This module provides SSL configuration for remote MCP connections (SSE and Streamable HTTP)
  5  to allow connecting to MCP servers with self-signed certificates or custom CA bundles.
  6  """
  7  
  8  import logging
  9  import os
 10  from dataclasses import dataclass
 11  
 12  import httpx
 13  
 14  log = logging.getLogger(__name__)
 15  
 16  
 17  @dataclass
 18  class SslConfig:
 19      """SSL configuration for MCP connections.
 20  
 21      Attributes:
 22          verify: Whether to verify SSL certificates. Defaults to True.
 23                  Set to False to disable SSL verification (development only).
 24          ca_bundle: Optional path to a custom CA certificate bundle file.
 25                     When provided, this takes precedence over the verify setting.
 26      """
 27  
 28      verify: bool = True
 29      ca_bundle: str | None = None
 30  
 31      def __post_init__(self):
 32          """Validate the SSL configuration after initialization."""
 33          if self.ca_bundle is not None and not os.path.isfile(self.ca_bundle):
 34              raise ValueError(
 35                  f"SSL ca_bundle path does not exist or is not a file: {self.ca_bundle}"
 36              )
 37  
 38  
 39  def create_ssl_httpx_client_factory(ssl_config: SslConfig):
 40      """Create an httpx client factory with SSL configuration.
 41  
 42      This factory creates httpx.AsyncClient instances with custom SSL settings,
 43      following the MCP library's expected factory signature.
 44  
 45      Args:
 46          ssl_config: SSL configuration specifying verification settings.
 47  
 48      Returns:
 49          A factory function that creates configured httpx.AsyncClient instances.
 50      """
 51      # MCP default timeouts
 52      MCP_DEFAULT_TIMEOUT = 30.0
 53      MCP_DEFAULT_SSE_READ_TIMEOUT = 300.0
 54  
 55      def factory(
 56          headers: dict[str, str] | None = None,
 57          timeout: httpx.Timeout | None = None,
 58          auth: httpx.Auth | None = None,
 59      ) -> httpx.AsyncClient:
 60          """Create an httpx.AsyncClient with SSL configuration.
 61  
 62          Args:
 63              headers: Optional headers to include with all requests.
 64              timeout: Request timeout as httpx.Timeout object.
 65              auth: Optional authentication handler.
 66  
 67          Returns:
 68              Configured httpx.AsyncClient instance.
 69          """
 70          # Determine SSL verification setting
 71          # ca_bundle takes precedence if provided
 72          verify: bool | str = (
 73              ssl_config.ca_bundle if ssl_config.ca_bundle else ssl_config.verify
 74          )
 75  
 76          # Build kwargs following MCP defaults
 77          kwargs: dict = {
 78              "follow_redirects": True,
 79              "verify": verify,
 80          }
 81  
 82          # Handle timeout
 83          if timeout is None:
 84              kwargs["timeout"] = httpx.Timeout(
 85                  MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT
 86              )
 87          else:
 88              kwargs["timeout"] = timeout
 89  
 90          # Handle headers
 91          if headers is not None:
 92              kwargs["headers"] = headers
 93  
 94          # Handle authentication
 95          if auth is not None:
 96              kwargs["auth"] = auth
 97  
 98          return httpx.AsyncClient(**kwargs)
 99  
100      return factory