feat(mcp): add streamable HTTP MCP transport

docs/content/docs/mcp.mdx

@@ -115,6 +115,9 @@ headroom mcp uninstall
 
 # Debug mode
 headroom mcp serve --debug
+
+# Streamable HTTP mode
+headroom mcp serve --transport http --host 127.0.0.1 --port 8788 --path /mcp
 ```
 
 ## MCP host configuration
@@ -152,7 +155,9 @@ For multiple proxy instances, register one stdio MCP server per proxy URL:
 }
 ```
 
-Do not assume that a running proxy exposes an HTTP MCP endpoint at `/mcp`. If `http://127.0.0.1:<port>/mcp` returns `404`, use the stdio configuration above.
+If you need Streamable HTTP instead of stdio, run `headroom mcp serve --transport http` and point your MCP host at that endpoint. The default path is `/mcp`, and the default port is `8788`.
+
+Do not assume that a running proxy exposes an HTTP MCP endpoint at `/mcp`. The proxy serves its own API; it does not automatically provide the MCP HTTP transport.
 
 ### `command: "headroom"` fails to start
 

headroom/ccr/mcp_http.py

@@ -0,0 +1,78 @@
+"""Streamable HTTP transport helpers for the Headroom MCP server."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.types import Receive, Scope, Send
+
+
+def _normalize_path(path: str) -> str:
+    normalized = path.strip()
+    if not normalized.startswith("/"):
+        normalized = f"/{normalized}"
+    if normalized != "/":
+        normalized = normalized.rstrip("/")
+    return normalized
+
+
+class StreamableHTTPASGIApp:
+    """Delegate ASGI requests to the SDK session manager."""
+
+    def __init__(self, session_manager: StreamableHTTPSessionManager):
+        self.session_manager = session_manager
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        await self.session_manager.handle_request(scope, receive, send)
+
+
+def create_streamable_http_session_manager(server: Any) -> StreamableHTTPSessionManager:
+    """Build the SDK session manager for an existing MCP server."""
+    return StreamableHTTPSessionManager(app=server.server)
+
+
+def create_streamable_http_app(
+    session_manager: StreamableHTTPSessionManager,
+    *,
+    path: str,
+    debug: bool = False,
+) -> Starlette:
+    """Build the Starlette app that exposes the MCP server over HTTP."""
+    streamable_http_app = StreamableHTTPASGIApp(session_manager)
+    return Starlette(
+        debug=debug,
+        routes=[
+            Route(
+                _normalize_path(path),
+                endpoint=streamable_http_app,
+                methods=["GET", "POST", "DELETE", "OPTIONS"],
+            )
+        ],
+        lifespan=lambda app: session_manager.run(),
+    )
+
+
+async def serve_streamable_http(
+    server: Any,
+    *,
+    host: str,
+    port: int,
+    path: str,
+    debug: bool = False,
+) -> None:
+    """Serve the MCP server over Streamable HTTP with uvicorn."""
+    import uvicorn
+
+    session_manager = create_streamable_http_session_manager(server)
+    starlette_app = create_streamable_http_app(session_manager, path=path, debug=debug)
+    config = uvicorn.Config(
+        starlette_app,
+        host=host,
+        port=port,
+        log_level="debug" if debug else "warning",
+    )
+    uvicorn_server = uvicorn.Server(config)
+    await uvicorn_server.serve()

headroom/ccr/mcp_server.py

@@ -925,6 +925,24 @@ async def run_stdio(self) -> None:
                 self.server.create_initialization_options(),
             )
 
+    async def run_streamable_http(
+        self,
+        host: str,
+        port: int,
+        path: str,
+        debug: bool = False,
+    ) -> None:
+        """Run the server with Streamable HTTP transport."""
+        from .mcp_http import serve_streamable_http
+
+        await serve_streamable_http(
+            self,
+            host=host,
+            port=port,
+            path=path,
+            debug=debug,
+        )
+
     async def cleanup(self) -> None:
         """Clean up resources."""
         if self._http_client:

headroom/cli/mcp.py

@@ -21,6 +21,9 @@
 CLAUDE_CONFIG_DIR = Path.home() / ".claude"
 MCP_CONFIG_PATH = CLAUDE_CONFIG_DIR / "mcp.json"
 DEFAULT_PROXY_URL = "http://127.0.0.1:8787"
+DEFAULT_HTTP_HOST = "127.0.0.1"
+DEFAULT_HTTP_PORT = 8788
+DEFAULT_HTTP_PATH = "/mcp"
 
 
 def get_headroom_command() -> list[str]:
@@ -302,6 +305,32 @@ def mcp_status() -> None:
     envvar="HEADROOM_PROXY_URL",
     help=f"Headroom proxy URL (default: {DEFAULT_PROXY_URL})",
 )
+@click.option(
+    "--transport",
+    type=click.Choice(["stdio", "http"], case_sensitive=False),
+    default="stdio",
+    show_default=True,
+    help="Transport to use for headroom mcp serve",
+)
+@click.option(
+    "--host",
+    default=DEFAULT_HTTP_HOST,
+    show_default=True,
+    help="HTTP bind host",
+)
+@click.option(
+    "--port",
+    default=DEFAULT_HTTP_PORT,
+    show_default=True,
+    type=int,
+    help="HTTP bind port",
+)
+@click.option(
+    "--path",
+    default=DEFAULT_HTTP_PATH,
+    show_default=True,
+    help="HTTP endpoint path",
+)
 @click.option(
     "--direct",
     is_flag=True,
@@ -312,16 +341,26 @@ def mcp_status() -> None:
     is_flag=True,
     help="Enable debug logging",
 )
-def mcp_serve(proxy_url: str | None, direct: bool, debug: bool) -> None:
+def mcp_serve(
+    proxy_url: str | None,
+    transport: str,
+    host: str,
+    port: int,
+    path: str,
+    direct: bool,
+    debug: bool,
+) -> None:
     """Start the MCP server (called by Claude Code).
 
     \b
     This command is typically invoked by Claude Code via the MCP config,
-    not run directly. It starts the MCP server with stdio transport.
+    not run directly. It starts the MCP server with stdio by default or
+    Streamable HTTP when requested.
 
     \b
     For manual testing:
         headroom mcp serve --debug
+        headroom mcp serve --transport http --host 127.0.0.1 --port 8788 --path /mcp
     """
     import asyncio
     import logging
@@ -346,6 +385,8 @@ def mcp_serve(proxy_url: str | None, direct: bool, debug: bool) -> None:
             format="%(levelname)s: %(message)s",
         )
 
+    transport = transport.lower()
+
     # Use default if not specified
     effective_proxy_url = proxy_url or DEFAULT_PROXY_URL
 
@@ -359,7 +400,10 @@ def mcp_serve(proxy_url: str | None, direct: bool, debug: bool) -> None:
 
     async def run() -> None:
         try:
-            await server.run_stdio()
+            if transport == "http":
+                await server.run_streamable_http(host=host, port=port, path=path, debug=debug)
+            else:
+                await server.run_stdio()
         finally:
             await server.cleanup()
 

pyproject.toml

@@ -207,6 +207,8 @@ strands = [
 mcp = [
     "mcp>=1.0.0",
     "httpx>=0.24.0",
+    "starlette>=0.27.0",
+    "uvicorn>=0.23.0,<1.0",
 ]
 # Voice filler detection
 voice = [

tests/test_ccr_mcp_http.py

@@ -0,0 +1,51 @@
+"""Contract tests for the Headroom Streamable HTTP MCP transport."""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+pytest.importorskip("mcp")
+
+from mcp.client.session import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+from headroom.ccr.mcp_http import (
+    create_streamable_http_app,
+    create_streamable_http_session_manager,
+)
+from headroom.ccr.mcp_server import create_ccr_mcp_server
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+async def test_streamable_http_initialize_and_list_tools() -> None:
+    server = create_ccr_mcp_server()
+    session_manager = create_streamable_http_session_manager(server)
+    app = create_streamable_http_app(session_manager, path="/mcp")
+    transport = httpx.ASGITransport(app=app)
+
+    async with session_manager.run():
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://testserver"
+        ) as http_client:
+            async with streamable_http_client(
+                "http://testserver/mcp",
+                http_client=http_client,
+                terminate_on_close=False,
+            ) as (read_stream, write_stream, get_session_id):
+                async with ClientSession(read_stream, write_stream) as session:
+                    initialize_result = await session.initialize()
+                    list_tools_result = await session.list_tools()
+
+    tool_names = [tool.name for tool in list_tools_result.tools]
+    assert initialize_result.protocolVersion
+    assert get_session_id() is not None
+    assert "headroom_compress" in tool_names
+    assert "headroom_retrieve" in tool_names
+    assert "headroom_stats" in tool_names

tests/test_cli/test_mcp.py

@@ -8,7 +8,7 @@
 
 import json
 import sys
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from click.testing import CliRunner
@@ -257,9 +257,64 @@ def test_serve_help(self):
         result = runner.invoke(main, ["mcp", "serve", "--help"])
 
         assert result.exit_code == 0
+        assert "transport" in result.output
+        assert "host" in result.output
+        assert "port" in result.output
+        assert "path" in result.output
         assert "proxy-url" in result.output
         assert "debug" in result.output
 
+    def test_serve_defaults_to_stdio(self):
+        """Serve command defaults to stdio transport."""
+        fake_server = MagicMock()
+        fake_server.run_stdio = AsyncMock()
+        fake_server.run_streamable_http = AsyncMock()
+        fake_server.cleanup = AsyncMock()
+
+        runner = CliRunner()
+        with patch("headroom.ccr.mcp_server.create_ccr_mcp_server", return_value=fake_server):
+            result = runner.invoke(main, ["mcp", "serve"])
+
+        assert result.exit_code == 0
+        fake_server.run_stdio.assert_awaited_once()
+        fake_server.run_streamable_http.assert_not_awaited()
+        fake_server.cleanup.assert_awaited_once()
+
+    def test_serve_http_transport_uses_http_settings(self):
+        """Serve command uses HTTP transport when requested with mixed-case input."""
+        fake_server = MagicMock()
+        fake_server.run_stdio = AsyncMock()
+        fake_server.run_streamable_http = AsyncMock()
+        fake_server.cleanup = AsyncMock()
+
+        runner = CliRunner()
+        with patch("headroom.ccr.mcp_server.create_ccr_mcp_server", return_value=fake_server):
+            result = runner.invoke(
+                main,
+                [
+                    "mcp",
+                    "serve",
+                    "--transport",
+                    "HTTP",
+                    "--host",
+                    "0.0.0.0",
+                    "--port",
+                    "9191",
+                    "--path",
+                    "/mcp",
+                ],
+            )
+
+        assert result.exit_code == 0
+        fake_server.run_stdio.assert_not_awaited()
+        fake_server.run_streamable_http.assert_awaited_once_with(
+            host="0.0.0.0",
+            port=9191,
+            path="/mcp",
+            debug=False,
+        )
+        fake_server.cleanup.assert_awaited_once()
+
 
 @pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP SDK not installed")
 class TestMCPServerInitialization: