mcp: allow SSE messages with empty data (SEP-1699)
This commit is contained in:
parent
1fb6e204d7
commit
d0922abcd3
3 changed files with 59 additions and 1 deletions
|
|
@ -19,4 +19,3 @@ client:
|
||||||
- auth/client-credentials-jwt
|
- auth/client-credentials-jwt
|
||||||
- auth/client-credentials-basic
|
- auth/client-credentials-basic
|
||||||
- auth/pre-registration
|
- auth/pre-registration
|
||||||
- sse-retry
|
|
||||||
|
|
|
||||||
|
|
@ -1935,6 +1935,15 @@ func (c *streamableClientConn) processStream(ctx context.Context, requestSummary
|
||||||
reconnectDelay = time.Duration(n) * time.Millisecond
|
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"
|
// According to SSE spec, events with no name default to "message"
|
||||||
if evt.Name != "" && evt.Name != "message" {
|
if evt.Name != "" && evt.Name != "message" {
|
||||||
continue
|
continue
|
||||||
|
|
|
||||||
|
|
@ -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
|
// TestScanEventsPingFiltering is a unit test for the low-level event scanning
|
||||||
// with ping events to verify scanEvents properly parses all event types.
|
// with ping events to verify scanEvents properly parses all event types.
|
||||||
func TestScanEventsPingFiltering(t *testing.T) {
|
func TestScanEventsPingFiltering(t *testing.T) {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue