diff --git a/docs/protocol.md b/docs/protocol.md index a77604e8..286ada1f 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -187,6 +187,61 @@ reporting the honored subset, then streams every change notification on the same request, and finally closes with a `SubscriptionsListenResult` when it tears the subscription down. +To open a stream explicitly, use +[`CallCustomMethod`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#CallCustomMethod) +with the SDK's `SubscriptionsListenParams` and `SubscriptionsListenResult` types. +Despite its name, this helper can call standard methods: the SDK already +registers them, so do not call `AddSendingCustomMethod` for `subscriptions/listen`. +Use a session connected with protocol `2026-07-28` or later. The result type must +match the SDK's registered type; an arbitrary result struct is not supported. + +```go +listenCtx, cancel := context.WithCancel(ctx) +defer cancel() + +done := make(chan error, 1) +go func() { + _, err := mcp.CallCustomMethod[ + *mcp.SubscriptionsListenParams, + *mcp.SubscriptionsListenResult, + ](listenCtx, session, "subscriptions/listen", &mcp.SubscriptionsListenParams{ + Notifications: &mcp.NotificationSubscriptions{ + ToolsListChanged: true, + }, + }) + done <- err +}() + +// Later, stop observing notifications without closing the session. +cancel() +err := <-done // Normally context.Canceled. +``` + +The call waits for the final result or error, or for context cancellation; the +acknowledgement is a notification, not the result. The server can acknowledge an +empty subset and finish immediately. Cancellation stops observation, not the +underlying work being observed. To inspect the accepted filters, install +receiving middleware on the client before connecting: + +```go +client.AddReceivingMiddleware(func(next mcp.MethodHandler) mcp.MethodHandler { + return func(ctx context.Context, method string, req mcp.Request) (mcp.Result, error) { + if method == "notifications/subscriptions/acknowledged" { + if p, ok := req.GetParams().(*mcp.SubscriptionsAcknowledgedParams); ok && p != nil { + log.Printf("Accepted subscriptions: %+v", p.Notifications) + } + } + return next(ctx, method, req) + } +}) +``` + +Observe subsequent notifications in the same middleware, or use the existing +`ClientOptions` handlers. Configuring list-change handlers also opens an automatic +listen when the server supports them; avoid opening another listen for the same +filters. Prefer `ClientSession.Subscribe` and `Unsubscribe` for resource updates; +they manage subscription state in addition to sending requests. + ## Transports A diff --git a/internal/docs/protocol.src.md b/internal/docs/protocol.src.md index e1cb37af..f15c64d8 100644 --- a/internal/docs/protocol.src.md +++ b/internal/docs/protocol.src.md @@ -131,6 +131,61 @@ reporting the honored subset, then streams every change notification on the same request, and finally closes with a `SubscriptionsListenResult` when it tears the subscription down. +To open a stream explicitly, use +[`CallCustomMethod`](https://pkg.go.dev/github.com/modelcontextprotocol/go-sdk/mcp#CallCustomMethod) +with the SDK's `SubscriptionsListenParams` and `SubscriptionsListenResult` types. +Despite its name, this helper can call standard methods: the SDK already +registers them, so do not call `AddSendingCustomMethod` for `subscriptions/listen`. +Use a session connected with protocol `2026-07-28` or later. The result type must +match the SDK's registered type; an arbitrary result struct is not supported. + +```go +listenCtx, cancel := context.WithCancel(ctx) +defer cancel() + +done := make(chan error, 1) +go func() { + _, err := mcp.CallCustomMethod[ + *mcp.SubscriptionsListenParams, + *mcp.SubscriptionsListenResult, + ](listenCtx, session, "subscriptions/listen", &mcp.SubscriptionsListenParams{ + Notifications: &mcp.NotificationSubscriptions{ + ToolsListChanged: true, + }, + }) + done <- err +}() + +// Later, stop observing notifications without closing the session. +cancel() +err := <-done // Normally context.Canceled. +``` + +The call waits for the final result or error, or for context cancellation; the +acknowledgement is a notification, not the result. The server can acknowledge an +empty subset and finish immediately. Cancellation stops observation, not the +underlying work being observed. To inspect the accepted filters, install +receiving middleware on the client before connecting: + +```go +client.AddReceivingMiddleware(func(next mcp.MethodHandler) mcp.MethodHandler { + return func(ctx context.Context, method string, req mcp.Request) (mcp.Result, error) { + if method == "notifications/subscriptions/acknowledged" { + if p, ok := req.GetParams().(*mcp.SubscriptionsAcknowledgedParams); ok && p != nil { + log.Printf("Accepted subscriptions: %+v", p.Notifications) + } + } + return next(ctx, method, req) + } +}) +``` + +Observe subsequent notifications in the same middleware, or use the existing +`ClientOptions` handlers. Configuring list-change handlers also opens an automatic +listen when the server supports them; avoid opening another listen for the same +filters. Prefer `ClientSession.Subscribe` and `Unsubscribe` for resource updates; +they manage subscription state in addition to sending requests. + ## Transports A diff --git a/mcp/client.go b/mcp/client.go index f81d0c14..33023d60 100644 --- a/mcp/client.go +++ b/mcp/client.go @@ -1815,11 +1815,19 @@ func AddSendingCustomMethod[P paramsPtr[PT], R Result, PT any]( return nil } -// CallCustomMethod sends a custom (non-standard) JSON-RPC method to the -// server and decodes the response into R. +// CallCustomMethod sends a JSON-RPC request to the server and decodes the response +// into R. Prefer a dedicated ClientSession method when one is available: it may +// also provide validation, caching, or subscription management. // -// The method must have been registered on the session's client via -// [AddSendingCustomMethod]. +// Custom methods must first be registered with [AddSendingCustomMethod]. Standard +// MCP methods are already registered by the SDK. Use their corresponding SDK +// parameter and result pointer types; for example, subscriptions/listen uses +// *[SubscriptionsListenParams] and *[SubscriptionsListenResult]. R must match the +// registered result type. +// +// A subscriptions/listen call requires protocol 2026-07-28 or later and blocks +// until the stream ends or ctx is cancelled. Receiving middleware can observe +// its acknowledgement and subsequent notifications. func CallCustomMethod[P paramsPtr[PT], R Result, PT any]( ctx context.Context, cs *ClientSession, diff --git a/mcp/mcp_test.go b/mcp/mcp_test.go index b15cd7fb..4bca97c8 100644 --- a/mcp/mcp_test.go +++ b/mcp/mcp_test.go @@ -2848,6 +2848,155 @@ func TestSubscriptionsListen_Streamable(t *testing.T) { &StreamableClientTransport{Endpoint: httpServer.URL}, events) } +func TestCallCustomMethodSubscriptionsListen(t *testing.T) { + for _, transport := range []string{"in-memory", "streamable-json"} { + t.Run(transport, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + server := NewServer(testImpl, &ServerOptions{Capabilities: &ServerCapabilities{ + Tools: &ToolCapabilities{ListChanged: true}, + }}) + ended := make(chan struct{}) + server.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { + return func(ctx context.Context, method string, req Request) (Result, error) { + if method == methodSubscriptionsListen { + defer close(ended) + } + return next(ctx, method, req) + } + }) + var ct Transport + if transport == "in-memory" { + var st Transport + ct, st = NewInMemoryTransports() + ss, err := server.Connect(ctx, st, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ss.Close() }) + } else { + handler := NewStreamableHTTPHandler(func(*http.Request) *Server { return server }, + &StreamableHTTPOptions{Stateless: true, JSONResponse: true}) + httpServer := httptest.NewServer(handler) + t.Cleanup(httpServer.Close) + ct = &StreamableClientTransport{Endpoint: httpServer.URL} + } + // Middleware observes notifications without enabling the auto-listen. + client := NewClient(testImpl, nil) + events := make(chan subListenEvent, 4) + client.AddReceivingMiddleware(func(next MethodHandler) MethodHandler { + return func(ctx context.Context, method string, req Request) (Result, error) { + switch method { + case notificationSubscriptionsAck: + params := req.GetParams().(*SubscriptionsAcknowledgedParams) + if !params.Notifications.ToolsListChanged { + t.Error("tools/list_changed was not acknowledged") + } + events <- subListenEvent{"ack", fmt.Sprint(params.Meta[MetaKeySubscriptionID])} + case notificationToolListChanged: + events <- subListenEvent{"tool", fmt.Sprint(req.GetParams().GetMeta()[MetaKeySubscriptionID])} + } + return next(ctx, method, req) + } + }) + cs, err := client.Connect(ctx, ct, &ClientSessionOptions{ProtocolVersion: protocolVersion20260728}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cs.Close() }) + listenCtx, endListen := context.WithCancel(ctx) + defer endListen() + done := make(chan error, 1) + go func() { + _, err := CallCustomMethod[*SubscriptionsListenParams, *SubscriptionsListenResult](listenCtx, cs, + methodSubscriptionsListen, &SubscriptionsListenParams{Notifications: &NotificationSubscriptions{ToolsListChanged: true}}) + done <- err + }() + ack := waitSubListenEvent(t, events, "ack") + if ack.id == "" || ack.id == "" { + t.Fatal("acknowledgement is missing its subscription ID") + } + select { + case err := <-done: + t.Fatalf("listen returned after acknowledgement: %v", err) + default: + } + AddTool(server, &Tool{Name: "added"}, sayHi) + if notification := waitSubListenEvent(t, events, "tool"); notification.id != ack.id { + t.Errorf("notification subscription ID=%s, want %s", notification.id, ack.id) + } + endListen() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Errorf("cancelled listen returned %v, want context.Canceled", err) + } + case <-ctx.Done(): + t.Fatal("cancelled listen did not return") + } + select { + case <-ended: + case <-ctx.Done(): + t.Fatal("server did not release the cancelled listen") + } + tools, err := cs.ListTools(ctx, nil) + if err != nil { + t.Fatalf("ListTools after cancellation: %v", err) + } + if len(tools.Tools) != 1 || tools.Tools[0].Name != "added" { + t.Fatalf("ListTools after cancellation returned %+v", tools.Tools) + } + }) + } +} + +func TestCallCustomMethodSubscriptionsListenResult(t *testing.T) { + for _, test := range []struct { + name string + params *SubscriptionsListenParams + code int64 + }{ + {"empty-filters", &SubscriptionsListenParams{Notifications: &NotificationSubscriptions{}}, 0}, + {"missing-filters", &SubscriptionsListenParams{}, jsonrpc.CodeInvalidParams}, + } { + t.Run(test.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + server := NewServer(testImpl, nil) + ct, st := NewInMemoryTransports() + ss, err := server.Connect(ctx, st, nil) + if err != nil { + t.Fatal(err) + } + defer ss.Close() + client := NewClient(testImpl, nil) + cs, err := client.Connect(ctx, ct, &ClientSessionOptions{ProtocolVersion: protocolVersion20260728}) + if err != nil { + t.Fatal(err) + } + defer cs.Close() + result, err := CallCustomMethod[*SubscriptionsListenParams, *SubscriptionsListenResult](ctx, cs, + methodSubscriptionsListen, test.params) + if test.code != 0 { + var rpcErr *jsonrpc.Error + if !errors.As(err, &rpcErr) || rpcErr.Code != test.code { + t.Fatalf("listen error = %v, want JSON-RPC code %d", err, test.code) + } + if result != nil { + t.Fatalf("failed listen returned result %+v", result) + } + return + } + if err != nil { + t.Fatal(err) + } + if result == nil || result.Meta[MetaKeySubscriptionID] == nil { + t.Fatalf("listen result is missing its subscription ID: %+v", result) + } + }) + } +} + // TestSubscriptionsListen_NoHandlersNoListen verifies that a new-protocol // client without any list-changed handlers registered does not open an // auto-listen on connect, and therefore does not receive any acknowledgment