diff --git a/cio/io.go b/cio/io.go
index a9c6d2b15..1f8abf5f0 100644
--- a/cio/io.go
+++ b/cio/io.go
@@ -275,3 +275,7 @@ func Load(set *FIFOSet) (IO, error) {
 		closers: []io.Closer{set},
 	}, nil
 }
+
+func (p *pipes) closers() []io.Closer {
+	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
+}
diff --git a/cio/io_unix.go b/cio/io_unix.go
index 3ab2a30b0..eb2ada80b 100644
--- a/cio/io_unix.go
+++ b/cio/io_unix.go
@@ -152,7 +152,3 @@ func NewDirectIO(ctx context.Context, fifos *FIFOSet) (*DirectIO, error) {
 		},
 	}, err
 }
-
-func (p *pipes) closers() []io.Closer {
-	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
-}
diff --git a/cio/io_windows.go b/cio/io_windows.go
index 5208f3eaa..4e5d18231 100644
--- a/cio/io_windows.go
+++ b/cio/io_windows.go
@@ -17,6 +17,7 @@
 package cio
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -144,3 +145,22 @@ func NewDirectIO(stdin io.WriteCloser, stdout, stderr io.ReadCloser, terminal bo
 		},
 	}
 }
+
+// NewDirectIOFromFIFOSet returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIOFromFIFOSet(ctx context.Context, stdin io.WriteCloser, stdout, stderr io.ReadCloser, fifos *FIFOSet) *DirectIO {
+	_, cancel := context.WithCancel(ctx)
+	pipes := pipes{
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+	}
+	return &DirectIO{
+		pipes: pipes,
+		cio: cio{
+			config:  fifos.Config,
+			closers: append(pipes.closers(), fifos),
+			cancel:  cancel,
+		},
+	}
+}