diff --git a/pkg/analysis_server/lib/src/analysis_server.dart b/pkg/analysis_server/lib/src/analysis_server.dart
index d0178e3..067f441 100644
--- a/pkg/analysis_server/lib/src/analysis_server.dart
+++ b/pkg/analysis_server/lib/src/analysis_server.dart
@@ -26,6 +26,7 @@
 import 'package:analysis_server/src/protocol_server.dart' as server;
 import 'package:analysis_server/src/server/crash_reporting_attachments.dart';
 import 'package:analysis_server/src/server/diagnostic_server.dart';
+import 'package:analysis_server/src/server/message_scheduler.dart';
 import 'package:analysis_server/src/server/performance.dart';
 import 'package:analysis_server/src/services/completion/completion_performance.dart';
 import 'package:analysis_server/src/services/correction/assist_internal.dart';
@@ -250,6 +251,10 @@
   /// A completer for [lspUninitialized].
   final Completer<void> _lspUninitializedCompleter = Completer<void>();
 
+  /// A scheduler that keeps track of all incoming messages and schedules them
+  /// for processing.
+  final MessageScheduler messageScheduler;
+
   AnalysisServer(
     this.options,
     this.sdkManager,
@@ -269,7 +274,9 @@
   })  : resourceProvider = OverlayResourceProvider(baseResourceProvider),
         pubApi = PubApi(instrumentationService, httpClient,
             Platform.environment['PUB_HOSTED_URL']),
-        producerGeneratorsForLintRules = AssistProcessor.computeLintRuleMap() {
+        producerGeneratorsForLintRules = AssistProcessor.computeLintRuleMap(),
+        messageScheduler = MessageScheduler() {
+    messageScheduler.setServer(this);
     // Set the default URI converter. This uses the resource providers path
     // context (unlike the initialized value) which allows tests to override it.
     uriConverter = ClientUriConverter.noop(baseResourceProvider.pathContext);
diff --git a/pkg/analysis_server/lib/src/legacy_analysis_server.dart b/pkg/analysis_server/lib/src/legacy_analysis_server.dart
index 5dd6d05..1b7daf5 100644
--- a/pkg/analysis_server/lib/src/legacy_analysis_server.dart
+++ b/pkg/analysis_server/lib/src/legacy_analysis_server.dart
@@ -86,6 +86,7 @@
 import 'package:analysis_server/src/server/diagnostic_server.dart';
 import 'package:analysis_server/src/server/error_notifier.dart';
 import 'package:analysis_server/src/server/features.dart';
+import 'package:analysis_server/src/server/message_scheduler.dart';
 import 'package:analysis_server/src/server/performance.dart';
 import 'package:analysis_server/src/server/sdk_configuration.dart';
 import 'package:analysis_server/src/services/completion/completion_state.dart';
@@ -606,7 +607,8 @@
   /// Handle a [request] that was read from the communication channel.
   void handleRequestOrResponse(RequestOrResponse requestOrResponse) {
     if (requestOrResponse is Request) {
-      handleRequest(requestOrResponse);
+      messageScheduler.add(LegacyMessage(request: requestOrResponse));
+      messageScheduler.notify();
     } else if (requestOrResponse is Response) {
       handleResponse(requestOrResponse);
     }
diff --git a/pkg/analysis_server/lib/src/lsp/lsp_analysis_server.dart b/pkg/analysis_server/lib/src/lsp/lsp_analysis_server.dart
index 9935437..52d392b 100644
--- a/pkg/analysis_server/lib/src/lsp/lsp_analysis_server.dart
+++ b/pkg/analysis_server/lib/src/lsp/lsp_analysis_server.dart
@@ -27,6 +27,7 @@
 import 'package:analysis_server/src/server/detachable_filesystem_manager.dart';
 import 'package:analysis_server/src/server/diagnostic_server.dart';
 import 'package:analysis_server/src/server/error_notifier.dart';
+import 'package:analysis_server/src/server/message_scheduler.dart';
 import 'package:analysis_server/src/server/performance.dart';
 import 'package:analysis_server/src/services/user_prompts/dart_fix_prompt_manager.dart';
 import 'package:analysis_server/src/utilities/extensions/flutter.dart';
@@ -191,7 +192,8 @@
     analysisDriverScheduler.start();
 
     _channelSubscription =
-        channel.listen(handleMessage, onDone: done, onError: socketError);
+        channel.listen(scheduleMessage, onDone: done, onError: socketError);
+
     if (AnalysisServer.supportsPlugins) {
       _pluginChangeSubscription =
           pluginManager.pluginsChanged.listen((_) => _onPluginsChanged());
@@ -757,6 +759,11 @@
     }
   }
 
+  void scheduleMessage(Message message) {
+    messageScheduler.add(LspMessage(message: message));
+    messageScheduler.notify();
+  }
+
   void sendErrorResponse(Message message, ResponseError error) {
     if (message is RequestMessage) {
       sendResponse(ResponseMessage(
diff --git a/pkg/analysis_server/lib/src/server/message_scheduler.dart b/pkg/analysis_server/lib/src/server/message_scheduler.dart
new file mode 100644
index 0000000..7d9948b
--- /dev/null
+++ b/pkg/analysis_server/lib/src/server/message_scheduler.dart
@@ -0,0 +1,92 @@
+// Copyright (c) 2024, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:collection';
+
+import 'package:analysis_server/lsp_protocol/protocol.dart' as lsp;
+import 'package:analysis_server/protocol/protocol.dart' as legacy;
+import 'package:analysis_server/src/analysis_server.dart';
+import 'package:analysis_server/src/legacy_analysis_server.dart';
+import 'package:analysis_server/src/lsp/lsp_analysis_server.dart';
+import 'package:analyzer/src/util/performance/operation_performance.dart';
+import 'package:meta/meta.dart';
+
+/// Represents a message from DTD (Dart Tooling Daemon).
+final class DtdMessage extends MessageObject {
+  final lsp.IncomingMessage message;
+  final Completer<Map<String, Object?>> completer;
+  final OperationPerformanceImpl performance;
+
+  DtdMessage(
+      {required this.message,
+      required this.completer,
+      required this.performance});
+}
+
+/// Represents a message in the Legacy protocol format.
+final class LegacyMessage extends MessageObject {
+  final legacy.Request request;
+
+  LegacyMessage({required this.request});
+}
+
+/// Represents a message in the LSP protocol format.
+final class LspMessage extends MessageObject {
+  final lsp.Message message;
+
+  LspMessage({required this.message});
+}
+
+/// Represents a message from a client, can be an IDE, DTD etc.
+sealed class MessageObject {}
+
+/// The [MessageScheduler] receives messages from all clients of the
+/// [AnalysisServer]. Clients can include IDE's (LSP and Legacy protocol), DTD,
+/// and the Diagnostic server. The [MessageSchedular] acts as a hub for all
+/// incoming messages and forwards the messages to the appropriate handlers.
+final class MessageScheduler {
+  /// The [AnalaysisServer] associated with the schedular.
+  late final AnalysisServer server;
+
+  /// A queue of incoming messages from all the clients of the [AnalysisServer].
+  final ListQueue<MessageObject> _incomingMessages = ListQueue<MessageObject>();
+
+  @visibleForTesting
+  ListQueue<MessageObject> get incomingMessages => _incomingMessages;
+
+  /// Add a message to the end of the incoming messages queue.
+  void add(MessageObject message) {
+    _incomingMessages.addLast(message);
+  }
+
+  /// Notify the [MessageSchedular] to process the messages queue.
+  void notify() async {
+    processMessages();
+  }
+
+  /// Dispatch the first message in the queue to be executed.
+  void processMessages() {
+    if (_incomingMessages.isEmpty) {
+      return;
+    }
+    var message = _incomingMessages.removeFirst();
+    switch (message) {
+      case LspMessage():
+        var lspMessage = message.message;
+        (server as LspAnalysisServer).handleMessage(lspMessage);
+      case LegacyMessage():
+        var request = message.request;
+        (server as LegacyAnalysisServer).handleRequest(request);
+      case DtdMessage():
+        server.dtd!.processMessage(
+            message.message, message.performance, message.completer);
+    }
+  }
+
+  /// Set the [AnalysisServer].
+  void setServer(AnalysisServer analysisServer) {
+    server = analysisServer;
+  }
+}
diff --git a/pkg/analysis_server/lib/src/services/dart_tooling_daemon/dtd_services.dart b/pkg/analysis_server/lib/src/services/dart_tooling_daemon/dtd_services.dart
index 425d832..b58af9b 100644
--- a/pkg/analysis_server/lib/src/services/dart_tooling_daemon/dtd_services.dart
+++ b/pkg/analysis_server/lib/src/services/dart_tooling_daemon/dtd_services.dart
@@ -6,10 +6,10 @@
 
 import 'package:analysis_server/lsp_protocol/protocol.dart';
 import 'package:analysis_server/src/analysis_server.dart';
-import 'package:analysis_server/src/lsp/client_capabilities.dart';
 import 'package:analysis_server/src/lsp/error_or.dart';
 import 'package:analysis_server/src/lsp/handlers/handler_states.dart';
 import 'package:analysis_server/src/lsp/handlers/handlers.dart';
+import 'package:analysis_server/src/server/message_scheduler.dart';
 import 'package:analysis_server/src/server/performance.dart';
 import 'package:analyzer/src/util/performance/operation_performance.dart';
 import 'package:dtd/dtd.dart';
@@ -63,6 +63,54 @@
 
   DtdConnectionState get state => _state;
 
+  /// Executes the LSP handler [messageHandler] with [params] and returns the
+  /// results as a map to provide back to DTD.
+  ///
+  /// If the handler fails, throws an [RpcException] to be propagated to the
+  /// client.
+  void processMessage(
+      IncomingMessage message,
+      OperationPerformanceImpl performance,
+      Completer<Map<String, Object?>> completer) async {
+    // (TODO:keertip) Lookup the right handler, execute and return results.
+    // For now, complete with exception.
+    completer.completeError(RpcException(
+      ErrorCodes.InvalidRequest.toJson(),
+      'DTD requests are not yet supported',
+    ));
+
+    // (TODO:keertip) Uncomment when lookup has been implemented
+    // var info = MessageInfo(
+    //   performance: performance,
+    //   // DTD clients requests are always executed with a fixed set of
+    //   // capabilities so that the responses don't change in format based on the
+    //   // owning editor.
+    //   clientCapabilities: fixedBasicLspClientCapabilities,
+    // );
+    // var token = NotCancelableToken(); // We don't currently support cancel.
+
+    // // Execute the handler.
+    // var result = await messageHandler.handleMessage(message, info, token);
+
+    // // Map the result (or error) on to what a DTD handler needs to return.
+    // return result.map(
+    //   // Map LSP errors on to equiv JSON-RPC errors for DTD.
+    //   (error) => throw RpcException(
+    //     error.code.toJson(),
+    //     error.message,
+    //     data: error.data,
+    //   ),
+    //   // DTD requires that all results are a Map and that they contain a
+    //   // 'type' field. This differs slightly from LSP where we could return a
+    //   // boolean (for example). This means we need to put the result in a
+    //   // field, which we're calling 'result'.
+    //   (result) => {
+    //     'type': result?.runtimeType.toString(),
+    //     'result': result,
+    //   },
+    // );
+  }
+
   /// Closes the connection to DTD and cleans up.
   void _close([DtdConnectionState state = DtdConnectionState.Disconnected]) {
     _state = state;
@@ -122,54 +170,26 @@
     }
   }
 
-  /// Executes the LSP handler [messageHandler] with [params] and returns the
-  /// results as a map to provide back to DTD.
-  ///
-  /// If the handler fails, throws an [RpcException] to be propagated to the
-  /// client.
+  /// The incoming request is sent to the [MessageScheduler] for execution.
+  /// A completer is returned which will be completed with the result of the
+  /// execution of the request by the corresponding [MessageHandler].
   Future<Map<String, Object?>> _executeLspHandler(
     MessageHandler<Object?, Object?, AnalysisServer> messageHandler,
     Parameters params,
     OperationPerformanceImpl performance,
   ) async {
-    // TODO(dantup): Currently the handler just runs immediately, but this
-    //  should interact with the scheduler in future.
-
     // Map the incoming request into types we use for LSP request handling.
     var message = IncomingMessage(
       jsonrpc: jsonRpcVersion,
       method: messageHandler.handlesMessage,
       params: params.asMap,
     );
-    var info = MessageInfo(
-      performance: performance,
-      // DTD clients requests are always executed with a fixed set of
-      // capabilities so that the responses don't change in format based on the
-      // owning editor.
-      clientCapabilities: fixedBasicLspClientCapabilities,
-    );
-    var token = NotCancelableToken(); // We don't currently support cancel.
-
-    // Execute the handler.
-    var result = await messageHandler.handleMessage(message, info, token);
-
-    // Map the result (or error) on to what a DTD handler needs to return.
-    return result.map(
-      // Map LSP errors on to equiv JSON-RPC errors for DTD.
-      (error) => throw RpcException(
-        error.code.toJson(),
-        error.message,
-        data: error.data,
-      ),
-      // DTD requires that all results are a Map and that they contain a
-      // 'type' field. This differs slightly from LSP where we could return a
-      // boolean (for example). This means we need to put the result in a
-      // field, which we're calling 'result'.
-      (result) => {
-        'type': result?.runtimeType.toString(),
-        'result': result,
-      },
-    );
+    var scheduler = _server.messageScheduler;
+    var completer = Completer<Map<String, Object?>>();
+    scheduler.add(DtdMessage(
+        message: message, performance: performance, completer: completer));
+    scheduler.notify();
+    return completer.future;
   }
 
   /// Handles an unexpected error occurring on the DTD connection by logging and
diff --git a/pkg/analysis_server/test/integration/server/message_scheduler_test.dart b/pkg/analysis_server/test/integration/server/message_scheduler_test.dart
new file mode 100644
index 0000000..1df1d6d
--- /dev/null
+++ b/pkg/analysis_server/test/integration/server/message_scheduler_test.dart
@@ -0,0 +1,119 @@
+// Copyright (c) 2024, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:collection';
+
+import 'package:analysis_server/lsp_protocol/protocol.dart' as lsp;
+import 'package:analysis_server/protocol/protocol.dart' as legacy;
+import 'package:analysis_server/protocol/protocol_constants.dart';
+import 'package:analysis_server/protocol/protocol_generated.dart';
+import 'package:analysis_server/src/server/message_scheduler.dart';
+import 'package:analyzer/src/util/performance/operation_performance.dart';
+import 'package:analyzer_utilities/testing/tree_string_sink.dart';
+import 'package:test/test.dart';
+import 'package:test_reflective_loader/test_reflective_loader.dart';
+
+void main() {
+  defineReflectiveSuite(() {
+    defineReflectiveTests(MessageSchedulerTest);
+  });
+}
+
+@reflectiveTest
+class MessageSchedulerTest {
+  late MessageScheduler messageScheduler;
+
+  DtdMessage get dtdMessage {
+    return DtdMessage(
+        message: lspRquest,
+        performance: OperationPerformanceImpl('<root>'),
+        completer: Completer());
+  }
+
+  legacy.Request get legacyRequest {
+    var params = AnalysisSetAnalysisRootsParams(['a', 'b', 'c'], ['d', 'e']);
+    return legacy.Request('1', ANALYSIS_REQUEST_SET_ANALYSIS_ROOTS,
+        params.toJson(clientUriConverter: null));
+  }
+
+  lsp.RequestMessage get lspRquest {
+    var params = {'processId': 'invalid'};
+    return lsp.RequestMessage(
+      id: lsp.Either2<int, String>.t1(1),
+      method: lsp.Method.initialize,
+      params: params,
+      jsonrpc: lsp.jsonRpcVersion,
+    );
+  }
+
+  void setUp() {
+    messageScheduler = MessageScheduler();
+  }
+
+  void test_addMultipleToQueue() {
+    messageScheduler.add(LegacyMessage(request: legacyRequest));
+    messageScheduler.add(LspMessage(message: lspRquest));
+    messageScheduler.add(dtdMessage);
+    _assertQueueContents(r'''
+incomingMessages
+  LegacyMessage
+    method: analysis.setAnalysisRoots
+  LspMessage
+    method: initialize
+  DtdMessage
+    method: initialize
+''');
+  }
+
+  void test_addSingleToQueue() {
+    messageScheduler.add(LspMessage(message: lspRquest));
+    _assertQueueContents(r'''
+incomingMessages
+  LspMessage
+    method: initialize
+''');
+  }
+
+  void _assertQueueContents(String expected) {
+    var actual = _getQueueContents(messageScheduler.incomingMessages);
+    if (actual != expected) {
+      print('-------- Actual --------');
+      print('$actual------------------------');
+    }
+    expect(actual, expected);
+  }
+
+  String _getQueueContents(ListQueue<MessageObject> queue) {
+    var buffer = StringBuffer();
+    var sink = TreeStringSink(sink: buffer, indent: '  ');
+    sink.writeln('incomingMessages');
+    while (queue.isNotEmpty) {
+      var message = queue.removeFirst();
+      switch (message) {
+        case DtdMessage():
+          sink.writelnWithIndent('DtdMessage');
+          sink.withIndent(() {
+            sink.writelnWithIndent(
+                'method: ${message.message.method.toString()}');
+          });
+        case LegacyMessage():
+          sink.writelnWithIndent('LegacyMessage');
+          sink.withIndent(() {
+            sink.writelnWithIndent(
+                'method: ${message.request.method.toString()}');
+          });
+        case LspMessage():
+          sink.writelnWithIndent('LspMessage');
+          var msg = message.message;
+          if (msg case lsp.RequestMessage()) {
+            sink.withIndent(() {
+              sink.writelnWithIndent('method: ${msg.method.toString()}');
+            });
+          }
+      }
+    }
+    return buffer.toString();
+  }
+}
diff --git a/pkg/analysis_server/test/integration/server/test_all.dart b/pkg/analysis_server/test/integration/server/test_all.dart
index d96283a..ea0a87e 100644
--- a/pkg/analysis_server/test/integration/server/test_all.dart
+++ b/pkg/analysis_server/test/integration/server/test_all.dart
@@ -7,6 +7,7 @@
 import 'blaze_changes_test.dart' as blaze_changes_test;
 import 'command_line_options_test.dart' as command_line_options_test;
 import 'get_version_test.dart' as get_version_test;
+import 'message_scheduler_test.dart' as scheduler_test;
 import 'set_subscriptions_invalid_service_test.dart'
     as set_subscriptions_invalid_service_test;
 import 'set_subscriptions_test.dart' as set_subscriptions_test;
@@ -18,6 +19,7 @@
     blaze_changes_test.main();
     command_line_options_test.main();
     get_version_test.main();
+    scheduler_test.main();
     set_subscriptions_test.main();
     set_subscriptions_invalid_service_test.main();
     shutdown_test.main();