diff --git a/conformance/baseline.yml b/conformance/baseline.yml index d731cd1..f0e1513 100644 --- a/conformance/baseline.yml +++ b/conformance/baseline.yml @@ -19,4 +19,3 @@ client: - auth/client-credentials-jwt - auth/client-credentials-basic - auth/pre-registration -- sse-retry diff --git a/mcp/streamable.go b/mcp/streamable.go index 893a706..c6b96bb 100644 --- a/mcp/streamable.go +++ b/mcp/streamable.go @@ -1935,6 +1935,15 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary reconnectDelay = time.Duration(n) * time.Millisecond } } + + // According to SSE specification + // (https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation) + // events with an empty data buffer are allowed. + // In MCP these can be priming events (SEP-1699) that carry only a Last-Event-ID for stream resumption. + if len(evt.Data) == 0 { + continue + } + // According to SSE spec, events with no name default to "message" if evt.Name != "" && evt.Name != "message" { continue diff --git a/mcp/streamable_test.go b/mcp/streamable_test.go index 05af216..f1f6200 100644 --- a/mcp/streamable_test.go +++ b/mcp/streamable_test.go @@ -2218,6 +2218,56 @@ collectLoop: } } +// TestProcessStreamPrimingEvent verifies that the streamable client correctly ignores +// SSE events with empty data buffers, which are used as priming events (e.g. SEP-1699). +func TestProcessStreamPrimingEvent(t *testing.T) { + // We create a mock response with a priming event (empty data, with an ID), + // followed by a normal event. + sseData := `id: 123 + +id: 124 +data: {"jsonrpc":"2.0","id":1,"result":{}} + +` + + ctx := t.Context() + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(sseData)), + } + + incoming := make(chan jsonrpc.Message, 10) + done := make(chan struct{}) + + conn := &streamableClientConn{ + ctx: ctx, + done: done, + incoming: incoming, + failed: make(chan struct{}), + logger: ensureLogger(nil), + } + + lastID, _, clientClosed := conn.processStream(ctx, "test", resp, nil) + + if clientClosed { + t.Fatalf("processStream was unexpectedly closed by client") + } + + if lastID != "124" { + t.Errorf("lastEventID = %q, want %q", lastID, "124") + } + + select { + case msg := <-incoming: + if res, ok := msg.(*jsonrpc.Response); !(ok && res.ID == jsonrpc2.Int64ID(1)) { + t.Errorf("got unexpected message: %v", msg) + } + default: + t.Errorf("expected a JSON-RPC message to be produced") + } +} + // TestScanEventsPingFiltering is a unit test for the low-level event scanning // with ping events to verify scanEvents properly parses all event types. func TestScanEventsPingFiltering(t *testing.T) {