diff --git a/doc/api/errors.md b/doc/api/errors.md
index 4d3829f78e7b95..3e8b9dfc33043a 100644
--- a/doc/api/errors.md
+++ b/doc/api/errors.md
@@ -3072,6 +3072,16 @@ added: v18.1.0
 The `Response` that has been passed to `WebAssembly.compileStreaming` or to
 `WebAssembly.instantiateStreaming` is not a valid WebAssembly response.
 
+<a id="ERR_WORKER_CONNECTION_REFUSED"></a>
+
+### `ERR_WORKER_CONNECTION_REFUSED`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread requested in [`connect()`][] refused the connection or has no connections listener provided.
+
 <a id="ERR_WORKER_INIT_FAILED"></a>
 
 ### `ERR_WORKER_INIT_FAILED`
@@ -3085,6 +3095,16 @@ The `Worker` initialization failed.
 The `execArgv` option passed to the `Worker` constructor contains
 invalid flags.
 
+<a id="ERR_WORKER_INVALID_ID"></a>
+
+### `ERR_WORKER_INVALID_ID`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread ID requested in [`connect()`][] is invalid.
+
 <a id="ERR_WORKER_NOT_RUNNING"></a>
 
 ### `ERR_WORKER_NOT_RUNNING`
@@ -3104,6 +3124,16 @@ The `Worker` instance terminated because it reached its memory limit.
 The path for the main script of a worker is neither an absolute path
 nor a relative path starting with `./` or `../`.
 
+<a id="ERR_WORKER_SAME_THREAD"></a>
+
+### `ERR_WORKER_SAME_THREAD`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+The thread id requested in [`connect()`][] is the current thread id.
+
 <a id="ERR_WORKER_UNSERIALIZABLE_ERROR"></a>
 
 ### `ERR_WORKER_UNSERIALIZABLE_ERROR`
@@ -3999,6 +4029,7 @@ An error occurred trying to allocate memory. This should never happen.
 [`Writable`]: stream.md#class-streamwritable
 [`child_process`]: child_process.md
 [`cipher.getAuthTag()`]: crypto.md#ciphergetauthtag
+[`connect()`]: worker_threads.md#workerconnecttarget-data-timeout
 [`crypto.getDiffieHellman()`]: crypto.md#cryptogetdiffiehellmangroupname
 [`crypto.scrypt()`]: crypto.md#cryptoscryptpassword-salt-keylen-options-callback
 [`crypto.scryptSync()`]: crypto.md#cryptoscryptsyncpassword-salt-keylen-options
diff --git a/doc/api/worker_threads.md b/doc/api/worker_threads.md
index e4e842c6d21365..a255195f202d75 100644
--- a/doc/api/worker_threads.md
+++ b/doc/api/worker_threads.md
@@ -61,6 +61,128 @@ Worker threads inherit non-process-specific options by default. Refer to
 [`Worker constructor options`][] to know how to customize worker thread options,
 specifically `argv` and `execArgv` options.
 
+## `worker.connect(target[, data][, timeout])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1.1 - Active development
+
+* `target` {number} The target thread ID.
+* `data` {any} Any arbitrary, cloneable JavaScript value.
+  The data will be passed as the second argument to the callback provided in the
+  target thread via [`worker.setConnectionsListener()`][].
+* `timeout` {number} Time to wait for the communication port to the target thread,
+  in milliseconds. By default it's `undefined`, which means wait forever.
+* Returns: {Promise} A promise for a `MessagePort`.
+
+Establishes a connection to another worker thread in the same process, returning a
+`MessagePort` that can be used for the communication.
+
+The target thread must have a connection listener setup via [`worker.setConnectionsListener()`][]
+otherwise the connection request will fail.
+
+This method does not support transferables in the `data` argument. If there is need to
+transfer objects between threads it can be accomplished via the returned port once the
+connection has been established.
+
+This method should be used when the target thread is not the direct
+parent or child of the current thread.
+If the two threads are parent-children, use the [`require('node:worker_threads').parentPort.postMessage()`][]
+and the [`worker.postMessage()`][] to let the threads communicate.
+
+If the messages never need transferable, use `BroadcastChannel` for the communication. Remember that
+they are one to many channels so if you need to narrow it down to a one to one channel you will have to
+ensure only two threads subscribe to a certain channel, for instance by using unique channel IDs.
+
+The example below shows the combined use of `connect` and [`worker.setConnectionsListener()`][]:
+it creates 10 nested threads, the last one will try to communicate with the third one.
+Since `MessagePort`s cannot be transferred in `BroadcastChannel`, the `connect` API was the only way
+to let the two threads talk.
+
+```mjs
+import { fileURLToPath } from 'node:url';
+import {
+  Worker,
+  connect,
+  setConnectionsListener,
+  threadId,
+  workerData,
+} from 'node:worker_threads';
+
+const level = workerData?.level ?? 0;
+const targetThread =
+  workerData?.targetThread ?? (level === 2 ? threadId : undefined);
+
+if (level < 10) {
+  const worker = new Worker(fileURLToPath(import.meta.url), {
+    workerData: { level: level + 1, targetThread },
+  });
+}
+
+if (level === 2) {
+  setConnectionsListener((sender, port, data) => {
+    port.on('message', (message) => {
+      console.log(`${sender} -> ${threadId}`, message);
+      port.postMessage({ message: 'pong', data });
+    });
+
+    return true;
+  });
+} else if (level === 10) {
+  const port = await connect(targetThread, { foo: 'bar' });
+
+  port.on('message', (message) => {
+    console.log(`${targetThread} -> ${threadId}`, message);
+    port.close();
+  });
+
+  port.postMessage({ message: 'ping' });
+}
+```
+
+```cjs
+const {
+  Worker,
+  connect,
+  setConnectionsListener,
+  threadId,
+  workerData,
+} = require('node:worker_threads');
+
+const level = workerData?.level ?? 0;
+const targetThread =
+  workerData?.targetThread ?? (level === 2 ? threadId : undefined);
+
+if (level < 10) {
+  const worker = new Worker(__filename, {
+    workerData: { level: level + 1, targetThread },
+  });
+}
+
+if (level === 2) {
+  setConnectionsListener((sender, port, data) => {
+    port.on('message', (message) => {
+      console.log(`${sender} -> ${threadId}`, message);
+      port.postMessage({ message: 'pong', data });
+    });
+
+    return true;
+  });
+} else if (level === 10) {
+  connect(targetThread, { foo: 'bar' }).then((port) => {
+    port.on('message', (message) => {
+      console.log(`${targetThread} -> ${threadId}`, message);
+      port.close();
+    });
+
+    port.postMessage({ message: 'ping' });
+
+  });
+}
+```
+
 ## `worker.getEnvironmentData(key)`
 
 <!-- YAML
@@ -325,6 +447,32 @@ new Worker('process.env.SET_IN_WORKER = "foo"', { eval: true, env: SHARE_ENV })
   });
 ```
 
+## `worker.setConnectionsListener(fn)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1.1 - Active development
+
+* `fn` {Function} A callback to be executed when [`worker.connect()`][] is called from another thread.
+  The function will receive the following arguments:
+
+  * `sender` {number} The other thread ID.
+  * `port` {MessagePort} The port than can be used to communicate with the other thread.
+  * `data` {any} The data passed as second argument to [`worker.connect()`][].
+
+The function must return `true` to accept the connection or any other value to
+refuse the connection. If the function returns a `Promise`, it will be awaited.
+
+Sets the callback that handles connection from other worker threads in the same process.
+If the callback is `null` or `undefined` then the current listener is removed.
+
+When no listeners are present (the default) all connection requests are immediately
+refused.
+
+See the example in [`worker.connect()`][] for more info on how to use this function and its callback.
+
 ## `worker.setEnvironmentData(key[, value])`
 
 <!-- YAML
@@ -1437,8 +1585,10 @@ thread spawned will spawn another until the application crashes.
 [`v8.getHeapSnapshot()`]: v8.md#v8getheapsnapshotoptions
 [`vm`]: vm.md
 [`worker.SHARE_ENV`]: #workershare_env
+[`worker.connect()`]: #workerconnecttarget-data-timeout
 [`worker.on('message')`]: #event-message_1
 [`worker.postMessage()`]: #workerpostmessagevalue-transferlist
+[`worker.setConnectionsListener()`]: #workersetconnectionslistenerfn
 [`worker.terminate()`]: #workerterminate
 [`worker.threadId`]: #workerthreadid_1
 [async-resource-worker-pool]: async_context.md#using-asyncresource-for-a-worker-thread-pool
diff --git a/lib/internal/errors.js b/lib/internal/errors.js
index f78be397d5dce2..b15e355e831ed6 100644
--- a/lib/internal/errors.js
+++ b/lib/internal/errors.js
@@ -1858,10 +1858,12 @@ E('ERR_VM_MODULE_NOT_MODULE',
 E('ERR_VM_MODULE_STATUS', 'Module status %s', Error);
 E('ERR_WASI_ALREADY_STARTED', 'WASI instance has already started', Error);
 E('ERR_WEBASSEMBLY_RESPONSE', 'WebAssembly response %s', TypeError);
+E('ERR_WORKER_CONNECTION_REFUSED', 'Connection refused from worker', Error);
 E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error);
 E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') =>
   `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`,
   Error);
+E('ERR_WORKER_INVALID_ID', 'Invalid worker id %d', Error);
 E('ERR_WORKER_NOT_RUNNING', 'Worker instance not running', Error);
 E('ERR_WORKER_OUT_OF_MEMORY',
   'Worker terminated due to reaching memory limit: %s', Error);
@@ -1876,6 +1878,7 @@ E('ERR_WORKER_PATH', (filename) =>
   ) +
   ` Received "${filename}"`,
   TypeError);
+E('ERR_WORKER_SAME_THREAD', 'Cannot connect to the same thread', Error);
 E('ERR_WORKER_UNSERIALIZABLE_ERROR',
   'Serializing an uncaught exception failed', Error);
 E('ERR_WORKER_UNSUPPORTED_OPERATION',
diff --git a/lib/internal/main/worker_thread.js b/lib/internal/main/worker_thread.js
index aa329b9fe04f15..eebed5b46ffa24 100644
--- a/lib/internal/main/worker_thread.js
+++ b/lib/internal/main/worker_thread.js
@@ -28,6 +28,7 @@ const {
   getEnvMessagePort,
 } = internalBinding('worker');
 
+const { processConnectionRequest } = require('internal/worker');
 const workerIo = require('internal/worker/io');
 const {
   messageTypes: {
@@ -40,6 +41,7 @@ const {
     // Messages that may be either received or posted
     STDIO_PAYLOAD,
     STDIO_WANTS_MORE_DATA,
+    CONNECT,
   },
   kStdioWantsMoreDataCallback,
 } = workerIo;
@@ -182,6 +184,8 @@ port.on('message', (message) => {
         break;
       }
     }
+  } else if (message.type === CONNECT) {
+    processConnectionRequest(message);
   } else if (message.type === STDIO_PAYLOAD) {
     const { stream, chunks } = message;
     ArrayPrototypeForEach(chunks, ({ chunk, encoding }) => {
diff --git a/lib/internal/worker.js b/lib/internal/worker.js
index 9eefaa97021756..2f55f5eb8d1bd3 100644
--- a/lib/internal/worker.js
+++ b/lib/internal/worker.js
@@ -5,8 +5,12 @@ const {
   ArrayPrototypeMap,
   ArrayPrototypePush,
   AtomicsAdd,
+  AtomicsNotify,
+  AtomicsStore,
+  AtomicsWaitAsync,
   Float64Array,
   FunctionPrototypeBind,
+  Int32Array,
   JSONStringify,
   MathMax,
   ObjectEntries,
@@ -34,12 +38,14 @@ const {
 
 const errorCodes = require('internal/errors').codes;
 const {
+  ERR_INVALID_ARG_TYPE,
+  ERR_INVALID_ARG_VALUE,
+  ERR_WORKER_CONNECTION_REFUSED,
+  ERR_WORKER_INVALID_EXEC_ARGV,
   ERR_WORKER_NOT_RUNNING,
   ERR_WORKER_PATH,
+  ERR_WORKER_SAME_THREAD,
   ERR_WORKER_UNSERIALIZABLE_ERROR,
-  ERR_WORKER_INVALID_EXEC_ARGV,
-  ERR_INVALID_ARG_TYPE,
-  ERR_INVALID_ARG_VALUE,
 } = errorCodes;
 
 const workerIo = require('internal/worker/io');
@@ -59,7 +65,12 @@ const {
 const { deserializeError } = require('internal/error_serdes');
 const { fileURLToPath, isURL, pathToFileURL } = require('internal/url');
 const { kEmptyObject } = require('internal/util');
-const { validateArray, validateString } = require('internal/validators');
+const {
+  validateArray,
+  validateFunction,
+  validateNumber,
+  validateString,
+} = require('internal/validators');
 const {
   throwIfBuildingSnapshot,
 } = require('internal/v8/startup_snapshot');
@@ -74,6 +85,8 @@ const {
   kCodeRangeSizeMb,
   kStackSizeMb,
   kTotalResourceLimitCount,
+  sendToWorker,
+  setMainPort,
 } = internalBinding('worker');
 
 const kHandle = Symbol('kHandle');
@@ -100,6 +113,14 @@ let cwdCounter;
 
 const environmentData = new SafeMap();
 
+// SharedArrayBuffer must always be Int32, so it's * 4.
+// We need one for the operation status (performing / performed) and one for the result (success / failure).
+const WORKER_MESSAGING_SHARED_DATA = 2 * 4;
+const WORKER_MESSAGING_STATUS_INDEX = 0;
+const WORKER_MESSAGING_RESULT_INDEX = 1;
+let connectionsListener = null;
+let mainPortWasSetup = false;
+
 // SharedArrayBuffers can be disabled with --enable-sharedarraybuffer-per-context.
 if (isMainThread && SharedArrayBuffer !== undefined) {
   cwdCounter = new Uint32Array(new SharedArrayBuffer(4));
@@ -527,6 +548,97 @@ function eventLoopUtilization(util1, util2) {
   );
 }
 
+function setConnectionsListener(fn) {
+  if (isMainThread && !mainPortWasSetup) {
+    setupMainPort();
+    mainPortWasSetup = true;
+  }
+
+  if (fn == null) {
+    connectionsListener = null;
+    return;
+  }
+
+  validateFunction(fn, 'fn');
+  connectionsListener = fn;
+}
+
+async function processConnectionRequest(request) {
+  const status = new Int32Array(request.memory);
+
+  try {
+    const result = await connectionsListener?.(request.sender, request.port, request.data);
+    AtomicsStore(status, WORKER_MESSAGING_RESULT_INDEX, result === true ? 0 : 1);
+  } catch (e) {
+    debug('connections listener rejected', e);
+    AtomicsStore(status, WORKER_MESSAGING_RESULT_INDEX, 2);
+  } finally {
+    AtomicsNotify(status, WORKER_MESSAGING_STATUS_INDEX, 1);
+  }
+}
+
+async function connect(target, data, timeout) {
+  if (typeof data === 'number' && typeof timeout === 'undefined') {
+    timeout = data;
+    data = undefined;
+  }
+
+  if (typeof timeout !== 'undefined') {
+    validateNumber(timeout, 'timeout', 0);
+  }
+
+  if (target === threadId) {
+    throw new ERR_WORKER_SAME_THREAD();
+  }
+
+  // Create a shared array to exchange the status and the result
+  const memory = new SharedArrayBuffer(WORKER_MESSAGING_SHARED_DATA);
+  const status = new Int32Array(memory);
+  const promise = AtomicsWaitAsync(status, WORKER_MESSAGING_STATUS_INDEX, 0, timeout).value;
+
+  // Create the channel and send it to the other thread
+  const { port1, port2 } = new MessageChannel();
+
+  // Make sure the event loop does not exist before this is completed
+  port1.ref();
+
+  // Send the message
+  try {
+    sendToWorker(target, { type: messageTypes.CONNECT, sender: threadId, port: port2, memory, data }, [port2]);
+  } finally {
+    port1.unref();
+  }
+
+  // Wait for the response
+  const value = await promise;
+
+  if (value === 'timed-out' || status[WORKER_MESSAGING_RESULT_INDEX] !== 0) {
+    port1.close();
+    port2.close();
+    throw new ERR_WORKER_CONNECTION_REFUSED();
+  }
+
+  return port1;
+}
+
+function setupMainPort() {
+  const { port1, port2 } = new MessageChannel();
+  setMainPort(port2);
+
+  // Set message management
+  port1.on('message', (message) => {
+    if (message.type === messageTypes.CONNECT) {
+      processConnectionRequest(message);
+    } else {
+      assert(message.type === messageTypes.CONNECT, `Unknown worker message type ${message.type}`);
+    }
+  });
+
+  // Never block the process on this channel
+  port1.unref();
+  port2.unref();
+}
+
 module.exports = {
   ownsProcessState,
   kIsOnline,
@@ -537,6 +649,9 @@ module.exports = {
   setEnvironmentData,
   getEnvironmentData,
   assignEnvironmentData,
+  setConnectionsListener,
+  processConnectionRequest,
+  connect,
   threadId,
   InternalWorker,
   Worker,
diff --git a/lib/internal/worker/io.js b/lib/internal/worker/io.js
index 3263ddb88194fc..f56312ab992c16 100644
--- a/lib/internal/worker/io.js
+++ b/lib/internal/worker/io.js
@@ -83,6 +83,7 @@ const messageTypes = {
   STDIO_PAYLOAD: 'stdioPayload',
   STDIO_WANTS_MORE_DATA: 'stdioWantsMoreData',
   LOAD_SCRIPT: 'loadScript',
+  CONNECT: 'connect',
 };
 
 // createFastMessageEvent skips webidl argument validation when the arguments
diff --git a/lib/worker_threads.js b/lib/worker_threads.js
index 084fa98908dc57..4db211f57a0419 100644
--- a/lib/worker_threads.js
+++ b/lib/worker_threads.js
@@ -6,8 +6,10 @@ const {
   resourceLimits,
   setEnvironmentData,
   getEnvironmentData,
+  connect,
   threadId,
   Worker,
+  setConnectionsListener,
 } = require('internal/worker');
 
 const {
@@ -40,4 +42,6 @@ module.exports = {
   BroadcastChannel,
   setEnvironmentData,
   getEnvironmentData,
+  setConnectionsListener,
+  connect,
 };
diff --git a/src/node_errors.h b/src/node_errors.h
index 0a74373cf5d333..ef33cea7dbf531 100644
--- a/src/node_errors.h
+++ b/src/node_errors.h
@@ -102,6 +102,7 @@ void OOMErrorHandler(const char* location, const v8::OOMDetails& details);
   V(ERR_VM_MODULE_LINK_FAILURE, Error)                                         \
   V(ERR_WASI_NOT_STARTED, Error)                                               \
   V(ERR_WORKER_INIT_FAILED, Error)                                             \
+  V(ERR_WORKER_INVALID_ID, Error)                                              \
   V(ERR_PROTO_ACCESS, Error)
 
 #define V(code, type)                                                          \
diff --git a/src/node_messaging.h b/src/node_messaging.h
index 6dd34b8bd5a407..a4d6b914e77a3a 100644
--- a/src/node_messaging.h
+++ b/src/node_messaging.h
@@ -19,6 +19,12 @@ class MessagePort;
 
 typedef MaybeStackBuffer<v8::Local<v8::Value>, 8> TransferList;
 
+// Get a transfer list out of an array of Javascript objects.
+bool GetTransferList(Environment* env,
+                     v8::Local<v8::Context> context,
+                     v8::Local<v8::Value> transfer_list_v,
+                     TransferList* transfer_list_out);
+
 // Used to represent the in-flight structure of an object that is being
 // transferred or cloned using postMessage().
 class TransferData : public MemoryRetainer {
@@ -288,7 +294,7 @@ class MessagePort : public HandleWrap {
   // alone is often not enough, since the backing C++ MessagePort object may
   // have been deleted already. For all intents and purposes, an object with a
   // NULL pointer to the C++ MessagePort object is also detached.
-  inline bool IsDetached() const;
+  bool IsDetached() const;
 
   BaseObject::TransferMode GetTransferMode() const override;
   std::unique_ptr<TransferData> TransferForMessaging() override;
diff --git a/src/node_worker.cc b/src/node_worker.cc
index bc43cb42934ba9..d2d9a98a8e2e48 100644
--- a/src/node_worker.cc
+++ b/src/node_worker.cc
@@ -6,6 +6,7 @@
 #include "node_buffer.h"
 #include "node_errors.h"
 #include "node_external_reference.h"
+#include "node_messaging.h"
 #include "node_options-inl.h"
 #include "node_perf.h"
 #include "node_snapshot_builder.h"
@@ -46,6 +47,8 @@ namespace node {
 namespace worker {
 
 constexpr double kMB = 1024 * 1024;
+std::unordered_map<uint64_t, MessagePort*> message_ports;
+Mutex Worker::messagePortsMutex;
 
 Worker::Worker(Environment* env,
                Local<Object> wrap,
@@ -63,22 +66,21 @@ Worker::Worker(Environment* env,
       name_(name),
       env_vars_(env_vars),
       embedder_preload_(env->embedder_preload()),
+      parent_port_(MessagePort::New(env, env->context())),
       snapshot_data_(snapshot_data) {
   Debug(this, "Creating new worker instance with thread id %llu",
         thread_id_.id);
 
-  // Set up everything that needs to be set up in the parent environment.
-  MessagePort* parent_port = MessagePort::New(env, env->context());
-  if (parent_port == nullptr) {
+  if (parent_port_ == nullptr) {
     // This can happen e.g. because execution is terminating.
     return;
   }
 
   child_port_data_ = std::make_unique<MessagePortData>(nullptr);
-  MessagePort::Entangle(parent_port, child_port_data_.get());
+  MessagePort::Entangle(parent_port_, child_port_data_.get());
 
   object()
-      ->Set(env->context(), env->message_port_string(), parent_port->object())
+      ->Set(env->context(), env->message_port_string(), parent_port_->object())
       .Check();
 
   object()->Set(env->context(),
@@ -660,7 +662,6 @@ void Worker::StartThread(const FunctionCallbackInfo<Value>& args) {
   Worker* w;
   ASSIGN_OR_RETURN_UNWRAP(&w, args.This());
   Mutex::ScopedLock lock(w->mutex_);
-
   w->stopped_ = false;
 
   if (w->resource_limits_[kStackSizeMb] > 0) {
@@ -685,6 +686,14 @@ void Worker::StartThread(const FunctionCallbackInfo<Value>& args) {
     // gcc 6.3 detect undefined behaviour when there shouldn't be any.
     // gcc 7+ handles this well.
     Worker* w = static_cast<Worker*>(arg);
+
+    // Store a reference to the worker so that it can be used
+    // for inter thread communication
+    {
+      Mutex::ScopedLock message_ports_lock(messagePortsMutex);
+      message_ports[w->thread_id_.id] = w->parent_port_;
+    }
+
     const uintptr_t stack_top = reinterpret_cast<uintptr_t>(&arg);
 
     // Leave a few kilobytes just to make sure we're within limits and have
@@ -775,6 +784,13 @@ void Worker::Exit(ExitCode code,
                   const char* error_code,
                   const char* error_message) {
   Mutex::ScopedLock lock(mutex_);
+
+  // Remove reference to the port used for inter-thread communication
+  {
+    Mutex::ScopedLock message_ports_lock(messagePortsMutex);
+    message_ports.erase(thread_id_.id);
+  }
+
   Debug(this,
         "Worker %llu called Exit(%d, %s, %s)",
         thread_id_.id,
@@ -918,6 +934,65 @@ void GetEnvMessagePort(const FunctionCallbackInfo<Value>& args) {
   }
 }
 
+void SetMainPort(const FunctionCallbackInfo<Value>& args) {
+  Mutex::ScopedLock lock(Worker::messagePortsMutex);
+
+  if (args.Length() == 0 || args[0]->IsNullOrUndefined()) {
+    message_ports.erase(0);
+    return;
+  }
+
+  MessagePort* port;
+  ASSIGN_OR_RETURN_UNWRAP(&port, args[0]);
+
+  message_ports[0] = port;
+}
+
+void SendToWorker(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  Local<Context> context = env->context();
+  v8::Isolate* isolate = args.GetIsolate();
+
+  // args[0] is the target worker thread id
+  // args[1] is the message to send via PostMessage
+  // args[2] is the transfer list to send via PostMessage, it is optional
+  CHECK_GT(args.Length(), 1);
+
+  CHECK(args[0]->IsUint32());
+  auto tid = args[0].As<v8::Uint32>()->Value();
+
+  MessagePort* port = nullptr;
+  {
+    Mutex::ScopedLock lock(Worker::messagePortsMutex);
+
+    auto it = message_ports.find(tid);
+    if (it == message_ports.end()) {
+      std::string message = SPrintF("Invalid worker id %d", tid);
+      THROW_ERR_WORKER_INVALID_ID(isolate, message.c_str());
+      return;
+    }
+
+    port = it->second;
+  }
+
+  TransferList transfer_list;
+  // Not all messages require a transfer_list
+  if (args.Length() > 2) {
+    if (!GetTransferList(env, context, args[2], &transfer_list)) {
+      THROW_ERR_INVALID_ARG_TYPE(isolate, "Invalid transfer list");
+      return;
+    }
+  }
+
+  if (port == nullptr || port->IsDetached()) {
+    std::string message = SPrintF("Invalid worker id %d", tid);
+    THROW_ERR_WORKER_INVALID_ID(isolate, message.c_str());
+    return;
+  }
+
+  port->PostMessage(env, context, args[1], transfer_list);
+}
+
 void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
                                       Local<ObjectTemplate> target) {
   Isolate* isolate = isolate_data->isolate();
@@ -957,6 +1032,8 @@ void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
   }
 
   SetMethod(isolate, target, "getEnvMessagePort", GetEnvMessagePort);
+  SetMethod(isolate, target, "sendToWorker", SendToWorker);
+  SetMethod(isolate, target, "setMainPort", SetMainPort);
 }
 
 void CreateWorkerPerContextProperties(Local<Object> target,
@@ -1001,6 +1078,8 @@ void CreateWorkerPerContextProperties(Local<Object> target,
 
 void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(GetEnvMessagePort);
+  registry->Register(SetMainPort);
+  registry->Register(SendToWorker);
   registry->Register(Worker::New);
   registry->Register(Worker::StartThread);
   registry->Register(Worker::StopThread);
diff --git a/src/node_worker.h b/src/node_worker.h
index 07fd7b460654e1..896bb5fe84bee8 100644
--- a/src/node_worker.h
+++ b/src/node_worker.h
@@ -77,6 +77,9 @@ class Worker : public AsyncWrap {
   static void LoopIdleTime(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void LoopStartTime(const v8::FunctionCallbackInfo<v8::Value>& args);
 
+  // this mutex is to synchronize access to message_ports calls
+  static Mutex messagePortsMutex;
+
  private:
   bool CreateEnvMessagePort(Environment* env);
   static size_t NearHeapLimit(void* data, size_t current_heap_limit,
@@ -131,6 +134,9 @@ class Worker : public AsyncWrap {
   // destroyed alongwith the worker thread.
   Environment* env_ = nullptr;
 
+  // The parent port used to communicate with the worker
+  MessagePort* parent_port_ = nullptr;
+
   const SnapshotData* snapshot_data_ = nullptr;
   friend class WorkerThreadData;
 };
diff --git a/test/parallel/test-http-multiple-headers.js b/test/parallel/test-http-multiple-headers.js
index 8f52f817273c52..627efdd90cc1cc 100644
--- a/test/parallel/test-http-multiple-headers.js
+++ b/test/parallel/test-http-multiple-headers.js
@@ -1,6 +1,5 @@
 'use strict';
 
-// TODO@PI: Run all tests
 const common = require('../common');
 const assert = require('assert');
 const { createServer, request } = require('http');
diff --git a/test/parallel/test-worker-connection-errors.js b/test/parallel/test-worker-connection-errors.js
new file mode 100644
index 00000000000000..33f3ed5e164535
--- /dev/null
+++ b/test/parallel/test-worker-connection-errors.js
@@ -0,0 +1,93 @@
+'use strict';
+
+const common = require('../common');
+const { once } = require('node:events');
+const {
+  connect,
+  setConnectionsListener,
+  workerData,
+  Worker,
+  parentPort,
+  threadId,
+} = require('node:worker_threads');
+const { rejects } = require('node:assert');
+
+const level = workerData?.level ?? 0;
+
+async function test() {
+  await rejects(common.mustCall(function() {
+    return connect(threadId);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_SAME_THREAD',
+  });
+
+  await rejects(common.mustCall(function() {
+    return connect(123);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_INVALID_ID',
+  });
+
+  // The first worker will refuse connection as there is no listener at all
+  const worker1 = new Worker(__filename, { workerData: { shouldHaveListener: false, level: level + 1 } });
+
+  // The second worker will refuse connection as the listener returns false
+  const worker2 = new Worker(__filename, { workerData: { shouldHaveListener: true, level: level + 1 } });
+
+  // The second worker will refuse connection as the listener throws
+  const worker3 = new Worker(
+    __filename,
+    { workerData: { shouldHaveListener: true, shouldListenerFail: true, level: level + 1 } },
+  );
+
+  await Promise.all([
+    once(worker1, 'message'),
+    once(worker2, 'message'),
+    once(worker3, 'message'),
+  ]);
+
+  await rejects(common.mustCall(function() {
+    return connect(worker1.threadId);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_CONNECTION_REFUSED',
+  });
+
+  await rejects(common.mustCall(function() {
+    return connect(worker2.threadId);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_CONNECTION_REFUSED',
+  });
+
+  await rejects(common.mustCall(function() {
+    return connect(worker3.threadId);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_CONNECTION_REFUSED',
+  });
+
+  worker1.postMessage('success');
+  worker2.postMessage('success');
+  worker3.postMessage('success');
+}
+
+if (level === 0) {
+  test();
+} else {
+  if (workerData.shouldHaveListener) {
+    setConnectionsListener(common.mustCall(function() {
+      if (workerData.shouldListenerFail) {
+        throw new Error('KABOOM');
+      }
+
+      return false;
+    }));
+  }
+
+  parentPort.postMessage('ready');
+
+  // Keep the workers alive for the time being
+  parentPort.once('message', common.mustCall());
+}
diff --git a/test/parallel/test-worker-connection-timeout.js b/test/parallel/test-worker-connection-timeout.js
new file mode 100644
index 00000000000000..d161a00aadb215
--- /dev/null
+++ b/test/parallel/test-worker-connection-timeout.js
@@ -0,0 +1,50 @@
+'use strict';
+
+const common = require('../common');
+const { once } = require('node:events');
+const {
+  connect,
+  setConnectionsListener,
+  workerData,
+  Worker,
+  parentPort,
+} = require('node:worker_threads');
+const { rejects } = require('node:assert');
+
+const level = workerData?.level ?? 0;
+
+async function test() {
+  const worker = new Worker(__filename, { workerData: { shouldHaveListener: false, level: level + 1 } });
+  await once(worker, 'message');
+
+  await rejects(common.mustCall(function() {
+    return connect(worker.threadId, 1000);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_CONNECTION_REFUSED',
+  });
+
+  await rejects(common.mustCall(function() {
+    return connect(worker.threadId, { foo: 'bar' }, 1000);
+  }), {
+    name: 'Error',
+    code: 'ERR_WORKER_CONNECTION_REFUSED',
+  });
+
+  worker.postMessage('success');
+}
+
+if (level === 0) {
+  test();
+} else {
+  setConnectionsListener(common.mustCall(function() {
+    return new Promise(common.mustCall(function(resolve) {
+      setTimeout(resolve, common.platformTimeout(2000));
+    }));
+  }, 2));
+
+  parentPort.postMessage('ready');
+
+  // Keep the workers alive for the time being
+  parentPort.once('message', common.mustCall());
+}
diff --git a/test/parallel/test-worker-connection.js b/test/parallel/test-worker-connection.js
new file mode 100644
index 00000000000000..fb6e4659abfe84
--- /dev/null
+++ b/test/parallel/test-worker-connection.js
@@ -0,0 +1,113 @@
+'use strict';
+
+const common = require('../common');
+const Countdown = require('../common/countdown');
+const {
+  connect,
+  setConnectionsListener,
+  workerData,
+  Worker,
+  threadId,
+  parentPort,
+} = require('node:worker_threads');
+const { strictEqual, deepStrictEqual } = require('node:assert');
+const { once } = require('node:events');
+
+// Spawn threads on three levels: 1 main thread, two children, four grand childrens. 7 threads total, max id = 6
+const MAX_LEVEL = 2;
+const MAX_THREAD = 6;
+
+// This is to allow the test to run in --worker mode
+const mainThread = workerData?.mainThread ?? threadId;
+const level = workerData?.level ?? 0;
+
+const channel = new BroadcastChannel('nodejs:test-worker-connection');
+let completed;
+
+if (level === 0) {
+  completed = new Countdown(MAX_THREAD + 1, () => {
+    channel.postMessage('exit');
+    channel.close();
+  });
+}
+
+async function createChildren() {
+  const worker = new Worker(__filename, { workerData: { mainThread, level: level + 1 } });
+  await once(worker, 'message');
+}
+
+async function ping() {
+  let target;
+  do {
+    target = mainThread + Math.floor(Math.random() * MAX_THREAD);
+  } while (target === threadId);
+  const port = await connect(target, { level });
+
+  port.on('message', common.mustCall(function(message) {
+    deepStrictEqual(message, { message: 'pong', source: target, destination: threadId });
+    port.close();
+
+    if (level === 0) {
+      completed.dec();
+    } else {
+      channel.postMessage('end');
+    }
+  }));
+
+  port.postMessage({ message: 'ping', source: threadId, destination: target });
+}
+
+// Do not use mustCall here as the thread might not receive any connection request
+setConnectionsListener(function(source, port, data) {
+  // Let's verify the source hierarchy
+  // Given we do depth first, the level is 1 for thread 1 and 4, 2 for other threads
+  if (source !== mainThread) {
+    const currentThread = source - mainThread;
+    strictEqual(data.level, (currentThread === 1 || currentThread === 4) ? 1 : 2);
+  } else {
+    strictEqual(data.level, 0);
+  }
+
+  // Verify communication
+  port.on('message', common.mustCall(function(message) {
+    deepStrictEqual(message, { message: 'ping', source, destination: threadId });
+    port.postMessage({ message: 'pong', source: threadId, destination: source });
+    port.close();
+  }));
+
+  return true;
+});
+
+async function test() {
+  if (level < MAX_LEVEL) {
+    await createChildren();
+    await createChildren();
+  }
+
+  channel.onmessage = function(message) {
+    switch (message.data) {
+      case 'start':
+        ping();
+        break;
+      case 'end':
+        if (level === 0) {
+          completed.dec();
+        }
+        break;
+      case 'exit':
+        channel.close();
+        break;
+    }
+  };
+
+  if (level > 0) {
+    const currentThread = threadId - mainThread;
+    strictEqual(level, (currentThread === 1 || currentThread === 4) ? 1 : 2);
+    parentPort.postMessage({ type: 'ready', threadId });
+  } else {
+    channel.postMessage('start');
+    ping();
+  }
+}
+
+test();