Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 17 additions & 7 deletions internal/jsonrpc2/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,9 @@ type Connection struct {
writer Writer
handler Handler

onInternalError func(error)
onDone func()
onInternalError func(error)
onResponseWriteError func(*Request, error)
onDone func()
}

// inFlightState records the state of the incoming and outgoing calls on a
Expand Down Expand Up @@ -219,6 +220,11 @@ type ConnectionConfig struct {
OnDone func() // optional
OnInternalError func(error) // optional

// OnResponseWriteError, if set, is called when the response to an incoming
// call could not be written, for example because the peer went away before
// the handler finished. The response is lost: the peer never receives it.
OnResponseWriteError func(*Request, error) // optional

// PropagateCancellation controls whether cancellation of the context
// passed to [NewConnection] is observable by request handlers.
//
Expand All @@ -244,11 +250,12 @@ func NewConnection(ctx context.Context, cfg ConnectionConfig) *Connection {
}

c := &Connection{
state: inFlightState{closer: cfg.Closer},
done: make(chan struct{}),
writer: cfg.Writer,
onDone: cfg.OnDone,
onInternalError: cfg.OnInternalError,
state: inFlightState{closer: cfg.Closer},
done: make(chan struct{}),
writer: cfg.Writer,
onDone: cfg.OnDone,
onInternalError: cfg.OnInternalError,
onResponseWriteError: cfg.OnResponseWriteError,
}
c.handler = cfg.Bind(c)
c.start(ctx, cfg.Reader, cfg.Preempter)
Expand Down Expand Up @@ -759,6 +766,9 @@ func (c *Connection) processResult(from any, req *incomingRequest, result any, e
delete(s.incomingByID, req.ID)
})
writeErr := c.write(notDone{req.ctx}, response)
if writeErr != nil && c.onResponseWriteError != nil {
c.onResponseWriteError(req.Request, writeErr)
}
if err == nil {
err = writeErr
}
Expand Down
3 changes: 3 additions & 0 deletions mcp/transport.go
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,9 @@ func connect[H handler, State any](ctx context.Context, t Transport, b binder[H,
OnInternalError: func(err error) {
logger.Error("jsonrpc2 internal error", "error", err)
},
OnResponseWriteError: func(req *jsonrpc2.Request, err error) {
logger.Warn("failed to write response", "method", req.Method, "id", req.ID.Raw(), "error", err)
},
PropagateCancellation: propagateCancellation,
})
assert(preempter.conn != nil, "unbound preempter")
Expand Down
42 changes: 42 additions & 0 deletions mcp/transport_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,12 @@ import (
"errors"
"fmt"
"io"
"log/slog"
"strings"
"testing"
"time"

"github.com/google/jsonschema-go/jsonschema"
"github.com/modelcontextprotocol/go-sdk/internal/jsonrpc2"
"github.com/modelcontextprotocol/go-sdk/jsonrpc"
)
Expand Down Expand Up @@ -132,6 +134,46 @@ func connectionClosedTests() []connectionClosedTest {
}
}

// TestServerLogsUndeliveredResponse is a regression test for #1147: when the
// client goes away before a tool handler finishes, the server logs that the
// response could not be delivered instead of dropping it silently.
func TestServerLogsUndeliveredResponse(t *testing.T) {
ctx := context.Background()
var buf bytes.Buffer
server := NewServer(testImpl, &ServerOptions{
Logger: slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelWarn})),
})
started := make(chan struct{})
server.AddTool(&Tool{Name: "slow", InputSchema: &jsonschema.Schema{Type: "object"}},
func(ctx context.Context, _ *CallToolRequest) (*CallToolResult, error) {
close(started)
<-ctx.Done() // cancelled once the client's connection is gone
return &CallToolResult{}, nil
})
serverIn, clientOut := io.Pipe()
clientIn, serverOut := io.Pipe()
ss, err := server.Connect(ctx, &IOTransport{Reader: serverIn, Writer: serverOut}, nil)
if err != nil {
t.Fatal(err)
}
cs, err := NewClient(testImpl, nil).Connect(ctx, &IOTransport{Reader: clientIn, Writer: clientOut}, nil)
if err != nil {
t.Fatal(err)
}
defer cs.Close()
go cs.CallTool(ctx, &CallToolParams{Name: "slow"})
<-started
// The client goes away abruptly, as one that gives up on a slow call does.
clientOut.Close()
clientIn.Close()
ss.Wait()

got := buf.String()
if !strings.Contains(got, `level=WARN msg="failed to write response" method=tools/call id=`) {
t.Errorf("server log does not report the undelivered response:\n%s", got)
}
}

func checkConnectionClosedError(t *testing.T, err error, causes ...error) {
t.Helper()
for _, want := range append([]error{ErrConnectionClosed}, causes...) {
Expand Down
Loading