From 3906b0106c827d50da8aea77ad4aefe63174f7cf Mon Sep 17 00:00:00 2001 From: Dayna Blackwell Date: Sat, 29 Aug 2026 09:10:10 -0700 Subject: [PATCH] mcp: keep the stdio session alive on a malformed JSON frame The stdio transport feeds os.Stdin into a single streaming json.Decoder. A syntactically malformed frame makes Decode return a *json.SyntaxError, and the read goroutine returned on that error, so one bad frame terminated the whole session. A streaming decoder also cannot resynchronize on its own: a syntax error poisons its buffered stream state. Per JSON-RPC 2.0, a parse error should be answered with a -32700 response and the session should continue. That is what mark3labs/mcp-go does (it reads newline-delimited frames and unmarshals each independently), and it is the same recoverable-decode direction taken for empty-method requests in #1000. The stdio read loop now: - replies with a -32700 parse-error response (id: null) for a malformed frame, - resynchronizes to the next newline-delimited frame and keeps reading, - terminates only on a genuine EOF or I/O error, as before. An EOF immediately after a malformed frame (with or without a trailing newline) still ends the session cleanly. Only *json.SyntaxError is recovered; the existing "invalid trailing data" handling is unchanged. Tests cover a malformed frame followed by a valid request (the request is still delivered and a -32700 is written), consecutive malformed frames, and malformed frames at end-of-stream. Verified with -race. Surfaced by an MCP conformance study run against a downstream server (blackwell-systems/agent-lsp#14). The behavior is a property of this transport, shared by every server built on the SDK's stdio transport. Fixes #1209 --- mcp/transport.go | 72 +++++++++++++++++++++++---- mcp/transport_test.go | 113 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 10 deletions(-) diff --git a/mcp/transport.go b/mcp/transport.go index 1cf03251..ca0f711d 100644 --- a/mcp/transport.go +++ b/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) diff --git a/mcp/transport_test.go b/mcp/transport_test.go index 52228d01..bd2acc93 100644 --- a/mcp/transport_test.go +++ b/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()) + } + }) + } +}