diff --git a/lib/mock/mock-utils.js b/lib/mock/mock-utils.js
index 5b6b5bd79cf..9e8c4a15f79 100644
--- a/lib/mock/mock-utils.js
+++ b/lib/mock/mock-utils.js
@@ -18,6 +18,8 @@ const {
 } = require('node:util')
 const { InvalidArgumentError } = require('../core/errors')
 
+const requestAborted = Symbol('request aborted')
+
 function matchValue (match, value) {
   if (typeof match === 'string') {
     return match === value
@@ -295,13 +297,9 @@ function mockDispatch (opts, handler) {
 
   mockDispatch.timesInvoked++
 
-  // Here's where we resolve a callback if a callback is present for the dispatch data.
-  if (mockDispatch.data.callback) {
-    mockDispatch.data = { ...mockDispatch.data, ...mockDispatch.data.callback(opts) }
-  }
-
   // Parse mockDispatch data
-  const { data: { statusCode, data, headers, trailers, error }, delay, persist } = mockDispatch
+  const { data: response, delay, persist } = mockDispatch
+  const { error } = response
   const { timesInvoked, times } = mockDispatch
 
   // If it's used up and not persistent, mark as consumed
@@ -346,32 +344,73 @@ function mockDispatch (opts, handler) {
     }
   }
 
+  let replyOpts = opts
+  const dispatches = this[kDispatches]
+
   // Call onRequestStart to allow the handler to receive the controller
   handler.onRequestStart?.(controller, null)
+  if (aborted) {
+    return true
+  }
 
-  // Handle the request with a delay if necessary
-  if (typeof delay === 'number' && delay > 0) {
-    timer = setTimeout(() => {
-      timer = null
-      handleReply(this[kDispatches])
-    }, delay)
-  } else {
-    handleReply(this[kDispatches])
+  const requestBody = dispatchRequestBody(opts.body, handler, controller, () => aborted)
+
+  if (isPromise(requestBody)) {
+    requestBody.then((body) => {
+      if (body === requestAborted) {
+        return
+      }
+      if (body !== opts.body) {
+        replyOpts = { ...opts, body }
+      }
+      sendReply()
+    }, (error) => controller.abort(error))
+    return true
+  }
+
+  if (requestBody === requestAborted) {
+    return true
+  }
+
+  if (requestBody !== opts.body) {
+    replyOpts = { ...opts, body: requestBody }
   }
 
-  function handleReply (mockDispatches, _data = data) {
+  sendReply()
+
+  function sendReply () {
+    // Handle the request with a delay if necessary
+    if (typeof delay === 'number' && delay > 0) {
+      timer = setTimeout(() => {
+        timer = null
+        handleReply(dispatches)
+      }, delay)
+    } else {
+      handleReply(dispatches)
+    }
+  }
+
+  function handleReply (mockDispatches, _response = response) {
     // Don't send response if the request was aborted
     if (aborted) {
       return
     }
 
+    if (_response.callback) {
+      const { callback, ...responseDefaults } = _response
+      mockDispatch.data = { ...responseDefaults, ...callback(replyOpts) }
+      return handleReply(mockDispatches, mockDispatch.data)
+    }
+
+    const { statusCode, data, headers, trailers } = _response
+
     // fetch's HeadersList is a 1D string array
     const optsHeaders = Array.isArray(opts.headers)
       ? buildHeadersFromArray(opts.headers)
       : opts.headers
-    const body = typeof _data === 'function'
-      ? _data({ ...opts, headers: optsHeaders })
-      : _data
+    const body = typeof data === 'function'
+      ? data({ ...replyOpts, headers: optsHeaders })
+      : data
 
     // util.types.isPromise is likely needed for jest.
     if (isPromise(body)) {
@@ -380,7 +419,7 @@ function mockDispatch (opts, handler) {
       // synchronously throw the error, which breaks some tests.
       // Rather, we wait for the callback to resolve if it is a
       // promise, and then re-run handleReply with the new body.
-      return body.then((newData) => handleReply(mockDispatches, newData))
+      return body.then((newData) => handleReply(mockDispatches, { ..._response, data: newData }))
     }
 
     // Check again if aborted after async body resolution
@@ -405,6 +444,96 @@ function mockDispatch (opts, handler) {
   return true
 }
 
+function dispatchRequestBody (body, handler, controller, isAborted) {
+  if (typeof handler.onBodySent !== 'function' && typeof handler.onRequestSent !== 'function') {
+    return body
+  }
+
+  if (body == null) {
+    return callOnRequestSent(handler, controller, isAborted) ? body : requestAborted
+  }
+
+  if (body && typeof body[Symbol.asyncIterator] === 'function') {
+    return dispatchAsyncIterableBody(body, handler, controller, isAborted)
+  }
+
+  if (isIterableBody(body)) {
+    const chunks = []
+
+    for (const chunk of body) {
+      if (isAborted()) {
+        return requestAborted
+      }
+      chunks.push(chunk)
+      if (!callOnBodySent(handler, controller, chunk) || isAborted()) {
+        return requestAborted
+      }
+    }
+
+    return callOnRequestSent(handler, controller, isAborted) ? chunks : requestAborted
+  }
+
+  if (isAborted()) {
+    return requestAborted
+  }
+  if (!callOnBodySent(handler, controller, body)) {
+    return requestAborted
+  }
+
+  return callOnRequestSent(handler, controller, isAborted) ? body : requestAborted
+}
+
+async function dispatchAsyncIterableBody (body, handler, controller, isAborted) {
+  const chunks = []
+
+  for await (const chunk of body) {
+    if (isAborted()) {
+      return requestAborted
+    }
+    chunks.push(chunk)
+    if (!callOnBodySent(handler, controller, chunk) || isAborted()) {
+      return requestAborted
+    }
+  }
+
+  if (!callOnRequestSent(handler, controller, isAborted)) {
+    return requestAborted
+  }
+
+  return {
+    async * [Symbol.asyncIterator] () {
+      yield * chunks
+    }
+  }
+}
+
+function callOnBodySent (handler, controller, chunk) {
+  try {
+    handler.onBodySent?.(chunk)
+    return true
+  } catch (error) {
+    controller.abort(error)
+    return false
+  }
+}
+
+function callOnRequestSent (handler, controller, isAborted) {
+  try {
+    handler.onRequestSent?.()
+    return !isAborted()
+  } catch (error) {
+    controller.abort(error)
+    return false
+  }
+}
+
+function isIterableBody (body) {
+  return typeof body !== 'string' &&
+    !Buffer.isBuffer(body) &&
+    !ArrayBuffer.isView(body) &&
+    typeof body[Symbol.iterator] === 'function'
+}
+
 function buildMockDispatch () {
   const agent = this[kMockAgent]
   const origin = this[kOrigin]
diff --git a/test/mock-agent.js b/test/mock-agent.js
index 500a8833116..4beac0c4c94 100644
--- a/test/mock-agent.js
+++ b/test/mock-agent.js
@@ -229,6 +229,594 @@ describe('MockAgent - dispatch', () => {
       onResponseError () {}
     }))
   })
+
+  test('should call request body lifecycle hooks', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: 'hello'
+      }, {
+        onRequestStart () {
+          events.push('start')
+        },
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData () {},
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['start', 'body:hello', 'sent'])
+  })
+
+  test('should call request sent lifecycle hook for null bodies', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+    const chunks = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    await new Promise((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        reject(new Error('mock response did not complete'))
+      }, 1000)
+
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: null
+      }, {
+        onRequestStart () {
+          events.push('start')
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData (_controller, chunk) {
+          chunks.push(chunk)
+        },
+        onResponseEnd () {
+          clearTimeout(timeout)
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          clearTimeout(timeout)
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['start', 'sent'])
+    t.assert.strictEqual(Buffer.concat(chunks).toString(), 'hello')
+  })
+
+  test('should call request body lifecycle hooks for async iterable bodies', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    async function * body () {
+      yield Buffer.from('he')
+      yield Buffer.from('llo')
+    }
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: body()
+      }, {
+        onRequestStart () {
+          events.push('start')
+        },
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData () {},
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['start', 'body:he', 'body:llo', 'sent'])
+  })
+
+  test('should call request sent hook for requests without a body', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'GET'
+    }).reply(200, 'hello')
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'GET'
+      }, {
+        onRequestStart () {
+          events.push('start')
+        },
+        onBodySent () {
+          events.push('body')
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData () {},
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['start', 'sent'])
+  })
+
+  test('should report async iterable request body errors', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const expected = new Error('fail')
+    const events = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    async function * body () {
+      yield Buffer.from('he')
+      throw expected
+    }
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: body()
+      }, {
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          reject(new Error('request should not be marked sent'))
+        },
+        onResponseStart () {
+          reject(new Error('response should not start'))
+        },
+        onResponseData () {},
+        onResponseEnd () {},
+        onResponseError (_controller, error) {
+          try {
+            t.assert.strictEqual(error, expected)
+            resolve()
+          } catch (assertionError) {
+            reject(assertionError)
+          }
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['body:he'])
+  })
+
+  test('should pass replayed async iterable bodies to reply option callbacks', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+    let callbackBody
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(({ body }) => {
+      events.push('callback')
+      callbackBody = body
+      return { statusCode: 200, data: 'hello' }
+    })
+
+    async function * body () {
+      yield Buffer.from('he')
+      yield Buffer.from('llo')
+    }
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: body()
+      }, {
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData () {},
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    const chunks = []
+    for await (const chunk of callbackBody) {
+      chunks.push(chunk)
+    }
+
+    t.assert.deepStrictEqual(events, ['body:he', 'body:llo', 'sent', 'callback'])
+    t.assert.strictEqual(Buffer.concat(chunks).toString(), 'hello')
+  })
+
+  test('should match request bodies when lifecycle hooks are present', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST',
+      body: /hello/
+    }).reply(200, 'matched')
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: 'hello=there'
+      }, {
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData (_controller, chunk) {
+          events.push(`response:${chunk.toString()}`)
+        },
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['body:hello=there', 'sent', 'response:matched'])
+  })
+
+  test('should replay async iterable request bodies to reply callbacks after lifecycle hooks', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const events = []
+    const response = []
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, async ({ body }) => {
+      const chunks = []
+
+      for await (const chunk of body) {
+        chunks.push(chunk)
+      }
+
+      return Buffer.concat(chunks).toString()
+    })
+
+    async function * body () {
+      yield Buffer.from('he')
+      yield Buffer.from('llo')
+    }
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: body()
+      }, {
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+        },
+        onResponseStart () {},
+        onResponseData (_controller, chunk) {
+          response.push(chunk)
+        },
+        onResponseEnd () {
+          resolve()
+        },
+        onResponseError (_controller, error) {
+          reject(error)
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['body:he', 'body:llo', 'sent'])
+    t.assert.strictEqual(Buffer.concat(response).toString(), 'hello')
+  })
+
+  test('should report request body lifecycle hook errors', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const expected = new Error('fail')
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: 'hello'
+      }, {
+        onBodySent () {
+          throw expected
+        },
+        onRequestSent () {
+          reject(new Error('request should not be marked sent'))
+        },
+        onResponseStart () {
+          reject(new Error('response should not start'))
+        },
+        onResponseData () {},
+        onResponseEnd () {},
+        onResponseError (_controller, error) {
+          try {
+            t.assert.strictEqual(error, expected)
+            resolve()
+          } catch (assertionError) {
+            reject(assertionError)
+          }
+        }
+      })
+    })
+  })
+
+  test('should not send delayed replies after request sent aborts', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const expected = new Error('fail')
+    const events = []
+    let requestController
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello').delay(20)
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: 'hello'
+      }, {
+        onRequestStart (controller) {
+          requestController = controller
+        },
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+        },
+        onRequestSent () {
+          events.push('sent')
+          requestController.abort(expected)
+        },
+        onResponseStart () {
+          reject(new Error('response should not start'))
+        },
+        onResponseData () {},
+        onResponseEnd () {},
+        onResponseError (_controller, error) {
+          try {
+            t.assert.strictEqual(error, expected)
+            setTimeout(resolve, 40)
+          } catch (assertionError) {
+            reject(assertionError)
+          }
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['body:hello', 'sent'])
+  })
+
+  test('should stop reading async iterable request bodies after body hook aborts', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const expected = new Error('fail')
+    const events = []
+    let requestController
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    async function * body () {
+      events.push('pull:he')
+      yield Buffer.from('he')
+      events.push('pull:llo')
+      yield Buffer.from('llo')
+    }
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: body()
+      }, {
+        onRequestStart (controller) {
+          requestController = controller
+        },
+        onBodySent (chunk) {
+          events.push(`body:${chunk.toString()}`)
+          requestController.abort(expected)
+        },
+        onRequestSent () {
+          reject(new Error('request should not be marked sent'))
+        },
+        onResponseStart () {
+          reject(new Error('response should not start'))
+        },
+        onResponseData () {},
+        onResponseEnd () {},
+        onResponseError (_controller, error) {
+          try {
+            t.assert.strictEqual(error, expected)
+            resolve()
+          } catch (assertionError) {
+            reject(assertionError)
+          }
+        }
+      })
+    })
+
+    t.assert.deepStrictEqual(events, ['pull:he', 'body:he'])
+  })
+
+  test('should not send request body lifecycle hooks after request start aborts', async (t) => {
+    const baseUrl = 'http://localhost:9999'
+    const expected = new Error('fail')
+
+    const mockAgent = new MockAgent()
+    after(() => mockAgent.close())
+
+    const mockPool = mockAgent.get(baseUrl)
+    mockPool.intercept({
+      path: '/foo',
+      method: 'POST'
+    }).reply(200, 'hello')
+
+    await new Promise((resolve, reject) => {
+      mockAgent.dispatch({
+        origin: baseUrl,
+        path: '/foo',
+        method: 'POST',
+        body: 'hello'
+      }, {
+        onRequestStart (controller) {
+          controller.abort(expected)
+        },
+        onBodySent () {
+          reject(new Error('body should not be sent'))
+        },
+        onRequestSent () {
+          reject(new Error('request should not be sent'))
+        },
+        onResponseStart () {
+          reject(new Error('response should not start'))
+        },
+        onResponseData () {},
+        onResponseEnd () {},
+        onResponseError (_controller, error) {
+          try {
+            t.assert.strictEqual(error, expected)
+            resolve()
+          } catch (assertionError) {
+            reject(assertionError)
+          }
+        }
+      })
+    })
+  })
 })
 
 test('MockAgent - .close should clean up registered pools', async (t) => {