diff --git a/api/next.txtpb b/api/next.txtpb
index f379b101..b423e686 100644
--- a/api/next.txtpb
+++ b/api/next.txtpb
@@ -669,3 +669,525 @@ file: {
     is_syntax_unspecified: false
   }
 }
+file: {
+  name: "proto/nerdbox/types/transfer/v1/datastream.proto"
+  package: "containerd.nerdbox.types.transfer.v1"
+  message_type: {
+    name: "ReadStream"
+    field: {
+      name: "stream"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "stream"
+    }
+    field: {
+      name: "media_type"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "mediaType"
+    }
+  }
+  message_type: {
+    name: "WriteStream"
+    field: {
+      name: "stream"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "stream"
+    }
+    field: {
+      name: "media_type"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "mediaType"
+    }
+  }
+  options: {
+    go_package: "github.com/containerd/nerdbox/api/types/transfer/v1;transfer"
+  }
+  source_code_info: {
+    location: {
+      span: 16
+      span: 0
+      span: 36
+      span: 1
+    }
+    location: {
+      path: 12
+      span: 16
+      span: 0
+      span: 18
+      leading_detached_comments: "\nCopyright The containerd Authors.\n\nLicensed under the Apache License, Version 2.0 (the \"License\");\nyou may not use this file except in compliance with the License.\nYou may obtain a copy of the License at\n\nhttp://www.apache.org/licenses/LICENSE-2.0\n\nUnless required by applicable law or agreed to in writing, software\ndistributed under the License is distributed on an \"AS IS\" BASIS,\nWITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\nSee the License for the specific language governing permissions and\nlimitations under the License.\n"
+    }
+    location: {
+      path: 2
+      span: 18
+      span: 0
+      span: 45
+    }
+    location: {
+      path: 8
+      span: 20
+      span: 0
+      span: 83
+    }
+    location: {
+      path: 8
+      path: 11
+      span: 20
+      span: 0
+      span: 83
+    }
+    location: {
+      path: 4
+      path: 0
+      span: 25
+      span: 0
+      span: 28
+      span: 1
+      leading_comments: " ReadStream carries data from the client to the server (import\n direction). The client sends data through the stream and the\n server reads it.\n"
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 1
+      span: 25
+      span: 8
+      span: 18
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      span: 26
+      span: 8
+      span: 26
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 5
+      span: 26
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 1
+      span: 26
+      span: 15
+      span: 21
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 3
+      span: 26
+      span: 24
+      span: 25
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      span: 27
+      span: 8
+      span: 30
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 5
+      span: 27
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 1
+      span: 27
+      span: 15
+      span: 25
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 3
+      span: 27
+      span: 28
+      span: 29
+    }
+    location: {
+      path: 4
+      path: 1
+      span: 33
+      span: 0
+      span: 36
+      span: 1
+      leading_comments: " WriteStream carries data from the server to the client (export\n direction). The server writes data into the stream and the\n client receives it.\n"
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 1
+      span: 33
+      span: 8
+      span: 19
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 0
+      span: 34
+      span: 8
+      span: 26
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 0
+      path: 5
+      span: 34
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 0
+      path: 1
+      span: 34
+      span: 15
+      span: 21
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 0
+      path: 3
+      span: 34
+      span: 24
+      span: 25
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 1
+      span: 35
+      span: 8
+      span: 30
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 1
+      path: 5
+      span: 35
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 1
+      path: 1
+      span: 35
+      span: 15
+      span: 25
+    }
+    location: {
+      path: 4
+      path: 1
+      path: 2
+      path: 1
+      path: 3
+      span: 35
+      span: 28
+      span: 29
+    }
+  }
+  syntax: "proto3"
+  buf_extension: {
+    is_import: false
+    is_syntax_unspecified: false
+  }
+}
+file: {
+  name: "proto/nerdbox/types/transfer/v1/filesystem.proto"
+  package: "containerd.nerdbox.types.transfer.v1"
+  message_type: {
+    name: "ContainerFilesystem"
+    field: {
+      name: "container_id"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "containerId"
+    }
+    field: {
+      name: "path"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "path"
+    }
+    field: {
+      name: "no_walk"
+      number: 3
+      label: LABEL_OPTIONAL
+      type: TYPE_BOOL
+      json_name: "noWalk"
+    }
+    field: {
+      name: "preserve_ownership"
+      number: 4
+      label: LABEL_OPTIONAL
+      type: TYPE_BOOL
+      json_name: "preserveOwnership"
+    }
+  }
+  options: {
+    go_package: "github.com/containerd/nerdbox/api/types/transfer/v1;transfer"
+  }
+  source_code_info: {
+    location: {
+      span: 16
+      span: 0
+      span: 38
+      span: 1
+    }
+    location: {
+      path: 12
+      span: 16
+      span: 0
+      span: 18
+      leading_detached_comments: "\nCopyright The containerd Authors.\n\nLicensed under the Apache License, Version 2.0 (the \"License\");\nyou may not use this file except in compliance with the License.\nYou may obtain a copy of the License at\n\nhttp://www.apache.org/licenses/LICENSE-2.0\n\nUnless required by applicable law or agreed to in writing, software\ndistributed under the License is distributed on an \"AS IS\" BASIS,\nWITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\nSee the License for the specific language governing permissions and\nlimitations under the License.\n"
+    }
+    location: {
+      path: 2
+      span: 18
+      span: 0
+      span: 45
+    }
+    location: {
+      path: 8
+      span: 20
+      span: 0
+      span: 83
+    }
+    location: {
+      path: 8
+      path: 11
+      span: 20
+      span: 0
+      span: 83
+    }
+    location: {
+      path: 4
+      path: 0
+      span: 25
+      span: 0
+      span: 38
+      span: 1
+      leading_comments: " ContainerFilesystem represents a path within a running container's\n filesystem. It acts as either a source or destination in a transfer\n operation, identifying the container and path for archive operations.\n"
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 1
+      span: 25
+      span: 8
+      span: 27
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      span: 26
+      span: 8
+      span: 32
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 5
+      span: 26
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 1
+      span: 26
+      span: 15
+      span: 27
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 0
+      path: 3
+      span: 26
+      span: 30
+      span: 31
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      span: 27
+      span: 8
+      span: 24
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 5
+      span: 27
+      span: 8
+      span: 14
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 1
+      span: 27
+      span: 15
+      span: 19
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 1
+      path: 3
+      span: 27
+      span: 22
+      span: 23
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 2
+      span: 32
+      span: 8
+      span: 25
+      leading_comments: " When true and path is a directory, return only the directory entry\n itself without walking into its contents. This is useful for\n stat-like operations where only the directory's metadata is needed.\n"
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 2
+      path: 5
+      span: 32
+      span: 8
+      span: 12
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 2
+      path: 1
+      span: 32
+      span: 13
+      span: 20
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 2
+      path: 3
+      span: 32
+      span: 23
+      span: 24
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 3
+      span: 37
+      span: 8
+      span: 36
+      leading_comments: " When true, preserve the UID/GID from tar headers when extracting\n files. When false, extracted files are owned by the extracting\n process.\n"
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 3
+      path: 5
+      span: 37
+      span: 8
+      span: 12
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 3
+      path: 1
+      span: 37
+      span: 13
+      span: 31
+    }
+    location: {
+      path: 4
+      path: 0
+      path: 2
+      path: 3
+      path: 3
+      span: 37
+      span: 34
+      span: 35
+    }
+  }
+  syntax: "proto3"
+  buf_extension: {
+    is_import: false
+    is_syntax_unspecified: false
+  }
+}
diff --git a/api/proto/nerdbox/types/transfer/v1/datastream.proto b/api/proto/nerdbox/types/transfer/v1/datastream.proto
new file mode 100644
index 00000000..b6ae8048
--- /dev/null
+++ b/api/proto/nerdbox/types/transfer/v1/datastream.proto
@@ -0,0 +1,37 @@
+/*
+	Copyright The containerd Authors.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+		http://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+syntax = "proto3";
+
+package containerd.nerdbox.types.transfer.v1;
+
+option go_package = "github.com/containerd/nerdbox/api/types/transfer/v1;transfer";
+
+// ReadStream carries data from the client to the server (import
+// direction). The client sends data through the stream and the
+// server reads it.
+message ReadStream {
+	string stream = 1;
+	string media_type = 2;
+}
+
+// WriteStream carries data from the server to the client (export
+// direction). The server writes data into the stream and the
+// client receives it.
+message WriteStream {
+	string stream = 1;
+	string media_type = 2;
+}
diff --git a/api/proto/nerdbox/types/transfer/v1/filesystem.proto b/api/proto/nerdbox/types/transfer/v1/filesystem.proto
new file mode 100644
index 00000000..695a12b9
--- /dev/null
+++ b/api/proto/nerdbox/types/transfer/v1/filesystem.proto
@@ -0,0 +1,39 @@
+/*
+	Copyright The containerd Authors.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+		http://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+syntax = "proto3";
+
+package containerd.nerdbox.types.transfer.v1;
+
+option go_package = "github.com/containerd/nerdbox/api/types/transfer/v1;transfer";
+
+// ContainerFilesystem represents a path within a running container's
+// filesystem. It acts as either a source or destination in a transfer
+// operation, identifying the container and path for archive operations.
+message ContainerFilesystem {
+	string container_id = 1;
+	string path = 2;
+
+	// When true and path is a directory, return only the directory entry
+	// itself without walking into its contents. This is useful for
+	// stat-like operations where only the directory's metadata is needed.
+	bool no_walk = 3;
+
+	// When true, preserve the UID/GID from tar headers when extracting
+	// files. When false, extracted files are owned by the extracting
+	// process.
+	bool preserve_ownership = 4;
+}
diff --git a/api/types/transfer/v1/datastream.pb.go b/api/types/transfer/v1/datastream.pb.go
new file mode 100644
index 00000000..425fec56
--- /dev/null
+++ b/api/types/transfer/v1/datastream.pb.go
@@ -0,0 +1,252 @@
+//
+//Copyright The containerd Authors.
+//
+//Licensed under the Apache License, Version 2.0 (the "License");
+//you may not use this file except in compliance with the License.
+//You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+//Unless required by applicable law or agreed to in writing, software
+//distributed under the License is distributed on an "AS IS" BASIS,
+//WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//See the License for the specific language governing permissions and
+//limitations under the License.
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.28.1
+// 	protoc        (unknown)
+// source: proto/nerdbox/types/transfer/v1/datastream.proto
+
+package transfer
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// ReadStream carries data from the client to the server (import
+// direction). The client sends data through the stream and the
+// server reads it.
+type ReadStream struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Stream    string `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`
+	MediaType string `protobuf:"bytes,2,opt,name=media_type,json=mediaType,proto3" json:"media_type,omitempty"`
+}
+
+func (x *ReadStream) Reset() {
+	*x = ReadStream{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ReadStream) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ReadStream) ProtoMessage() {}
+
+func (x *ReadStream) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ReadStream.ProtoReflect.Descriptor instead.
+func (*ReadStream) Descriptor() ([]byte, []int) {
+	return file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ReadStream) GetStream() string {
+	if x != nil {
+		return x.Stream
+	}
+	return ""
+}
+
+func (x *ReadStream) GetMediaType() string {
+	if x != nil {
+		return x.MediaType
+	}
+	return ""
+}
+
+// WriteStream carries data from the server to the client (export
+// direction). The server writes data into the stream and the
+// client receives it.
+type WriteStream struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Stream    string `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`
+	MediaType string `protobuf:"bytes,2,opt,name=media_type,json=mediaType,proto3" json:"media_type,omitempty"`
+}
+
+func (x *WriteStream) Reset() {
+	*x = WriteStream{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *WriteStream) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WriteStream) ProtoMessage() {}
+
+func (x *WriteStream) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WriteStream.ProtoReflect.Descriptor instead.
+func (*WriteStream) Descriptor() ([]byte, []int) {
+	return file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *WriteStream) GetStream() string {
+	if x != nil {
+		return x.Stream
+	}
+	return ""
+}
+
+func (x *WriteStream) GetMediaType() string {
+	if x != nil {
+		return x.MediaType
+	}
+	return ""
+}
+
+var File_proto_nerdbox_types_transfer_v1_datastream_proto protoreflect.FileDescriptor
+
+var file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDesc = []byte{
+	0x0a, 0x30, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6e, 0x65, 0x72, 0x64, 0x62, 0x6f, 0x78, 0x2f,
+	0x74, 0x79, 0x70, 0x65, 0x73, 0x2f, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2f, 0x76,
+	0x31, 0x2f, 0x64, 0x61, 0x74, 0x61, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x12, 0x24, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x64, 0x2e, 0x6e,
+	0x65, 0x72, 0x64, 0x62, 0x6f, 0x78, 0x2e, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x74, 0x72, 0x61,
+	0x6e, 0x73, 0x66, 0x65, 0x72, 0x2e, 0x76, 0x31, 0x22, 0x43, 0x0a, 0x0a, 0x52, 0x65, 0x61, 0x64,
+	0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x1d,
+	0x0a, 0x0a, 0x6d, 0x65, 0x64, 0x69, 0x61, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x09, 0x6d, 0x65, 0x64, 0x69, 0x61, 0x54, 0x79, 0x70, 0x65, 0x22, 0x44, 0x0a,
+	0x0b, 0x57, 0x72, 0x69, 0x74, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x16, 0x0a, 0x06,
+	0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74,
+	0x72, 0x65, 0x61, 0x6d, 0x12, 0x1d, 0x0a, 0x0a, 0x6d, 0x65, 0x64, 0x69, 0x61, 0x5f, 0x74, 0x79,
+	0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6d, 0x65, 0x64, 0x69, 0x61, 0x54,
+	0x79, 0x70, 0x65, 0x42, 0x3e, 0x5a, 0x3c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
+	0x6d, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x64, 0x2f, 0x6e, 0x65, 0x72,
+	0x64, 0x62, 0x6f, 0x78, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2f, 0x74,
+	0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2f, 0x76, 0x31, 0x3b, 0x74, 0x72, 0x61, 0x6e, 0x73,
+	0x66, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescOnce sync.Once
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescData = file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDesc
+)
+
+func file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescGZIP() []byte {
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescOnce.Do(func() {
+		file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescData)
+	})
+	return file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDescData
+}
+
+var file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_proto_nerdbox_types_transfer_v1_datastream_proto_goTypes = []interface{}{
+	(*ReadStream)(nil),  // 0: containerd.nerdbox.types.transfer.v1.ReadStream
+	(*WriteStream)(nil), // 1: containerd.nerdbox.types.transfer.v1.WriteStream
+}
+var file_proto_nerdbox_types_transfer_v1_datastream_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
+}
+
+func init() { file_proto_nerdbox_types_transfer_v1_datastream_proto_init() }
+func file_proto_nerdbox_types_transfer_v1_datastream_proto_init() {
+	if File_proto_nerdbox_types_transfer_v1_datastream_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ReadStream); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*WriteStream); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_proto_nerdbox_types_transfer_v1_datastream_proto_goTypes,
+		DependencyIndexes: file_proto_nerdbox_types_transfer_v1_datastream_proto_depIdxs,
+		MessageInfos:      file_proto_nerdbox_types_transfer_v1_datastream_proto_msgTypes,
+	}.Build()
+	File_proto_nerdbox_types_transfer_v1_datastream_proto = out.File
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_rawDesc = nil
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_goTypes = nil
+	file_proto_nerdbox_types_transfer_v1_datastream_proto_depIdxs = nil
+}
diff --git a/api/types/transfer/v1/filesystem.pb.go b/api/types/transfer/v1/filesystem.pb.go
new file mode 100644
index 00000000..d42342b8
--- /dev/null
+++ b/api/types/transfer/v1/filesystem.pb.go
@@ -0,0 +1,204 @@
+//
+//Copyright The containerd Authors.
+//
+//Licensed under the Apache License, Version 2.0 (the "License");
+//you may not use this file except in compliance with the License.
+//You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+//Unless required by applicable law or agreed to in writing, software
+//distributed under the License is distributed on an "AS IS" BASIS,
+//WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//See the License for the specific language governing permissions and
+//limitations under the License.
+
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.28.1
+// 	protoc        (unknown)
+// source: proto/nerdbox/types/transfer/v1/filesystem.proto
+
+package transfer
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// ContainerFilesystem represents a path within a running container's
+// filesystem. It acts as either a source or destination in a transfer
+// operation, identifying the container and path for archive operations.
+type ContainerFilesystem struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	ContainerID string `protobuf:"bytes,1,opt,name=container_id,json=containerId,proto3" json:"container_id,omitempty"`
+	Path        string `protobuf:"bytes,2,opt,name=path,proto3" json:"path,omitempty"`
+	// When true and path is a directory, return only the directory entry
+	// itself without walking into its contents. This is useful for
+	// stat-like operations where only the directory's metadata is needed.
+	NoWalk bool `protobuf:"varint,3,opt,name=no_walk,json=noWalk,proto3" json:"no_walk,omitempty"`
+	// When true, preserve the UID/GID from tar headers when extracting
+	// files. When false, extracted files are owned by the extracting
+	// process.
+	PreserveOwnership bool `protobuf:"varint,4,opt,name=preserve_ownership,json=preserveOwnership,proto3" json:"preserve_ownership,omitempty"`
+}
+
+func (x *ContainerFilesystem) Reset() {
+	*x = ContainerFilesystem{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_proto_nerdbox_types_transfer_v1_filesystem_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ContainerFilesystem) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ContainerFilesystem) ProtoMessage() {}
+
+func (x *ContainerFilesystem) ProtoReflect() protoreflect.Message {
+	mi := &file_proto_nerdbox_types_transfer_v1_filesystem_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ContainerFilesystem.ProtoReflect.Descriptor instead.
+func (*ContainerFilesystem) Descriptor() ([]byte, []int) {
+	return file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ContainerFilesystem) GetContainerID() string {
+	if x != nil {
+		return x.ContainerID
+	}
+	return ""
+}
+
+func (x *ContainerFilesystem) GetPath() string {
+	if x != nil {
+		return x.Path
+	}
+	return ""
+}
+
+func (x *ContainerFilesystem) GetNoWalk() bool {
+	if x != nil {
+		return x.NoWalk
+	}
+	return false
+}
+
+func (x *ContainerFilesystem) GetPreserveOwnership() bool {
+	if x != nil {
+		return x.PreserveOwnership
+	}
+	return false
+}
+
+var File_proto_nerdbox_types_transfer_v1_filesystem_proto protoreflect.FileDescriptor
+
+var file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDesc = []byte{
+	0x0a, 0x30, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6e, 0x65, 0x72, 0x64, 0x62, 0x6f, 0x78, 0x2f,
+	0x74, 0x79, 0x70, 0x65, 0x73, 0x2f, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x2f, 0x76,
+	0x31, 0x2f, 0x66, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x12, 0x24, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x64, 0x2e, 0x6e,
+	0x65, 0x72, 0x64, 0x62, 0x6f, 0x78, 0x2e, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x74, 0x72, 0x61,
+	0x6e, 0x73, 0x66, 0x65, 0x72, 0x2e, 0x76, 0x31, 0x22, 0x94, 0x01, 0x0a, 0x13, 0x43, 0x6f, 0x6e,
+	0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x46, 0x69, 0x6c, 0x65, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d,
+	0x12, 0x21, 0x0a, 0x0c, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x5f, 0x69, 0x64,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65,
+	0x72, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x04, 0x70, 0x61, 0x74, 0x68, 0x12, 0x17, 0x0a, 0x07, 0x6e, 0x6f, 0x5f, 0x77, 0x61,
+	0x6c, 0x6b, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x06, 0x6e, 0x6f, 0x57, 0x61, 0x6c, 0x6b,
+	0x12, 0x2d, 0x0a, 0x12, 0x70, 0x72, 0x65, 0x73, 0x65, 0x72, 0x76, 0x65, 0x5f, 0x6f, 0x77, 0x6e,
+	0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x18, 0x04, 0x20, 0x01, 0x28, 0x08, 0x52, 0x11, 0x70, 0x72,
+	0x65, 0x73, 0x65, 0x72, 0x76, 0x65, 0x4f, 0x77, 0x6e, 0x65, 0x72, 0x73, 0x68, 0x69, 0x70, 0x42,
+	0x3e, 0x5a, 0x3c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x6f,
+	0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x64, 0x2f, 0x6e, 0x65, 0x72, 0x64, 0x62, 0x6f, 0x78,
+	0x2f, 0x61, 0x70, 0x69, 0x2f, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2f, 0x74, 0x72, 0x61, 0x6e, 0x73,
+	0x66, 0x65, 0x72, 0x2f, 0x76, 0x31, 0x3b, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x66, 0x65, 0x72, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescOnce sync.Once
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescData = file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDesc
+)
+
+func file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescGZIP() []byte {
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescOnce.Do(func() {
+		file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescData)
+	})
+	return file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDescData
+}
+
+var file_proto_nerdbox_types_transfer_v1_filesystem_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_proto_nerdbox_types_transfer_v1_filesystem_proto_goTypes = []interface{}{
+	(*ContainerFilesystem)(nil), // 0: containerd.nerdbox.types.transfer.v1.ContainerFilesystem
+}
+var file_proto_nerdbox_types_transfer_v1_filesystem_proto_depIdxs = []int32{
+	0, // [0:0] is the sub-list for method output_type
+	0, // [0:0] is the sub-list for method input_type
+	0, // [0:0] is the sub-list for extension type_name
+	0, // [0:0] is the sub-list for extension extendee
+	0, // [0:0] is the sub-list for field type_name
+}
+
+func init() { file_proto_nerdbox_types_transfer_v1_filesystem_proto_init() }
+func file_proto_nerdbox_types_transfer_v1_filesystem_proto_init() {
+	if File_proto_nerdbox_types_transfer_v1_filesystem_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_proto_nerdbox_types_transfer_v1_filesystem_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ContainerFilesystem); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_proto_nerdbox_types_transfer_v1_filesystem_proto_goTypes,
+		DependencyIndexes: file_proto_nerdbox_types_transfer_v1_filesystem_proto_depIdxs,
+		MessageInfos:      file_proto_nerdbox_types_transfer_v1_filesystem_proto_msgTypes,
+	}.Build()
+	File_proto_nerdbox_types_transfer_v1_filesystem_proto = out.File
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_rawDesc = nil
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_goTypes = nil
+	file_proto_nerdbox_types_transfer_v1_filesystem_proto_depIdxs = nil
+}
diff --git a/cmd/containerd-shim-nerdbox-v1/main.go b/cmd/containerd-shim-nerdbox-v1/main.go
index 1a22109e..8ac41e48 100644
--- a/cmd/containerd-shim-nerdbox-v1/main.go
+++ b/cmd/containerd-shim-nerdbox-v1/main.go
@@ -24,7 +24,9 @@ import (
 	"github.com/containerd/nerdbox/internal/shim/manager"
 
 	_ "github.com/containerd/nerdbox/plugins/shim/sandbox"
+	_ "github.com/containerd/nerdbox/plugins/shim/streaming"
 	_ "github.com/containerd/nerdbox/plugins/shim/task"
+	_ "github.com/containerd/nerdbox/plugins/shim/transfer"
 	_ "github.com/containerd/nerdbox/plugins/vm/libkrun"
 )
 
diff --git a/cmd/vminitd/main.go b/cmd/vminitd/main.go
index 1272da3f..858a4499 100644
--- a/cmd/vminitd/main.go
+++ b/cmd/vminitd/main.go
@@ -47,6 +47,7 @@ import (
 
 	_ "github.com/containerd/nerdbox/plugins/services/bundle"
 	_ "github.com/containerd/nerdbox/plugins/services/system"
+	_ "github.com/containerd/nerdbox/plugins/services/transfer"
 
 	_ "github.com/containerd/nerdbox/plugins/vminit/events"
 	_ "github.com/containerd/nerdbox/plugins/vminit/streaming"
diff --git a/go.mod b/go.mod
index bf0647cc..60c80944 100644
--- a/go.mod
+++ b/go.mod
@@ -27,6 +27,7 @@ require (
 	github.com/vishvananda/netns v0.0.5
 	golang.org/x/sync v0.19.0
 	golang.org/x/sys v0.41.0
+	google.golang.org/grpc v1.79.1
 	google.golang.org/protobuf v1.36.11
 )
 
@@ -36,6 +37,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cilium/ebpf v0.16.0 // indirect
 	github.com/containerd/continuity v0.4.5 // indirect
+	github.com/containerd/platforms v1.0.0-rc.2 // indirect
 	github.com/coreos/go-systemd/v22 v22.7.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
@@ -46,6 +48,7 @@ require (
 	github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/josharian/native v1.1.0 // indirect
+	github.com/klauspost/compress v1.18.4 // indirect
 	github.com/mdlayher/packet v1.1.2 // indirect
 	github.com/mdlayher/socket v0.5.1 // indirect
 	github.com/moby/sys/mountinfo v0.7.2 // indirect
@@ -65,7 +68,6 @@ require (
 	golang.org/x/net v0.48.0 // indirect
 	golang.org/x/text v0.32.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect
-	google.golang.org/grpc v1.79.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
diff --git a/go.sum b/go.sum
index 13dcf7a7..0917a5fd 100644
--- a/go.sum
+++ b/go.sum
@@ -29,6 +29,8 @@ github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
 github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
 github.com/containerd/otelttrpc v0.1.0 h1:UOX68eVTE8H/T45JveIg+I22Ev2aFj4qPITCmXsskjw=
 github.com/containerd/otelttrpc v0.1.0/go.mod h1:XhoA2VvaGPW1clB2ULwrBZfXVuEWuyOd2NUD1IM0yTg=
+github.com/containerd/platforms v1.0.0-rc.2 h1:0SPgaNZPVWGEi4grZdV8VRYQn78y+nm6acgLGv/QzE4=
+github.com/containerd/platforms v1.0.0-rc.2/go.mod h1:J71L7B+aiM5SdIEqmd9wp6THLVRzJGXfNuWCZCllLA4=
 github.com/containerd/plugin v1.0.0 h1:c8Kf1TNl6+e2TtMHZt+39yAPDbouRH9WAToRjex483Y=
 github.com/containerd/plugin v1.0.0/go.mod h1:hQfJe5nmWfImiqT1q8Si3jLv3ynMUIBB47bQ+KexvO8=
 github.com/containerd/ttrpc v1.2.8 h1:xbVu6D4qF2jihdh9rDVOKqUMiFBQk6YctTdo1zk087Y=
@@ -104,6 +106,8 @@ github.com/jsimonetti/rtnetlink/v2 v2.0.1 h1:xda7qaHDSVOsADNouv7ukSuicKZO7GgVUCX
 github.com/jsimonetti/rtnetlink/v2 v2.0.1/go.mod h1:7MoNYNbb3UaDHtF8udiJo/RH6VsTKP1pqKLUTVCvToE=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
+github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
diff --git a/integration/test.sh b/integration/test.sh
index 580603c1..72e223de 100755
--- a/integration/test.sh
+++ b/integration/test.sh
@@ -32,6 +32,7 @@ fi
 tests=(
     "TestSystemInfo"
     "TestStreamInitialization"
+    "TestTransferEcho"
 )
 
 for test in "${tests[@]}"; do
diff --git a/integration/transfer_test.go b/integration/transfer_test.go
new file mode 100644
index 00000000..ec8fe49f
--- /dev/null
+++ b/integration/transfer_test.go
@@ -0,0 +1,200 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package integration
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+
+	transferapi "github.com/containerd/containerd/api/services/transfer/v1"
+	"github.com/containerd/containerd/v2/core/streaming"
+	"github.com/containerd/typeurl/v2"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/containerd/nerdbox/internal/transfer"
+	"github.com/containerd/nerdbox/internal/vm"
+)
+
+func TestTransferEcho(t *testing.T) {
+	runWithVM(t, func(t *testing.T, i vm.Instance) {
+		ctx := t.Context()
+		client := i.Client()
+
+		// Generate random test data
+		testData := make([]byte, 64*1024) // 64KB
+		if _, err := rand.Read(testData); err != nil {
+			t.Fatal("failed to generate test data:", err)
+		}
+
+		// Create a stream creator backed by VM streams
+		sc := &vmStreamCreator{ctx: ctx, instance: i}
+
+		// Source: ReadStream sending testData
+		src := transfer.NewReadStream(bytes.NewReader(testData), "application/octet-stream")
+
+		// Destination: WriteStream receiving into a buffer.
+		// Use a signaling writer so we can wait for all stream data to
+		// arrive before checking — the ReceiveStream goroutine may still
+		// be writing after the Transfer RPC returns.
+		var received bytes.Buffer
+		dstWriter := &doneWriter{Writer: &received, done: make(chan struct{})}
+		dst := transfer.NewWriteStream(dstWriter, "application/octet-stream")
+
+		// Marshal both (this creates streams and starts data pump goroutines)
+		srcAny, err := marshalTransferAny(ctx, src, sc)
+		if err != nil {
+			t.Fatal("failed to marshal source:", err)
+		}
+		dstAny, err := marshalTransferAny(ctx, dst, sc)
+		if err != nil {
+			t.Fatal("failed to marshal destination:", err)
+		}
+
+		// Call Transfer via TTRPC
+		tc := transferapi.NewTTRPCTransferClient(client)
+		if _, err := tc.Transfer(ctx, &transferapi.TransferRequest{
+			Source:      srcAny,
+			Destination: dstAny,
+		}); err != nil {
+			t.Fatal("transfer failed:", err)
+		}
+
+		// Wait for the receive goroutine to finish draining the stream.
+		<-dstWriter.done
+
+		// Verify
+		if !bytes.Equal(received.Bytes(), testData) {
+			t.Fatalf("data mismatch: sent %d bytes, received %d bytes", len(testData), received.Len())
+		}
+		t.Logf("echo transfer: %d bytes transferred successfully", len(testData))
+	})
+}
+
+// streamMarshaler is the interface implemented by types that need to create
+// streams during marshaling (e.g. ReadStream, WriteStream).
+type streamMarshaler interface {
+	MarshalAny(context.Context, streaming.StreamCreator) (typeurl.Any, error)
+}
+
+// marshalTransferAny marshals a transfer type, using the stream creator if
+// the type implements streamMarshaler, otherwise using plain typeurl marshal.
+func marshalTransferAny(ctx context.Context, v any, sc streaming.StreamCreator) (*anypb.Any, error) {
+	var a typeurl.Any
+	var err error
+	if sm, ok := v.(streamMarshaler); ok {
+		a, err = sm.MarshalAny(ctx, sc)
+	} else {
+		a, err = typeurl.MarshalAny(v)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &anypb.Any{
+		TypeUrl: a.GetTypeUrl(),
+		Value:   a.GetValue(),
+	}, nil
+}
+
+// vmStreamCreator implements streaming.StreamCreator by creating vsock
+// connections to the VM and wrapping them with the same length-prefixed
+// proto framing used by the vminitd streaming service.
+type vmStreamCreator struct {
+	ctx      context.Context
+	instance vm.Instance
+}
+
+func (sc *vmStreamCreator) Create(ctx context.Context, id string) (streaming.Stream, error) {
+	conn, err := sc.instance.StartStream(ctx, id)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create stream %q: %w", id, err)
+	}
+	return &framedStream{conn: conn}, nil
+}
+
+// framedStream implements streaming.Stream over a net.Conn using
+// length-prefixed proto framing (matching the vminitd vsockStream protocol).
+type framedStream struct {
+	conn net.Conn
+}
+
+func (s *framedStream) Send(a typeurl.Any) error {
+	data, err := proto.Marshal(typeurl.MarshalProto(a))
+	if err != nil {
+		return fmt.Errorf("failed to marshal stream message: %w", err)
+	}
+	if err := binary.Write(s.conn, binary.BigEndian, uint32(len(data))); err != nil {
+		return fmt.Errorf("failed to write frame length: %w", err)
+	}
+	if _, err := s.conn.Write(data); err != nil {
+		return fmt.Errorf("failed to write frame data: %w", err)
+	}
+	return nil
+}
+
+func (s *framedStream) Recv() (typeurl.Any, error) {
+	var length uint32
+	if err := binary.Read(s.conn, binary.BigEndian, &length); err != nil {
+		return nil, err
+	}
+	data := make([]byte, length)
+	if _, err := io.ReadFull(s.conn, data); err != nil {
+		return nil, fmt.Errorf("failed to read frame data: %w", err)
+	}
+	var a anypb.Any
+	if err := proto.Unmarshal(data, &a); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal stream message: %w", err)
+	}
+	return &a, nil
+}
+
+func (s *framedStream) Close() error {
+	// Use half-close (shutdown write) instead of full close. SendStream
+	// calls Close() after sending all data; a full close can discard
+	// buffered data the VM hasn't read yet. Shutdown SHUT_WR signals
+	// EOF to the reader while letting buffered data drain.
+	if sc, ok := s.conn.(interface{ CloseWrite() error }); ok {
+		return sc.CloseWrite()
+	}
+	return s.conn.Close()
+}
+
+// nopWriteCloser wraps an io.Writer with a no-op Close method.
+type nopWriteCloser struct {
+	io.Writer
+}
+
+func (nopWriteCloser) Close() error { return nil }
+
+// doneWriter wraps an io.Writer and signals on the done channel when
+// Close is called. This lets callers wait for the ReceiveStream
+// goroutine (which calls Close after draining all data) to finish.
+type doneWriter struct {
+	io.Writer
+	done chan struct{}
+}
+
+func (w *doneWriter) Close() error {
+	close(w.done)
+	return nil
+}
diff --git a/internal/transfer/containerfs.go b/internal/transfer/containerfs.go
new file mode 100644
index 00000000..06ff3bf7
--- /dev/null
+++ b/internal/transfer/containerfs.go
@@ -0,0 +1,229 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package transfer
+
+import (
+	"archive/tar"
+	"context"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"strings"
+
+	ctransfer "github.com/containerd/containerd/v2/core/transfer"
+	"github.com/containerd/errdefs"
+)
+
+const mediaTypeTar = "application/x-tar"
+
+// NewContainerFSTransferrer returns a Transferrer that handles
+// ContainerFilesystem + ReadStream/WriteStream transfer pairs.
+func NewContainerFSTransferrer(bundleDir string) ctransfer.Transferrer {
+	return &containerFSTransferrer{bundleDir: bundleDir}
+}
+
+type containerFSTransferrer struct {
+	bundleDir string
+}
+
+func (t *containerFSTransferrer) Transfer(ctx context.Context, src, dst any, opts ...ctransfer.Opt) error {
+	switch s := src.(type) {
+	case *ContainerFilesystem:
+		// Copy-from: ContainerFilesystem -> WriteStream
+		d, ok := dst.(*WriteStream)
+		if !ok {
+			return errdefs.ErrNotImplemented
+		}
+		rootfs := filepath.Join(t.bundleDir, s.ContainerID, "rootfs")
+		w := d.Writer(ctx)
+		defer w.Close()
+		return writePath(rootfs, s.Path, w, d.MediaType, s.NoWalk)
+
+	case *ReadStream:
+		// Copy-to: ReadStream -> ContainerFilesystem
+		d, ok := dst.(*ContainerFilesystem)
+		if !ok {
+			return errdefs.ErrNotImplemented
+		}
+		rootfs := filepath.Join(t.bundleDir, d.ContainerID, "rootfs")
+		r := s.Reader(ctx)
+		return readPath(r, rootfs, d.Path, s.MediaType, d.PreserveOwnership)
+	}
+
+	return errdefs.ErrNotImplemented
+}
+
+// writePath creates a tar archive from the given path within rootfs
+// and writes it to w. When noWalk is true and path is a directory,
+// only the directory entry itself is included without walking into it.
+func writePath(rootfs, path string, w io.Writer, mediaType string, noWalk bool) error {
+	if mediaType != mediaTypeTar {
+		return fmt.Errorf("unsupported media type %q: %w", mediaType, errdefs.ErrNotImplemented)
+	}
+
+	srcPath := filepath.Join(rootfs, filepath.Clean("/"+path))
+
+	fi, err := os.Lstat(srcPath)
+	if err != nil {
+		return fmt.Errorf("failed to stat %s: %w", path, err)
+	}
+
+	tw := tar.NewWriter(w)
+	defer tw.Close()
+
+	if !fi.IsDir() || noWalk {
+		return writeTarEntry(tw, srcPath, fi, filepath.Base(srcPath))
+	}
+
+	return filepath.WalkDir(srcPath, func(filePath string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		info, err := d.Info()
+		if err != nil {
+			return err
+		}
+
+		// Compute relative path for the tar header
+		rel, err := filepath.Rel(srcPath, filePath)
+		if err != nil {
+			return err
+		}
+		if rel == "." {
+			rel = filepath.Base(srcPath)
+		} else {
+			rel = filepath.Join(filepath.Base(srcPath), rel)
+		}
+
+		return writeTarEntry(tw, filePath, info, rel)
+	})
+}
+
+func writeTarEntry(tw *tar.Writer, filePath string, fi os.FileInfo, name string) error {
+	header, err := tar.FileInfoHeader(fi, "")
+	if err != nil {
+		return err
+	}
+	header.Name = name
+
+	// Resolve symlink target
+	if fi.Mode()&os.ModeSymlink != 0 {
+		link, err := os.Readlink(filePath)
+		if err != nil {
+			return err
+		}
+		header.Linkname = link
+	}
+
+	if err := tw.WriteHeader(header); err != nil {
+		return err
+	}
+
+	if fi.Mode().IsRegular() {
+		f, err := os.Open(filePath)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		if _, err := io.Copy(tw, f); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// readPath reads a tar archive from r and extracts it to the given path
+// within rootfs. When preserveOwnership is true, extracted files have
+// their UID/GID set from the tar headers.
+func readPath(r io.Reader, rootfs, path, mediaType string, preserveOwnership bool) error {
+	if mediaType != mediaTypeTar {
+		return fmt.Errorf("unsupported media type %q: %w", mediaType, errdefs.ErrNotImplemented)
+	}
+
+	dstPath := filepath.Join(rootfs, filepath.Clean("/"+path))
+
+	tr := tar.NewReader(r)
+	for {
+		header, err := tr.Next()
+		if err == io.EOF {
+			return nil
+		}
+		if err != nil {
+			return fmt.Errorf("failed to read tar header: %w", err)
+		}
+
+		target := filepath.Join(dstPath, filepath.Clean("/"+header.Name))
+
+		// Ensure the target is within the destination directory
+		if !strings.HasPrefix(target, filepath.Clean(dstPath)+string(os.PathSeparator)) && target != filepath.Clean(dstPath) {
+			return fmt.Errorf("tar entry %q would escape destination", header.Name)
+		}
+
+		if err := extractTarEntry(target, header, tr, preserveOwnership); err != nil {
+			return err
+		}
+	}
+}
+
+func extractTarEntry(target string, header *tar.Header, r io.Reader, preserveOwnership bool) error {
+	switch header.Typeflag {
+	case tar.TypeDir:
+		if err := os.MkdirAll(target, os.FileMode(header.Mode)); err != nil {
+			return err
+		}
+	case tar.TypeReg:
+		if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
+			return err
+		}
+		f, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(header.Mode))
+		if err != nil {
+			return err
+		}
+		if _, err := io.Copy(f, r); err != nil {
+			f.Close()
+			return err
+		}
+		if err := f.Close(); err != nil {
+			return err
+		}
+	case tar.TypeSymlink:
+		if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
+			return err
+		}
+		if err := os.Symlink(header.Linkname, target); err != nil {
+			return err
+		}
+	case tar.TypeLink:
+		if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
+			return err
+		}
+		if err := os.Link(header.Linkname, target); err != nil {
+			return err
+		}
+	}
+
+	if preserveOwnership {
+		if err := os.Lchown(target, header.Uid, header.Gid); err != nil {
+			return fmt.Errorf("failed to chown %s: %w", target, err)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/transfer/echo.go b/internal/transfer/echo.go
new file mode 100644
index 00000000..7bbd2efc
--- /dev/null
+++ b/internal/transfer/echo.go
@@ -0,0 +1,52 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package transfer
+
+import (
+	"context"
+	"io"
+
+	ctransfer "github.com/containerd/containerd/v2/core/transfer"
+	"github.com/containerd/errdefs"
+)
+
+// NewEchoTransferrer returns a Transferrer that copies data from a
+// ReadStream source to a WriteStream destination. This is useful for
+// testing the streaming pipeline end-to-end.
+func NewEchoTransferrer() ctransfer.Transferrer {
+	return &echoTransferrer{}
+}
+
+type echoTransferrer struct{}
+
+func (t *echoTransferrer) Transfer(ctx context.Context, src, dst any, opts ...ctransfer.Opt) error {
+	s, ok := src.(*ReadStream)
+	if !ok {
+		return errdefs.ErrNotImplemented
+	}
+	d, ok := dst.(*WriteStream)
+	if !ok {
+		return errdefs.ErrNotImplemented
+	}
+
+	r := s.Reader(ctx)
+	w := d.Writer(ctx)
+	defer w.Close()
+
+	_, err := io.Copy(w, r)
+	return err
+}
diff --git a/internal/transfer/types.go b/internal/transfer/types.go
new file mode 100644
index 00000000..d3291a69
--- /dev/null
+++ b/internal/transfer/types.go
@@ -0,0 +1,178 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+// Package transfer defines transfer types for container filesystem operations.
+package transfer
+
+import (
+	"context"
+	"io"
+
+	"github.com/containerd/containerd/v2/core/streaming"
+	tplugins "github.com/containerd/containerd/v2/core/transfer/plugins"
+	tstreaming "github.com/containerd/containerd/v2/core/transfer/streaming"
+	"github.com/containerd/typeurl/v2"
+
+	transferpb "github.com/containerd/nerdbox/api/types/transfer/v1"
+)
+
+func init() {
+	tplugins.Register(&transferpb.ContainerFilesystem{}, ContainerFilesystem{})
+	tplugins.Register(&transferpb.ReadStream{}, ReadStream{})
+	tplugins.Register(&transferpb.WriteStream{}, WriteStream{})
+}
+
+// ContainerFilesystem represents a path within a running container's
+// filesystem. It acts as either a source or destination in a transfer
+// operation, identifying the container and path for archive operations.
+type ContainerFilesystem struct {
+	ContainerID       string
+	Path              string
+	NoWalk            bool
+	PreserveOwnership bool
+}
+
+// MarshalAny marshals the ContainerFilesystem to a typeurl.Any.
+func (cf *ContainerFilesystem) MarshalAny(ctx context.Context, sm streaming.StreamCreator) (typeurl.Any, error) {
+	return typeurl.MarshalAny(&transferpb.ContainerFilesystem{
+		ContainerID:       cf.ContainerID,
+		Path:              cf.Path,
+		NoWalk:            cf.NoWalk,
+		PreserveOwnership: cf.PreserveOwnership,
+	})
+}
+
+// UnmarshalAny unmarshals a ContainerFilesystem from a typeurl.Any.
+func (cf *ContainerFilesystem) UnmarshalAny(ctx context.Context, sg streaming.StreamGetter, a typeurl.Any) error {
+	var p transferpb.ContainerFilesystem
+	if err := typeurl.UnmarshalTo(a, &p); err != nil {
+		return err
+	}
+	cf.ContainerID = p.ContainerID
+	cf.Path = p.Path
+	cf.NoWalk = p.NoWalk
+	cf.PreserveOwnership = p.PreserveOwnership
+	return nil
+}
+
+// ReadStream carries data from the client to the server (import
+// direction). The client sends data through the stream and the server
+// reads it.
+type ReadStream struct {
+	MediaType string
+	stream    streaming.Stream
+	reader    io.Reader // client-side: set by constructor, used by MarshalAny
+}
+
+// NewReadStream creates a ReadStream that will send data from r to the
+// server during marshaling.
+func NewReadStream(r io.Reader, mediaType string) *ReadStream {
+	return &ReadStream{MediaType: mediaType, reader: r}
+}
+
+// MarshalAny marshals the ReadStream, creating a streaming connection
+// and starting a goroutine that sends data from the reader.
+func (s *ReadStream) MarshalAny(ctx context.Context, sm streaming.StreamCreator) (typeurl.Any, error) {
+	sid := tstreaming.GenerateID("data")
+	stream, err := sm.Create(ctx, sid)
+	if err != nil {
+		return nil, err
+	}
+
+	go tstreaming.SendStream(ctx, s.reader, stream)
+
+	return typeurl.MarshalAny(&transferpb.ReadStream{
+		Stream:    sid,
+		MediaType: s.MediaType,
+	})
+}
+
+// UnmarshalAny unmarshals a ReadStream from a typeurl.Any, recovering
+// the stream from the StreamGetter.
+func (s *ReadStream) UnmarshalAny(ctx context.Context, sg streaming.StreamGetter, a typeurl.Any) error {
+	var p transferpb.ReadStream
+	if err := typeurl.UnmarshalTo(a, &p); err != nil {
+		return err
+	}
+	stream, err := sg.Get(ctx, p.Stream)
+	if err != nil {
+		return err
+	}
+	s.stream = stream
+	s.MediaType = p.MediaType
+	return nil
+}
+
+// Reader returns an io.Reader that consumes data sent by the client.
+func (s *ReadStream) Reader(ctx context.Context) io.Reader {
+	return tstreaming.ReceiveStream(ctx, s.stream)
+}
+
+// WriteStream carries data from the server to the client (export
+// direction). The server writes data into the stream and the client
+// receives it.
+type WriteStream struct {
+	MediaType string
+	stream    streaming.Stream
+	writer    io.WriteCloser // client-side: set by constructor, used by MarshalAny
+}
+
+// NewWriteStream creates a WriteStream that will receive data from the
+// server into w during marshaling.
+func NewWriteStream(w io.WriteCloser, mediaType string) *WriteStream {
+	return &WriteStream{MediaType: mediaType, writer: w}
+}
+
+// MarshalAny marshals the WriteStream, creating a streaming connection
+// and starting a goroutine that receives data into the writer.
+func (s *WriteStream) MarshalAny(ctx context.Context, sm streaming.StreamCreator) (typeurl.Any, error) {
+	sid := tstreaming.GenerateID("data")
+	stream, err := sm.Create(ctx, sid)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		io.Copy(s.writer, tstreaming.ReceiveStream(ctx, stream))
+		s.writer.Close()
+	}()
+
+	return typeurl.MarshalAny(&transferpb.WriteStream{
+		Stream:    sid,
+		MediaType: s.MediaType,
+	})
+}
+
+// UnmarshalAny unmarshals a WriteStream from a typeurl.Any, recovering
+// the stream from the StreamGetter.
+func (s *WriteStream) UnmarshalAny(ctx context.Context, sg streaming.StreamGetter, a typeurl.Any) error {
+	var p transferpb.WriteStream
+	if err := typeurl.UnmarshalTo(a, &p); err != nil {
+		return err
+	}
+	stream, err := sg.Get(ctx, p.Stream)
+	if err != nil {
+		return err
+	}
+	s.stream = stream
+	s.MediaType = p.MediaType
+	return nil
+}
+
+// Writer returns an io.WriteCloser that sends data to the client.
+func (s *WriteStream) Writer(ctx context.Context) io.WriteCloser {
+	return tstreaming.WriteByteStream(ctx, s.stream)
+}
diff --git a/plugins/services/transfer/service.go b/plugins/services/transfer/service.go
new file mode 100644
index 00000000..9ee4dc73
--- /dev/null
+++ b/plugins/services/transfer/service.go
@@ -0,0 +1,123 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package transfer
+
+import (
+	"context"
+
+	transferapi "github.com/containerd/containerd/api/services/transfer/v1"
+	"github.com/containerd/containerd/v2/core/streaming"
+	"github.com/containerd/containerd/v2/core/transfer"
+	tplugins "github.com/containerd/containerd/v2/core/transfer/plugins"
+	"github.com/containerd/errdefs"
+	"github.com/containerd/log"
+	"github.com/containerd/plugin"
+	"github.com/containerd/plugin/registry"
+	"github.com/containerd/ttrpc"
+	"github.com/containerd/typeurl/v2"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	itransfer "github.com/containerd/nerdbox/internal/transfer"
+	"github.com/containerd/nerdbox/plugins"
+)
+
+// streamGetterProvider is implemented by the vsock streaming plugin.
+type streamGetterProvider interface {
+	StreamGetter() streaming.StreamGetter
+}
+
+func init() {
+	registry.Register(&plugin.Registration{
+		Type: plugins.TTRPCPlugin,
+		ID:   "transfer",
+		Requires: []plugin.Type{
+			plugins.StreamingPlugin,
+		},
+		InitFn: func(ic *plugin.InitContext) (any, error) {
+			sp, err := ic.GetByID(plugins.StreamingPlugin, "vsock")
+			if err != nil {
+				return nil, err
+			}
+			sgp := sp.(streamGetterProvider)
+
+			bundleDir := ic.Properties[plugins.PropertyBundleDir]
+
+			return &service{
+				streamGetter: sgp.StreamGetter(),
+				transferrers: []transfer.Transferrer{
+					itransfer.NewContainerFSTransferrer(bundleDir),
+					itransfer.NewEchoTransferrer(),
+				},
+			}, nil
+		},
+	})
+}
+
+type service struct {
+	streamGetter streaming.StreamGetter
+	transferrers []transfer.Transferrer
+}
+
+func (s *service) RegisterTTRPC(server *ttrpc.Server) error {
+	transferapi.RegisterTTRPCTransferService(server, s)
+	return nil
+}
+
+func (s *service) Transfer(ctx context.Context, req *transferapi.TransferRequest) (*emptypb.Empty, error) {
+	src, err := s.convertAny(ctx, req.Source)
+	if err != nil {
+		return nil, err
+	}
+	dst, err := s.convertAny(ctx, req.Destination)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, t := range s.transferrers {
+		if err = t.Transfer(ctx, src, dst); err == nil {
+			return &emptypb.Empty{}, nil
+		} else if !errdefs.IsNotImplemented(err) {
+			return nil, err
+		}
+		log.G(ctx).WithError(err).Debugf("transfer not implemented for %T to %T", src, dst)
+	}
+	return nil, status.Errorf(codes.Unimplemented, "method Transfer not implemented for %s to %s", req.Source.GetTypeUrl(), req.Destination.GetTypeUrl())
+}
+
+type streamUnmarshaler interface {
+	UnmarshalAny(context.Context, streaming.StreamGetter, typeurl.Any) error
+}
+
+func (s *service) convertAny(ctx context.Context, a typeurl.Any) (any, error) {
+	obj, err := tplugins.ResolveType(a)
+	if err != nil {
+		if errdefs.IsNotFound(err) {
+			return typeurl.UnmarshalAny(a)
+		}
+		return nil, err
+	}
+	switch v := obj.(type) {
+	case streamUnmarshaler:
+		err = v.UnmarshalAny(ctx, s.streamGetter, a)
+		return obj, err
+	default:
+		err = typeurl.UnmarshalTo(a, obj)
+		return obj, err
+	}
+}
diff --git a/plugins/shim/streaming/plugin.go b/plugins/shim/streaming/plugin.go
new file mode 100644
index 00000000..8e163e3e
--- /dev/null
+++ b/plugins/shim/streaming/plugin.go
@@ -0,0 +1,183 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streaming
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"io"
+
+	streamapi "github.com/containerd/containerd/api/services/streaming/v1"
+	ptypes "github.com/containerd/containerd/v2/pkg/protobuf/types"
+	cplugins "github.com/containerd/containerd/v2/plugins"
+	"github.com/containerd/log"
+	"github.com/containerd/plugin"
+	"github.com/containerd/plugin/registry"
+	"github.com/containerd/ttrpc"
+	typeurl "github.com/containerd/typeurl/v2"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/containerd/nerdbox/internal/shim/sandbox"
+	"github.com/containerd/nerdbox/plugins"
+)
+
+func init() {
+	registry.Register(&plugin.Registration{
+		Type: cplugins.TTRPCPlugin,
+		ID:   "streaming",
+		Requires: []plugin.Type{
+			plugins.SandboxPlugin,
+		},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			sb, err := ic.GetSingle(plugins.SandboxPlugin)
+			if err != nil {
+				return nil, err
+			}
+
+			return &service{
+				sb: sb.(sandbox.Sandbox),
+			}, nil
+		},
+	})
+}
+
+type service struct {
+	sb sandbox.Sandbox
+}
+
+func (s *service) RegisterTTRPC(server *ttrpc.Server) error {
+	streamapi.RegisterTTRPCStreamingService(server, s)
+	return nil
+}
+
+func (s *service) Stream(ctx context.Context, srv streamapi.TTRPCStreaming_StreamServer) error {
+	// Receive the StreamInit message with the stream ID
+	a, err := srv.Recv()
+	if err != nil {
+		return err
+	}
+	var i streamapi.StreamInit
+	if err := typeurl.UnmarshalTo(a, &i); err != nil {
+		return err
+	}
+
+	log.G(ctx).WithField("stream", i.ID).Debug("creating stream bridge")
+
+	// Create a stream connection to the VM, passing through the stream ID
+	vmConn, err := s.sb.StartStream(ctx, i.ID)
+	if err != nil {
+		return fmt.Errorf("failed to start vm stream: %w", err)
+	}
+	defer vmConn.Close()
+
+	log.G(ctx).WithField("stream", i.ID).Debug("stream bridge established")
+
+	// Send ack back to containerd client
+	e, _ := typeurl.MarshalAnyToProto(&ptypes.Empty{})
+	if err := srv.Send(e); err != nil {
+		return err
+	}
+
+	// Start bidirectional bridge between TTRPC and VM.
+	// Messages are forwarded as length-prefixed proto frames.
+	done := make(chan error, 2)
+
+	// TTRPC -> VM: receive typeurl.Any from containerd, frame and write to VM
+	go func() {
+		err := bridgeTTRPCToVM(srv, vmConn)
+		// Half-close the write side so the VM sees EOF on its reads
+		// while still allowing data to flow back from VM -> TTRPC.
+		if cw, ok := vmConn.(interface{ CloseWrite() error }); ok {
+			cw.CloseWrite()
+		}
+		done <- err
+	}()
+
+	// VM -> TTRPC: read framed messages from VM, send to containerd
+	go func() {
+		done <- bridgeVMToTTRPC(vmConn, srv)
+	}()
+
+	// Wait for both bridge directions to finish or context cancellation.
+	// We must not return early after just one direction finishes, because:
+	// 1. Returning closes the TTRPC server stream which can race with
+	//    other in-flight RPCs (e.g. Transfer) on the same connection.
+	// 2. Closing vmConn eagerly can truncate in-flight data that the
+	//    VM hasn't read yet.
+	// Instead, wait for both to finish naturally. For unidirectional
+	// streams, one direction will block until context cancellation
+	// (shim shutdown), which is correct — the stream stays alive as
+	// long as the connection does.
+	for n := 0; n < 2; n++ {
+		select {
+		case err := <-done:
+			if err != nil {
+				log.G(ctx).WithError(err).WithField("stream", i.ID).Debug("stream bridge direction ended")
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+
+	return nil
+}
+
+// bridgeTTRPCToVM reads typeurl.Any messages from the TTRPC stream and
+// writes them as length-prefixed proto frames to the VM connection.
+func bridgeTTRPCToVM(srv streamapi.TTRPCStreaming_StreamServer, conn io.Writer) error {
+	for {
+		a, err := srv.Recv()
+		if err != nil {
+			return err
+		}
+
+		data, err := proto.Marshal(typeurl.MarshalProto(a))
+		if err != nil {
+			return fmt.Errorf("failed to marshal for vm: %w", err)
+		}
+		if err := binary.Write(conn, binary.BigEndian, uint32(len(data))); err != nil {
+			return fmt.Errorf("failed to write frame length to vm: %w", err)
+		}
+		if _, err := conn.Write(data); err != nil {
+			return fmt.Errorf("failed to write frame data to vm: %w", err)
+		}
+	}
+}
+
+// bridgeVMToTTRPC reads length-prefixed proto frames from the VM
+// connection and sends them as typeurl.Any messages on the TTRPC stream.
+func bridgeVMToTTRPC(conn io.Reader, srv streamapi.TTRPCStreaming_StreamServer) error {
+	for {
+		var length uint32
+		if err := binary.Read(conn, binary.BigEndian, &length); err != nil {
+			return err
+		}
+		data := make([]byte, length)
+		if _, err := io.ReadFull(conn, data); err != nil {
+			return fmt.Errorf("failed to read frame data from vm: %w", err)
+		}
+		var a anypb.Any
+		if err := proto.Unmarshal(data, &a); err != nil {
+			return fmt.Errorf("failed to unmarshal from vm: %w", err)
+		}
+		if err := srv.Send(&a); err != nil {
+			return err
+		}
+	}
+}
diff --git a/plugins/shim/transfer/plugin.go b/plugins/shim/transfer/plugin.go
new file mode 100644
index 00000000..c95e1e65
--- /dev/null
+++ b/plugins/shim/transfer/plugin.go
@@ -0,0 +1,72 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package transfer
+
+import (
+	"context"
+	"fmt"
+
+	transferapi "github.com/containerd/containerd/api/services/transfer/v1"
+	"github.com/containerd/log"
+	"github.com/containerd/plugin"
+	"github.com/containerd/plugin/registry"
+	"github.com/containerd/ttrpc"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/containerd/nerdbox/internal/shim/sandbox"
+	"github.com/containerd/nerdbox/plugins"
+)
+
+func init() {
+	registry.Register(&plugin.Registration{
+		Type: plugins.TTRPCPlugin,
+		ID:   "transfer",
+		Requires: []plugin.Type{
+			plugins.SandboxPlugin,
+		},
+		InitFn: func(ic *plugin.InitContext) (interface{}, error) {
+			sb, err := ic.GetSingle(plugins.SandboxPlugin)
+			if err != nil {
+				return nil, err
+			}
+			return &service{
+				sandbox: sb.(sandbox.Sandbox),
+			}, nil
+		},
+	})
+}
+
+type service struct {
+	sandbox sandbox.Sandbox
+}
+
+func (s *service) RegisterTTRPC(ts *ttrpc.Server) error {
+	transferapi.RegisterTTRPCTransferService(ts, s)
+	return nil
+}
+
+func (s *service) Transfer(ctx context.Context, req *transferapi.TransferRequest) (*emptypb.Empty, error) {
+	log.G(ctx).Debug("transfer: forwarding to vminitd")
+
+	client, err := s.sandbox.Client()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get vminitd client: %w", err)
+	}
+
+	vmTransfer := transferapi.NewTTRPCTransferClient(client)
+	return vmTransfer.Transfer(ctx, req)
+}
diff --git a/plugins/vminit/streaming/plugin.go b/plugins/vminit/streaming/plugin.go
index 7f3c04f1..ffebdca5 100644
--- a/plugins/vminit/streaming/plugin.go
+++ b/plugins/vminit/streaming/plugin.go
@@ -25,13 +25,17 @@ import (
 	"net"
 	"sync"
 
+	"github.com/containerd/containerd/v2/core/streaming"
 	"github.com/containerd/containerd/v2/pkg/shutdown"
 	cplugins "github.com/containerd/containerd/v2/plugins"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
 	"github.com/containerd/plugin"
 	"github.com/containerd/plugin/registry"
+	"github.com/containerd/typeurl/v2"
 	"github.com/mdlayher/vsock"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/containerd/nerdbox/plugins"
 )
@@ -182,3 +186,68 @@ func (s *service) Get(id string) (io.ReadWriteCloser, error) {
 	delete(s.streams, id)
 	return conn, nil
 }
+
+// StreamGetter returns a streaming.StreamGetter that looks up streams by
+// their string stream ID.
+func (s *service) StreamGetter() streaming.StreamGetter {
+	return &streamGetter{s: s}
+}
+
+type streamGetter struct {
+	s *service
+}
+
+func (sg *streamGetter) Get(ctx context.Context, name string) (streaming.Stream, error) {
+	sg.s.mu.Lock()
+	conn, ok := sg.s.streams[name]
+	if !ok {
+		sg.s.mu.Unlock()
+		return nil, fmt.Errorf("stream %q not found: %w", name, errdefs.ErrNotFound)
+	}
+	// Remove from map so the stream is exclusively owned by the caller.
+	// The caller is responsible for closing the stream.
+	delete(sg.s.streams, name)
+	sg.s.mu.Unlock()
+	return &vsockStream{conn: conn}, nil
+}
+
+// vsockStream wraps a net.Conn with length-prefixed proto framing to
+// implement the streaming.Stream interface. Each message is framed as
+// a 4-byte big-endian length prefix followed by serialized proto bytes.
+type vsockStream struct {
+	conn net.Conn
+}
+
+func (s *vsockStream) Send(a typeurl.Any) error {
+	data, err := proto.Marshal(typeurl.MarshalProto(a))
+	if err != nil {
+		return fmt.Errorf("failed to marshal stream message: %w", err)
+	}
+	if err := binary.Write(s.conn, binary.BigEndian, uint32(len(data))); err != nil {
+		return fmt.Errorf("failed to write frame length: %w", err)
+	}
+	if _, err := s.conn.Write(data); err != nil {
+		return fmt.Errorf("failed to write frame data: %w", err)
+	}
+	return nil
+}
+
+func (s *vsockStream) Recv() (typeurl.Any, error) {
+	var length uint32
+	if err := binary.Read(s.conn, binary.BigEndian, &length); err != nil {
+		return nil, err
+	}
+	data := make([]byte, length)
+	if _, err := io.ReadFull(s.conn, data); err != nil {
+		return nil, fmt.Errorf("failed to read frame data: %w", err)
+	}
+	var a anypb.Any
+	if err := proto.Unmarshal(data, &a); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal stream message: %w", err)
+	}
+	return &a, nil
+}
+
+func (s *vsockStream) Close() error {
+	return s.conn.Close()
+}