feat(MCPServer): send log message notifications to specific client

mcp/types.go

@@ -5,8 +5,8 @@ package mcp
 import (
 	"encoding/json"
 	"fmt"
-	"strconv"
 	"maps"
+	"strconv"
 
 	"github.com/yosida95/uritemplate/v3"
 )
@@ -67,6 +67,10 @@ const (
 	// MethodNotificationToolsListChanged notifies when the list of available tools changes.
 	// https://spec.modelcontextprotocol.io/specification/2024-11-05/server/tools/list_changed/
 	MethodNotificationToolsListChanged = "notifications/tools/list_changed"
+
+	// MethodNotificationMessage notifies when severs send log messages.
+	// https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/logging#log-message-notifications
+	MethodNotificationMessage MCPMethod = "notifications/message"
 )
 
 type URITemplate struct {
@@ -734,6 +738,30 @@ const (
 	LoggingLevelEmergency LoggingLevel = "emergency"
 )
 
+var (
+	// Map logging level constants to numerical codes as specified in RFC-5424
+	levelToSeverity = func() map[LoggingLevel]int {
+		return map[LoggingLevel]int{
+			LoggingLevelEmergency: 0,
+			LoggingLevelAlert:     1,
+			LoggingLevelCritical:  2,
+			LoggingLevelError:     3,
+			LoggingLevelWarning:   4,
+			LoggingLevelNotice:    5,
+			LoggingLevelInfo:      6,
+			LoggingLevelDebug:     7,
+		}
+	}()
+)
+
+// Allows is a helper function that decides a message could be sent to client or not according to the logging level
+func (subscribedLevel LoggingLevel) Allows(currentLevel LoggingLevel) (bool, error) {
+	if _, ok := levelToSeverity[currentLevel]; !ok {
+		return false, fmt.Errorf("illegal message logging level:%s", currentLevel)
+	}
+	return levelToSeverity[subscribedLevel] >= levelToSeverity[currentLevel], nil
+}
+
 /* Sampling */
 
 // CreateMessageRequest is a request from the server to sample an LLM via the

mcp/utils.go

@@ -171,7 +171,7 @@ func NewLoggingMessageNotification(
 ) LoggingMessageNotification {
 	return LoggingMessageNotification{
 		Notification: Notification{
-			Method: "notifications/message",
+			Method: string(MethodNotificationMessage),
 		},
 		Params: struct {
 			Level  LoggingLevel `json:"level"`

server/errors.go

@@ -17,7 +17,7 @@ var (
 	ErrSessionExists              	= errors.New("session already exists")
 	ErrSessionNotInitialized      	= errors.New("session not properly initialized")
 	ErrSessionDoesNotSupportTools 	= errors.New("session does not support per-session tools")
-	ErrSessionDoesNotSupportLogging = errors.New("session does not support setting logging level")
+	ErrSessionDoesNotSupportLogging = errors.New("session does not support setting and getting logging level")
 
 	// Notification-related errors
 	ErrNotificationNotInitialized = errors.New("notification channel not initialized")

server/session.go

@@ -343,3 +343,43 @@ func (s *MCPServer) DeleteSessionTools(sessionID string, names ...string) error
 
 	return nil
 }
+
+// SendLogMessageToClient sends a log message notification to the current client.
+// The client sesssion is expected to be an instance of server.SessionWithLogging to support setting and getting minimum log level
+func(s *MCPServer) SendLogMessageToClient(ctx context.Context, msg mcp.LoggingMessageNotification) error {
+	if s.capabilities.logging == nil || !(*s.capabilities.logging) {
+		return fmt.Errorf("server does not support emitting log message notifications")
+	}
+
+	clientSession := ClientSessionFromContext(ctx)
+	if clientSession == nil || !clientSession.Initialized() {
+		return ErrSessionNotInitialized
+	}
+
+	logSession, ok := clientSession.(SessionWithLogging)
+	if !ok {
+		return ErrSessionDoesNotSupportLogging
+	}
+
+	// Servers send notifications containing severity levels, optional logger names, and arbitrary JSON-serializable data.
+	// see <https://modelcontextprotocol.io/specification/2025-03-26/server/utilities/logging>
+	if msg.Params.Level == "" || msg.Params.Data == nil {
+		return fmt.Errorf("invalid log message without level or data")
+	}
+
+	clientLogLevel := logSession.GetLogLevel()
+	allowed, err := clientLogLevel.Allows(msg.Params.Level)
+	if err != nil {
+		return err
+	}
+	if !allowed {
+		return fmt.Errorf("message level(%s) is lower than client level(%s)", msg.Params.Level, clientLogLevel)
+	}
+
+	params := map[string]any {
+		"level":	msg.Params.Level,
+		"data":		msg.Params.Data,
+		"logger":	msg.Params.Logger,
+	}
+	return s.SendNotificationToClient(ctx, msg.Method, params)
+}

server/session_test.go

@@ -104,7 +104,7 @@ type sessionTestClientWithLogging struct {
 	sessionID           string
 	notificationChannel chan mcp.JSONRPCNotification
 	initialized         bool
-	loggingLevel 		atomic.Value
+	loggingLevel        atomic.Value
 }
 
 func (f *sessionTestClientWithLogging) SessionID() string {
@@ -136,9 +136,9 @@ func (f *sessionTestClientWithLogging) GetLogLevel() mcp.LoggingLevel {
 
 // Verify that all implementations satisfy their respective interfaces
 var (
-	_ ClientSession 			= (*sessionTestClient)(nil)
-	_ SessionWithTools 			= (*sessionTestClientWithTools)(nil)
-	_ SessionWithLogging		= (*sessionTestClientWithLogging)(nil)
+	_ ClientSession      = (*sessionTestClient)(nil)
+	_ SessionWithTools   = (*sessionTestClientWithTools)(nil)
+	_ SessionWithLogging = (*sessionTestClientWithLogging)(nil)
 )
 
 func TestSessionWithTools_Integration(t *testing.T) {
@@ -1039,6 +1039,187 @@ func TestMCPServer_SetLevel(t *testing.T) {
 
 	// Check logging level
 	if session.GetLogLevel() != mcp.LoggingLevelCritical {
-		t.Errorf("Expected critical level, got %v", session.GetLogLevel())
+		t.Errorf("Expected critical level, got %s", session.GetLogLevel())
 	}
-}
+}
+
+func TestMCPServer_SendLogMessageToClientDisabled(t *testing.T) {
+	// Create server without logging capability
+	server := NewMCPServer("test-server", "1.0.0")
+
+	// Create and initialize a session
+	sessionChan := make(chan mcp.JSONRPCNotification, 10)
+	session := &sessionTestClientWithLogging{
+		sessionID:           "session-1",
+		notificationChannel: sessionChan,
+	}
+	session.Initialize()
+
+	// Mock a request context
+	ctx := server.WithContext(context.Background(), session)
+
+	// Try to send a log message to client when capability is disabled
+	require.Error(t, server.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelCritical, "test logger", "test data")))
+}
+
+func TestMCPServer_SendLogMessageToClient(t *testing.T) {
+	// Prepare a log message
+	logMsg := mcp.NewLoggingMessageNotification(
+		mcp.LoggingLevelAlert,
+		"test logger",
+		"test data",
+	)
+
+	tests := []struct {
+		name           string
+		contextPrepare func(context.Context, *MCPServer) context.Context
+		validate       func(*testing.T, context.Context, *MCPServer)
+	}{
+		{
+			name: "no active session",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				return ctx
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				require.Error(t, srv.SendLogMessageToClient(ctx, logMsg))
+			},
+		},
+		{
+			name: "uninit session",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithLogging{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 10),
+					initialized:         false,
+				}
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				require.Error(t, srv.SendLogMessageToClient(ctx, logMsg))
+				_, ok := ClientSessionFromContext(ctx).(*sessionTestClientWithLogging)
+				require.True(t, ok, "session not found or of incorrect type")
+			},
+		},
+		{
+			name: "session not supports logging",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithTools{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 10),
+					initialized:         false,
+				}
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				require.Error(t, srv.SendLogMessageToClient(ctx, logMsg))
+			},
+		},
+		{
+			name: "invalid log messages without level or data",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithLogging{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 10),
+					initialized:         false,
+				}
+				logSession.Initialize()
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				// Invalid message without level
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification("", "test logger", "test data")))
+				// Invalid message with illegal level
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevel("invalid level"), "test logger", "test data")))
+				// Invalid message without data
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelCritical, "test logger", nil)))
+			},
+		},
+		{
+			name: "active session",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithLogging{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 10),
+					initialized:         false,
+				}
+				logSession.Initialize()
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				for range 10 {
+					require.NoError(t, srv.SendLogMessageToClient(ctx, logMsg))
+				}
+				session, ok := ClientSessionFromContext(ctx).(*sessionTestClientWithLogging)
+				require.True(t, ok, "session not found or of incorrect type")
+				for range 10 {
+					select {
+					case msg := <-session.notificationChannel:
+						assert.Equal(t, string(mcp.MethodNotificationMessage), msg.Method)
+						assert.Equal(t, mcp.LoggingLevelAlert, msg.Params.AdditionalFields["level"])
+						assert.Equal(t, "test logger", msg.Params.AdditionalFields["logger"])
+						assert.Equal(t, "test data", msg.Params.AdditionalFields["data"])
+					default:
+						t.Errorf("log message not sent")
+					}
+				}
+			},
+		},
+		{
+			name: "session with blocked channel",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithLogging{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 1),
+					initialized:         false,
+				}
+				logSession.Initialize()
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				require.NoError(t, srv.SendLogMessageToClient(ctx, logMsg))
+				require.Error(t, srv.SendLogMessageToClient(ctx, logMsg))
+			},
+		},
+		{
+			name: "send log messages of different levels",
+			contextPrepare: func(ctx context.Context, srv *MCPServer) context.Context {
+				logSession := &sessionTestClientWithLogging{
+					sessionID:           "test",
+					notificationChannel: make(chan mcp.JSONRPCNotification, 10),
+					initialized:         false,
+				}
+				logSession.Initialize()
+				// Set client log level to "Error"
+				logSession.SetLogLevel(mcp.LoggingLevelError)
+				return srv.WithContext(ctx, logSession)
+			},
+			validate: func(t *testing.T, ctx context.Context, srv *MCPServer) {
+				// Log messages of higher level than client level could be sent
+				require.NoError(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelEmergency, "test logger", "")))
+				require.NoError(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelAlert, "test logger", "")))
+				require.NoError(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelCritical, "test logger", "")))
+				require.NoError(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelError, "test logger", "")))
+
+				// Log messages of lower level than client level could not be sent
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelWarning, "test logger", "")))
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelNotice, "test logger", "")))
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelInfo, "test logger", "")))
+				require.Error(t, srv.SendLogMessageToClient(ctx, mcp.NewLoggingMessageNotification(mcp.LoggingLevelDebug, "test logger", "")))
+
+				logSession, ok := ClientSessionFromContext(ctx).(*sessionTestClientWithLogging)
+				require.True(t, ok, "session not found or of incorrect type")
+
+				// Confirm four log messages were received
+				require.Equal(t, len(logSession.notificationChannel), 4)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := NewMCPServer("test-server", "1.0.0", WithLogging())
+			ctx := tt.contextPrepare(context.Background(), server)
+
+			tt.validate(t, ctx, server)
+		})
+	}
+}