mcp: keep the stdio session alive on a malformed JSON frame

mcp/transport.go

@@ -5,6 +5,7 @@
 package mcp
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"errors"
@@ -477,6 +478,25 @@ type msgOrErr struct {
 	err error
 }
 
+// errParseError signals from the read loop to [ioConn.Read] that a frame could
+// not be parsed as JSON. The underlying stream is intact: Read replies with a
+// JSON-RPC parse error (-32700) and the read loop resynchronizes to the next
+// newline-delimited frame, keeping the session alive rather than ending it on a
+// single malformed message (per JSON-RPC 2.0).
+var errParseError = errors.New("jsonrpc2: parse error")
+
+// parseErrorResponse is a JSON-RPC 2.0 parse error with a null id, which is the
+// correct id for a frame too malformed to recover one from.
+var parseErrorResponse = []byte(`{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"parse error"}}` + "\n")
+
+// writeParseError writes a -32700 parse-error response, serialized with Write.
+func (t *ioConn) writeParseError() error {
+	t.writeMu.Lock()
+	defer t.writeMu.Unlock()
+	_, err := t.rwc.Write(parseErrorResponse)
+	return err
+}
+
 func newIOConn(rwc io.ReadWriteCloser) *ioConn {
 	var (
 		incoming = make(chan msgOrErr)
@@ -507,6 +527,27 @@ func newIOConn(rwc io.ReadWriteCloser) *ioConn {
 					err = readErr
 				}
 			}
+			// A JSON syntax error means this frame is malformed, but the stream
+			// itself is fine. Signal Read to reply with -32700, resynchronize to
+			// the next newline-delimited frame, and keep reading, rather than
+			// ending the session on a single bad message.
+			var syntaxErr *json.SyntaxError
+			if errors.As(err, &syntaxErr) {
+				select {
+				case incoming <- msgOrErr{err: errParseError}:
+				case <-closed:
+					return
+				}
+				// Skip the malformed frame up to the next newline, then continue
+				// with a fresh decoder over the remaining stream. If there is no
+				// newline before EOF, the next Decode returns io.EOF and the loop
+				// ends through the normal path below (which unblocks Read).
+				resync := bufio.NewReader(io.MultiReader(dec.Buffered(), rwc))
+				_, _ = resync.ReadBytes('\n')
+				dec = json.NewDecoder(resync)
+				continue
+			}
+
 			select {
 			case incoming <- msgOrErr{msg: raw, err: err}:
 			case <-closed:
@@ -619,18 +660,29 @@ func (t *ioConn) Read(ctx context.Context) (jsonrpc.Message, error) {
 	}
 
 	var raw json.RawMessage
-	select {
-	case <-ctx.Done():
-		return nil, ctx.Err()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+
+		case v := <-t.incoming:
+			if errors.Is(v.err, errParseError) {
+				// Malformed frame: reply -32700 and keep the session alive. The
+				// read loop has already resynchronized to the next frame.
+				if werr := t.writeParseError(); werr != nil {
+					return nil, werr
+				}
+				continue
+			}
+			if v.err != nil {
+				return nil, v.err
+			}
+			raw = v.msg
 
-	case v := <-t.incoming:
-		if v.err != nil {
-			return nil, v.err
+		case <-t.closed:
+			return nil, io.EOF
 		}
-		raw = v.msg
-
-	case <-t.closed:
-		return nil, io.EOF
+		break
 	}
 
 	msgs, batch, err := readBatch(raw)

mcp/transport_test.go

@@ -5,10 +5,13 @@
 package mcp
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
 	"github.com/modelcontextprotocol/go-sdk/jsonrpc"
@@ -167,3 +170,113 @@ func TestIOConnRead_EmptyMethod(t *testing.T) {
 		t.Errorf("ID = %v, want 5", req.ID.Raw())
 	}
 }
+
+// bufWriteCloser is a concurrency-safe io.WriteCloser over a bytes.Buffer.
+type bufWriteCloser struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *bufWriteCloser) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *bufWriteCloser) Close() error { return nil }
+
+func (b *bufWriteCloser) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+// TestIOConnRead_MalformedFrameRecovers verifies that a syntactically malformed
+// frame does not end the session: the transport replies with a JSON-RPC parse
+// error (-32700) and resynchronizes to the next newline-delimited frame, so the
+// following valid request is still delivered. This matches JSON-RPC 2.0 and the
+// behavior of other MCP server libraries; previously a single bad frame
+// terminated the stdio session.
+func TestIOConnRead_MalformedFrameRecovers(t *testing.T) {
+	out := &bufWriteCloser{}
+	tr := newIOConn(rwc{
+		rc: io.NopCloser(strings.NewReader("{bad json\n" + `{"jsonrpc":"2.0","id":7,"method":"ping"}` + "\n")),
+		wc: out,
+	})
+	t.Cleanup(func() { tr.Close() })
+
+	// The next Read must skip the malformed frame and return the valid one.
+	msg, err := tr.Read(context.Background())
+	if err != nil {
+		t.Fatalf("Read after malformed frame: error = %v, want nil (session must survive)", err)
+	}
+	req, ok := msg.(*jsonrpc.Request)
+	if !ok {
+		t.Fatalf("message type = %T, want *jsonrpc.Request", msg)
+	}
+	if req.Method != "ping" || req.ID != jsonrpc2.Int64ID(7) {
+		t.Errorf("recovered request = {method:%q id:%v}, want {ping 7}", req.Method, req.ID.Raw())
+	}
+
+	// A -32700 parse-error response must have been written for the bad frame.
+	if got := out.String(); !strings.Contains(got, "-32700") || !strings.Contains(got, "parse error") {
+		t.Errorf("expected a -32700 parse-error response, wrote: %q", got)
+	}
+}
+
+// TestIOConnRead_MalformedFrameEdgeCases covers consecutive malformed frames and
+// malformed frames at end-of-stream (with and without a trailing newline). Each
+// Read is guarded by a timeout so a regression that hangs the session is caught
+// as a failure rather than hanging the suite.
+func TestIOConnRead_MalformedFrameEdgeCases(t *testing.T) {
+	const ping = `{"jsonrpc":"2.0","id":9,"method":"ping"}`
+	tests := []struct {
+		name       string
+		input      string
+		wantMethod string // "" means expect io.EOF (no valid frame follows)
+	}{
+		{"two malformed then valid", "{bad1\nalso bad}\n" + ping + "\n", "ping"},
+		{"malformed then EOF with newline", "{bad\n", ""},
+		{"malformed then EOF without newline", "{bad", ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := &bufWriteCloser{}
+			tr := newIOConn(rwc{rc: io.NopCloser(strings.NewReader(tt.input)), wc: out})
+			t.Cleanup(func() { tr.Close() })
+
+			type result struct {
+				msg jsonrpc.Message
+				err error
+			}
+			done := make(chan result, 1)
+			go func() {
+				m, e := tr.Read(context.Background())
+				done <- result{m, e}
+			}()
+
+			select {
+			case r := <-done:
+				if tt.wantMethod == "" {
+					if r.err != io.EOF {
+						t.Errorf("err = %v, want io.EOF (session should end cleanly after the bad frame)", r.err)
+					}
+				} else {
+					if r.err != nil {
+						t.Fatalf("err = %v, want nil", r.err)
+					}
+					req, ok := r.msg.(*jsonrpc.Request)
+					if !ok || req.Method != tt.wantMethod {
+						t.Errorf("got %#v, want a %q request", r.msg, tt.wantMethod)
+					}
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("Read blocked: the session neither recovered nor terminated")
+			}
+
+			if !strings.Contains(out.String(), "-32700") {
+				t.Errorf("expected a -32700 response for the malformed frame, wrote: %q", out.String())
+			}
+		})
+	}
+}