feat: add streamable HTTP transport to MCP server (#1266)

* feat: add streamable HTTP transport to MCP server

The MCP server only supported the legacy SSE transport (GET /sse + POST
/message), deprecated in MCP spec 2025-03-26. Claude Code uses the newer
streamable HTTP transport, causing connection failures.

Add a /mcp endpoint implementing the streamable HTTP spec:
- POST initialize creates a session with Mcp-Session-Id header
- POST with session dispatches JSON-RPC and returns synchronous JSON
- DELETE terminates a session
- Legacy SSE endpoints preserved for backwards compatibility

Both transports share one MCPServer instance via a new public Dispatch()
method that wraps the existing handleRequest/handleNotification logic.

* fix: address review feedback on streamable HTTP handler

- Use mime.ParseMediaType for Content-Type check to accept
  "application/json; charset=utf-8" (CodeRabbit, claude[bot])
- Dispatch initialize before creating session to prevent orphans
  on failed initialize (CodeRabbit)
- Add background session eviction goroutine (1h idle timeout)
  to clean up sessions from disconnected clients (claude[bot])
- Extract "initialize" string literal to constant (goconst lint)
- Add Close() method and defer in main.go
- Add tests: failed initialize creates no session, charset tolerance

* fix: harden streamable HTTP handler (CodeRabbit review round 2)

- Limit request body to 1 MB via io.LimitReader to prevent memory
  exhaustion from oversized payloads
- Redact session IDs in log messages (show last 8 chars only) to
  reduce hijack risk if logs are exposed
- Guard against nil dispatch response in initialize path

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

deploy/demo/Caddyfile

@@ -30,8 +30,11 @@ demo.meridianhub.cloud, *.demo.meridianhub.cloud {
 		reverse_proxy dex:5556
 	}
 
-	# MCP Server: SSE transport + message endpoint
+	# MCP Server: streamable HTTP + legacy SSE transport
 	# NOTE: Requires mcp-server container to be running; routes will 502 until deployed.
+	handle /mcp {
+		reverse_proxy mcp-server:8090
+	}
 	handle /sse {
 		reverse_proxy mcp-server:8090
 	}

services/mcp-server/cmd/main.go

@@ -154,6 +154,11 @@ func runSSE(logger *slog.Logger, cfg server.Config) error {
 
 	srv := server.New(sseTr, cfg, logger)
 
+	// Streamable HTTP transport (MCP spec 2025-03-26).
+	// Shares the same MCPServer instance so tools/resources/prompts are identical.
+	streamableHandler := transport.NewStreamableHTTPHandler(srv, logger)
+	defer streamableHandler.Close()
+
 	mux := http.NewServeMux()
 
 	// OAuth 2.1 endpoints (optional — enabled via MCP_OAUTH_ENABLED=true).
@@ -180,9 +185,11 @@ func runSSE(logger *slog.Logger, cfg server.Config) error {
 		validator := &passthroughValidator{logger: logger}
 		bearerMW := mcpauth.NewBearerMiddleware(validator, meta)
 
+		mux.Handle("/mcp", bearerMW.Handler(streamableHandler))
 		mux.Handle("/sse", bearerMW.Handler(http.HandlerFunc(sseTr.HandleSSE)))
 		mux.Handle("/message", bearerMW.Handler(http.HandlerFunc(sseTr.HandleMessage)))
 	} else {
+		mux.Handle("/mcp", streamableHandler)
 		mux.HandleFunc("/sse", sseTr.HandleSSE)
 		mux.HandleFunc("/message", sseTr.HandleMessage)
 	}

services/mcp-server/internal/server/server.go

@@ -166,6 +166,21 @@ func (s *MCPServer) RegisterTool(tool Tool, handler ToolHandler) {
 	s.handlers[tool.Name] = handler
 }
 
+// Dispatch handles a single JSON-RPC message and returns the response.
+// For notifications it returns nil (no response required by JSON-RPC spec).
+// For non-request/non-notification messages it returns an invalid-request error.
+// This method is safe to call from multiple goroutines concurrently.
+func (s *MCPServer) Dispatch(ctx context.Context, msg *transport.JSONRPCMessage) *transport.JSONRPCMessage {
+	if msg.IsNotification() {
+		s.handleNotification(msg)
+		return nil
+	}
+	if !msg.IsRequest() {
+		return transport.NewErrorResponse(msg.ID, transport.CodeInvalidRequest, "invalid message")
+	}
+	return s.handleRequest(ctx, msg)
+}
+
 // Run starts the server's message processing loop. It blocks until the context
 // is cancelled or an unrecoverable error occurs.
 func (s *MCPServer) Run(ctx context.Context) error {
@@ -182,18 +197,11 @@ func (s *MCPServer) Run(ctx context.Context) error {
 			return fmt.Errorf("read message: %w", err)
 		}
 
-		if msg.IsNotification() {
-			s.handleNotification(msg)
+		resp := s.Dispatch(ctx, msg)
+		if resp == nil {
 			continue
 		}
-
-		if !msg.IsRequest() {
-			s.logger.Warn("ignoring non-request message")
-			continue
-		}
-
-		response := s.handleRequest(ctx, msg)
-		if err := s.transport.WriteMessage(ctx, response); err != nil {
+		if err := s.transport.WriteMessage(ctx, resp); err != nil {
 			s.logger.Error("failed to write response", "error", err)
 		}
 	}

services/mcp-server/internal/transport/streamable.go

@@ -0,0 +1,257 @@
+package transport
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"mime"
+	"net/http"
+	"sync"
+	"time"
+)
+
+const (
+	methodInitialize = "initialize"
+	// sessionIdleTimeout is how long a session can be idle before eviction.
+	sessionIdleTimeout = 1 * time.Hour
+	// sessionEvictInterval is how often the background goroutine sweeps idle sessions.
+	sessionEvictInterval = 5 * time.Minute
+	// maxRequestBodySize is the maximum size of a JSON-RPC request body (1 MB).
+	maxRequestBodySize = 1 << 20
+)
+
+// Dispatcher handles a single JSON-RPC message and returns the response.
+// Notifications return nil (no response required).
+type Dispatcher interface {
+	Dispatch(ctx context.Context, msg *JSONRPCMessage) *JSONRPCMessage
+}
+
+// StreamableHTTPHandler implements the MCP streamable HTTP transport
+// (spec 2025-03-26). Clients POST JSON-RPC messages to a single endpoint
+// and receive synchronous JSON responses.
+type StreamableHTTPHandler struct {
+	dispatcher Dispatcher
+	sessions   map[string]*streamSession
+	mu         sync.RWMutex
+	logger     *slog.Logger
+	stop       chan struct{}
+}
+
+type streamSession struct {
+	id       string
+	created  time.Time
+	lastUsed time.Time
+}
+
+// NewStreamableHTTPHandler creates a handler for the MCP streamable HTTP transport.
+// Call Close to stop the background session eviction goroutine.
+func NewStreamableHTTPHandler(dispatcher Dispatcher, logger *slog.Logger) *StreamableHTTPHandler {
+	h := &StreamableHTTPHandler{
+		dispatcher: dispatcher,
+		sessions:   make(map[string]*streamSession),
+		logger:     logger,
+		stop:       make(chan struct{}),
+	}
+	go h.evictLoop()
+	return h
+}
+
+// Close stops the background session eviction goroutine.
+func (h *StreamableHTTPHandler) Close() {
+	select {
+	case <-h.stop:
+	default:
+		close(h.stop)
+	}
+}
+
+func (h *StreamableHTTPHandler) evictLoop() {
+	ticker := time.NewTicker(sessionEvictInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			h.evictIdleSessions()
+		case <-h.stop:
+			return
+		}
+	}
+}
+
+func (h *StreamableHTTPHandler) evictIdleSessions() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for id, sess := range h.sessions {
+		if time.Since(sess.lastUsed) > sessionIdleTimeout {
+			delete(h.sessions, id)
+			h.logger.Info("evicted idle streamable HTTP session", "session_id", redactSessionID(id))
+		}
+	}
+}
+
+// ServeHTTP routes requests by HTTP method.
+func (h *StreamableHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch r.Method {
+	case http.MethodPost:
+		h.handlePost(w, r)
+	case http.MethodDelete:
+		h.handleDelete(w, r)
+	default:
+		w.Header().Set("Allow", "POST, DELETE")
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+	}
+}
+
+func (h *StreamableHTTPHandler) handlePost(w http.ResponseWriter, r *http.Request) {
+	if !isJSONContentType(r) {
+		http.Error(w, "Content-Type must be application/json", http.StatusUnsupportedMediaType)
+		return
+	}
+
+	var msg JSONRPCMessage
+	if err := json.NewDecoder(io.LimitReader(r.Body, maxRequestBodySize)).Decode(&msg); err != nil {
+		writeJSONError(w, nil, CodeParseError, "invalid JSON")
+		return
+	}
+
+	// For initialize requests, create a new session.
+	if msg.Method == methodInitialize {
+		h.handleInitialize(w, r, &msg)
+		return
+	}
+
+	// All other messages require a valid session.
+	sessionID := r.Header.Get("Mcp-Session-Id")
+	if sessionID == "" {
+		http.Error(w, "Mcp-Session-Id header required", http.StatusBadRequest)
+		return
+	}
+
+	h.mu.RLock()
+	sess, exists := h.sessions[sessionID]
+	h.mu.RUnlock()
+
+	if !exists {
+		http.Error(w, "unknown session", http.StatusNotFound)
+		return
+	}
+
+	h.mu.Lock()
+	sess.lastUsed = time.Now()
+	h.mu.Unlock()
+
+	resp := h.dispatcher.Dispatch(r.Context(), &msg)
+	if resp == nil {
+		// Notification — no response body.
+		w.WriteHeader(http.StatusAccepted)
+		return
+	}
+
+	writeJSON(w, resp)
+}
+
+func (h *StreamableHTTPHandler) handleInitialize(w http.ResponseWriter, r *http.Request, msg *JSONRPCMessage) {
+	// Dispatch first — only create the session if initialize succeeds.
+	resp := h.dispatcher.Dispatch(r.Context(), msg)
+	if resp == nil || resp.Error != nil {
+		if resp != nil {
+			writeJSON(w, resp)
+		} else {
+			http.Error(w, "internal server error", http.StatusInternalServerError)
+		}
+		return
+	}
+
+	sessionID := generateSessionID()
+
+	h.mu.Lock()
+	now := time.Now()
+	h.sessions[sessionID] = &streamSession{
+		id:       sessionID,
+		created:  now,
+		lastUsed: now,
+	}
+	h.mu.Unlock()
+
+	h.logger.Info("streamable HTTP session created", "session_id", redactSessionID(sessionID))
+
+	w.Header().Set("Mcp-Session-Id", sessionID)
+	writeJSON(w, resp)
+}
+
+func (h *StreamableHTTPHandler) handleDelete(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.Header.Get("Mcp-Session-Id")
+	if sessionID == "" {
+		http.Error(w, "Mcp-Session-Id header required", http.StatusBadRequest)
+		return
+	}
+
+	h.mu.Lock()
+	_, exists := h.sessions[sessionID]
+	if exists {
+		delete(h.sessions, sessionID)
+	}
+	h.mu.Unlock()
+
+	if !exists {
+		http.Error(w, "unknown session", http.StatusNotFound)
+		return
+	}
+
+	h.logger.Info("streamable HTTP session terminated", "session_id", redactSessionID(sessionID))
+	w.WriteHeader(http.StatusAccepted)
+}
+
+// SessionCount returns the number of active sessions.
+func (h *StreamableHTTPHandler) SessionCount() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.sessions)
+}
+
+func writeJSON(w http.ResponseWriter, msg *JSONRPCMessage) {
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(msg); err != nil {
+		// Headers already sent; log but can't send error response.
+		_ = err
+	}
+}
+
+func writeJSONError(w http.ResponseWriter, id json.RawMessage, code int, message string) {
+	resp := NewErrorResponse(id, code, message)
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusBadRequest)
+	_ = json.NewEncoder(w).Encode(resp)
+}
+
+// isJSONContentType checks whether the request Content-Type is application/json,
+// tolerating optional parameters like charset (e.g. "application/json; charset=utf-8").
+func isJSONContentType(r *http.Request) bool {
+	ct := r.Header.Get("Content-Type")
+	mediaType, _, err := mime.ParseMediaType(ct)
+	if err != nil {
+		return false
+	}
+	return mediaType == "application/json"
+}
+
+func generateSessionID() string {
+	b := make([]byte, 16)
+	if _, err := rand.Read(b); err != nil {
+		panic("crypto/rand failed: " + err.Error())
+	}
+	return hex.EncodeToString(b)
+}
+
+// redactSessionID returns a truncated session ID safe for logging.
+// Shows only the last 8 characters to aid debugging without exposing
+// the full value (which acts as authorization material).
+func redactSessionID(id string) string {
+	if len(id) <= 8 {
+		return "***"
+	}
+	return "..." + id[len(id)-8:]
+}