From 77e5f12fdd0acc0d4c4ae6df9c63fe9b413f9dfe Mon Sep 17 00:00:00 2001 From: Serob Nahapetyan Date: Fri, 13 Nov 2026 14:49:14 +0000 Subject: [PATCH] add ServerSession.NotifyResourceUpdated --- docs/server.md | 5 + internal/docs/server.src.md | 5 + mcp/mcp_test.go | 212 +++++++++++++++++++++++++++++++----- mcp/server.go | 38 ++++++- 4 files changed, 226 insertions(+), 34 deletions(-) diff --git a/docs/server.md b/docs/server.md index be1817cb..d87eb2d9 100644 --- a/docs/server.md +++ b/docs/server.md @@ -402,6 +402,11 @@ Publish a change with which notifies every session subscribed to that URI. The notification carries only the URI, not the new contents, so a client that wants them reads the resource again. +To notify a single session instead, call +[`ServerSession.NotifyResourceUpdated`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerSession.NotifyResourceUpdated). +It only notifies sessions on protocol version `2026-07-28` or later, once on +each of the session's +[`subscriptions/listen`](protocol.md#subscriptions) streams that name the URI. ```go func Example_resourceSubscription() { diff --git a/internal/docs/server.src.md b/internal/docs/server.src.md index ac8e7751..1455a9a0 100644 --- a/internal/docs/server.src.md +++ b/internal/docs/server.src.md @@ -140,6 +140,11 @@ Publish a change with which notifies every session subscribed to that URI. The notification carries only the URI, not the new contents, so a client that wants them reads the resource again. +To notify a single session instead, call +[`ServerSession.NotifyResourceUpdated`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#ServerSession.NotifyResourceUpdated). +It only notifies sessions on protocol version `2026-07-28` or later, once on +each of the session's +[`subscriptions/listen`](protocol.md#subscriptions) streams that name the URI. %include ../../mcp/server_example_test.go subscribe - diff --git a/mcp/mcp_test.go b/mcp/mcp_test.go index 7a99bf87..f5300bff 100644 --- a/mcp/mcp_test.go +++ b/mcp/mcp_test.go @@ -3154,6 +3154,28 @@ func resourceSubServer(t *testing.T, subCh, unsubCh chan string) *Server { return s } +// resourceSubClient returns a client wired to push every listen +// acknowledgment and every resource update it receives into events, tagged +// with the kind ("ack" or "updated") and the subscription ID from _meta. +func resourceSubClient(events chan subListenEvent) *Client { + c := NewClient(testImpl, &ClientOptions{ + ResourceUpdatedHandler: func(_ context.Context, req *ResourceUpdatedNotificationRequest) { + events <- subListenEvent{"updated", fmt.Sprint(req.Params.Meta[MetaKeySubscriptionID])} + }, + }) + c.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { + return func(ctx context.Context, method string, req Request) (Result, error) { + if method == notificationSubscriptionsAck { + if cr, ok := req.(*ClientRequest[*SubscriptionsAcknowledgedParams]); ok && cr.Params != nil { + events <- subListenEvent{"ack", fmt.Sprint(cr.Params.Meta[MetaKeySubscriptionID])} + } + } + return next(ctx, method, req) + } + }) + return c +} + // resourceSubEvent is one delivered notifications/resources/updated. type resourceSubEvent struct { uri string @@ -3325,6 +3347,164 @@ func TestResourceSubscriptions_InMemory(t *testing.T) { } } +// TestNotifyResourceUpdated verifies that ServerSession.NotifyResourceUpdated +// notifies only the session it is called on, only for a URI that session is +// subscribed to, and only on protocol version 2026-07-28 or later. +func TestNotifyResourceUpdated(t *testing.T) { + const r1, r2 = "file:///r1", "file:///r2" + for _, tc := range []struct { + name string + version string // protocol version of session A + subscribe string // URI session A subscribes to, if any + notify string // URI session A is notified about + wantA bool + }{ + {"subscribed", protocolVersion20260728, r1, r1, true}, + {"other URI", protocolVersion20260728, r1, r2, false}, + {"not subscribed", protocolVersion20260728, "", r1, false}, + {"legacy", protocolVersion20251125, r1, r1, false}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + server := resourceSubServer(t, make(chan string, 8), make(chan string, 8)) + + // open connects a client and subscribes it to uri, if set. It returns + // the client's events and its server session. + open := func(version, uri string) (chan subListenEvent, *ServerSession) { + t.Helper() + events := make(chan subListenEvent, 8) + ct, st := NewInMemoryTransports() + ss, err := server.Connect(context.Background(), st, nil) + if err != nil { + t.Fatalf("server connect: %v", err) + } + t.Cleanup(func() { ss.Close() }) + cs, err := resourceSubClient(events).Connect(context.Background(), ct, &ClientSessionOptions{ProtocolVersion: version}) + if err != nil { + t.Fatalf("client connect: %v", err) + } + t.Cleanup(func() { cs.Close() }) + if uri != "" { + if err := cs.Subscribe(ctx, &SubscribeParams{URI: uri}); err != nil { + t.Fatalf("subscribe %s: %v", uri, err) + } + } + return events, ss + } + evA, ssA := open(tc.version, tc.subscribe) + evB, _ := open(protocolVersion20260728, r1) + + var ackA subListenEvent + if tc.subscribe != "" && tc.version >= protocolVersion20260728 { + ackA = waitSubListenEvent(t, evA, "ack") + } + waitSubListenEvent(t, evB, "ack") + + if err := ssA.NotifyResourceUpdated(ctx, &ResourceUpdatedNotificationParams{URI: tc.notify}); err != nil { + t.Fatalf("NotifyResourceUpdated: %v", err) + } + + if tc.wantA { + if got := waitSubListenEvent(t, evA, "updated"); got.id != ackA.id { + t.Errorf("session A: update subscription ID = %s, want %s", got.id, ackA.id) + } + } + select { + case e := <-evA: + t.Fatalf("session A received unexpected event %q (id=%s)", e.kind, e.id) + case e := <-evB: + t.Fatalf("session B received unexpected event %q (id=%s)", e.kind, e.id) + case <-time.After(notificationDelay * 5): + } + }) + } +} + +// TestNotifyResourceUpdated_StreamableStateless verifies that, on a stateless HTTP +// server, NotifyResourceUpdated reaches only that client, even when the caller's +// ctx carries the ID of an unrelated request. +func TestNotifyResourceUpdated_StreamableStateless(t *testing.T) { + const uri = "file:///r1" + ctx := t.Context() + server := resourceSubServer(t, make(chan string, 8), make(chan string, 8)) + + listenSessions := make(chan *ServerSession, 2) + server.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { + return func(ctx context.Context, method string, req Request) (Result, error) { + if method == methodSubscriptionsListen { + select { + case listenSessions <- req.GetSession().(*ServerSession): + default: + } + } + return next(ctx, method, req) + } + }) + handler := NewStreamableHTTPHandler(func(*http.Request) *Server { return server }, &StreamableHTTPOptions{Stateless: true}) + httpServer := httptest.NewServer(mustNotPanic(t, handler)) + t.Cleanup(httpServer.Close) + + // open connects a client and subscribes it to uri. It returns the client's + // events, the session serving its listen, and the listen's acknowledgment. + open := func(t *testing.T) (chan subListenEvent, *ServerSession) { + t.Helper() + events := make(chan subListenEvent, 16) + c := resourceSubClient(events) + cs, err := c.Connect(ctx, &StreamableClientTransport{Endpoint: httpServer.URL}, &ClientSessionOptions{ProtocolVersion: protocolVersion20260728}) + if err != nil { + t.Fatalf("client connect: %v", err) + } + t.Cleanup(func() { cs.Close() }) + + if err := cs.Subscribe(ctx, &SubscribeParams{URI: uri}); err != nil { + t.Fatalf("subscribe %s: %v", uri, err) + } + select { + case ss := <-listenSessions: + return events, ss + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the client's listen") + return nil, nil + } + } + evA, ssA := open(t) + evB, _ := open(t) + ackA := waitSubListenEvent(t, evA, "ack") + waitSubListenEvent(t, evB, "ack") + + // Notify with a ctx carrying another request's ID, as a tool handler's ctx + // would. The notification must still go out on the listen's stream. + otherID, err := jsonrpc.MakeID("other-request") + if err != nil { + t.Fatal(err) + } + otherCtx := context.WithValue(ctx, idContextKey{}, otherID) + + for _, tc := range []struct { + name string + ctx context.Context + }{ + {"plain ctx", ctx}, + {"context of another request", otherCtx}, + } { + t.Run(tc.name, func(t *testing.T) { + if err := ssA.NotifyResourceUpdated(tc.ctx, &ResourceUpdatedNotificationParams{URI: uri}); err != nil { + t.Fatalf("NotifyResourceUpdated: %v", err) + } + if got := waitSubListenEvent(t, evA, "updated"); got.id != ackA.id { + t.Errorf("session A: update subscription id = %s, want %s", got.id, ackA.id) + } + select { + case e := <-evA: + t.Fatalf("session A received a second event: %q", e.kind) + case e := <-evB: + t.Fatalf("session B received unexpected event %q", e.kind) + case <-time.After(notificationDelay * 5): + } + }) + } +} + // TestResourceSubscriptions_Subscribe_Idempotent verifies that calling // Subscribe twice for the same URI in the same session is a no-op for the // second call: it returns nil without invoking SubscribeHandler again and @@ -3896,21 +4076,7 @@ func TestSubscriptionsListen_ResubscribeOutlivesEarlierTeardown(t *testing.T) { defer ss.Close() events := make(chan subListenEvent, 16) - c := NewClient(testImpl, &ClientOptions{ - ResourceUpdatedHandler: func(_ context.Context, req *ResourceUpdatedNotificationRequest) { - events <- subListenEvent{"updated", fmt.Sprint(req.Params.Meta[MetaKeySubscriptionID])} - }, - }) - c.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { - return func(ctx context.Context, method string, req Request) (Result, error) { - if method == notificationSubscriptionsAck { - if cr, ok := req.(*ClientRequest[*SubscriptionsAcknowledgedParams]); ok && cr.Params != nil { - events <- subListenEvent{"ack", fmt.Sprint(cr.Params.Meta[MetaKeySubscriptionID])} - } - } - return next(ctx, method, req) - } - }) + c := resourceSubClient(events) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() cs, err := c.Connect(ctx, ct, &ClientSessionOptions{ProtocolVersion: protocolVersion20260728}) @@ -3998,21 +4164,7 @@ func newURIListens(t *testing.T, uri string) *uriListens { t.Cleanup(func() { ss.Close() }) f.ss = ss - c := NewClient(testImpl, &ClientOptions{ - ResourceUpdatedHandler: func(_ context.Context, req *ResourceUpdatedNotificationRequest) { - f.events <- subListenEvent{"updated", fmt.Sprint(req.Params.Meta[MetaKeySubscriptionID])} - }, - }) - c.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { - return func(ctx context.Context, method string, req Request) (Result, error) { - if method == notificationSubscriptionsAck { - if cr, ok := req.(*ClientRequest[*SubscriptionsAcknowledgedParams]); ok && cr.Params != nil { - f.events <- subListenEvent{"ack", fmt.Sprint(cr.Params.Meta[MetaKeySubscriptionID])} - } - } - return next(ctx, method, req) - } - }) + c := resourceSubClient(f.events) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) t.Cleanup(cancel) f.ctx = ctx diff --git a/mcp/server.go b/mcp/server.go index 6811e5d9..16145db8 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -1712,12 +1712,34 @@ func (ss *ServerSession) NotifyResourceListChanged(ctx context.Context, params * }) } +// NotifyResourceUpdated sends a "notifications/resources/updated" +// notification to the client of this session, about the resource identified +// by params.URI. Unlike [Server.ResourceUpdated], other sessions subscribed +// to the URI are not notified. +// +// The notification is sent once on each of the session's open +// subscriptions/listen streams that name the URI, and not at all if there +// are none. Only clients on protocol version 2026-07-28 or later open such +// streams, so nothing is sent to earlier sessions. +func (ss *ServerSession) NotifyResourceUpdated(ctx context.Context, params *ResourceUpdatedNotificationParams) error { + if ss.negotiatedLegacyProtocol() { + return nil + } + server := ss.server + + server.mu.Lock() + ids := slices.Clone(server.resourceSubscriptions[params.URI][ss]) + server.mu.Unlock() + + makeParams := func() Params { + p := *params + return &p + } + return ss.notifyListens(ctx, notificationResourceUpdated, ids, makeParams) +} + // notifyListChanged implements the Notify*ListChanged methods for the given // list-changed notification method. -// -// makeParams is called once per notification sent and must return a new value -// each time, because each notification carries the ID of the listen it is sent -// on in its _meta (see [injectMetaSubscriptionID]). func (ss *ServerSession) notifyListChanged(ctx context.Context, method string, makeParams func() Params) error { server := ss.server @@ -1732,6 +1754,14 @@ func (ss *ServerSession) notifyListChanged(ctx context.Context, method string, m ids = slices.Clone(server.resourceChangeSubscriptions[ss]) } server.mu.Unlock() + + return ss.notifyListens(ctx, method, ids, makeParams) +} + +// notifyListens sends method once on each of the given subscriptions/listen +// requests of this session, stamping each listen's ID into _meta. +// makeParams must return a new value on each call. +func (ss *ServerSession) notifyListens(ctx context.Context, method string, ids []jsonrpc.ID, makeParams func() Params) error { var errs []error for _, id := range ids { params := makeParams()