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/mcp/streamable.go b/mcp/streamable.go index a7ab2275..3087c46e 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,10 +2331,11 @@ 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. An independent new-protocol request only ends + // 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 @@ -2378,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 } @@ -2414,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. @@ -2428,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 @@ -2437,6 +2444,38 @@ func (c *streamableClientConn) fail(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: @@ -2544,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. @@ -2568,10 +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) && 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 } @@ -2612,7 +2655,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 @@ -2693,16 +2736,16 @@ 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)) + c.endCall(forCall, fmt.Errorf("%s: failed to read body: %v", requestSummary, err)) return } msg, err := jsonrpc.DecodeMessage(body) if err != nil { - c.fail(fmt.Errorf("%s: failed to decode response: %v", requestSummary, err)) + c.endCall(forCall, fmt.Errorf("%s: failed to decode response: %v", requestSummary, err)) return } select { @@ -2751,7 +2794,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 } @@ -2764,14 +2807,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 } } @@ -2834,9 +2877,9 @@ 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)) + c.endCall(forCall, fmt.Errorf("%s: %v", requestSummary, err)) return "", 0, true } @@ -2868,7 +2911,7 @@ 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)) + c.endCall(forCall, fmt.Errorf("%s: failed to decode event: %v", requestSummary, err)) return "", 0, true } @@ -2897,14 +2940,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.failCall(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 ae2319b2..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,26 +1854,53 @@ 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. 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, + wantErrRejected: true, + 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. - name: "400 with plain-text body", - callStatus: http.StatusBadRequest, - callBody: "Bad Request", - wantErrRejected: false, - wantSessionAlive: false, + // 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 @@ -1886,9 +1915,34 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { { // 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)", + 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, }, @@ -1896,12 +1950,20 @@ func TestStreamableClientHandlerErrorPropagation(t *testing.T) { 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. @@ -1918,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) { @@ -1934,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) }