Merge remote-tracking branch 'origin/main' into stateless-http

# Conflicts:
#	mcp/server.go
#	mcp/streamable.go

Author: ln-12

design/design.md

@@ -748,13 +748,26 @@ Server sessions also support the spec methods `ListResources` and `ListResourceT
 
 #### Subscriptions
 
-ClientSessions can manage change notifications on particular resources:
+##### Client-Side Usage
+
+Use the Subscribe and Unsubscribe methods on a ClientSession to start or stop receiving updates for a specific resource.
 
 ```go
 func (*ClientSession) Subscribe(context.Context, *SubscribeParams) error
 func (*ClientSession) Unsubscribe(context.Context, *UnsubscribeParams) error
 ```
 
+To process incoming update notifications, you must provide a ResourceUpdatedHandler in your ClientOptions. The SDK calls this function automatically whenever the server sends a notification for a resource you're subscribed to.
+
+```go
+type ClientOptions struct {
+  ...
+  ResourceUpdatedHandler func(context.Context, *ClientSession, *ResourceUpdatedNotificationParams)
+}
+```
+
+##### Server-Side Implementation
+
 The server does not implement resource subscriptions. It passes along subscription requests to the user, and supplies a method to notify clients of changes. It tracks which sessions have subscribed to which resources so the user doesn't have to.
 
 If a server author wants to support resource subscriptions, they must provide handlers to be called when clients subscribe and unsubscribe. It is an error to provide only one of these handlers.
@@ -772,7 +785,7 @@ type ServerOptions struct {
 User code should call `ResourceUpdated` when a subscribed resource changes.
 
 ```go
-func (*Server) ResourceUpdated(context.Context, *ResourceUpdatedNotification) error
+func (*Server) ResourceUpdated(context.Context, *ResourceUpdatedNotificationParams) error
 ```
 
 The server routes these notifications to the server sessions that subscribed to the resource.

mcp/client.go

@@ -60,6 +60,7 @@ type ClientOptions struct {
 	ToolListChangedHandler      func(context.Context, *ClientSession, *ToolListChangedParams)
 	PromptListChangedHandler    func(context.Context, *ClientSession, *PromptListChangedParams)
 	ResourceListChangedHandler  func(context.Context, *ClientSession, *ResourceListChangedParams)
+	ResourceUpdatedHandler      func(context.Context, *ClientSession, *ResourceUpdatedNotificationParams)
 	LoggingMessageHandler       func(context.Context, *ClientSession, *LoggingMessageParams)
 	ProgressNotificationHandler func(context.Context, *ClientSession, *ProgressNotificationParams)
 	// If non-zero, defines an interval for regular "ping" requests.
@@ -293,6 +294,7 @@ var clientMethodInfos = map[string]methodInfo{
 	notificationToolListChanged:     newMethodInfo(clientMethod((*Client).callToolChangedHandler)),
 	notificationPromptListChanged:   newMethodInfo(clientMethod((*Client).callPromptChangedHandler)),
 	notificationResourceListChanged: newMethodInfo(clientMethod((*Client).callResourceChangedHandler)),
+	notificationResourceUpdated:     newMethodInfo(clientMethod((*Client).callResourceUpdatedHandler)),
 	notificationLoggingMessage:      newMethodInfo(clientMethod((*Client).callLoggingHandler)),
 	notificationProgress:            newMethodInfo(sessionMethod((*ClientSession).callProgressNotificationHandler)),
 }
@@ -386,6 +388,20 @@ func (cs *ClientSession) Complete(ctx context.Context, params *CompleteParams) (
 	return handleSend[*CompleteResult](ctx, cs, methodComplete, orZero[Params](params))
 }
 
+// Subscribe sends a "resources/subscribe" request to the server, asking for
+// notifications when the specified resource changes.
+func (cs *ClientSession) Subscribe(ctx context.Context, params *SubscribeParams) error {
+	_, err := handleSend[*emptyResult](ctx, cs, methodSubscribe, orZero[Params](params))
+	return err
+}
+
+// Unsubscribe sends a "resources/unsubscribe" request to the server, cancelling
+// a previous subscription.
+func (cs *ClientSession) Unsubscribe(ctx context.Context, params *UnsubscribeParams) error {
+	_, err := handleSend[*emptyResult](ctx, cs, methodUnsubscribe, orZero[Params](params))
+	return err
+}
+
 func (c *Client) callToolChangedHandler(ctx context.Context, s *ClientSession, params *ToolListChangedParams) (Result, error) {
 	return callNotificationHandler(ctx, c.opts.ToolListChangedHandler, s, params)
 }
@@ -398,6 +414,10 @@ func (c *Client) callResourceChangedHandler(ctx context.Context, s *ClientSessio
 	return callNotificationHandler(ctx, c.opts.ResourceListChangedHandler, s, params)
 }
 
+func (c *Client) callResourceUpdatedHandler(ctx context.Context, s *ClientSession, params *ResourceUpdatedNotificationParams) (Result, error) {
+	return callNotificationHandler(ctx, c.opts.ResourceUpdatedHandler, s, params)
+}
+
 func (c *Client) callLoggingHandler(ctx context.Context, cs *ClientSession, params *LoggingMessageParams) (Result, error) {
 	if h := c.opts.LoggingMessageHandler; h != nil {
 		h(ctx, cs, params)

mcp/event.go

@@ -153,10 +153,9 @@ func scanEvents(r io.Reader) iter.Seq2[Event, error] {
 //
 // All of an EventStore's methods must be safe for use by multiple goroutines.
 type EventStore interface {
-	// AppendEvent appends data for an outgoing event to given stream, which is part of the
-	// given session. It returns the index of the event in the stream, suitable for constructing
-	// an event ID to send to the client.
-	AppendEvent(_ context.Context, sessionID string, _ StreamID, data []byte) (int, error)
+	// Append appends data for an outgoing event to given stream, which is part of the
+	// given session.
+	Append(_ context.Context, sessionID string, _ StreamID, data []byte) error
 
 	// After returns an iterator over the data for the given session and stream, beginning
 	// just after the given index.
@@ -165,16 +164,15 @@ type EventStore interface {
 	// dropped; it must not return partial results.
 	After(_ context.Context, sessionID string, _ StreamID, index int) iter.Seq2[[]byte, error]
 
-	// StreamClosed informs the store that the given stream is finished.
-	// A store cannot rely on this method being called for cleanup. It should institute
-	// additional mechanisms, such as timeouts, to reclaim storage.
-	StreamClosed(_ context.Context, sessionID string, streamID StreamID) error
-
 	// SessionClosed informs the store that the given session is finished, along
 	// with all of its streams.
 	// A store cannot rely on this method being called for cleanup. It should institute
 	// additional mechanisms, such as timeouts, to reclaim storage.
+	//
 	SessionClosed(_ context.Context, sessionID string) error
+
+	// There is no StreamClosed method. A server doesn't know when a stream is finished, because
+	// the client can always send a GET with a Last-Event-ID referring to the stream.
 }
 
 // A dataList is a list of []byte.
@@ -210,15 +208,6 @@ func (dl *dataList) removeFirst() int {
 	return r
 }
 
-// lastIndex returns the index of the last data item in dl.
-// It panics if there are none.
-func (dl *dataList) lastIndex() int {
-	if len(dl.data) == 0 {
-		panic("empty dataList")
-	}
-	return dl.first + len(dl.data) - 1
-}
-
 // A MemoryEventStore is an [EventStore] backed by memory.
 type MemoryEventStore struct {
 	mu       sync.Mutex
@@ -267,9 +256,8 @@ func NewMemoryEventStore(opts *MemoryEventStoreOptions) *MemoryEventStore {
 	}
 }
 
-// AppendEvent implements [EventStore.AppendEvent] by recording data
-// in memory.
-func (s *MemoryEventStore) AppendEvent(_ context.Context, sessionID string, streamID StreamID, data []byte) (int, error) {
+// Append implements [EventStore.Append] by recording data in memory.
+func (s *MemoryEventStore) Append(_ context.Context, sessionID string, streamID StreamID, data []byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -288,9 +276,13 @@ func (s *MemoryEventStore) AppendEvent(_ context.Context, sessionID string, stre
 	s.purge()
 	dl.appendData(data)
 	s.nBytes += len(data)
-	return dl.lastIndex(), nil
+	return nil
 }
 
+// ErrEventsPurged is the error that [EventStore.After] should return if the event just after the
+// index is no longer available.
+var ErrEventsPurged = errors.New("data purged")
+
 // After implements [EventStore.After].
 func (s *MemoryEventStore) After(_ context.Context, sessionID string, streamID StreamID, index int) iter.Seq2[[]byte, error] {
 	// Return the data items to yield.
@@ -306,10 +298,12 @@ func (s *MemoryEventStore) After(_ context.Context, sessionID string, streamID S
 		if !ok {
 			return nil, fmt.Errorf("MemoryEventStore.After: unknown stream ID %v in session %q", streamID, sessionID)
 		}
-		if dl.first > index {
-			return nil, fmt.Errorf("MemoryEventStore.After: data purged at index %d, stream ID %v, session %q", index, streamID, sessionID)
+		start := index + 1
+		if dl.first > start {
+			return nil, fmt.Errorf("MemoryEventStore.After: index %d, stream ID %v, session %q: %w",
+				index, streamID, sessionID, ErrEventsPurged)
 		}
-		return slices.Clone(dl.data[index-dl.first:]), nil
+		return slices.Clone(dl.data[start-dl.first:]), nil
 	}
 
 	return func(yield func([]byte, error) bool) {
@@ -326,26 +320,6 @@ func (s *MemoryEventStore) After(_ context.Context, sessionID string, streamID S
 	}
 }
 
-// StreamClosed implements [EventStore.StreamClosed].
-func (s *MemoryEventStore) StreamClosed(_ context.Context, sessionID string, streamID StreamID) error {
-	if sessionID == "" {
-		panic("empty sessionID")
-	}
-
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	sm := s.store[sessionID]
-	dl := sm[streamID]
-	s.nBytes -= dl.size
-	delete(sm, streamID)
-	if len(sm) == 0 {
-		delete(s.store, sessionID)
-	}
-	s.validate()
-	return nil
-}
-
 // SessionClosed implements [EventStore.SessionClosed].
 func (s *MemoryEventStore) SessionClosed(_ context.Context, sessionID string) error {
 	s.mu.Lock()

mcp/event_test.go

@@ -105,7 +105,7 @@ func TestMemoryEventStoreState(t *testing.T) {
 	ctx := context.Background()
 
 	appendEvent := func(s *MemoryEventStore, sess string, str StreamID, data string) {
-		if _, err := s.AppendEvent(ctx, sess, str, []byte(data)); err != nil {
+		if err := s.Append(ctx, sess, str, []byte(data)); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -127,18 +127,6 @@ func TestMemoryEventStoreState(t *testing.T) {
 			"S1 1 first=0 d1 d3; S1 2 first=0 d2; S2 8 first=0 d4",
 			8,
 		},
-		{
-			"stream close",
-			func(s *MemoryEventStore) {
-				appendEvent(s, "S1", 1, "d1")
-				appendEvent(s, "S1", 2, "d2")
-				appendEvent(s, "S1", 1, "d3")
-				appendEvent(s, "S2", 8, "d4")
-				s.StreamClosed(ctx, "S1", 1)
-			},
-			"S1 2 first=0 d2; S2 8 first=0 d4",
-			4,
-		},
 		{
 			"session close",
 			func(s *MemoryEventStore) {
@@ -218,10 +206,10 @@ func TestMemoryEventStoreAfter(t *testing.T) {
 	ctx := context.Background()
 	s := NewMemoryEventStore(nil)
 	s.SetMaxBytes(4)
-	s.AppendEvent(ctx, "S1", 1, []byte("d1"))
-	s.AppendEvent(ctx, "S1", 1, []byte("d2"))
-	s.AppendEvent(ctx, "S1", 1, []byte("d3"))
-	s.AppendEvent(ctx, "S1", 2, []byte("d4")) // will purge d1
+	s.Append(ctx, "S1", 1, []byte("d1"))
+	s.Append(ctx, "S1", 1, []byte("d2"))
+	s.Append(ctx, "S1", 1, []byte("d3"))
+	s.Append(ctx, "S1", 2, []byte("d4")) // will purge d1
 	want := "S1 1 first=1 d2 d3; S1 2 first=0 d4"
 	if got := s.debugString(); got != want {
 		t.Fatalf("got state %q, want %q", got, want)
@@ -234,10 +222,10 @@ func TestMemoryEventStoreAfter(t *testing.T) {
 		want      []string
 		wantErr   string // if non-empty, error should contain this string
 	}{
-		{"S1", 1, 0, nil, "purge"},
-		{"S1", 1, 1, []string{"d2", "d3"}, ""},
-		{"S1", 1, 2, []string{"d3"}, ""},
-		{"S1", 2, 0, []string{"d4"}, ""},
+		{"S1", 1, 0, []string{"d2", "d3"}, ""},
+		{"S1", 1, 1, []string{"d3"}, ""},
+		{"S1", 1, 2, nil, ""},
+		{"S1", 2, 0, nil, ""},
 		{"S1", 3, 0, nil, "unknown stream ID"},
 		{"S2", 0, 0, nil, "unknown session ID"},
 	} {

mcp/mcp_test.go

@@ -60,7 +60,7 @@ func TestEndToEnd(t *testing.T) {
 
 	// Channels to check if notification callbacks happened.
 	notificationChans := map[string]chan int{}
-	for _, name := range []string{"initialized", "roots", "tools", "prompts", "resources", "progress_server", "progress_client"} {
+	for _, name := range []string{"initialized", "roots", "tools", "prompts", "resources", "progress_server", "progress_client", "resource_updated", "subscribe", "unsubscribe"} {
 		notificationChans[name] = make(chan int, 1)
 	}
 	waitForNotification := func(t *testing.T, name string) {
@@ -78,6 +78,14 @@ func TestEndToEnd(t *testing.T) {
 		ProgressNotificationHandler: func(context.Context, *ServerSession, *ProgressNotificationParams) {
 			notificationChans["progress_server"] <- 0
 		},
+		SubscribeHandler: func(context.Context, *SubscribeParams) error {
+			notificationChans["subscribe"] <- 0
+			return nil
+		},
+		UnsubscribeHandler: func(context.Context, *UnsubscribeParams) error {
+			notificationChans["unsubscribe"] <- 0
+			return nil
+		},
 	}
 	s := NewServer(testImpl, sopts)
 	AddTool(s, &Tool{
@@ -128,6 +136,9 @@ func TestEndToEnd(t *testing.T) {
 		ProgressNotificationHandler: func(context.Context, *ClientSession, *ProgressNotificationParams) {
 			notificationChans["progress_client"] <- 0
 		},
+		ResourceUpdatedHandler: func(context.Context, *ClientSession, *ResourceUpdatedNotificationParams) {
+			notificationChans["resource_updated"] <- 0
+		},
 	}
 	c := NewClient(testImpl, opts)
 	rootAbs, err := filepath.Abs(filepath.FromSlash("testdata/files"))
@@ -421,6 +432,37 @@ func TestEndToEnd(t *testing.T) {
 		waitForNotification(t, "progress_server")
 	})
 
+	t.Run("resource_subscriptions", func(t *testing.T) {
+		err := cs.Subscribe(ctx, &SubscribeParams{
+			URI: "test",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		waitForNotification(t, "subscribe")
+		s.ResourceUpdated(ctx, &ResourceUpdatedNotificationParams{
+			URI: "test",
+		})
+		waitForNotification(t, "resource_updated")
+		err = cs.Unsubscribe(ctx, &UnsubscribeParams{
+			URI: "test",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		waitForNotification(t, "unsubscribe")
+
+		// Verify the client does not receive the update after unsubscribing.
+		s.ResourceUpdated(ctx, &ResourceUpdatedNotificationParams{
+			URI: "test",
+		})
+		select {
+		case <-notificationChans["resource_updated"]:
+			t.Fatalf("resource updated after unsubscription")
+		case <-time.After(time.Second):
+		}
+	})
+
 	// Disconnect.
 	cs.Close()
 	clientWG.Wait()
@@ -617,7 +659,7 @@ func TestCancellation(t *testing.T) {
 		return nil, nil
 	}
 	_, cs := basicConnection(t, func(s *Server) {
-		s.AddTool(&Tool{Name: "slow"}, slowRequest)
+		s.AddTool(&Tool{Name: "slow", InputSchema: &jsonschema.Schema{}}, slowRequest)
 	})
 	defer cs.Close()
 

mcp/protocol.go

@@ -859,6 +859,38 @@ type ToolListChangedParams struct {
 func (x *ToolListChangedParams) GetProgressToken() any  { return getProgressToken(x) }
 func (x *ToolListChangedParams) SetProgressToken(t any) { setProgressToken(x, t) }
 
+// Sent from the client to request resources/updated notifications from the
+// server whenever a particular resource changes.
+type SubscribeParams struct {
+	// This property is reserved by the protocol to allow clients and servers to
+	// attach additional metadata to their responses.
+	Meta `json:"_meta,omitempty"`
+	// The URI of the resource to subscribe to.
+	URI string `json:"uri"`
+}
+
+// Sent from the client to request cancellation of resources/updated
+// notifications from the server. This should follow a previous
+// resources/subscribe request.
+type UnsubscribeParams struct {
+	// This property is reserved by the protocol to allow clients and servers to
+	// attach additional metadata to their responses.
+	Meta `json:"_meta,omitempty"`
+	// The URI of the resource to unsubscribe from.
+	URI string `json:"uri"`
+}
+
+// A notification from the server to the client, informing it that a resource
+// has changed and may need to be read again. This should only be sent if the
+// client previously sent a resources/subscribe request.
+type ResourceUpdatedNotificationParams struct {
+	// This property is reserved by the protocol to allow clients and servers to
+	// attach additional metadata to their responses.
+	Meta `json:"_meta,omitempty"`
+	// The URI of the resource that has been updated. This might be a sub-resource of the one that the client actually subscribed to.
+	URI string `json:"uri"`
+}
+
 // TODO(jba): add CompleteRequest and related types.
 
 // TODO(jba): add ElicitRequest and related types.