Add Streamable HTTP protocol support and support for self-signed certs

README.md

@@ -129,6 +129,84 @@ If you want to use the tools inside of [claude desktop](https://claude.ai/downlo
 
 To add new MCP servers, edit the config.json file.
 
+### MCP Server Types
+
+MCP-Bridge supports three transport types for MCP servers.
+
+**stdio** — runs a local command:
+```json
+"fetch": {
+    "command": "uvx",
+    "args": ["mcp-server-fetch"]
+}
+```
+
+**SSE** — connects to a remote server over Server-Sent Events:
+```json
+"my-server": {
+    "url": "http://example.com/sse"
+}
+```
+
+**Streamable HTTP** — connects to a remote server using the newer Streamable HTTP transport (e.g. Home Assistant). Requires `"transport": "streamable_http"`:
+```json
+"homeassistant": {
+    "url": "http://homeassistant.local:8123/api/mcp",
+    "transport": "streamable_http"
+}
+```
+
+Both SSE and Streamable HTTP support a `headers` field for passing authentication:
+```json
+"homeassistant": {
+    "url": "https://homeassistant.local:8123/api/mcp",
+    "headers": {
+        "Authorization": "Bearer <your-long-lived-access-token>"
+    },
+    "transport": "streamable_http"
+}
+```
+
+#### Self-signed certificates (Docker)
+
+If your MCP server uses a self-signed or private CA certificate, the Docker container won't trust it by default. The recommended fix is to mount your host's CA bundle (which should already include your custom cert after running `update-ca-certificates`) and point Python's SSL to it:
+
+**1. Add the volume mount and environment variable to `compose.yml`:**
+```yaml
+services:
+  mcp-bridge:
+    environment:
+      - SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
+    volumes:
+      - /etc/ssl/certs/ca-certificates.crt:/etc/ssl/certs/ca-certificates.crt:ro
+```
+
+**2. Use `ssl_verify` in `config.json` to explicitly point to the bundle:**
+```json
+"homeassistant": {
+    "url": "https://homeassistant.local:8123/api/mcp",
+    "headers": {
+        "Authorization": "Bearer <your-long-lived-access-token>"
+    },
+    "ssl_verify": "/etc/ssl/certs/ca-certificates.crt",
+    "transport": "streamable_http"
+}
+```
+
+Make sure your CA certificate is trusted on the host first:
+```bash
+sudo cp /path/to/MyCA.crt /usr/local/share/ca-certificates/
+sudo update-ca-certificates
+```
+
+### Docker DNS issues
+If your tools or MCP servers (mainly ones running in docker) are having name resolution issues, you may need to hardcode DNS servers in your /etc/docker/daemon.json and restart docker.service. for example:
+```json
+{
+    "dns": ["192.168.2.2", "1.1.1.1"],
+}
+```
+
 ### API Key Authentication
 
 MCP-Bridge supports API key authentication to secure your server. To enable this feature, add something like this to your `config.json` file:

mcp_bridge/config/final.py

@@ -39,12 +39,21 @@ class Sampling(BaseModel):
 
 
 class SSEMCPServer(BaseModel):
-    # TODO: expand this once I find a good definition for this
     url: str = Field(description="URL of the MCP server")
+    headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send with requests")
+    ssl_verify: bool | str = Field(default=True, description="SSL verification: True, False, or path to a CA bundle file")
+    transport: Literal["sse"] = Field(default="sse")
+
+
+class StreamableHTTPMCPServer(BaseModel):
+    url: str = Field(description="URL of the MCP server")
+    headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send with requests")
+    ssl_verify: bool | str = Field(default=True, description="SSL verification: True, False, or path to a CA bundle file")
+    transport: Literal["streamable_http"]
 
 
 MCPServer = Annotated[
-    Union[StdioServerParameters, SSEMCPServer, DockerMCPServer],
+    Union[StdioServerParameters, SSEMCPServer, StreamableHTTPMCPServer, DockerMCPServer],
     Field(description="MCP server configuration"),
 ]
 

mcp_bridge/mcp_clients/McpClientManager.py

@@ -5,13 +5,14 @@
 from mcpx.client.transports.docker import DockerMCPServer
 
 from mcp_bridge.config import config
-from mcp_bridge.config.final import SSEMCPServer
+from mcp_bridge.config.final import SSEMCPServer, StreamableHTTPMCPServer
 
 from .DockerClient import DockerClient
 from .SseClient import SseClient
+from .StreamableHttpClient import StreamableHttpClient
 from .StdioClient import StdioClient
 
-client_types = Union[StdioClient, SseClient, DockerClient]
+client_types = Union[StdioClient, SseClient, StreamableHttpClient, DockerClient]
 
 
 class MCPClientManager:
@@ -36,11 +37,15 @@ async def construct_client(self, name, server_config) -> client_types:
             return client
 
         if isinstance(server_config, SSEMCPServer):
-            # TODO: implement sse client
             client = SseClient(name, server_config)  # type: ignore
             await client.start()
             return client
-
+
+        if isinstance(server_config, StreamableHTTPMCPServer):
+            client = StreamableHttpClient(name, server_config)  # type: ignore
+            await client.start()
+            return client
+
         if isinstance(server_config, DockerMCPServer):
             client = DockerClient(name, server_config)
             await client.start()

mcp_bridge/mcp_clients/SseClient.py

@@ -1,4 +1,5 @@
 import asyncio
+import httpx
 from mcp.client.sse import sse_client
 from mcp_bridge.config import config
 from mcp_bridge.config.final import SSEMCPServer
@@ -16,22 +17,41 @@ def __init__(self, name: str, config: SSEMCPServer) -> None:
         self.config = config
 
     async def _maintain_session(self):
-        async with sse_client(self.config.url) as client:
-            async with McpClientSession(*client) as session:
-                await session.initialize()
-                logger.debug(f"finished initialise session for {self.name}")
-                self.session = session
-
-                try:
-                    while True:
-                        await asyncio.sleep(10)
-                        if config.logging.log_server_pings:
-                            logger.debug(f"pinging session for {self.name}")
-
-                        await session.send_ping()
-
-                except Exception as exc:
-                    logger.error(f"ping failed for {self.name}: {exc}")
-                    self.session = None
+        ssl_verify = self.config.ssl_verify
+
+        if ssl_verify is not True:
+            # sse_client doesn't expose ssl options, so briefly patch httpx.AsyncClient
+            # to inject the verify setting before it creates its internal client.
+            _original_init = httpx.AsyncClient.__init__
+
+            def _patched_init(client_self, *args, **kwargs):
+                kwargs.setdefault("verify", ssl_verify)
+                _original_init(client_self, *args, **kwargs)
+
+            httpx.AsyncClient.__init__ = _patched_init
+
+        try:
+            async with sse_client(
+                self.config.url, headers=self.config.headers or None
+            ) as client:
+                async with McpClientSession(*client) as session:
+                    await session.initialize()
+                    logger.debug(f"finished initialise session for {self.name}")
+                    self.session = session
+
+                    try:
+                        while True:
+                            await asyncio.sleep(10)
+                            if config.logging.log_server_pings:
+                                logger.debug(f"pinging session for {self.name}")
+
+                            await session.send_ping()
+
+                    except Exception as exc:
+                        logger.error(f"ping failed for {self.name}: {exc}")
+                        self.session = None
+        finally:
+            if ssl_verify is not True:
+                httpx.AsyncClient.__init__ = _original_init
 
         logger.debug(f"exiting session for {self.name}")

mcp_bridge/mcp_clients/StreamableHttpClient.py

@@ -0,0 +1,56 @@
+import asyncio
+import httpx
+from mcp.client.streamable_http import streamablehttp_client
+from mcp_bridge.config import config
+from mcp_bridge.config.final import StreamableHTTPMCPServer
+from mcp_bridge.mcp_clients.session import McpClientSession
+from .AbstractClient import GenericMcpClient
+from loguru import logger
+
+
+class StreamableHttpClient(GenericMcpClient):
+    config: StreamableHTTPMCPServer
+
+    def __init__(self, name: str, config: StreamableHTTPMCPServer) -> None:
+        super().__init__(name=name)
+        self.config = config
+
+    async def _maintain_session(self):
+        ssl_verify = self.config.ssl_verify
+
+        if ssl_verify is not True:
+            # streamablehttp_client doesn't expose ssl options, so briefly patch
+            # httpx.AsyncClient to inject the verify setting.
+            _original_init = httpx.AsyncClient.__init__
+            def _patched_init(client_self, *args, **kwargs):
+                kwargs.setdefault("verify", ssl_verify)
+                _original_init(client_self, *args, **kwargs)
+            httpx.AsyncClient.__init__ = _patched_init
+
+        try:
+            async with streamablehttp_client(
+                self.config.url,
+                headers=self.config.headers or None,
+            ) as streams:
+                # SDK may yield (read, write) or (read, write, session_id_cb)
+                read_stream, write_stream = streams[0], streams[1]
+                async with McpClientSession(read_stream, write_stream) as session:
+                    await session.initialize()
+                    logger.debug(f"finished initialise session for {self.name}")
+                    self.session = session
+
+                    try:
+                        while True:
+                            await asyncio.sleep(10)
+                            if config.logging.log_server_pings:
+                                logger.debug(f"pinging session for {self.name}")
+                            await session.send_ping()
+
+                    except Exception as exc:
+                        logger.error(f"ping failed for {self.name}: {exc}")
+                        self.session = None
+        finally:
+            if ssl_verify is not True:
+                httpx.AsyncClient.__init__ = _original_init
+
+        logger.debug(f"exiting session for {self.name}")

mcp_bridge/mcp_clients/session.py

@@ -40,30 +40,11 @@ def __init__(
             read_timeout_seconds=read_timeout_seconds,
         )
 
-    async def __aenter__(self):
-        session = await super().__aenter__()
-        self._task_group.start_soon(self._consume_messages)
-        return session
-
-    async def _consume_messages(self):
-        try:
-            async for message in self.incoming_messages:
-                try:
-                    if isinstance(message, Exception):
-                        logger.error(f"Received exception in message stream: {message}")
-                    elif isinstance(message, RequestResponder):                        
-                        logger.debug(f"Received request: {message.request}")                        
-                    elif isinstance(message, types.ServerNotification):
-                        if isinstance(message.root, types.LoggingMessageNotification):
-                            logger.debug(f"Received notification from server: {message.root.params}")                        
-                        else:
-                            logger.debug(f"Received notification from server: {message}")                        
-                    else:
-                        logger.debug(f"Received notification: {message}")
-                except Exception as e:
-                    logger.exception(f"Error processing message: {e}")
-        except Exception as e:
-            logger.exception(f"Message consumer task failed: {e}")
+    async def _received_notification(self, notification: types.ServerNotification) -> None:
+        if isinstance(notification.root, types.LoggingMessageNotification):
+            logger.debug(f"Received notification from server: {notification.root.params}")
+        else:
+            logger.debug(f"Received notification from server: {notification}")
 
     async def initialize(self) -> types.InitializeResult:
         result = await self.send_request(

pyproject.toml

@@ -11,7 +11,7 @@ dependencies = [
     "httpx-sse>=0.4.0",
     "lmos-openai-types",
     "loguru>=0.7.3",
-    "mcp>=1.2.0,<=1.7.1",
+    "mcp>=1.8.0,<=1.9.3",
     "mcpx[docker]>=0.1.1",
     "opentelemetry-api>=1.33.1",
     "opentelemetry-exporter-otlp>=1.33.1",