mcp: make NotificationSubscriptions a mandatory field (#1050)
Fixes #1037
This commit is contained in:
parent
ec564f9ccb
commit
d651e2d2e1
4 changed files with 56 additions and 5 deletions
|
|
@ -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},
|
||||
},
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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() {}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
Loading…
Reference in a new issue