diff --git a/beacon/engine/types.go b/beacon/engine/types.go
index 5c94e67de1..71ee44d877 100644
--- a/beacon/engine/types.go
+++ b/beacon/engine/types.go
@@ -157,6 +157,11 @@ type BlobAndProofV2 struct {
 	CellProofs []hexutil.Bytes `json:"proofs"` // proofs MUST contain exactly CELLS_PER_EXT_BLOB cell proofs.
 }
 
+type BlobCellsAndProofsV1 struct {
+	BlobCells []hexutil.Bytes `json:"blob_cells"`
+	Proofs    []hexutil.Bytes `json:"proofs"`
+}
+
 // JSON type overrides for ExecutionPayloadEnvelope.
 type executionPayloadEnvelopeMarshaling struct {
 	BlockValue *hexutil.Big
diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go
index 7328eba460..1da727be71 100644
--- a/core/txpool/blobpool/blobpool.go
+++ b/core/txpool/blobpool/blobpool.go
@@ -1750,7 +1750,85 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash, version byte) ([]*kzg4844.Blo
 	return blobs, commitments, proofs, nil
 }
 
-// AvailableBlobs returns the number of blobs that are available in the subpool.
+// GetBlobCells returns cells for the given versioned blob hashes,
+// filtered by the requested cell indices(mask).
+// Each entry in the result corresponds to one vhash. Nil entries mean the blob
+// was not available.
+func (p *BlobPool) GetBlobCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]*kzg4844.Cell, [][]*kzg4844.Proof, error) {
+	var (
+		cells  = make([][]*kzg4844.Cell, len(vhashes))
+		proofs = make([][]*kzg4844.Proof, len(vhashes))
+		vindex = make(map[common.Hash][]int) // Indices of versioned hashes in the request
+		filled = make(map[common.Hash]struct{})
+	)
+	for i, h := range vhashes {
+		vindex[h] = append(vindex[h], i)
+	}
+	requestedIndices := mask.Indices()
+
+	for _, vhash := range vhashes {
+		if _, ok := filled[vhash]; ok {
+			continue
+		}
+		p.lock.RLock()
+		txID, exists := p.lookup.storeidOfBlob(vhash)
+		p.lock.RUnlock()
+		if !exists {
+			continue
+		}
+		data, err := p.store.Get(txID)
+		if err != nil {
+			continue
+		}
+		var pooledTx pooledBlobTx
+		if err := rlp.DecodeBytes(data, &pooledTx); err != nil {
+			continue
+		}
+		sidecar := pooledTx.Sidecar
+		if sidecar == nil {
+			continue
+		}
+		tx := pooledTx.Transaction
+		cellsPerBlob := sidecar.Custody.OneCount()
+		storedIndices := sidecar.Custody.Indices()
+
+		for blobIdx, hash := range tx.BlobHashes() {
+			indices, ok := vindex[hash]
+			if !ok {
+				continue
+			}
+			filled[hash] = struct{}{}
+
+			blobCells := make([]*kzg4844.Cell, len(requestedIndices))
+			blobProofs := make([]*kzg4844.Proof, len(requestedIndices))
+
+			for i, cellIdx := range requestedIndices {
+				pos := -1
+				for k, storedIdx := range storedIndices {
+					if storedIdx == cellIdx {
+						pos = k
+						break
+					}
+				}
+				if pos >= 0 {
+					cell := sidecar.Cells[blobIdx*cellsPerBlob+pos]
+					blobCells[i] = &cell
+					proofIdx := blobIdx*kzg4844.CellProofsPerBlob + int(cellIdx)
+					if proofIdx < len(sidecar.Proofs) {
+						proof := sidecar.Proofs[proofIdx]
+						blobProofs[i] = &proof
+					}
+				}
+			}
+			for _, idx := range indices {
+				cells[idx] = blobCells
+				proofs[idx] = blobProofs
+			}
+		}
+	}
+	return cells, proofs, nil
+}
+
 func (p *BlobPool) AvailableBlobs(vhashes []common.Hash) int {
 	available := 0
 	for _, vhash := range vhashes {
diff --git a/eth/catalyst/api.go b/eth/catalyst/api.go
index d71edeac6b..6aaa287eac 100644
--- a/eth/catalyst/api.go
+++ b/eth/catalyst/api.go
@@ -660,6 +660,66 @@ func (api *ConsensusAPI) getBlobs(hashes []common.Hash, v2 bool) ([]*engine.Blob
 	return res, nil
 }
 
+// GetBlobsV4 returns cell-level blob data from the transaction pool.
+// V4 returns only the requested cells as specified by the indices_bitarray.
+func (api *ConsensusAPI) GetBlobsV4(hashes []common.Hash, indicesBitarray hexutil.Bytes) ([]*engine.BlobCellsAndProofsV1, error) {
+	head := api.eth.BlockChain().CurrentHeader()
+	// Sparse blobpool is not necessarily coupled with the Amsterdam fork and
+	// can technically be supported after the Osaka fork
+	// (where cell proofs are introduced).
+	if api.config().LatestFork(head.Time) < forks.Osaka {
+		return nil, nil
+	}
+	if len(hashes) > 128 {
+		return nil, engine.TooLargeRequest.With(fmt.Errorf("requested blob count too large: %v", len(hashes)))
+	}
+	if len(indicesBitarray) != 16 {
+		return nil, engine.InvalidParams.With(fmt.Errorf("indices_bitarray must be 16 bytes, got %d", len(indicesBitarray)))
+	}
+	var mask types.CustodyBitmap
+	copy(mask[:], indicesBitarray)
+	cells, proofs, err := api.eth.BlobTxPool().GetBlobCells(hashes, mask)
+	if err != nil {
+		return nil, engine.InvalidParams.With(err)
+	}
+	var (
+		res      = make([]*engine.BlobCellsAndProofsV1, len(hashes))
+		hitCount int
+	)
+	getBlobsRequestedCounter.Inc(int64(len(hashes)))
+	for i := range hashes {
+		if cells[i] == nil || proofs[i] == nil {
+			continue
+		}
+		hitCount++
+		blobCells := make([]hexutil.Bytes, len(cells[i]))
+		for j, cell := range cells[i] {
+			if cell != nil {
+				blobCells[j] = cell[:]
+			}
+		}
+		blobProofs := make([]hexutil.Bytes, len(proofs[i]))
+		for j, proof := range proofs[i] {
+			if proof != nil {
+				blobProofs[j] = proof[:]
+			}
+		}
+		res[i] = &engine.BlobCellsAndProofsV1{
+			BlobCells: blobCells,
+			Proofs:    blobProofs,
+		}
+	}
+	getBlobsAvailableCounter.Inc(int64(hitCount))
+	if hitCount == len(hashes) {
+		getBlobsRequestCompleteHit.Inc(1)
+	} else if hitCount > 0 {
+		getBlobsRequestPartialHit.Inc(1)
+	} else {
+		getBlobsRequestMiss.Inc(1)
+	}
+	return res, nil
+}
+
 // Helper for NewPayload* methods.
 var invalidStatus = engine.PayloadStatusV1{Status: engine.INVALID}