diff --git a/internal/jsonrpc2/conn.go b/internal/jsonrpc2/conn.go index 94da6ae3..09afde79 100644 --- a/internal/jsonrpc2/conn.go +++ b/internal/jsonrpc2/conn.go @@ -714,6 +714,10 @@ func (c *Connection) processResult(from any, req *incomingRequest, result any, e } response, respErr := NewResponse(req.ID, result, err) + if respErr != nil { + err = c.internalErrorf("%#v returned a malformed result for %q: %w", from, req.Method, respErr) + response, _ = NewResponse(req.ID, nil, err) + } // The caller could theoretically reuse the request's ID as soon as we've // sent the response, so ensure that it is removed from the incoming map @@ -721,13 +725,9 @@ func (c *Connection) processResult(from any, req *incomingRequest, result any, e c.updateInFlight(func(s *inFlightState) { delete(s.incomingByID, req.ID) }) - if respErr == nil { - writeErr := c.write(notDone{req.ctx}, response) - if err == nil { - err = writeErr - } - } else { - err = c.internalErrorf("%#v returned a malformed result for %q: %w", from, req.Method, respErr) + writeErr := c.write(notDone{req.ctx}, response) + if err == nil { + err = writeErr } } else { // req is a notification if result != nil { diff --git a/internal/jsonrpc2/conn_test.go b/internal/jsonrpc2/conn_test.go index afeb4c02..064dd0e7 100644 --- a/internal/jsonrpc2/conn_test.go +++ b/internal/jsonrpc2/conn_test.go @@ -5,11 +5,79 @@ package jsonrpc2 import ( + "context" "errors" "io" + "math" + "strings" "testing" ) +type fakeReader struct { + msgs chan Message +} + +func (r fakeReader) Read(ctx context.Context) (Message, error) { + select { + case m := <-r.msgs: + return m, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +type fakeWriter struct{ out chan Message } + +func (w fakeWriter) Write(_ context.Context, m Message) error { + w.out <- m + return nil +} + +type nopCloser struct{} + +func (nopCloser) Close() error { return nil } + +// TestUnmarshalableResultSendsError verifies that a result that can't be +// marshaled is reported to the log and answered with a generic error response. +func TestUnmarshalableResultSendsError(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + call, err := NewCall(Int64ID(7), "m", nil) + if err != nil { + t.Fatal(err) + } + r := fakeReader{msgs: make(chan Message, 1)} + r.msgs <- call + w := fakeWriter{out: make(chan Message, 1)} + internalErrs := make(chan error, 1) + NewConnection(ctx, ConnectionConfig{ + Reader: r, + Writer: w, + Closer: nopCloser{}, + Bind: func(*Connection) Handler { + return HandlerFunc(func(context.Context, *Request) (any, error) { + return math.Inf(1), nil + }) + }, + OnInternalError: func(err error) { internalErrs <- err }, + }) + + resp, ok := (<-w.out).(*Response) + if !ok { + t.Fatal("want a Response") + } + if resp.ID != Int64ID(7) { + t.Errorf("response ID = %v, want 7", resp.ID) + } + var werr *WireError + if !errors.As(resp.Error, &werr) || werr.Code != -32603 || !strings.Contains(resp.Error.Error(), "malformed result") { + t.Errorf("response error = %v, want generic -32603 error", resp.Error) + } + if err := <-internalErrs; err == nil { + t.Error("want internal error to be logged") + } +} + // TestShuttingDownWrapsWriteError verifies that when a connection shuts down // because its write side failed, the returned error preserves the underlying // write error in its chain so callers can classify it with errors.Is (for diff --git a/mcp/streamable_test.go b/mcp/streamable_test.go index 5ad4ae10..aa19aa3a 100644 --- a/mcp/streamable_test.go +++ b/mcp/streamable_test.go @@ -14,6 +14,7 @@ import ( "fmt" "io" "maps" + "math" "net" "net/http" "net/http/cookiejar" @@ -1757,6 +1758,43 @@ func TestStreamableStateless(t *testing.T) { testClientCompatibility(t, statelessHandler) } +// TestStreamableUnencodableToolResult verifies that a tool result that cannot +// be marshaled produces a JSON-RPC error instead of hanging the request. +func TestStreamableUnencodableToolResult(t *testing.T) { + for _, stateless := range []bool{true, false} { + t.Run(fmt.Sprintf("stateless=%v", stateless), func(t *testing.T) { + server := NewServer(testImpl, nil) + AddTool(server, &Tool{Name: "bad"}, func(context.Context, *CallToolRequest, struct{}) (*CallToolResult, any, error) { + return &CallToolResult{ + Content: []Content{&TextContent{Text: "inf"}}, + StructuredContent: map[string]any{"value": math.Inf(1)}, + }, nil, nil + }) + handler := NewStreamableHTTPHandler(func(*http.Request) *Server { return server }, + &StreamableHTTPOptions{Stateless: stateless}) + httpServer := httptest.NewServer(mustNotPanic(t, handler)) + defer httpServer.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cs, err := NewClient(testImpl, nil).Connect(ctx, &StreamableClientTransport{Endpoint: httpServer.URL}, nil) + if err != nil { + t.Fatal(err) + } + defer cs.Close() + + _, err = cs.CallTool(ctx, &CallToolParams{Name: "bad", Arguments: map[string]any{}}) + var wireErr *jsonrpc.Error + if !errors.As(err, &wireErr) { + t.Fatalf("CallTool() error = %v, want a JSON-RPC error", err) + } + if wireErr.Code != jsonrpc.CodeInternalError { + t.Errorf("error code = %d, want %d", wireErr.Code, jsonrpc.CodeInternalError) + } + }) + } +} + // TestStreamableStateless_RejectsLegacyOn2026OnlyServer verifies that a // stateless server configured with [ServerOptions.SupportedProtocolVersions] // set to only ["2026-07-28"] rejects legacy-format requests (no _meta, no