mcp: support SSE polling and server disconnects (SEP-1699)

docs/rough_edges.md

@@ -11,6 +11,10 @@ v2.
 
   **Workaround**: `Open` may be implemented as a no-op.
 
+- `Event` need not have been exported: it's an implementation detail of the SSE
+  and streamable transports. Also the 'Name' field is a misnomer: it should be
+  'event'.
+
 - Enforcing valid tool names: with
   [SEP-986](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/986)
   landing after the SDK was at v1, we missed an opportunity to panic on invalid

internal/docs/rough_edges.src.md

@@ -10,6 +10,10 @@ v2.
 
   **Workaround**: `Open` may be implemented as a no-op.
 
+- `Event` need not have been exported: it's an implementation detail of the SSE
+  and streamable transports. Also the 'Name' field is a misnomer: it should be
+  'event'.
+
 - Enforcing valid tool names: with
   [SEP-986](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/986)
   landing after the SDK was at v1, we missed an opportunity to panic on invalid

mcp/event.go

@@ -29,14 +29,15 @@ const validateMemoryEventStore = false
 // An Event is a server-sent event.
 // See https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields.
 type Event struct {
-	Name string // the "event" field
-	ID   string // the "id" field
-	Data []byte // the "data" field
+	Name  string // the "event" field
+	ID    string // the "id" field
+	Data  []byte // the "data" field
+	Retry string // the "retry" field
 }
 
 // Empty reports whether the Event is empty.
 func (e Event) Empty() bool {
-	return e.Name == "" && e.ID == "" && len(e.Data) == 0
+	return e.Name == "" && e.ID == "" && len(e.Data) == 0 && e.Retry == ""
 }
 
 // writeEvent writes the event to w, and flushes.
@@ -48,6 +49,9 @@ func writeEvent(w io.Writer, evt Event) (int, error) {
 	if evt.ID != "" {
 		fmt.Fprintf(&b, "id: %s\n", evt.ID)
 	}
+	if evt.Retry != "" {
+		fmt.Fprintf(&b, "retry: %s\n", evt.Retry)
+	}
 	fmt.Fprintf(&b, "data: %s\n\n", string(evt.Data))
 	n, err := w.Write(b.Bytes())
 	if f, ok := w.(http.Flusher); ok {
@@ -73,6 +77,7 @@ func scanEvents(r io.Reader) iter.Seq2[Event, error] {
 		eventKey = []byte("event")
 		idKey    = []byte("id")
 		dataKey  = []byte("data")
+		retryKey = []byte("retry")
 	)
 
 	return func(yield func(Event, error) bool) {
@@ -119,6 +124,8 @@ func scanEvents(r io.Reader) iter.Seq2[Event, error] {
 				evt.Name = strings.TrimSpace(string(after))
 			case bytes.Equal(before, idKey):
 				evt.ID = strings.TrimSpace(string(after))
+			case bytes.Equal(before, retryKey):
+				evt.Retry = strings.TrimSpace(string(after))
 			case bytes.Equal(before, dataKey):
 				data := bytes.TrimSpace(after)
 				if dataBuf != nil {
@@ -191,12 +198,8 @@ type dataList struct {
 }
 
 func (dl *dataList) appendData(d []byte) {
-	// If we allowed empty data, we would consume memory without incrementing the size.
-	// We could of course account for that, but we keep it simple and assume there is no
-	// empty data.
-	if len(d) == 0 {
-		panic("empty data item")
-	}
+	// Empty data consumes memory but doesn't increment size. However, it should
+	// be rare.
 	dl.data = append(dl.data, d)
 	dl.size += len(d)
 }

mcp/shared.go

@@ -432,6 +432,14 @@ type ServerRequest[P Params] struct {
 type RequestExtra struct {
 	TokenInfo *auth.TokenInfo // bearer token info (e.g. from OAuth) if any
 	Header    http.Header     // header from HTTP request, if any
+
+	// CloseStream closes the current request stream, if the current transport
+	// supports replaying requests.
+	//
+	// If reconnectAfter is nonzero, it signals to the client to reconnect after
+	// the given duration. Otherwise, clients may determine their own
+	// reconnection policy.
+	CloseStream func(reconnectAfter time.Duration)
 }
 
 func (*ClientRequest[P]) isRequest() {}