Add support for message filters

This commit adds support for server-side message filters that can intercept
and process incoming and outgoing JSON-RPC messages. It also includes:

- Message filter infrastructure (McpMessageFilter, McpMessageHandler)
- MessageContext for accessing message metadata during filter execution
- Extended RequestContext with message filter support
- McpServerBuilderExtensions for registering message filters
- Comprehensive tests for message filter functionality

Author: halter73

src/ModelContextProtocol.Core/Authentication/ClientOAuthProvider.cs

@@ -260,9 +260,6 @@ private async Task<string> GetAccessTokenAsync(HttpResponseMessage response, boo
         // Get auth server metadata
         var authServerMetadata = await GetAuthServerMetadataAsync(selectedAuthServer, cancellationToken).ConfigureAwait(false);
 
-        // Store auth server metadata for future refresh operations
-        _authServerMetadata = authServerMetadata;
-
         // The existing access token must be invalid to have resulted in a 401 response, but refresh might still work.
         var resourceUri = GetRequiredResourceUri(protectedResourceMetadata);
 
@@ -296,6 +293,9 @@ await _tokenCache.GetTokensAsync(cancellationToken).ConfigureAwait(false) is { R
             }
         }
 
+        // Store auth server metadata for future refresh operations
+        _authServerMetadata = authServerMetadata;
+
         // Perform the OAuth flow
         return await InitiateAuthorizationCodeFlowAsync(protectedResourceMetadata, authServerMetadata, cancellationToken).ConfigureAwait(false);
     }

src/ModelContextProtocol.Core/Client/McpClientImpl.cs

@@ -58,7 +58,15 @@ internal McpClientImpl(ITransport transport, string endpointName, McpClientOptio
 
         RegisterHandlers(options, notificationHandlers, requestHandlers);
 
-        _sessionHandler = new McpSessionHandler(isServer: false, transport, endpointName, requestHandlers, notificationHandlers, _logger);
+        _sessionHandler = new McpSessionHandler(
+            isServer: false,
+            transport,
+            endpointName,
+            requestHandlers,
+            notificationHandlers,
+            incomingMessageFilter: null,
+            outgoingMessageFilter: null,
+            _logger);
     }
 
     private void RegisterHandlers(McpClientOptions options, NotificationHandlers notificationHandlers, RequestHandlers requestHandlers)

src/ModelContextProtocol.Core/JsonRpcMessageFilter.cs

@@ -0,0 +1,10 @@
+using ModelContextProtocol.Protocol;
+
+namespace ModelContextProtocol;
+
+/// <summary>
+/// Represents a filter that wraps the processing of incoming JSON-RPC messages.
+/// </summary>
+/// <param name="next">The next handler in the pipeline.</param>
+/// <returns>A wrapped handler that processes messages and optionally delegates to the next handler.</returns>
+internal delegate Func<JsonRpcMessage, CancellationToken, Task> JsonRpcMessageFilter(Func<JsonRpcMessage, CancellationToken, Task> next);

src/ModelContextProtocol.Core/McpSessionHandler.cs

@@ -66,6 +66,8 @@ internal static bool SupportsPrimingEvent(string? protocolVersion)
     private readonly ITransport _transport;
     private readonly RequestHandlers _requestHandlers;
     private readonly NotificationHandlers _notificationHandlers;
+    private readonly JsonRpcMessageFilter _incomingMessageFilter;
+    private readonly JsonRpcMessageFilter _outgoingMessageFilter;
     private readonly long _sessionStartingTimestamp = Stopwatch.GetTimestamp();
 
     private readonly DistributedContextPropagator _propagator = DistributedContextPropagator.Current;
@@ -95,13 +97,17 @@ internal static bool SupportsPrimingEvent(string? protocolVersion)
     /// <param name="endpointName">The name of the endpoint for logging and debug purposes.</param>
     /// <param name="requestHandlers">A collection of request handlers.</param>
     /// <param name="notificationHandlers">A collection of notification handlers.</param>
+    /// <param name="incomingMessageFilter">A filter that wraps incoming message processing. Takes the next handler and returns a wrapped handler. If null, a passthrough filter is used.</param>
+    /// <param name="outgoingMessageFilter">A filter that wraps outgoing message processing. Takes the next handler and returns a wrapped handler. If null, a passthrough filter is used.</param>
     /// <param name="logger">The logger.</param>
     public McpSessionHandler(
         bool isServer,
         ITransport transport,
         string endpointName,
         RequestHandlers requestHandlers,
         NotificationHandlers notificationHandlers,
+        JsonRpcMessageFilter? incomingMessageFilter,
+        JsonRpcMessageFilter? outgoingMessageFilter,
         ILogger logger)
     {
         Throw.IfNull(transport);
@@ -120,7 +126,9 @@ public McpSessionHandler(
         EndpointName = endpointName;
         _requestHandlers = requestHandlers;
         _notificationHandlers = notificationHandlers;
-        _logger = logger ?? NullLogger.Instance;
+        _incomingMessageFilter = incomingMessageFilter ?? (next => next);
+        _outgoingMessageFilter = outgoingMessageFilter ?? (next => next);
+        _logger = logger;
         LogSessionCreated(EndpointName, _sessionId, _transportKind);
     }
 
@@ -309,36 +317,16 @@ private async Task HandleMessageAsync(JsonRpcMessage message, CancellationToken
                 AddTags(ref tags, activity, message, method, target);
             }
 
-            switch (message)
+            var filteredHandler = _incomingMessageFilter(async (msg, ct) =>
             {
-                case JsonRpcRequest request:
-                    LogRequestHandlerCalled(EndpointName, request.Method);
-                    long requestStartingTimestamp = Stopwatch.GetTimestamp();
-                    try
-                    {
-                        var result = await HandleRequest(request, cancellationToken).ConfigureAwait(false);
-                        LogRequestHandlerCompleted(EndpointName, request.Method, GetElapsed(requestStartingTimestamp).TotalMilliseconds);
-                        AddResponseTags(ref tags, activity, result, method);
-                    }
-                    catch (Exception ex)
-                    {
-                        LogRequestHandlerException(EndpointName, request.Method, GetElapsed(requestStartingTimestamp).TotalMilliseconds, ex);
-                        throw;
-                    }
-                    break;
-
-                case JsonRpcNotification notification:
-                    await HandleNotification(notification, cancellationToken).ConfigureAwait(false);
-                    break;
-
-                case JsonRpcMessageWithId messageWithId:
-                    HandleMessageWithId(message, messageWithId);
-                    break;
+                var result = await HandleMessageCoreAsync(msg, ct).ConfigureAwait(false);
+                if (addTags && result is not null)
+                {
+                    AddResponseTags(ref tags, activity, result, method);
+                }
+            });
 
-                default:
-                    LogEndpointHandlerUnexpectedMessageType(EndpointName, message.GetType().Name);
-                    break;
-            }
+            await filteredHandler(message, cancellationToken).ConfigureAwait(false);
         }
         catch (Exception e) when (addTags)
         {
@@ -351,7 +339,40 @@ private async Task HandleMessageAsync(JsonRpcMessage message, CancellationToken
         }
     }
 
-    private async Task HandleNotification(JsonRpcNotification notification, CancellationToken cancellationToken)
+    private async Task<JsonNode?> HandleMessageCoreAsync(JsonRpcMessage message, CancellationToken cancellationToken)
+    {
+        switch (message)
+        {
+            case JsonRpcRequest request:
+                LogRequestHandlerCalled(EndpointName, request.Method);
+                long requestStartingTimestamp = Stopwatch.GetTimestamp();
+                try
+                {
+                    var result = await HandleRequestAsync(request, cancellationToken).ConfigureAwait(false);
+                    LogRequestHandlerCompleted(EndpointName, request.Method, GetElapsed(requestStartingTimestamp).TotalMilliseconds);
+                    return result;
+                }
+                catch (Exception ex)
+                {
+                    LogRequestHandlerException(EndpointName, request.Method, GetElapsed(requestStartingTimestamp).TotalMilliseconds, ex);
+                    throw;
+                }
+
+            case JsonRpcNotification notification:
+                await HandleNotificationAsync(notification, cancellationToken).ConfigureAwait(false);
+                return null;
+
+            case JsonRpcMessageWithId messageWithId:
+                HandleMessageWithId(message, messageWithId);
+                return null;
+
+            default:
+                LogEndpointHandlerUnexpectedMessageType(EndpointName, message.GetType().Name);
+                return null;
+        }
+    }
+
+    private async Task HandleNotificationAsync(JsonRpcNotification notification, CancellationToken cancellationToken)
     {
         // Special-case cancellation to cancel a pending operation. (We'll still subsequently invoke a user-specified handler if one exists.)
         if (notification.Method == NotificationMethods.CancelledNotification)
@@ -387,7 +408,7 @@ private void HandleMessageWithId(JsonRpcMessage message, JsonRpcMessageWithId me
         }
     }
 
-    private async Task<JsonNode?> HandleRequest(JsonRpcRequest request, CancellationToken cancellationToken)
+    private async Task<JsonNode?> HandleRequestAsync(JsonRpcRequest request, CancellationToken cancellationToken)
     {
         if (!_requestHandlers.TryGetValue(request.Method, out var handler))
         {
@@ -586,26 +607,31 @@ public async Task SendMessageAsync(JsonRpcMessage message, CancellationToken can
                 AddTags(ref tags, activity, message, method, target);
             }
 
-            if (_logger.IsEnabled(LogLevel.Trace))
+            var filteredHandler = _outgoingMessageFilter(async (msg, ct) =>
             {
-                LogSendingMessageSensitive(EndpointName, JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage));
-            }
-            else
-            {
-                LogSendingMessage(EndpointName);
-            }
+                if (_logger.IsEnabled(LogLevel.Trace))
+                {
+                    LogSendingMessageSensitive(EndpointName, JsonSerializer.Serialize(msg, McpJsonUtilities.JsonContext.Default.JsonRpcMessage));
+                }
+                else
+                {
+                    LogSendingMessage(EndpointName);
+                }
 
-            await SendToRelatedTransportAsync(message, cancellationToken).ConfigureAwait(false);
+                await SendToRelatedTransportAsync(msg, ct).ConfigureAwait(false);
 
-            // If the sent notification was a cancellation notification, cancel the pending request's await, as either the
-            // server won't be sending a response, or per the specification, the response should be ignored. There are inherent
-            // race conditions here, so it's possible and allowed for the operation to complete before we get to this point.
-            if (message is JsonRpcNotification { Method: NotificationMethods.CancelledNotification } notification &&
-                GetCancelledNotificationParams(notification.Params) is CancelledNotificationParams cn &&
-                _pendingRequests.TryRemove(cn.RequestId, out var tcs))
-            {
-                tcs.TrySetCanceled(default);
-            }
+                // If the sent notification was a cancellation notification, cancel the pending request's await, as either the
+                // server won't be sending a response, or per the specification, the response should be ignored. There are inherent
+                // race conditions here, so it's possible and allowed for the operation to complete before we get to this point.
+                if (msg is JsonRpcNotification { Method: NotificationMethods.CancelledNotification } notification &&
+                    GetCancelledNotificationParams(notification.Params) is CancelledNotificationParams cn &&
+                    _pendingRequests.TryRemove(cn.RequestId, out var tcs))
+                {
+                    tcs.TrySetCanceled(default);
+                }
+            });
+
+            await filteredHandler(message, cancellationToken).ConfigureAwait(false);
         }
         catch (Exception ex) when (addTags)
         {

src/ModelContextProtocol.Core/Server/McpMessageFilter.cs

@@ -0,0 +1,30 @@
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Delegate type for applying filters to JSON-RPC messages.
+/// </summary>
+/// <param name="next">The next message handler in the pipeline.</param>
+/// <returns>The next message handler wrapped with the filter.</returns>
+/// <remarks>
+/// <para>
+/// Message filters allow you to intercept and process JSON-RPC messages before they reach
+/// their respective handlers (incoming) or before they are sent (outgoing). This is useful for implementing
+/// cross-cutting concerns that need to apply to all message types, such as logging, authentication, rate limiting,
+/// redaction, or request tracing.
+/// </para>
+/// <para>
+/// Filters are applied in the order they are registered, with the first registered filter being the outermost.
+/// Each filter receives the next handler in the pipeline and can choose to:
+/// <list type="bullet">
+/// <item><description>Call the next handler to continue processing (await next(context, cancellationToken))</description></item>
+/// <item><description>Skip the default handlers entirely by not calling next</description></item>
+/// <item><description>Perform operations before and/or after calling next</description></item>
+/// <item><description>Catch and handle exceptions from inner handlers</description></item>
+/// </list>
+/// </para>
+/// <para>
+/// For request-specific filters, use <see cref="McpRequestFilter{TParams, TResult}"/> instead.
+/// </para>
+/// </remarks>
+public delegate McpMessageHandler McpMessageFilter(
+    McpMessageHandler next);

src/ModelContextProtocol.Core/Server/McpMessageHandler.cs

@@ -0,0 +1,21 @@
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Delegate type for handling incoming JSON-RPC messages.
+/// </summary>
+/// <param name="context">The message context containing the JSON-RPC message and other metadata.</param>
+/// <param name="cancellationToken">A cancellation token to cancel the operation.</param>
+/// <returns>A task representing the asynchronous operation.</returns>
+/// <remarks>
+/// <para>
+/// This delegate can handle any type of JSON-RPC message, including requests, notifications, responses, and errors.
+/// Use this for implementing cross-cutting concerns that need to intercept all message types,
+/// such as logging, authentication, rate limiting, or request tracing.
+/// </para>
+/// <para>
+/// For request-specific handling, use <see cref="McpRequestHandler{TParams, TResult}"/> instead.
+/// </para>
+/// </remarks>
+public delegate Task McpMessageHandler(
+    MessageContext context,
+    CancellationToken cancellationToken);

src/ModelContextProtocol.Core/Server/McpServerFilters.cs

@@ -11,6 +11,39 @@ namespace ModelContextProtocol.Server;
 /// </remarks>
 public sealed class McpServerFilters
 {
+    /// <summary>
+    /// Gets the filters for all incoming JSON-RPC messages.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// These filters intercept all incoming JSON-RPC messages before they are processed by the server,
+    /// including requests, notifications, responses, and errors. The filters can perform logging,
+    /// authentication, rate limiting, or other cross-cutting concerns that apply to all message types.
+    /// </para>
+    /// <para>
+    /// Message filters are applied before request-specific filters. If a message filter does not call
+    /// the next handler in the pipeline, the default handlers will not be executed.
+    /// </para>
+    /// </remarks>
+    public List<McpMessageFilter> IncomingMessageFilters { get; } = [];
+
+    /// <summary>
+    /// Gets the filters for all outgoing JSON-RPC messages.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// These filters intercept all outgoing JSON-RPC messages before they are sent to the client,
+    /// including responses, notifications, and errors. The filters can perform logging,
+    /// redaction, auditing, or other cross-cutting concerns that apply to all message types.
+    /// </para>
+    /// <para>
+    /// If a message filter does not call the next handler in the pipeline, the message will not be sent.
+    /// Filters may also call the next handler multiple times with different messages to emit additional
+    /// server-to-client messages.
+    /// </para>
+    /// </remarks>
+    public List<McpMessageFilter> OutgoingMessageFilters { get; } = [];
+
     /// <summary>
     /// Gets the filters for the list-tools handler pipeline.
     /// </summary>

src/ModelContextProtocol.Core/Server/McpServerImpl.cs

@@ -118,7 +118,17 @@ void Register<TPrimitive>(McpServerPrimitiveCollection<TPrimitive>? collection,
         }
 
         // And initialize the session.
-        _sessionHandler = new McpSessionHandler(isServer: true, _sessionTransport, _endpointName!, _requestHandlers, _notificationHandlers, _logger);
+        var incomingMessageFilter = BuildMessageFilterPipeline(options.Filters.IncomingMessageFilters);
+        var outgoingMessageFilter = BuildMessageFilterPipeline(options.Filters.OutgoingMessageFilters);
+        _sessionHandler = new McpSessionHandler(
+            isServer: true,
+            _sessionTransport,
+            _endpointName!,
+            _requestHandlers,
+            _notificationHandlers,
+            incomingMessageFilter,
+            outgoingMessageFilter,
+            _logger);
     }
 
     /// <inheritdoc/>
@@ -875,6 +885,37 @@ private static McpRequestHandler<TParams, TResult> BuildFilterPipeline<TParams,
         return current;
     }
 
+    private JsonRpcMessageFilter BuildMessageFilterPipeline(List<McpMessageFilter> filters)
+    {
+        if (filters.Count == 0)
+        {
+            return next => next;
+        }
+
+        return next =>
+        {
+            // Build the handler chain from the filters.
+            // The innermost handler calls the provided 'next' delegate with the message from the context.
+            McpMessageHandler baseHandler = async (context, cancellationToken) =>
+            {
+                await next(context.JsonRpcMessage, cancellationToken).ConfigureAwait(false);
+            };
+
+            var current = baseHandler;
+            for (int i = filters.Count - 1; i >= 0; i--)
+            {
+                current = filters[i](current);
+            }
+
+            // Return the handler that creates a MessageContext and invokes the pipeline.
+            return async (message, cancellationToken) =>
+            {
+                var context = new MessageContext(new DestinationBoundMcpServer(this, message.Context?.RelatedTransport), message);
+                await current(context, cancellationToken).ConfigureAwait(false);
+            };
+        };
+    }
+
     private void UpdateEndpointNameWithClientInfo()
     {
         if (ClientInfo is null)