diff --git a/mcp/client.go b/mcp/client.go index 0a6c906..52c1e06 100644 --- a/mcp/client.go +++ b/mcp/client.go @@ -330,7 +330,9 @@ func (c *Client) Connect(ctx context.Context, t Transport, opts *ClientSessionOp if hc, ok := cs.mcpConn.(clientConnection); ok { hc.sessionUpdated(cs.state) } - subscribeParams := &SubscriptionsListenParams{} + subscribeParams := &SubscriptionsListenParams{ + Notifications: &NotificationSubscriptions{}, + } if c.opts.ToolListChangedHandler != nil { subscribeParams.Notifications.ToolsListChanged = true } @@ -1393,7 +1395,7 @@ func (cs *ClientSession) Subscribe(ctx context.Context, params *SubscribeParams) } return cs.subscriptionsListen(listenCtx, &SubscriptionsListenParams{ - Notifications: NotificationSubscriptions{ + Notifications: &NotificationSubscriptions{ ResourceSubscriptions: []string{uri}, }, }) diff --git a/mcp/mcp_test.go b/mcp/mcp_test.go index c0a9cfe..97be926 100644 --- a/mcp/mcp_test.go +++ b/mcp/mcp_test.go @@ -2652,6 +2652,48 @@ func TestSubscriptionsListen_NoHandlersNoListen(t *testing.T) { } } +// TestSubscriptionsListen_MissingNotifications verifies that a +// subscriptions/listen request without the required "notifications" field +// is rejected with an invalid params error. +func TestSubscriptionsListen_MissingNotifications(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + server := newSubListenServer() + _, st := NewInMemoryTransports() + ss, err := server.Connect(ctx, st, nil) + if err != nil { + t.Fatalf("server connect: %v", err) + } + defer ss.Close() + + // Invoke the server handler directly with a nil Notifications field, + // simulating a request whose params omit the required field on the wire. + // The client-side subscriptionsListen does not await the RPC response + // (the call's lifetime is the notification stream), so we assert the + // server-side behavior at the handler level. + id, err := jsonrpc.MakeID("test-1") + if err != nil { + t.Fatalf("MakeID: %v", err) + } + reqCtx := context.WithValue(ctx, idContextKey{}, id) + req := &SubscriptionsListenRequest{ + Session: ss, + Params: &SubscriptionsListenParams{}, // Notifications is nil + } + _, err = server.subscriptionsListen(reqCtx, req) + if err == nil { + t.Fatal("expected error for missing notifications field, got nil") + } + var jerr *jsonrpc.Error + if !errors.As(err, &jerr) { + t.Fatalf("expected *jsonrpc.Error, got %T: %v", err, err) + } + if jerr.Code != jsonrpc.CodeInvalidParams { + t.Errorf("error code = %d, want %d", jerr.Code, jsonrpc.CodeInvalidParams) + } +} + // resourceSubServer builds a server that advertises resource subscriptions // and records every Subscribe/Unsubscribe handler invocation through chans. func resourceSubServer(t *testing.T, subCh, unsubCh chan string) *Server { diff --git a/mcp/protocol.go b/mcp/protocol.go index c187741..a68ef42 100644 --- a/mcp/protocol.go +++ b/mcp/protocol.go @@ -2063,7 +2063,7 @@ type SubscriptionsListenParams struct { Meta `json:"_meta,omitempty"` // Notifications declares which notification types the client wants to // receive on this stream. - Notifications NotificationSubscriptions `json:"notifications"` + Notifications *NotificationSubscriptions `json:"notifications"` } func (x *SubscriptionsListenParams) isParams() {} diff --git a/mcp/server.go b/mcp/server.go index 21b5a72..5fa57f7 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -1169,6 +1169,10 @@ func (s *Server) subscriptionsListen(ctx context.Context, req *SubscriptionsList return nil, fmt.Errorf("%w: subscriptions/listen requires a request ID", jsonrpc2.ErrInvalidRequest) } + if req.Params.Notifications == nil { + return nil, fmt.Errorf("%w: missing required 'notifications' field", jsonrpc2.ErrInvalidParams) + } + allowed := s.allowedSubscriptions(req.Params.Notifications) s.mu.Lock() if allowed.ToolsListChanged { @@ -1217,11 +1221,14 @@ func (s *Server) subscriptionsListen(ctx context.Context, req *SubscriptionsList return nil, fmt.Errorf("sending subscriptions/acknowledged: %w", err) } - <-ctx.Done() + // If there are any active subscriptions, we block until the context is cancelled. Otherwise, we return immediately. + if len(allowed.ResourceSubscriptions) > 0 || allowed.ToolsListChanged || allowed.PromptsListChanged || allowed.ResourcesListChanged { + <-ctx.Done() + } return &emptyResult{}, nil } -func (s *Server) allowedSubscriptions(want NotificationSubscriptions) NotificationSubscriptions { +func (s *Server) allowedSubscriptions(want *NotificationSubscriptions) NotificationSubscriptions { caps := s.capabilities() agreed := NotificationSubscriptions{} if want.ToolsListChanged && caps.Tools != nil && caps.Tools.ListChanged {