From d80f8d393ded56907ba5f4f7a848b1712eed6855 Mon Sep 17 00:00:00 2001 From: Ukjae Jeong Date: Thu, 8 Oct 2026 14:46:06 +0100 Subject: [PATCH 1/5] mcp: prevent a failed request from breaking a stateless streamable client One failed call broke the streamable client connection, even with stateless, even after the server recovered. Signed-off-by: Ukjae Jeong --- internal/jsonrpc2/conn.go | 18 ++++++++ mcp/streamable.go | 86 ++++++++++++++++++++++++++--------- mcp/streamable_client_test.go | 21 ++++----- 3 files changed, 92 insertions(+), 33 deletions(-) diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go index 94da6ae3..36319acd 100644 --- a/internal/jsonrpc2/conn.go +++ b/internal/jsonrpc2/conn.go @@ -199,6 +199,21 @@ type Writer interface { Write(context.Context, Message) error } +// NonFatal marks an error returned by [Writer.Write] as a failure of the +// written message only: the connection remains usable. Unlike [ErrRejected], +// the caller receives err itself. +func NonFatal(err error) error { + if err == nil { + return nil + } + return &nonFatalError{err} +} + +type nonFatalError struct{ err error } + +func (e *nonFatalError) Error() string { return e.err.Error() } +func (e *nonFatalError) Unwrap() error { return e.err } + // A ConnectionConfig configures a bidirectional jsonrpc2 connection. type ConnectionConfig struct { Reader Reader // required @@ -769,6 +784,9 @@ func (c *Connection) write(ctx context.Context, msg Message) error { if err == nil { err = c.writer.Write(ctx, msg) } + if nf, ok := err.(*nonFatalError); ok { + return nf.err + } // For cancelled or rejected requests, we don't set the writeErr (which would // break the connection). They can just be returned to the caller. diff --git a/mcp/streamable.go b/mcp/streamable.go index 6e28051b..c6798f81 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -2315,7 +2315,8 @@ type streamableClientConn struct { // Logical reads are distributed across multiple http requests. Whenever any // of them fails to process their response, we must break the connection, by - // failing the pending Read. + // failing the pending Read. Without a session, only that call is ended + // instead (see [streamableClientConn.endCall]). // // Achieve this by storing the failure message, and signalling when reads are // broken. See also [streamableClientConn.fail] and @@ -2427,6 +2428,14 @@ func (c *streamableClientConn) fail(err error) { } } +// endCall ends call with err, leaving the connection usable. +func (c *streamableClientConn) endCall(call *jsonrpc.Request, err error) { + select { + case c.incoming <- &jsonrpc2.Response{ID: call.ID, Error: err}: + case <-c.done: + } +} + func (c *streamableClientConn) failure() error { select { case <-c.failed: @@ -2558,6 +2567,8 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e // plain text 404 or 400), which checkResponse cannot classify as a // per-call rejection on its own. err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected) + } else if !errors.Is(err, jsonrpc2.ErrRejected) && c.SessionID() == "" { + return jsonrpc2.NonFatal(err) } else if !errors.Is(err, jsonrpc2.ErrRejected) { // Only fail the connection for non-transient errors. // Transient errors (wrapped with ErrRejected) should not break the connection. @@ -2602,7 +2613,7 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e contentType := baseMediaType(resp.Header.Get("Content-Type")) switch contentType { case "application/json": - go c.handleJSON(requestSummary, resp) + go c.handleJSON(requestSummary, resp, forCall) case "text/event-stream": var forCall *jsonrpc.Request @@ -2617,7 +2628,11 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e default: resp.Body.Close() - return fmt.Errorf("%s: unsupported content type %q", requestSummary, contentType) + err := fmt.Errorf("%s: unsupported content type %q", requestSummary, contentType) + if c.SessionID() == "" { + return jsonrpc2.NonFatal(err) + } + return err } return nil } @@ -2683,16 +2698,26 @@ func protocolVersionFromMessage(msg jsonrpc.Message) string { return v } -func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response) { +func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response, forCall *jsonrpc.Request) { body, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { - c.fail(fmt.Errorf("%s: failed to read body: %v", requestSummary, err)) + err = fmt.Errorf("%s: failed to read body: %v", requestSummary, err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } return } msg, err := jsonrpc.DecodeMessage(body) if err != nil { - c.fail(fmt.Errorf("%s: failed to decode response: %v", requestSummary, err)) + err = fmt.Errorf("%s: failed to decode response: %v", requestSummary, err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } return } select { @@ -2741,7 +2766,12 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str retriesWithoutProgress++ if retriesWithoutProgress > c.maxRetries { if ctx.Err() == nil { - c.fail(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID)) + err := fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } } return } @@ -2751,17 +2781,26 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str newResp, err := c.connectSSE(ctx, lastEventID, reconnectDelay, false) if err != nil { // If the client didn't cancel this request, any failure to execute it - // breaks the logical MCP session. + // breaks the logical MCP session, if there is one. if ctx.Err() == nil { - // All reconnection attempts failed: fail the connection. - c.fail(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err)) + // All reconnection attempts failed. + err = fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } } return } resp = newResp if err := c.checkResponse(ctx, requestSummary, resp); err != nil { - c.fail(err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } return } } @@ -2824,9 +2863,14 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary } // Malformed events are hard errors that indicate corrupted data or protocol - // violations. These should fail the connection permanently. + // violations, so the stream is not resumed. if errors.Is(err, errMalformedEvent) { - c.fail(fmt.Errorf("%s: %v", requestSummary, err)) + err = fmt.Errorf("%s: %v", requestSummary, err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } return "", 0, true } @@ -2858,7 +2902,12 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary msg, err := jsonrpc.DecodeMessage(evt.Data) if err != nil { - c.fail(fmt.Errorf("%s: failed to decode event: %v", requestSummary, err)) + err = fmt.Errorf("%s: failed to decode event: %v", requestSummary, err) + if forCall != nil && c.SessionID() == "" { + c.endCall(forCall, err) + } else { + c.fail(err) + } return "", 0, true } @@ -2887,14 +2936,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary // Note that this is different from the cancellation case above, since the // caller is still waiting for a response that will never come. if lastEventID == "" && forCall != nil { - errmsg := &jsonrpc2.Response{ - ID: forCall.ID, - Error: fmt.Errorf("request terminated without response"), - } - select { - case c.incoming <- errmsg: - case <-c.done: - } + c.endCall(forCall, fmt.Errorf("request terminated without response")) } return lastEventID, reconnectDelay, false } diff --git a/mcp/streamable_client_test.go b/mcp/streamable_client_test.go index cd141260..c47b0321 100644 --- a/mcp/streamable_client_test.go +++ b/mcp/streamable_client_test.go @@ -1854,24 +1854,23 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { }, { // Same server response, but the user opted out via - // MCPGODEBUG=noprotocolerrorbody=1. Pre-fix behavior: the - // session is torn down. + // MCPGODEBUG=noprotocolerrorbody=1. The body is ignored, but + // without a session the failure still fails only the call. name: "400 with JSON-RPC error body (noprotocolerrorbody=1)", callStatus: http.StatusBadRequest, callBody: jsonRPCErrorBody(1, CodeMissingRequiredClientCapabilities, "missing capability"), disableBodyDecode: true, wantErrRejected: false, - wantSessionAlive: false, + wantSessionAlive: true, }, { - // Legacy server returns plain-text 400 with no JSON-RPC - // body: cannot be classified as a per-call rejection, so the - // session is still torn down. + // Plain-text 400 with no JSON-RPC body: without a session, + // only the call fails. name: "400 with plain-text body", callStatus: http.StatusBadRequest, callBody: "Bad Request", wantErrRejected: false, - wantSessionAlive: false, + wantSessionAlive: true, }, { // SEP-2575: server returns -32601 for an unimplemented @@ -1884,13 +1883,13 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { wantSessionAlive: true, }, { - // A bare 404 with no JSON-RPC body means the session has - // been terminated. Session must not survive. - name: "404 with empty body (terminated session)", + // A bare 404 with no JSON-RPC body: there is no session to + // terminate, so only the call fails. + name: "404 with empty body", callStatus: http.StatusNotFound, callBody: "", wantErrRejected: false, - wantSessionAlive: false, + wantSessionAlive: true, }, } From c91e3345484d663a366935c006902b95bacdad18 Mon Sep 17 00:00:00 2001 From: Ukjae Jeong Date: Thu, 15 Oct 2026 10:20:01 +0100 Subject: [PATCH 2/5] mcp: address comment Signed-off-by: Ukjae Jeong --- docs/mcpgodebug.md | 5 ++ internal/docs/mcpgodebug.src.md | 5 ++ internal/jsonrpc2/conn.go | 18 ------ mcp/streamable.go | 60 +++++++++---------- mcp/streamable_client_test.go | 100 ++++++++++++++++++++++++++++---- 5 files changed, 127 insertions(+), 61 deletions(-) diff --git a/docs/mcpgodebug.md b/docs/mcpgodebug.md index 01e96285..f463c6b7 100644 --- a/docs/mcpgodebug.md +++ b/docs/mcpgodebug.md @@ -49,6 +49,11 @@ Options listed below were added and will be removed in the 1.11.0 version of the tells a client nothing and is indistinguishable from an unset code. See issue #1367. +- `fatalnewprotocolerrors` added. If set to `1`, a streamable HTTP client + permanently fails its connection when an independent `2026-07-28` request + fails, restoring the previous behavior. By default, the failure is returned + only to the affected call. + ### 1.8.0 Options listed below were added and will be removed in the 1.9.0 version of the SDK. diff --git a/internal/docs/mcpgodebug.src.md b/internal/docs/mcpgodebug.src.md index 0a75d81d..2129e370 100644 --- a/internal/docs/mcpgodebug.src.md +++ b/internal/docs/mcpgodebug.src.md @@ -48,6 +48,11 @@ Options listed below were added and will be removed in the 1.11.0 version of the tells a client nothing and is indistinguishable from an unset code. See issue #1367. +- `fatalnewprotocolerrors` added. If set to `1`, a streamable HTTP client + permanently fails its connection when an independent `2026-07-28` request + fails, restoring the previous behavior. By default, the failure is returned + only to the affected call. + ### 1.8.0 Options listed below were added and will be removed in the 1.9.0 version of the SDK. diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go index 36319acd..94da6ae3 100644 --- a/internal/jsonrpc2/conn.go +++ b/internal/jsonrpc2/conn.go @@ -199,21 +199,6 @@ type Writer interface { Write(context.Context, Message) error } -// NonFatal marks an error returned by [Writer.Write] as a failure of the -// written message only: the connection remains usable. Unlike [ErrRejected], -// the caller receives err itself. -func NonFatal(err error) error { - if err == nil { - return nil - } - return &nonFatalError{err} -} - -type nonFatalError struct{ err error } - -func (e *nonFatalError) Error() string { return e.err.Error() } -func (e *nonFatalError) Unwrap() error { return e.err } - // A ConnectionConfig configures a bidirectional jsonrpc2 connection. type ConnectionConfig struct { Reader Reader // required @@ -784,9 +769,6 @@ func (c *Connection) write(ctx context.Context, msg Message) error { if err == nil { err = c.writer.Write(ctx, msg) } - if nf, ok := err.(*nonFatalError); ok { - return nf.err - } // For cancelled or rejected requests, we don't set the writeErr (which would // break the connection). They can just be returned to the caller. diff --git a/mcp/streamable.go b/mcp/streamable.go index 1080315c..1b4a28b3 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -347,6 +347,12 @@ var allowsessionsinstateless = mcpgodebug.Value("allowsessionsinstateless") // permanently fails the connection. var noprotocolerrorbody = mcpgodebug.Value("noprotocolerrorbody") +// fatalnewprotocolerrors is a compatibility parameter that restores the +// previous behavior of permanently failing the connection when an independent +// new-protocol request fails. The option will be removed in the 1.11.0 version +// of the SDK. +var fatalnewprotocolerrors = mcpgodebug.Value("fatalnewprotocolerrors") + // plaintextstatefulrejection is a compatibility parameter that restores the // previous behavior of a stateful [StreamableHTTPHandler] when it receives a // request carrying SEP-2575 per-request metadata (i.e. an @@ -2325,8 +2331,8 @@ type streamableClientConn struct { // Logical reads are distributed across multiple http requests. Whenever any // of them fails to process their response, we must break the connection, by - // failing the pending Read. Without a session, only that call is ended - // instead (see [streamableClientConn.endCall]). + // failing the pending Read. An independent new-protocol request only ends + // that call instead (see [streamableClientConn.endCall]). // // Achieve this by storing the failure message, and signalling when reads are // broken. See also [streamableClientConn.fail] and @@ -2577,8 +2583,10 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e // plain text 404 or 400), which checkResponse cannot classify as a // per-call rejection on its own. err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected) - } else if !errors.Is(err, jsonrpc2.ErrRejected) && c.SessionID() == "" { - return jsonrpc2.NonFatal(err) + } else if !errors.Is(err, jsonrpc2.ErrRejected) && canFailIndependently(msg) { + // An independent request failure applies only to that request. Wrap it with + // ErrRejected so jsonrpc2 does not treat the connection as broken. + err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected) } else if !errors.Is(err, jsonrpc2.ErrRejected) { // Only fail the connection for non-transient errors. // Transient errors (wrapped with ErrRejected) should not break the connection. @@ -2638,11 +2646,7 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e default: resp.Body.Close() - err := fmt.Errorf("%s: unsupported content type %q", requestSummary, contentType) - if c.SessionID() == "" { - return jsonrpc2.NonFatal(err) - } - return err + return fmt.Errorf("%s: unsupported content type %q", requestSummary, contentType) } return nil } @@ -2708,12 +2712,18 @@ func protocolVersionFromMessage(msg jsonrpc.Message) string { return v } +// canFailIndependently reports whether msg carries SEP-2575 per-request +// metadata and the caller has not opted out of independent failure handling. +func canFailIndependently(msg jsonrpc.Message) bool { + return fatalnewprotocolerrors != "1" && protocolVersionFromMessage(msg) != "" +} + func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response, forCall *jsonrpc.Request) { body, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { err = fmt.Errorf("%s: failed to read body: %v", requestSummary, err) - if forCall != nil && c.SessionID() == "" { + if canFailIndependently(forCall) { c.endCall(forCall, err) } else { c.fail(err) @@ -2723,7 +2733,7 @@ func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Resp msg, err := jsonrpc.DecodeMessage(body) if err != nil { err = fmt.Errorf("%s: failed to decode response: %v", requestSummary, err) - if forCall != nil && c.SessionID() == "" { + if canFailIndependently(forCall) { c.endCall(forCall, err) } else { c.fail(err) @@ -2776,12 +2786,7 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str retriesWithoutProgress++ if retriesWithoutProgress > c.maxRetries { if ctx.Err() == nil { - err := fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID) - if forCall != nil && c.SessionID() == "" { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.fail(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID)) } return } @@ -2791,26 +2796,17 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str newResp, err := c.connectSSE(ctx, lastEventID, reconnectDelay, false) if err != nil { // If the client didn't cancel this request, any failure to execute it - // breaks the logical MCP session, if there is one. + // breaks the logical MCP session. if ctx.Err() == nil { - // All reconnection attempts failed. - err = fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err) - if forCall != nil && c.SessionID() == "" { - c.endCall(forCall, err) - } else { - c.fail(err) - } + // All reconnection attempts failed: fail the connection. + c.fail(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err)) } return } resp = newResp if err := c.checkResponse(ctx, requestSummary, resp); err != nil { - if forCall != nil && c.SessionID() == "" { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.fail(err) return } } @@ -2876,7 +2872,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary // violations, so the stream is not resumed. if errors.Is(err, errMalformedEvent) { err = fmt.Errorf("%s: %v", requestSummary, err) - if forCall != nil && c.SessionID() == "" { + if canFailIndependently(forCall) { c.endCall(forCall, err) } else { c.fail(err) @@ -2913,7 +2909,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary msg, err := jsonrpc.DecodeMessage(evt.Data) if err != nil { err = fmt.Errorf("%s: failed to decode event: %v", requestSummary, err) - if forCall != nil && c.SessionID() == "" { + if canFailIndependently(forCall) { c.endCall(forCall, err) } else { c.fail(err) diff --git a/mcp/streamable_client_test.go b/mcp/streamable_client_test.go index 89653f2a..16bd6774 100644 --- a/mcp/streamable_client_test.go +++ b/mcp/streamable_client_test.go @@ -1840,6 +1840,8 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { callStatus int // HTTP status returned for the tools/call callBody string // response body for the tools/call ("" for empty; may be a JSON-RPC error) disableBodyDecode bool // if true, sets noprotocolerrorbody="1" for the duration of the test + fatalNewErrors bool // if true, sets fatalnewprotocolerrors="1" for the duration of the test + protocolVersion string // defaults to protocolVersion20260728 wantErrRejected bool // whether the returned error should match errors.Is(err, jsonrpc2.ErrRejected) wantSessionAlive bool // whether a subsequent call on the same session should succeed }{ @@ -1852,6 +1854,31 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { wantErrRejected: true, wantSessionAlive: true, }, + { + // Plain-text 400 with no JSON-RPC body: without a session, + // only the call fails. + name: "400 with plain-text body", + callStatus: http.StatusBadRequest, + callBody: "Bad Request", + wantErrRejected: true, + wantSessionAlive: true, + }, + { + name: "400 with plain-text body (legacy protocol)", + callStatus: http.StatusBadRequest, + callBody: "Bad Request", + protocolVersion: protocolVersion20251125, + wantErrRejected: false, + wantSessionAlive: false, + }, + { + name: "400 with plain-text body (fatalnewprotocolerrors=1)", + callStatus: http.StatusBadRequest, + callBody: "Bad Request", + fatalNewErrors: true, + wantErrRejected: false, + wantSessionAlive: false, + }, { // Same server response, but the user opted out via // MCPGODEBUG=noprotocolerrorbody=1. The body is ignored, but @@ -1860,17 +1887,20 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { callStatus: http.StatusBadRequest, callBody: jsonRPCErrorBody(1, CodeMissingRequiredClientCapabilities, "missing capability"), disableBodyDecode: true, - wantErrRejected: false, + wantErrRejected: true, wantSessionAlive: true, }, { - // Plain-text 400 with no JSON-RPC body: without a session, - // only the call fails. - name: "400 with plain-text body", - callStatus: http.StatusBadRequest, - callBody: "Bad Request", - wantErrRejected: false, - wantSessionAlive: true, + // Same server response, but the user opted out via + // MCPGODEBUG=noprotocolerrorbody=1 and fatalnewprotocolerrors=1. + // Pre-fix behavior: the session is torn down. + name: "400 with JSON-RPC error body (noprotocolerrorbody=1, fatalnewprotocolerrors=1)", + callStatus: http.StatusBadRequest, + callBody: jsonRPCErrorBody(1, CodeMissingRequiredClientCapabilities, "missing capability"), + disableBodyDecode: true, + fatalNewErrors: true, + wantErrRejected: false, + wantSessionAlive: false, }, { // SEP-2575: server returns -32601 for an unimplemented @@ -1882,25 +1912,58 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { wantErrRejected: true, wantSessionAlive: true, }, + { + // A bare 404 with no JSON-RPC body means the session has + // been terminated. Session must not survive. + name: "404 with empty body (terminated session, fatalnewprotocolerrors=1)", + callStatus: http.StatusNotFound, + callBody: "", + fatalNewErrors: true, + wantErrRejected: false, + wantSessionAlive: false, + }, { // A bare 404 with no JSON-RPC body: there is no session to // terminate, so only the call fails. name: "404 with empty body", callStatus: http.StatusNotFound, callBody: "", + wantErrRejected: true, + wantSessionAlive: true, + }, + { + name: "200 with invalid JSON body", + callStatus: http.StatusOK, + callBody: "not JSON", wantErrRejected: false, wantSessionAlive: true, }, + { + name: "200 with invalid JSON body (fatalnewprotocolerrors=1)", + callStatus: http.StatusOK, + callBody: "not JSON", + fatalNewErrors: true, + wantErrRejected: false, + wantSessionAlive: false, + }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { + protocolVersion := test.protocolVersion + if protocolVersion == "" { + protocolVersion = protocolVersion20260728 + } if test.disableBodyDecode { prev := noprotocolerrorbody noprotocolerrorbody = "1" t.Cleanup(func() { noprotocolerrorbody = prev }) } - + if test.fatalNewErrors { + prev := fatalnewprotocolerrors + fatalnewprotocolerrors = "1" + t.Cleanup(func() { fatalnewprotocolerrors = prev }) + } // Track how many tools/call requests we've served: first // returns the error under test, subsequent calls succeed so // we can observe whether the session is still alive. @@ -1917,6 +1980,18 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { return jsonBody(t, resp(r.ID.Raw().(int64), discoverResult, nil)), http.StatusOK }, }, + {"POST", "", methodInitialize, ""}: { + header: header{"Content-Type": "application/json"}, + responseFunc: func(r *jsonrpc.Request) (string, int) { + return jsonBody(t, &jsonrpc.Response{ID: r.ID, Result: mustMarshal(initResult)}), http.StatusOK + }, + optional: true, + }, + {"POST", "", notificationInitialized, ""}: { + status: http.StatusAccepted, + wantProtocolVersion: protocolVersion20251125, + optional: true, + }, {"POST", "", "tools/call", ""}: { header: header{"Content-Type": "application/json"}, responseFunc: func(r *jsonrpc.Request) (string, int) { @@ -1933,9 +2008,12 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { httpServer := httptest.NewServer(fake) defer httpServer.Close() - transport := &StreamableClientTransport{Endpoint: httpServer.URL} + transport := &StreamableClientTransport{ + Endpoint: httpServer.URL, + DisableStandaloneSSE: protocolVersion < protocolVersion20260728, + } client := NewClient(testImpl, nil) - session, err := client.Connect(ctx, transport, &ClientSessionOptions{ProtocolVersion: protocolVersion20260728}) + session, err := client.Connect(ctx, transport, &ClientSessionOptions{ProtocolVersion: protocolVersion}) if err != nil { t.Fatalf("Connect: %v", err) } From ca927d669790f1a5881e7c0c6871ab4e82b2fffa Mon Sep 17 00:00:00 2001 From: Ukjae Jeong Date: Thu, 15 Oct 2026 14:06:53 +0100 Subject: [PATCH 3/5] Update mcp/streamable.go Co-authored-by: Guglielmo Colombo --- mcp/streamable.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/mcp/streamable.go b/mcp/streamable.go index 1b4a28b3..0600be1f 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -2714,8 +2714,14 @@ func protocolVersionFromMessage(msg jsonrpc.Message) string { // canFailIndependently reports whether msg carries SEP-2575 per-request // metadata and the caller has not opted out of independent failure handling. -func canFailIndependently(msg jsonrpc.Message) bool { - return fatalnewprotocolerrors != "1" && protocolVersionFromMessage(msg) != "" +func (c *streamableClientConn) canFailIndependently(msg jsonrpc.Message) bool { + if fatalnewprotocolerrors == "1" { + return false + } + if pv := protocolVersionFromMessage(msg); pv != "" { + return pv >= protocolVersion20260728 + } + return c.initializedResult != nil && c.initializedResult.ProtocolVersion >= protocolVersion20260728 } func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response, forCall *jsonrpc.Request) { From ca9094d95c014c0b777088f3abaac2bf13952ff9 Mon Sep 17 00:00:00 2001 From: Ukjae Jeong Date: Thu, 15 Oct 2026 14:22:46 +0100 Subject: [PATCH 4/5] mcp: address comment Signed-off-by: Ukjae Jeong --- mcp/streamable.go | 90 ++++++++++++++++++++++------------------------- 1 file changed, 43 insertions(+), 47 deletions(-) diff --git a/mcp/streamable.go b/mcp/streamable.go index 0600be1f..512e6403 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -2332,10 +2332,10 @@ type streamableClientConn struct { // Logical reads are distributed across multiple http requests. Whenever any // of them fails to process their response, we must break the connection, by // failing the pending Read. An independent new-protocol request only ends - // that call instead (see [streamableClientConn.endCall]). + // that call instead (see [streamableClientConn.failCall]). // // Achieve this by storing the failure message, and signalling when reads are - // broken. See also [streamableClientConn.fail] and + // broken. See also [streamableClientConn.failConnection] and // [streamableClientConn.failure]. failOnce sync.Once _failure error @@ -2385,7 +2385,7 @@ func (c *streamableClientConn) connectStandaloneSSE() { // If the client didn't cancel the request, and failure breaks the logical // session. if c.ctx.Err() == nil { - c.fail(fmt.Errorf("standalone SSE request failed (session ID: %v): %v", c.sessionID, err)) + c.failConnection(fmt.Errorf("standalone SSE request failed (session ID: %v): %v", c.sessionID, err)) } return } @@ -2421,13 +2421,13 @@ func (c *streamableClientConn) connectStandaloneSSE() { } summary := "standalone SSE stream" if err := c.checkResponse(c.ctx, summary, resp); err != nil { - c.fail(err) + c.failConnection(err) return } go c.handleSSE(c.ctx, summary, resp, nil) } -// fail handles an asynchronous error while reading. +// failConnection handles an asynchronous error while reading. // // If err is non-nil, it is terminal, and subsequent (or pending) Reads will // fail. @@ -2435,7 +2435,7 @@ func (c *streamableClientConn) connectStandaloneSSE() { // If err wraps ErrSessionMissing, the failure indicates that the session is no // longer present on the server, and no final DELETE will be performed when // closing the connection. -func (c *streamableClientConn) fail(err error) { +func (c *streamableClientConn) failConnection(err error) { if err != nil { c.failOnce.Do(func() { c._failure = err @@ -2444,14 +2444,38 @@ func (c *streamableClientConn) fail(err error) { } } -// endCall ends call with err, leaving the connection usable. -func (c *streamableClientConn) endCall(call *jsonrpc.Request, err error) { +// failCall ends call with err, leaving the connection usable. +func (c *streamableClientConn) failCall(call *jsonrpc.Request, err error) { select { case c.incoming <- &jsonrpc2.Response{ID: call.ID, Error: err}: case <-c.done: } } +// canFailIndependently reports whether msg belongs to a new-protocol +// interaction and the caller has not opted out of independent failure handling. +func (c *streamableClientConn) canFailIndependently(msg jsonrpc.Message) bool { + if fatalnewprotocolerrors == "1" { + return false + } + if pv := protocolVersionFromMessage(msg); pv != "" { + return pv >= protocolVersion20260728 + } + c.mu.Lock() + defer c.mu.Unlock() + return c.initializedResult != nil && c.initializedResult.ProtocolVersion >= protocolVersion20260728 +} + +// endCall fails only the call when it can fail independently, or fails the +// connection otherwise. +func (c *streamableClientConn) endCall(call *jsonrpc.Request, err error) { + if call != nil && c.canFailIndependently(call) { + c.failCall(call, err) + } else { + c.failConnection(err) + } +} + func (c *streamableClientConn) failure() error { select { case <-c.failed: @@ -2559,7 +2583,7 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e // source for whether the caller abandoned the request. ctxErr := ctx.Err() if errors.Is(ctxErr, context.Canceled) || errors.Is(ctxErr, context.DeadlineExceeded) { - c.fail(fmt.Errorf("%s: authorization cancelled: %w", requestSummary, err)) + c.failConnection(fmt.Errorf("%s: authorization cancelled: %w", requestSummary, err)) } // Wrap with ErrRejected so the jsonrpc2 connection doesn't set writeErr // and permanently break the connection. @@ -2583,14 +2607,14 @@ func (c *streamableClientConn) Write(ctx context.Context, msg jsonrpc.Message) e // plain text 404 or 400), which checkResponse cannot classify as a // per-call rejection on its own. err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected) - } else if !errors.Is(err, jsonrpc2.ErrRejected) && canFailIndependently(msg) { + } else if !errors.Is(err, jsonrpc2.ErrRejected) && c.canFailIndependently(msg) { // An independent request failure applies only to that request. Wrap it with // ErrRejected so jsonrpc2 does not treat the connection as broken. err = fmt.Errorf("%w: %w", err, jsonrpc2.ErrRejected) } else if !errors.Is(err, jsonrpc2.ErrRejected) { // Only fail the connection for non-transient errors. // Transient errors (wrapped with ErrRejected) should not break the connection. - c.fail(err) + c.failConnection(err) } return err } @@ -2712,38 +2736,18 @@ func protocolVersionFromMessage(msg jsonrpc.Message) string { return v } -// canFailIndependently reports whether msg carries SEP-2575 per-request -// metadata and the caller has not opted out of independent failure handling. -func (c *streamableClientConn) canFailIndependently(msg jsonrpc.Message) bool { - if fatalnewprotocolerrors == "1" { - return false - } - if pv := protocolVersionFromMessage(msg); pv != "" { - return pv >= protocolVersion20260728 - } - return c.initializedResult != nil && c.initializedResult.ProtocolVersion >= protocolVersion20260728 -} - func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Response, forCall *jsonrpc.Request) { body, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { err = fmt.Errorf("%s: failed to read body: %v", requestSummary, err) - if canFailIndependently(forCall) { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.endCall(forCall, err) return } msg, err := jsonrpc.DecodeMessage(body) if err != nil { err = fmt.Errorf("%s: failed to decode response: %v", requestSummary, err) - if canFailIndependently(forCall) { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.endCall(forCall, err) return } select { @@ -2792,7 +2796,7 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str retriesWithoutProgress++ if retriesWithoutProgress > c.maxRetries { if ctx.Err() == nil { - c.fail(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID)) + c.failConnection(fmt.Errorf("%s: exceeded %d retries without progress (session ID: %v)", requestSummary, c.maxRetries, c.sessionID)) } return } @@ -2805,14 +2809,14 @@ func (c *streamableClientConn) handleSSE(ctx context.Context, requestSummary str // breaks the logical MCP session. if ctx.Err() == nil { // All reconnection attempts failed: fail the connection. - c.fail(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err)) + c.failConnection(fmt.Errorf("%s: failed to reconnect (session ID: %v): %v", requestSummary, c.sessionID, err)) } return } resp = newResp if err := c.checkResponse(ctx, requestSummary, resp); err != nil { - c.fail(err) + c.failConnection(err) return } } @@ -2878,11 +2882,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary // violations, so the stream is not resumed. if errors.Is(err, errMalformedEvent) { err = fmt.Errorf("%s: %v", requestSummary, err) - if canFailIndependently(forCall) { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.endCall(forCall, err) return "", 0, true } @@ -2915,11 +2915,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary msg, err := jsonrpc.DecodeMessage(evt.Data) if err != nil { err = fmt.Errorf("%s: failed to decode event: %v", requestSummary, err) - if canFailIndependently(forCall) { - c.endCall(forCall, err) - } else { - c.fail(err) - } + c.endCall(forCall, err) return "", 0, true } @@ -2948,7 +2944,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary // Note that this is different from the cancellation case above, since the // caller is still waiting for a response that will never come. if lastEventID == "" && forCall != nil { - c.endCall(forCall, fmt.Errorf("request terminated without response")) + c.failCall(forCall, fmt.Errorf("request terminated without response")) } return lastEventID, reconnectDelay, false } From 1d844b4e13708f9fbe5b3b1cf1961710ffedf84f Mon Sep 17 00:00:00 2001 From: Ukjae Jeong Date: Thu, 15 Oct 2026 14:25:13 +0100 Subject: [PATCH 5/5] mcp: simplify calls Signed-off-by: Ukjae Jeong --- mcp/streamable.go | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/mcp/streamable.go b/mcp/streamable.go index 512e6403..3087c46e 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -2740,14 +2740,12 @@ func (c *streamableClientConn) handleJSON(requestSummary string, resp *http.Resp body, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { - err = fmt.Errorf("%s: failed to read body: %v", requestSummary, err) - c.endCall(forCall, err) + c.endCall(forCall, fmt.Errorf("%s: failed to read body: %v", requestSummary, err)) return } msg, err := jsonrpc.DecodeMessage(body) if err != nil { - err = fmt.Errorf("%s: failed to decode response: %v", requestSummary, err) - c.endCall(forCall, err) + c.endCall(forCall, fmt.Errorf("%s: failed to decode response: %v", requestSummary, err)) return } select { @@ -2881,8 +2879,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary // Malformed events are hard errors that indicate corrupted data or protocol // violations, so the stream is not resumed. if errors.Is(err, errMalformedEvent) { - err = fmt.Errorf("%s: %v", requestSummary, err) - c.endCall(forCall, err) + c.endCall(forCall, fmt.Errorf("%s: %v", requestSummary, err)) return "", 0, true } @@ -2914,8 +2911,7 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary msg, err := jsonrpc.DecodeMessage(evt.Data) if err != nil { - err = fmt.Errorf("%s: failed to decode event: %v", requestSummary, err) - c.endCall(forCall, err) + c.endCall(forCall, fmt.Errorf("%s: failed to decode event: %v", requestSummary, err)) return "", 0, true }