mcp: jsonrpc response fix

internal/mcpproxy/handlers.go

@@ -6,6 +6,7 @@
 package mcpproxy
 
 import (
+	"bytes"
 	"cmp"
 	"context"
 	"encoding/base64"
@@ -768,53 +769,60 @@ func copyProxyHeaders(resp *http.Response, w http.ResponseWriter) {
 func (m *mcpRequestContext) proxyResponseBody(ctx context.Context, s *session, w http.ResponseWriter, resp *http.Response,
 	req *jsonrpc.Request, backend filterapi.MCPBackend,
 ) error {
+	// Some backends (e.g. Slack MCP) send SSE data despite Content-Type: application/json.
+	// Try to decode as a single JSON-RPC message first; if that fails, fall through to the
+	// SSE parser using the already-read bytes.
+	var sseReader io.Reader = resp.Body
 	if resp.Header.Get("Content-Type") == "application/json" {
 		body, err := io.ReadAll(resp.Body)
 		if err != nil {
 			m.l.Error("failed to read response body", slog.String("error", err.Error()))
 			return err
 		}
-		_msg, err := jsonrpc.DecodeMessage(body)
-		if err != nil {
-			m.l.Error("failed to decode JSON-RPC message from response body", slog.String("error", err.Error()))
-			return err
-		}
-
-		var responseError error
-		switch msg := _msg.(type) {
-		case *jsonrpc.Request:
-			if err = m.maybeServerToClientRequestModify(ctx, msg, backend.Name); err != nil {
-				m.l.Error("failed to modify server->client request", slog.String("error", err.Error()))
-				return err
-			}
-			body, _ = jsonrpc.EncodeMessage(msg)
-		case *jsonrpc.Response:
-			if req != nil {
-				if err = m.maybeResponseModify(ctx, req, msg, backend.Name); err != nil {
-					m.l.Error("failed to modify response", slog.String("error", err.Error()))
+		_msg, ok := tryDecodeJSONRPCMessage(body)
+		if ok {
+			var responseError error
+			switch msg := _msg.(type) {
+			case *jsonrpc.Request:
+				if err = m.maybeServerToClientRequestModify(ctx, msg, backend.Name); err != nil {
+					m.l.Error("failed to modify server->client request", slog.String("error", err.Error()))
 					return err
 				}
-				msg.ID = req.ID
-
-				// Check if this is a JSON-RPC error response
-				if msg.Error != nil {
-					responseError = msg.Error
-				} else if toolErr := checkToolCallError(req, msg, backend.Name); toolErr != nil {
-					// Check if this is a tools/call response with isError=true
-					responseError = toolErr
-				}
-
 				body, _ = jsonrpc.EncodeMessage(msg)
+			case *jsonrpc.Response:
+				if req != nil {
+					if err = m.maybeResponseModify(ctx, req, msg, backend.Name); err != nil {
+						m.l.Error("failed to modify response", slog.String("error", err.Error()))
+						return err
+					}
+					msg.ID = req.ID
+
+					// Check if this is a JSON-RPC error response
+					if msg.Error != nil {
+						responseError = msg.Error
+					} else if toolErr := checkToolCallError(req, msg, backend.Name); toolErr != nil {
+						// Check if this is a tools/call response with isError=true
+						responseError = toolErr
+					}
+
+					body, _ = jsonrpc.EncodeMessage(msg)
+				}
+				m.recordResponse(ctx, msg)
 			}
-			m.recordResponse(ctx, msg)
-		}
 
-		// We need to update the content length since we might have modified the ID.
-		w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body)))
-		w.WriteHeader(resp.StatusCode)
-		_, _ = w.Write(body)
+			// We need to update the content length since we might have modified the ID.
+			w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body)))
+			w.WriteHeader(resp.StatusCode)
+			_, _ = w.Write(body)
 
-		return responseError
+			return responseError
+		}
+		// Body claimed application/json but isn't valid JSON-RPC (e.g. some backends
+		// send SSE data despite the content type). Fall through to the SSE parser
+		// using the already-read body bytes since resp.Body is now drained.
+		m.l.Info("response Content-Type is application/json but body is not valid JSON-RPC, falling back to SSE parsing",
+			slog.String("backend", backend.Name))
+		sseReader = bytes.NewReader(body)
 	}
 
 	// io.Copy won't flush until the end, which doesn't happen for streaming responses.
@@ -825,7 +833,7 @@ func (m *mcpRequestContext) proxyResponseBody(ctx context.Context, s *session, w
 	w.WriteHeader(resp.StatusCode)
 	// For single-backend operations, metrics are recorded in the defer of servePOST,
 	// so we don't need to track startAt in events here.
-	parser := newSSEEventParser(resp.Body, backend.Name)
+	parser := newSSEEventParser(sseReader, backend.Name)
 
 	// Collect errors from multiple events to return them all to the caller
 	var responseErrors []error
@@ -1603,7 +1611,8 @@ func (m *mcpRequestContext) handleSetLoggingLevel(ctx context.Context, s *sessio
 
 // mergeToolsList merges the list of tools from all backends and prepare the response message to be sent back to the client.
 func (m *mcpRequestContext) mergeToolsList(s *session, responses []broadCastResponse[mcp.ListToolsResult]) mcp.ListToolsResult {
-	resp := mcp.ListToolsResult{}
+	// Use a non-nil empty slice so JSON encodes as [] not null; some clients reject tools:null.
+	resp := mcp.ListToolsResult{Tools: make([]*mcp.Tool, 0)}
 	route := m.routes[s.route]
 	if route == nil {
 		// This should never happen as the route must have been validated when the session is created.

internal/mcpproxy/handlers_test.go

@@ -900,6 +900,29 @@ func TestProxyResponseBody_JSONResponse(t *testing.T) {
 	require.Contains(t, rr.Body.String(), id.Raw())
 }
 
+func TestProxyResponseBody_JSONResponseWithBOM(t *testing.T) {
+	proxy := newTestMCPProxy()
+
+	id := mustJSONRPCRequestID()
+	resp := &jsonrpc.Response{ID: id, Result: []byte(`{"test": "bom"}`)}
+	body, err := jsonrpc.EncodeMessage(resp)
+	require.NoError(t, err)
+
+	bomBody := append([]byte{0xEF, 0xBB, 0xBF}, body...)
+	httpResp := &http.Response{
+		Header:     http.Header{"Content-Type": []string{"application/json"}},
+		Body:       io.NopCloser(bytes.NewReader(bomBody)),
+		StatusCode: http.StatusOK,
+	}
+
+	rr := httptest.NewRecorder()
+
+	proxy.proxyResponseBody(t.Context(), nil, rr, httpResp, &jsonrpc.Request{ID: id}, filterapi.MCPBackend{Name: "mybackend"}) //nolint:errcheck
+
+	require.Contains(t, rr.Body.String(), "bom")
+	require.Contains(t, rr.Body.String(), id.Raw())
+}
+
 func TestProxyResponseBody_SSEResponse(t *testing.T) {
 	proxy := newTestMCPProxy()
 

internal/mcpproxy/mcpproxy.go

@@ -301,9 +301,19 @@ func (m *mcpRequestContext) initializeSession(ctx context.Context, routeName fil
 		}
 
 		var rawMsg jsonrpc.Message
-		switch resp.Header.Get("Content-Type") {
-		case "text/event-stream":
-			parser := newSSEEventParser(resp.Body, backend.Name)
+		var sseReader io.Reader = resp.Body
+		if resp.Header.Get("Content-Type") != "text/event-stream" {
+			body, _ := io.ReadAll(resp.Body)
+			msg, ok := tryDecodeJSONRPCMessage(body)
+			if ok {
+				rawMsg = msg
+			} else {
+				// Not valid JSON-RPC; fall through to SSE parser with the already-read bytes.
+				sseReader = bytes.NewReader(body)
+			}
+		}
+		if rawMsg == nil {
+			parser := newSSEEventParser(sseReader, backend.Name)
 			for {
 				event, parseErr := parser.next()
 				// TODO: handle reconnect. We need to re-arrange the event ID so that it will also contain the backend name and the original session ID.
@@ -324,15 +334,6 @@ func (m *mcpRequestContext) initializeSession(ctx context.Context, routeName fil
 					break
 				}
 			}
-		default:
-			// Handle JSON response.
-			body, _ := io.ReadAll(resp.Body)
-			// Decode the JSON-RPC message.
-			rawMsg, err = jsonrpc.DecodeMessage(body)
-			if err != nil {
-				m.l.Warn("Failed to decode MCP message", slog.String("error", err.Error()))
-				return nil, fmt.Errorf("failed to decode MCP message: %w", err)
-			}
 		}
 
 		msg, ok := rawMsg.(*jsonrpc.Response)

internal/mcpproxy/session.go

@@ -439,28 +439,29 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b
 	}
 
 	if httpResp.Header.Get("Content-Type") == "application/json" {
-		// This is not an SSE response, but only a single JSON response. Convert it as an event and
-		// send it to the channel.
+		// Try to decode as a single JSON-RPC message first.
 		var respBody []byte
 		respBody, err = io.ReadAll(bodyReader)
 		if err != nil {
 			return fmt.Errorf("failed to read MCP response body: %w", err)
 		}
-		var msg jsonrpc.Message
-		msg, err = jsonrpc.DecodeMessage(respBody)
-		if err != nil {
-			return fmt.Errorf("failed to decode jsonrpc message from MCP response body: %w", err)
-		}
-		eventChan <- &backendEvent{
-			sseEvent: &sseEvent{
-				backend:  backend.Name,
-				event:    "message",
-				id:       "", // No event ID in this case.
-				messages: []jsonrpc.Message{msg},
-			},
-			startAt: startAt,
+		msg, ok := tryDecodeJSONRPCMessage(respBody)
+		if ok {
+			eventChan <- &backendEvent{
+				sseEvent: &sseEvent{
+					backend:  backend.Name,
+					event:    "message",
+					id:       "", // No event ID in this case.
+					messages: []jsonrpc.Message{msg},
+				},
+				startAt: startAt,
+			}
+			return nil
 		}
-		return nil
+		// Body claimed application/json but isn't valid JSON-RPC (e.g. some backends
+		// send SSE data despite the content type). Fall through to the SSE parser
+		// using the already-read body bytes.
+		bodyReader = bytes.NewReader(respBody)
 	}
 
 	// io.Copy won't flush until the end, which doesn't happen for streaming responses.

internal/mcpproxy/session_test.go

@@ -569,6 +569,42 @@ func TestSendRequestPerBackend_BrotliDecompression(t *testing.T) {
 	require.Equal(t, "ping", req.Method)
 }
 
+func TestSendRequestPerBackend_BOMPrefixedJSON(t *testing.T) {
+	id1, _ := jsonrpc.MakeID("1")
+	msg1, _ := jsonrpc.EncodeMessage(&jsonrpc.Response{ID: id1, Result: []byte(`{"ok":true}`)})
+
+	bomBody := append([]byte{0xEF, 0xBB, 0xBF}, msg1...)
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(bomBody)
+	}))
+	defer server.Close()
+
+	proxy := newTestMCPProxy()
+	proxy.backendListenerAddr = server.URL
+	s := &session{reqCtx: proxy}
+	ch := make(chan *backendEvent, 10)
+	ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+	defer cancel()
+	err := s.sendRequestPerBackend(ctx, ch, "route1", filterapi.MCPBackend{Name: "backend1"}, &compositeSessionEntry{
+		sessionID: "sess1",
+	}, http.MethodGet, nil)
+	require.NoError(t, err)
+	close(ch)
+	var events []*backendEvent
+	for e := range ch {
+		events = append(events, e)
+	}
+	require.Len(t, events, 1, "expected 1 event from BOM-prefixed JSON response")
+	require.Equal(t, "message", events[0].event)
+	require.Len(t, events[0].messages, 1)
+	resp, ok := events[0].messages[0].(*jsonrpc.Response)
+	require.True(t, ok)
+	require.Equal(t, id1, resp.ID)
+}
+
 func TestHandleNotificationsPerBackend_SSE(t *testing.T) {
 	// Provide two SSE events with valid JSON-RPC requests then close.
 	id1, _ := jsonrpc.MakeID("1")

internal/mcpproxy/sse.go

@@ -26,6 +26,10 @@ var (
 	sseLF   = []byte{'\n'}
 	sseCRLF = []byte{'\r', '\n'}
 	sseLFLF = []byte{'\n', '\n'}
+
+	// utf8BOM is the UTF-8 Byte Order Mark (U+FEFF). Some backends prepend this
+	// invisible sequence to response bodies, which breaks JSON decoding.
+	utf8BOM = []byte{0xEF, 0xBB, 0xBF}
 )
 
 // sseEventParser reads bytes from a reader and parses the SSE Events gracefully
@@ -41,6 +45,20 @@ func newSSEEventParser(r io.Reader, backend filterapi.MCPBackendName) sseEventPa
 	return sseEventParser{r: r, backend: backend}
 }
 
+// tryDecodeJSONRPCMessage attempts to decode the body as a single JSON-RPC message.
+// It strips a leading UTF-8 BOM and whitespace before decoding.
+// Returns the decoded message and true on success, or nil and false if the body
+// is not valid JSON-RPC (e.g. the backend sent SSE despite a JSON content type).
+func tryDecodeJSONRPCMessage(body []byte) (jsonrpc.Message, bool) {
+	body = bytes.TrimSpace(body)
+	body = bytes.TrimPrefix(body, utf8BOM)
+	msg, err := jsonrpc.DecodeMessage(body)
+	if err != nil {
+		return nil, false
+	}
+	return msg, true
+}
+
 // next reads the next SSE event from the stream.
 func (s *sseEventParser) next() (*sseEvent, error) {
 	for {

internal/mcpproxy/sse_test.go

@@ -24,6 +24,44 @@ func mustEncode(t *testing.T, m jsonrpc.Message) []byte {
 	return b
 }
 
+func TestTryDecodeJSONRPCMessage(t *testing.T) {
+	id, err := jsonrpc.MakeID("1")
+	require.NoError(t, err)
+	want := &jsonrpc.Response{ID: id, Result: []byte(`{"tools":[]}`)}
+	raw, err := jsonrpc.EncodeMessage(want)
+	require.NoError(t, err)
+
+	t.Run("valid JSON-RPC", func(t *testing.T) {
+		msg, ok := tryDecodeJSONRPCMessage(raw)
+		require.True(t, ok)
+		got, isResp := msg.(*jsonrpc.Response)
+		require.True(t, isResp)
+		require.Equal(t, want.ID, got.ID)
+	})
+	t.Run("BOM prefix stripped", func(t *testing.T) {
+		bomBody := append(append([]byte{}, utf8BOM...), raw...)
+		msg, ok := tryDecodeJSONRPCMessage(bomBody)
+		require.True(t, ok)
+		got, isResp := msg.(*jsonrpc.Response)
+		require.True(t, isResp)
+		require.Equal(t, want.ID, got.ID)
+	})
+	t.Run("leading whitespace stripped", func(t *testing.T) {
+		msg, ok := tryDecodeJSONRPCMessage(append([]byte("  \n"), raw...))
+		require.True(t, ok)
+		require.NotNil(t, msg)
+	})
+	t.Run("SSE body returns false", func(t *testing.T) {
+		sseBody := append(append([]byte("data: "), raw...), []byte("\n\n")...)
+		_, ok := tryDecodeJSONRPCMessage(sseBody)
+		require.False(t, ok)
+	})
+	t.Run("binary garbage returns false", func(t *testing.T) {
+		_, ok := tryDecodeJSONRPCMessage([]byte{0x13, 0x65, 0x70, 0x8c})
+		require.False(t, ok)
+	})
+}
+
 func TestSSEEventParser_SingleEvent(t *testing.T) {
 	id, err := jsonrpc.MakeID("1")
 	require.NoError(t, err)