diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
index 31684347cf33..2fc88c333e09 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/agentBindings.cpp
@@ -26,6 +26,7 @@
 #endif
 
 #include <nanobind/nanobind.h>
+#include <nanobind/ndarray.h>
 #include <nanobind/stl/optional.h>
 #include <nanobind/stl/string.h>
 #include <nanobind/stl/tuple.h>
@@ -83,6 +84,47 @@ NB_MODULE(tensorrt_llm_transfer_agent_binding, m)
                 new (self) kvc::MemoryDescs(type, std::move(descs));
             },
             nb::arg("type"), nb::arg("tuples"))
+        // Classmethod: batch construction from numpy arrays
+        .def_static(
+            "from_arrays",
+            [](kvc::MemoryType type, nb::ndarray<int64_t, nb::ndim<1>, nb::c_contig, nb::device::cpu> addrs,
+                nb::ndarray<int64_t, nb::ndim<1>, nb::c_contig, nb::device::cpu> sizes,
+                nb::ndarray<int32_t, nb::ndim<1>, nb::c_contig, nb::device::cpu> deviceIds)
+            {
+                size_t n = addrs.shape(0);
+                auto const* a = addrs.data();
+                auto const* s = sizes.data();
+                auto const* d = deviceIds.data();
+                std::vector<kvc::MemoryDesc> descs;
+                descs.reserve(n);
+                for (size_t i = 0; i < n; ++i)
+                {
+                    descs.emplace_back(
+                        static_cast<uintptr_t>(a[i]), static_cast<size_t>(s[i]), static_cast<uint32_t>(d[i]));
+                }
+                return kvc::MemoryDescs(type, std::move(descs));
+            },
+            nb::arg("type"), nb::arg("addrs"), nb::arg("sizes"), nb::arg("device_ids"),
+            nb::call_guard<nb::gil_scoped_release>())
+        // Classmethod: batch construction with uniform device_id (avoids np.full allocation)
+        .def_static(
+            "from_arrays_uniform_device",
+            [](kvc::MemoryType type, nb::ndarray<int64_t, nb::ndim<1>, nb::c_contig, nb::device::cpu> addrs,
+                nb::ndarray<int64_t, nb::ndim<1>, nb::c_contig, nb::device::cpu> sizes, uint32_t deviceId)
+            {
+                size_t n = addrs.shape(0);
+                auto const* a = addrs.data();
+                auto const* s = sizes.data();
+                std::vector<kvc::MemoryDesc> descs;
+                descs.reserve(n);
+                for (size_t i = 0; i < n; ++i)
+                {
+                    descs.emplace_back(static_cast<uintptr_t>(a[i]), static_cast<size_t>(s[i]), deviceId);
+                }
+                return kvc::MemoryDescs(type, std::move(descs));
+            },
+            nb::arg("type"), nb::arg("addrs"), nb::arg("sizes"), nb::arg("device_id"),
+            nb::call_guard<nb::gil_scoped_release>())
         .def_prop_ro("type", &kvc::MemoryDescs::getType)
         .def_prop_ro("descs", &kvc::MemoryDescs::getDescs);
 
@@ -105,9 +147,24 @@ NB_MODULE(tensorrt_llm_transfer_agent_binding, m)
             });
 
     // TransferRequest class
+    //
+    // NOTE: The constructor uses std::move to transfer ownership of src_descs / dst_descs
+    // into the TransferRequest.  This avoids an O(n) copy of the internal
+    // std::vector<MemoryDesc> (24 bytes * n).  For 40k descriptors this saves ~937 KB
+    // of memcpy and turns a ~58 us copy into an O(1) pointer swap (~0.4 us).
+    //
+    // IMPORTANT: After construction, the Python MemoryDescs objects passed as src_descs
+    // and dst_descs are left in a moved-from state — their internal descriptor list
+    // becomes empty.  Do NOT access them after passing to TransferRequest.
     nb::class_<kvc::TransferRequest>(m, "TransferRequest")
-        .def(nb::init<kvc::TransferOp, kvc::TransferDescs, kvc::TransferDescs, std::string const&,
-                 std::optional<kvc::SyncMessage>>(),
+        .def(
+            "__init__",
+            [](kvc::TransferRequest* self, kvc::TransferOp op, kvc::TransferDescs& srcDescs,
+                kvc::TransferDescs& dstDescs, std::string const& remoteName,
+                std::optional<kvc::SyncMessage> syncMessage) {
+                new (self) kvc::TransferRequest(
+                    op, std::move(srcDescs), std::move(dstDescs), remoteName, std::move(syncMessage));
+            },
             nb::arg("op"), nb::arg("src_descs"), nb::arg("dst_descs"), nb::arg("remote_name"),
             nb::arg("sync_message") = std::nullopt)
         .def_prop_ro("op", &kvc::TransferRequest::getOp)
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
index 018ac5d2dff3..bad3e184f983 100644
--- a/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
+++ b/cpp/tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.cpp
@@ -18,6 +18,7 @@
 #include "tensorrt_llm/executor/cache_transmission/nixl_utils/transferAgent.h"
 #include "tensorrt_llm/common/envUtils.h"
 #include "tensorrt_llm/common/logger.h"
+#include "tensorrt_llm/common/nvtxUtils.h"
 #include "tensorrt_llm/executor/transferAgent.h"
 #include "tensorrt_llm/runtime/utils/mpiUtils.h"
 
@@ -789,6 +790,7 @@ void NixlTransferAgent::invalidateRemoteAgent(std::string const& name)
     // Set TRTLLM_NIXL_ENABLE_COALESCE=1 to enable this optimization
     if (common::getEnvNixlEnableCoalesce())
     {
+        NVTX3_SCOPED_RANGE(coalesceTransferDescs_CreateXferReq);
         auto [coalescedSrc, coalescedDst] = NixlHelper::coalesceTransferDescs(splitSrc, splitDst);
         status
             = mRawAgent->createXferReq(NixlHelper::convert(request.getOp()), NixlHelper::convertXferDist(coalescedSrc),
@@ -796,6 +798,7 @@ void NixlTransferAgent::invalidateRemoteAgent(std::string const& name)
     }
     else
     {
+        NVTX3_SCOPED_RANGE(createXferReq);
         status = mRawAgent->createXferReq(NixlHelper::convert(request.getOp()), NixlHelper::convertXferDist(splitSrc),
             NixlHelper::convertXferDist(splitDst), request.getRemoteName(), handle, &mExtraParams);
     }
@@ -804,8 +807,10 @@ void NixlTransferAgent::invalidateRemoteAgent(std::string const& name)
         " rank: %d createXferReq failed with status: %s selfname: %s remoteAgent name: %s",
         mpi::MpiComm::world().getRank(), nixlEnumStrings::statusStr(status).c_str(), mName.c_str(),
         request.getRemoteName().c_str());
-
-    status = mRawAgent->postXferReq(handle, &mExtraParams);
+    {
+        NVTX3_SCOPED_RANGE(postXferReq);
+        status = mRawAgent->postXferReq(handle, &mExtraParams);
+    }
     return std::make_unique<NixlTransferStatus>(mRawAgent.get(), handle);
 }
 
@@ -932,6 +937,7 @@ MemoryDescs NixlTransferAgent::splitDescsFromRegistry(MemoryDescs const& descs)
 std::pair<MemoryDescs, MemoryDescs> NixlTransferAgent::splitTransferDescsFromRegistry(
     MemoryDescs const& srcDescs, MemoryDescs const& dstDescs) const
 {
+    NVTX3_SCOPED_RANGE(splitTransferDescsFromRegistry);
     if (srcDescs.getType() != MemoryType::kVRAM)
         return {srcDescs, dstDescs};
 
diff --git a/tensorrt_llm/_torch/disaggregation/base/agent.py b/tensorrt_llm/_torch/disaggregation/base/agent.py
index bf4899707e03..1aac24d98e75 100644
--- a/tensorrt_llm/_torch/disaggregation/base/agent.py
+++ b/tensorrt_llm/_torch/disaggregation/base/agent.py
@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from typing import List, NamedTuple, Optional, Tuple
 
+import numpy as np
+
 from tensorrt_llm import logger
 
 
@@ -27,10 +29,42 @@ class MemoryDesc(NamedTuple):
     device_id: int
 
 
-@dataclass
 class MemoryDescs:
-    type: str
-    descs: List[MemoryDesc]
+    """Describes a set of memory regions with a common type.
+
+    descs: List of (ptr, size, device_id) tuples.
+    """
+
+    __slots__ = ("type", "descs")
+
+    def __init__(self, type: str, descs: List[tuple[int, int, int]]):
+        self.type = type
+        self.descs = descs
+
+    @classmethod
+    def from_arrays(
+        cls, type: str, addrs: np.ndarray, sizes: np.ndarray, device_ids: np.ndarray
+    ) -> "MemoryDescs":
+        """Batch-construct from numpy arrays of addrs, sizes, device_ids.
+
+        Pure-Python fallback; the C++ binding overrides this with a version
+        that reads numpy raw pointers directly.
+        """
+        descs = np.stack([addrs, sizes, device_ids], axis=1).tolist()
+        return cls(type, [tuple(d) for d in descs])
+
+    @classmethod
+    def from_arrays_uniform_device(
+        cls, type: str, addrs: np.ndarray, sizes: np.ndarray, device_id: int
+    ) -> "MemoryDescs":
+        """Batch-construct from numpy arrays with a single device_id for all entries.
+
+        Pure-Python fallback; the C++ binding overrides this with a version
+        that reads numpy raw pointers directly.
+        """
+        dev_ids = np.full(addrs.size, device_id, dtype=np.int32)
+        descs = np.stack([addrs, sizes, dev_ids], axis=1).tolist()
+        return cls(type, [tuple(d) for d in descs])
 
 
 @dataclass
diff --git a/tensorrt_llm/_torch/disaggregation/base/region.py b/tensorrt_llm/_torch/disaggregation/base/region.py
index 83e10c6ad500..249bf742bc2b 100644
--- a/tensorrt_llm/_torch/disaggregation/base/region.py
+++ b/tensorrt_llm/_torch/disaggregation/base/region.py
@@ -3,6 +3,8 @@
 from enum import IntFlag, auto
 from typing import List, NamedTuple, Optional
 
+import numpy as np
+
 
 @dataclass(frozen=True)
 class IndexRange:
@@ -33,7 +35,7 @@ class MemRegion(NamedTuple):
 class MemRegionGroup(NamedTuple):
     """Describes a block of memory by starting pointer and size in bytes."""
 
-    ptrs: List[int]
+    ptrs: np.ndarray  # dtype=np.int64
     bytes_per_region: int
 
 
@@ -89,10 +91,10 @@ class RegionExtractorBase(ABC):
     """
 
     @abstractmethod
-    def extract(self, region_ids: Optional[List[int]] = None) -> List[SpecRegion]:
+    def extract(self, region_ids: Optional[np.ndarray] = None) -> List[SpecRegion]:
         """
         Args:
-            region_ids: (Optional) List of integer region identifiers to extract.
+            region_ids: (Optional) np.ndarray of integer region identifiers to extract.
         Returns:
             List of Regions for corresponding regions.
         """
diff --git a/tensorrt_llm/_torch/disaggregation/base/transfer.py b/tensorrt_llm/_torch/disaggregation/base/transfer.py
index 842530620cfa..8c8b6d60a0f1 100644
--- a/tensorrt_llm/_torch/disaggregation/base/transfer.py
+++ b/tensorrt_llm/_torch/disaggregation/base/transfer.py
@@ -6,6 +6,8 @@
 from enum import Enum
 from typing import List, Optional, cast
 
+import numpy as np
+
 from tensorrt_llm import DisaggregatedParams
 from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequest
 
@@ -46,9 +48,9 @@ class KVSlice:
 
     token_range: Optional[TokenRange] = None
     layer_range: Optional[LayerRange] = None
-    block_ids_per_layer_groups: List[List[int]] = field(
+    block_ids_per_layer_groups: List[np.ndarray] = field(
         default_factory=list
-    )  # Physical block IDs per layer group
+    )  # Physical block IDs per layer group, each np.ndarray(dtype=np.int64)
     is_last_slice: bool = False
 
 
diff --git a/tensorrt_llm/_torch/disaggregation/native/auxiliary.py b/tensorrt_llm/_torch/disaggregation/native/auxiliary.py
index 311131991fcc..a264b8d6c01b 100644
--- a/tensorrt_llm/_torch/disaggregation/native/auxiliary.py
+++ b/tensorrt_llm/_torch/disaggregation/native/auxiliary.py
@@ -3,6 +3,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
+import numpy as np
 import torch
 
 from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequest
@@ -10,25 +11,25 @@
 
 @dataclass
 class AuxBufferMeta:
-    ptrs: list[int]
-    size: list[int]
-    item_sizes: list[int] = field(default_factory=list)
+    ptrs: np.ndarray  # dtype=np.int64
+    size: np.ndarray  # dtype=np.int64
+    item_sizes: np.ndarray = field(default_factory=lambda: np.array([], dtype=np.int64))
     device: str = "cpu"
 
     def to_dict(self) -> dict[str, Any]:
         return {
-            "ptrs": self.ptrs,
-            "size": self.size,
-            "item_sizes": self.item_sizes,
+            "ptrs": self.ptrs.tolist(),
+            "size": self.size.tolist(),
+            "item_sizes": self.item_sizes.tolist(),
             "device": self.device,
         }
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> "AuxBufferMeta":
         return cls(
-            ptrs=data["ptrs"],
-            size=data["size"],
-            item_sizes=data.get("item_sizes", []),
+            ptrs=np.array(data["ptrs"], dtype=np.int64),
+            size=np.array(data["size"], dtype=np.int64),
+            item_sizes=np.array(data.get("item_sizes", []), dtype=np.int64),
             device=data.get("device", "cpu"),
         )
 
@@ -109,21 +110,30 @@ def __init__(self, max_slot_num: int, beam_width: int, max_draft_len: int, devic
         )
 
         self._meta = AuxBufferMeta(
-            ptrs=[
-                self._first_tokens_buffer.data_ptr(),
-                self._draft_tokens_buffer.data_ptr(),
-                self._token_counts_buffer.data_ptr(),
-            ],
-            size=[
-                self._first_tokens_buffer.numel() * self._first_tokens_buffer.element_size(),
-                self._draft_tokens_buffer.numel() * self._draft_tokens_buffer.element_size(),
-                self._token_counts_buffer.numel() * self._token_counts_buffer.element_size(),
-            ],
-            item_sizes=[
-                self._first_tokens_buffer[0].numel() * self._first_tokens_buffer.element_size(),
-                self._draft_tokens_buffer[0].numel() * self._draft_tokens_buffer.element_size(),
-                self._token_counts_buffer[0].numel() * self._token_counts_buffer.element_size(),
-            ],
+            ptrs=np.array(
+                [
+                    self._first_tokens_buffer.data_ptr(),
+                    self._draft_tokens_buffer.data_ptr(),
+                    self._token_counts_buffer.data_ptr(),
+                ],
+                dtype=np.int64,
+            ),
+            size=np.array(
+                [
+                    self._first_tokens_buffer.numel() * self._first_tokens_buffer.element_size(),
+                    self._draft_tokens_buffer.numel() * self._draft_tokens_buffer.element_size(),
+                    self._token_counts_buffer.numel() * self._token_counts_buffer.element_size(),
+                ],
+                dtype=np.int64,
+            ),
+            item_sizes=np.array(
+                [
+                    self._first_tokens_buffer[0].numel() * self._first_tokens_buffer.element_size(),
+                    self._draft_tokens_buffer[0].numel() * self._draft_tokens_buffer.element_size(),
+                    self._token_counts_buffer[0].numel() * self._token_counts_buffer.element_size(),
+                ],
+                dtype=np.int64,
+            ),
             device=self._device,
         )
 
diff --git a/tensorrt_llm/_torch/disaggregation/native/mixers/attention/peer.py b/tensorrt_llm/_torch/disaggregation/native/mixers/attention/peer.py
index a2f5049b9a3f..e9644d42a357 100644
--- a/tensorrt_llm/_torch/disaggregation/native/mixers/attention/peer.py
+++ b/tensorrt_llm/_torch/disaggregation/native/mixers/attention/peer.py
@@ -9,6 +9,7 @@
 )
 from tensorrt_llm._torch.disaggregation.native.rank_info import RankInfo
 from tensorrt_llm._torch.disaggregation.resource.utils import PoolRole
+from tensorrt_llm._utils import nvtx_range
 
 
 class IdentityMapper(RegionMapperBase):
@@ -23,21 +24,16 @@ class IdentityMapper(RegionMapperBase):
     dst_ptrs: [ D0 ] [ D1 ] [ D2 ] ...
     """
 
+    @nvtx_range("IdentityMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
-        assert len(src_group.ptrs) == len(dst_group.ptrs), (
-            f"Number of regions of src({len(src_group.ptrs)}) and dst({len(dst_group.ptrs)}) must match"
-        )
-        new_src = MemRegionGroup(
-            ptrs=list(src_group.ptrs), bytes_per_region=src_group.bytes_per_region
-        )
-        new_dst = MemRegionGroup(
-            ptrs=list(dst_group.ptrs), bytes_per_region=dst_group.bytes_per_region
+        assert src_group.ptrs.size == dst_group.ptrs.size, (
+            f"Number of regions of src({src_group.ptrs.size}) and dst({dst_group.ptrs.size}) must match"
         )
         return SpecRegionPair(
-            src=SpecRegion(memory=new_src, spec=src_regions.spec),
-            dst=SpecRegion(memory=new_dst, spec=dst_regions.spec),
+            src=SpecRegion(memory=src_group, spec=src_regions.spec),
+            dst=SpecRegion(memory=dst_group, spec=dst_regions.spec),
         )
 
 
@@ -94,14 +90,15 @@ def __init__(
             dst_layer_off, slot_size_per_layer=slot_size_per_layer
         )
 
+    @nvtx_range("HeadMatchMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
-        assert len(src_group.ptrs) == len(dst_group.ptrs), (
-            f"Number of regions of src({len(src_group.ptrs)}) and dst({len(dst_group.ptrs)}) must match"
+        assert src_group.ptrs.size == dst_group.ptrs.size, (
+            f"Number of regions of src({src_group.ptrs.size}) and dst({dst_group.ptrs.size}) must match"
         )
-        new_src_ptrs = [src_ptr + self._src_block_off for src_ptr in src_group.ptrs]
-        new_dst_ptrs = [dst_ptr + self._dst_block_off for dst_ptr in dst_group.ptrs]
+        new_src_ptrs = src_group.ptrs + self._src_block_off
+        new_dst_ptrs = dst_group.ptrs + self._dst_block_off
         new_src = MemRegionGroup(ptrs=new_src_ptrs, bytes_per_region=self._frag_size)
         new_dst = MemRegionGroup(ptrs=new_dst_ptrs, bytes_per_region=self._frag_size)
         return SpecRegionPair(
@@ -145,7 +142,6 @@ def __init__(
     ):
         self._ri = self_ri
         self._peer_ri = peer_ri
-        self._src_layer_off = src_layer_off
 
         kv_factor = self_ri.attention.kv_factor
         self_tp_per_dp = self_ri.tp_size_per_dp_group
@@ -170,36 +166,68 @@ def __init__(
             peer_tp_rank,
             self._bytes_cont_heads,
         )
-        self._layer_indices = np.arange(transfer_layers, dtype=np.int64)
-        self._kv_indices = np.arange(kv_factor, dtype=np.int64)
         self._peer_layer_off = peer_layer_off
 
+        # --- Pre-compute flat 1D offset arrays ---
+        #
+        # Each KV cache block (slot) is laid out as:
+        #
+        #   block_base ──► [layer_0 kv_0] [layer_0 kv_1] [layer_1 kv_0] [layer_1 kv_1] ...
+        #                   ◄─ layer_kv ─► ◄─ layer_kv ─►
+        #                   ◄────── layer_num (= layer_kv * kv_factor) ──────►
+        #
+        # To address fragment (layer=j, kv=k) within a block at base_ptr:
+        #
+        #   frag_ptr = base_ptr
+        #            + layer_num * (layer_off + j)    # skip to the right layer
+        #            + layer_kv  * k                  # skip to key or value
+        #            + head_off                       # head offset for TP mismatch
+        #
+        # The original code computed this as a 3D broadcast in map():
+        #   bases[:, None, None] + layer_offsets[None, :, None]
+        #                        + kv_offsets[None, None, :] + head_off
+        # producing shape (n_blocks, transfer_layers, kv_factor) then .ravel().
+        #
+        # Optimization: since the (layer, kv) offsets are independent of the
+        # per-call block base pointers, we pre-compute them here as a flat 1D
+        # array of length (transfer_layers * kv_factor).  At map() time we only
+        # need np.add.outer(bases, flat_offsets).ravel(), which produces the
+        # same result in the same C-order traversal (blocks outer, offsets inner)
+        # but with fewer intermediate allocations.
+        layer_indices = np.arange(transfer_layers, dtype=np.int64)
+        kv_indices = np.arange(kv_factor, dtype=np.int64)
+
+        src_layer_kv_num = self._get_layer_kv_num(self._ri)
+        src_layer_num = src_layer_kv_num * kv_factor
+        # Shape (transfer_layers, kv_factor) → ravel to 1D
+        self._src_flat_offsets = (
+            src_layer_num * (src_layer_off + layer_indices)[:, None]
+            + src_layer_kv_num * kv_indices[None, :]
+            + self._src_head_off
+        ).ravel()
+
+        dst_layer_kv_num = self._get_layer_kv_num(self._peer_ri)
+        dst_layer_num = dst_layer_kv_num * kv_factor
+        self._dst_flat_offsets = (
+            dst_layer_num * (peer_layer_off + layer_indices)[:, None]
+            + dst_layer_kv_num * kv_indices[None, :]
+            + self._dst_head_off
+        ).ravel()
+
+    @nvtx_range("HeadMismatchMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
-        assert len(src_group.ptrs) == len(dst_group.ptrs), (
-            f"Number of regions of src({len(src_group.ptrs)}) and dst({len(dst_group.ptrs)}) must match"
-        )
-        src_bases = np.array(src_group.ptrs, dtype=np.int64)
-        dst_bases = np.array(dst_group.ptrs, dtype=np.int64)
-        src_frags = self._get_frags(
-            bases=src_bases,
-            layer_indices=self._src_layer_off + self._layer_indices,
-            layer_kv_num=self._get_layer_kv_num(self._ri),
-            kv_indices=self._kv_indices,
-            head_off=self._src_head_off,
-            kv_factor=self._kv_indices.size,
+        assert src_group.ptrs.size == dst_group.ptrs.size, (
+            f"Number of regions of src({src_group.ptrs.size}) and dst({dst_group.ptrs.size}) must match"
         )
-        dst_frags = self._get_frags(
-            bases=dst_bases,
-            layer_indices=self._peer_layer_off + self._layer_indices,
-            layer_kv_num=self._get_layer_kv_num(self._peer_ri),
-            kv_indices=self._kv_indices,
-            head_off=self._dst_head_off,
-            kv_factor=self._kv_indices.size,
-        )
-        all_src_ptrs = [int(x) for x in src_frags.flatten()]
-        all_dst_ptrs = [int(x) for x in dst_frags.flatten()]
+        # np.add.outer(ptrs, offsets) produces every (base + offset) combination:
+        #   shape (n_blocks, transfer_layers * kv_factor)
+        # .ravel() flattens in C-order: for each block, emit all layer×kv fragments.
+        # This is equivalent to the original 3D broadcast + ravel, but the per-(layer,kv)
+        # offsets are pre-computed in __init__ so map() does a single vectorized add.
+        all_src_ptrs = np.add.outer(src_group.ptrs, self._src_flat_offsets).ravel()
+        all_dst_ptrs = np.add.outer(dst_group.ptrs, self._dst_flat_offsets).ravel()
         new_src = MemRegionGroup(ptrs=all_src_ptrs, bytes_per_region=self._bytes_cont_heads)
         new_dst = MemRegionGroup(ptrs=all_dst_ptrs, bytes_per_region=self._bytes_cont_heads)
         return SpecRegionPair(
@@ -232,16 +260,6 @@ def _get_layer_kv_num(ri: RankInfo) -> int:
             * ri.attention.element_bytes
         )
 
-    @staticmethod
-    def _get_frags(bases, layer_indices, layer_kv_num, kv_indices, head_off, kv_factor):
-        layer_num = layer_kv_num * kv_factor
-        return (
-            bases[:, None, None]
-            + layer_num * layer_indices[None, :, None]
-            + layer_kv_num * kv_indices[None, None, :]
-            + head_off
-        )
-
 
 class IndexerKCacheHeadMatchMapper(RegionMapperBase):
     """
@@ -269,14 +287,15 @@ def __init__(
         self._src_block_off = block_size_per_layer * src_layer_off
         self._dst_block_off = block_size_per_layer * dst_layer_off
 
+    @nvtx_range("IndexerKCacheHeadMatchMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
-        assert len(src_group.ptrs) == len(dst_group.ptrs), (
-            f"Number of regions of src({len(src_group.ptrs)}) and dst({len(dst_group.ptrs)}) must match"
+        assert src_group.ptrs.size == dst_group.ptrs.size, (
+            f"Number of regions of src({src_group.ptrs.size}) and dst({dst_group.ptrs.size}) must match"
         )
-        new_src_ptrs = [src_ptr + self._src_block_off for src_ptr in src_group.ptrs]
-        new_dst_ptrs = [dst_ptr + self._dst_block_off for dst_ptr in dst_group.ptrs]
+        new_src_ptrs = src_group.ptrs + self._src_block_off
+        new_dst_ptrs = dst_group.ptrs + self._dst_block_off
         new_src = MemRegionGroup(ptrs=new_src_ptrs, bytes_per_region=self._frag_size)
         new_dst = MemRegionGroup(ptrs=new_dst_ptrs, bytes_per_region=self._frag_size)
         return SpecRegionPair(
diff --git a/tensorrt_llm/_torch/disaggregation/native/mixers/ssm/peer.py b/tensorrt_llm/_torch/disaggregation/native/mixers/ssm/peer.py
index 5fa3ec8d01c8..1e59cf675884 100644
--- a/tensorrt_llm/_torch/disaggregation/native/mixers/ssm/peer.py
+++ b/tensorrt_llm/_torch/disaggregation/native/mixers/ssm/peer.py
@@ -1,11 +1,14 @@
 from typing import List
 
+import numpy as np
+
 from tensorrt_llm._torch.disaggregation.base.region import (
     MemRegionGroup,
     RegionMapperBase,
     SpecRegion,
     SpecRegionPair,
 )
+from tensorrt_llm._utils import nvtx_range
 
 
 class MambaHeadMatchMapper(RegionMapperBase):
@@ -34,6 +37,7 @@ def __init__(
         self._dst_layer_off = dst_layer_off
         self._block_bytes = block_bytes_per_layer
 
+    @nvtx_range("MambaHeadMatchMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
@@ -42,12 +46,12 @@ def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPai
         src_ptrs = src_group.ptrs[self._src_layer_off : self._src_layer_off + self._transfer_layers]
         dst_ptrs = dst_group.ptrs[self._dst_layer_off : self._dst_layer_off + self._transfer_layers]
 
-        assert len(src_ptrs) == len(dst_ptrs), (
-            f"Number of regions of src({len(src_ptrs)}) and dst({len(dst_ptrs)}) must match"
+        assert src_ptrs.size == dst_ptrs.size, (
+            f"Number of regions of src({src_ptrs.size}) and dst({dst_ptrs.size}) must match"
         )
 
-        new_src = MemRegionGroup(ptrs=list(src_ptrs), bytes_per_region=self._block_bytes)
-        new_dst = MemRegionGroup(ptrs=list(dst_ptrs), bytes_per_region=self._block_bytes)
+        new_src = MemRegionGroup(ptrs=src_ptrs, bytes_per_region=self._block_bytes)
+        new_dst = MemRegionGroup(ptrs=dst_ptrs, bytes_per_region=self._block_bytes)
         return SpecRegionPair(
             src=SpecRegion(memory=new_src, spec=src_regions.spec),
             dst=SpecRegion(memory=new_dst, spec=dst_regions.spec),
@@ -103,6 +107,7 @@ def __init__(
             self._bytes_cont_heads,
         )
 
+    @nvtx_range("MambaHeadMismatchMapper.map")
     def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPair:
         src_group = src_regions.memory
         dst_group = dst_regions.memory
@@ -114,14 +119,14 @@ def map(self, src_regions: SpecRegion, dst_regions: SpecRegion) -> SpecRegionPai
         dst_layer_ptrs = dst_group.ptrs[
             self._dst_layer_off : self._dst_layer_off + self._transfer_layers
         ]
-        if len(src_layer_ptrs) != len(dst_layer_ptrs):
+        if src_layer_ptrs.size != dst_layer_ptrs.size:
             raise ValueError(
-                f"Number of layer ptrs mismatch: src={len(src_layer_ptrs)}, dst={len(dst_layer_ptrs)}"
+                f"Number of layer ptrs mismatch: src={src_layer_ptrs.size}, dst={dst_layer_ptrs.size}"
             )
 
-        # Apply head offset to each layer's address
-        new_src_ptrs = [ptr + self._src_head_off for ptr in src_layer_ptrs]
-        new_dst_ptrs = [ptr + self._dst_head_off for ptr in dst_layer_ptrs]
+        # Apply head offset to each layer's address (vectorized)
+        new_src_ptrs = src_layer_ptrs + self._src_head_off
+        new_dst_ptrs = dst_layer_ptrs + self._dst_head_off
 
         new_src = MemRegionGroup(ptrs=new_src_ptrs, bytes_per_region=self._bytes_cont_heads)
         new_dst = MemRegionGroup(ptrs=new_dst_ptrs, bytes_per_region=self._bytes_cont_heads)
@@ -180,7 +185,11 @@ def __init__(
             self_tp_rank,
             peer_tp_rank,
         )
+        # Pre-compute offset arrays for vectorized map()
+        self._section_src_offs = np.array([p[0] for p in self._section_plans], dtype=np.int64)
+        self._section_dst_offs = np.array([p[1] for p in self._section_plans], dtype=np.int64)
 
+    @nvtx_range("ConvStateMismatchMapper.map")
     def map(
         self,
         src_regions: SpecRegion,
@@ -198,26 +207,33 @@ def map(
             self._dst_layer_off : self._dst_layer_off + self._transfer_layers
         ]
 
-        assert len(src_layer_ptrs) == len(dst_layer_ptrs), (
-            f"Number of layer ptrs mismatch: src={len(src_layer_ptrs)}, dst={len(dst_layer_ptrs)}"
+        assert src_layer_ptrs.size == dst_layer_ptrs.size, (
+            f"Number of layer ptrs mismatch: src={src_layer_ptrs.size}, dst={dst_layer_ptrs.size}"
         )
 
+        # Vectorized: broadcast all section offsets at once
+        # _section_offsets shape: (num_sections, 3) with (src_off, dst_off, transfer_bytes)
+        src_offs = self._section_src_offs  # (num_sections,)
+        dst_offs = self._section_dst_offs  # (num_sections,)
+
+        # (num_sections, num_layers) = (num_sections, 1) + (1, num_layers)
+        all_src_ptrs = src_offs[:, None] + src_layer_ptrs[None, :]
+        all_dst_ptrs = dst_offs[:, None] + dst_layer_ptrs[None, :]
+
         results: List[SpecRegionPair] = []
-        for src_off, dst_off, transfer_bytes in self._section_plans:
-            sec_src_ptrs = [ptr + src_off for ptr in src_layer_ptrs]
-            sec_dst_ptrs = [ptr + dst_off for ptr in dst_layer_ptrs]
+        for i, (_, _, transfer_bytes) in enumerate(self._section_plans):
             results.append(
                 SpecRegionPair(
                     src=SpecRegion(
                         memory=MemRegionGroup(
-                            ptrs=sec_src_ptrs,
+                            ptrs=all_src_ptrs[i],
                             bytes_per_region=transfer_bytes,
                         ),
                         spec=src_regions.spec,
                     ),
                     dst=SpecRegion(
                         memory=MemRegionGroup(
-                            ptrs=sec_dst_ptrs,
+                            ptrs=all_dst_ptrs[i],
                             bytes_per_region=transfer_bytes,
                         ),
                         spec=dst_regions.spec,
diff --git a/tensorrt_llm/_torch/disaggregation/native/transfer.py b/tensorrt_llm/_torch/disaggregation/native/transfer.py
index 31c9d63cc61e..7292b0a800c3 100644
--- a/tensorrt_llm/_torch/disaggregation/native/transfer.py
+++ b/tensorrt_llm/_torch/disaggregation/native/transfer.py
@@ -6,11 +6,12 @@
 import threading
 import time
 import weakref
-from dataclasses import asdict, dataclass
+from dataclasses import dataclass
 from enum import Enum
 from typing import List, Optional
 
 import msgpack
+import numpy as np
 import torch
 
 try:
@@ -22,7 +23,6 @@
 from tensorrt_llm import logger
 from tensorrt_llm._torch.disaggregation.base.agent import (
     BaseTransferAgent,
-    MemoryDesc,
     MemoryDescs,
     MemoryType,
     RegMemoryDescs,
@@ -66,17 +66,35 @@ class RecvReqInfo:
     sender_req_id: int
     instance_name: str
     instance_rank: int
-    block_ids_per_layer_groups: list[list[int]]
+    block_ids_per_layer_groups: list[
+        np.ndarray
+    ]  # Block IDs per layer group, each np.ndarray(dtype=np.int64)
     unique_rid: int
     start_token_idx: Optional[int] = None
     aux_slot: Optional[int] = None
 
     def to_bytes(self) -> bytes:
-        return msgpack.packb(asdict(self))
+        return msgpack.packb(
+            {
+                "sender_req_id": self.sender_req_id,
+                "instance_name": self.instance_name,
+                "instance_rank": self.instance_rank,
+                "block_ids_per_layer_groups": [
+                    arr.tobytes() for arr in self.block_ids_per_layer_groups
+                ],
+                "unique_rid": self.unique_rid,
+                "start_token_idx": self.start_token_idx,
+                "aux_slot": self.aux_slot,
+            }
+        )
 
     @classmethod
     def from_bytes(cls, data: bytes) -> "RecvReqInfo":
-        return cls(**msgpack.unpackb(data, raw=False))
+        d = msgpack.unpackb(data, raw=False)
+        d["block_ids_per_layer_groups"] = [
+            np.frombuffer(b, dtype=np.int64).copy() for b in d["block_ids_per_layer_groups"]
+        ]
+        return cls(**d)
 
 
 @dataclass
@@ -99,9 +117,9 @@ class WriteMeta:
     peer_rank: int
     peer_endpoint: str
     unique_rid: int
-    src_ptrs: List[int]
-    dst_ptrs: List[int]
-    sizes: List[int]
+    src_ptrs: np.ndarray  # dtype=np.int64
+    dst_ptrs: np.ndarray  # dtype=np.int64
+    sizes: np.ndarray  # dtype=np.int64
     dst_device_id: Optional[int] = None
     slice_id: Optional[int] = None
     is_last_slice: bool = False
@@ -299,13 +317,14 @@ def _process_task_queue(self, thread_idx: int):
     @staticmethod
     @nvtx_range("_make_agent_request")
     def _make_agent_request(write_meta: WriteMeta, device_id: int) -> "TransferRequest":
-        if not (len(write_meta.src_ptrs) == len(write_meta.dst_ptrs) == len(write_meta.sizes)):
+        if not (write_meta.src_ptrs.size == write_meta.dst_ptrs.size == write_meta.sizes.size):
             raise ValueError(
                 f"Pointer/size mismatch for unique_rid={write_meta.unique_rid}: "
-                f"{len(write_meta.src_ptrs)=}, "
-                f"{len(write_meta.dst_ptrs)=}, "
-                f"{len(write_meta.sizes)=}"
+                f"{write_meta.src_ptrs.size=}, "
+                f"{write_meta.dst_ptrs.size=}, "
+                f"{write_meta.sizes.size=}"
             )
+        n = write_meta.src_ptrs.size
         if write_meta.meta_type == WriteMetaType.AUX:
             src_dev, dst_dev, mem_type = 0, 0, MemoryType.DRAM
         else:
@@ -316,25 +335,27 @@ def _make_agent_request(write_meta: WriteMeta, device_id: int) -> "TransferReque
                 )
             src_dev, dst_dev, mem_type = device_id, write_meta.dst_device_id, MemoryType.VRAM
 
-        src_list = [
-            MemoryDesc(ptr, size, src_dev)
-            for ptr, size in zip(write_meta.src_ptrs, write_meta.sizes, strict=True)
-        ]
-        dst_list = [
-            MemoryDesc(ptr, size, dst_dev)
-            for ptr, size in zip(write_meta.dst_ptrs, write_meta.sizes, strict=True)
-        ]
+        if n == 0:
+            src_memory_descs = MemoryDescs(mem_type, [])
+            dst_memory_descs = MemoryDescs(mem_type, [])
+        else:
+            src_memory_descs = MemoryDescs.from_arrays_uniform_device(
+                mem_type, write_meta.src_ptrs, write_meta.sizes, src_dev
+            )
+            dst_memory_descs = MemoryDescs.from_arrays_uniform_device(
+                mem_type, write_meta.dst_ptrs, write_meta.sizes, dst_dev
+            )
+
+        # NOTE: TransferRequest moves (not copies) src/dst MemoryDescs internally.
+        # After this call, src_memory_descs and dst_memory_descs are in a moved-from
+        # state and must NOT be accessed again.
         return TransferRequest(
-            TransferOp.WRITE,  # type: ignore[arg-type]
-            MemoryDescs(mem_type, src_list),
-            MemoryDescs(mem_type, dst_list),
-            write_meta.peer_name,
-            None,
+            TransferOp.WRITE, src_memory_descs, dst_memory_descs, write_meta.peer_name, None
         )
 
     @nvtx_range("_deliver_kv_to_agent")
     def _deliver_kv_to_agent(self, write_meta: WriteMeta):
-        assert len(write_meta.src_ptrs) == len(write_meta.dst_ptrs) == len(write_meta.sizes), (
+        assert write_meta.src_ptrs.size == write_meta.dst_ptrs.size == write_meta.sizes.size, (
             f"WriteMeta ptr/size mismatch for unique_rid={write_meta.unique_rid}"
         )
 
@@ -360,7 +381,7 @@ def _deliver_kv_to_agent(self, write_meta: WriteMeta):
         task.status = TaskStatus.TRANSFERRING
 
         agent_result = AgentResult.SUCCESS
-        if write_meta.src_ptrs:
+        if write_meta.src_ptrs.size > 0:
             request = Sender._make_agent_request(write_meta, device_id=self._device_id)
             if timer:
                 timer.record_transfer_start(write_meta.peer_rank)
@@ -426,7 +447,7 @@ def _deliver_aux_to_agent(self, write_meta: WriteMeta):
             timer.record_push_end(write_meta.peer_rank)
 
         agent_result = AgentResult.SUCCESS
-        if write_meta.src_ptrs:
+        if write_meta.src_ptrs.size > 0:
             request = Sender._make_agent_request(write_meta, device_id=self._device_id)
             if timer:
                 timer.record_transfer_start(write_meta.peer_rank)
@@ -463,7 +484,9 @@ def _deliver_aux_to_agent(self, write_meta: WriteMeta):
             )
 
     @staticmethod
-    def _filter_kv_blocks(src_block_ids, dst_block_ids) -> tuple[list[int], list[int]]:
+    def _filter_kv_blocks(
+        src_block_ids: np.ndarray, dst_block_ids: np.ndarray
+    ) -> tuple[np.ndarray, np.ndarray]:
         # TODO: filter the kv block_ids according to the peer_overlap
         return src_block_ids, dst_block_ids
 
@@ -476,9 +499,16 @@ def _build_kv_write_meta(self, task: KVSendTask, req_info: RecvReqInfo) -> Write
         targets = self._registrar.get_peer_overlap(peer_ri, peer_ri.dp_rank)
         expected_transfers = len(targets.ranks)
 
-        src_frags: List[int] = []
-        dst_frags: List[int] = []
-        kv_sizes: List[int] = []
+        # Aggregate fragment pointers from all matching pool pairs.
+        # Each pool pair produces one or more region pairs (rp), each containing
+        # a numpy array of src/dst pointers and a uniform bytes_per_region.
+        src_frag_parts: list[np.ndarray] = []
+        dst_frag_parts: list[np.ndarray] = []
+        # Instead of calling np.full() per region pair to build a size array and
+        # then np.concatenate() all of them, we record (count, bytes_per_region)
+        # tuples and construct the final sizes array with a single np.repeat().
+        # For 48k+ items this avoids many small allocations in the hot loop.
+        size_specs: list[tuple[int, int]] = []
         dst_device_id = None
         if self._registrar.should_send_kv(targets, peer_ri):
             dst_device_id = peer_ri.device_id
@@ -490,11 +520,12 @@ def _build_kv_write_meta(self, task: KVSendTask, req_info: RecvReqInfo) -> Write
             dst_block_ids_per_groups = req_info.block_ids_per_layer_groups
             src_block_ids_per_groups = task._slice.block_ids_per_layer_groups
 
+            # Aggregate fragments from all matching pools using numpy concatenation
             for (self_lg, self_pi), (peer_lg, peer_pi) in pool_mapping.items():
                 src_block_ids = src_block_ids_per_groups[self_lg]
                 dst_block_ids = dst_block_ids_per_groups[peer_lg]
 
-                if len(src_block_ids) + 1 == len(dst_block_ids):
+                if src_block_ids.size + 1 == dst_block_ids.size:
                     # FIXME: this is a temporary solution, need to be fixed for the draft tokens
                     logger.warning(
                         "src_block_num is one less than dst_block_num, maybe it is due to draft tokens,"
@@ -515,14 +546,27 @@ def _build_kv_write_meta(self, task: KVSendTask, req_info: RecvReqInfo) -> Write
                 region_pair = mapper.map(src_region, dst_region)
                 region_pairs = region_pair if isinstance(region_pair, list) else [region_pair]
                 for rp in region_pairs:
-                    src_frags.extend(rp.src.memory.ptrs)  # type: ignore[attr-defined]
-                    dst_frags.extend(rp.dst.memory.ptrs)  # type: ignore[attr-defined]
-                    frag_size = rp.src.memory.bytes_per_region  # type: ignore[attr-defined]
-                    kv_sizes.extend([frag_size] * len(rp.src.memory.ptrs))  # type: ignore[attr-defined]
+                    src_frag_parts.append(rp.src.memory.ptrs)
+                    dst_frag_parts.append(rp.dst.memory.ptrs)
+                    size_specs.append((rp.src.memory.ptrs.size, rp.src.memory.bytes_per_region))
+
+        if src_frag_parts:
+            src_frags = np.concatenate(src_frag_parts)
+            dst_frags = np.concatenate(dst_frag_parts)
+            # Build the kv_sizes array in one shot: np.repeat expands each
+            # bytes_per_region value by its count, e.g.:
+            #   values=[4096, 8192], counts=[100, 200]
+            #   → [4096]*100 ++ [8192]*200
+            counts, values = zip(*size_specs)
+            kv_sizes = np.repeat(np.array(values, dtype=np.int64), counts)
+        else:
+            src_frags = np.array([], dtype=np.int64)
+            dst_frags = np.array([], dtype=np.int64)
+            kv_sizes = np.array([], dtype=np.int64)
 
         if timer:
             timer.record_prepare_args_end(peer_ri.instance_rank)
-            timer.record_transfer_sizes(peer_ri.instance_rank, sum(kv_sizes), len(dst_frags))
+            timer.record_transfer_sizes(peer_ri.instance_rank, int(kv_sizes.sum()), dst_frags.size)
 
         return WriteMeta(
             task_future=task.future,
@@ -546,7 +590,9 @@ def _build_aux_write_meta(self, task: AuxSendTask, req_info: RecvReqInfo) -> Wri
             timer.record_prepare_args_start(peer_ri.instance_rank)
         expected_transfers = len(self._registrar.get_peer_overlap(peer_ri, peer_ri.dp_rank).ranks)
 
-        src_ptrs, dst_ptrs, sizes = [], [], []
+        src_ptrs = np.array([], dtype=np.int64)
+        dst_ptrs = np.array([], dtype=np.int64)
+        sizes = np.array([], dtype=np.int64)
         if self._registrar.should_send_aux(peer_ri):
             src_aux_meta = self._registrar.self_rank_info.aux_meta
             peer_aux_meta = peer_ri.aux_meta
@@ -555,19 +601,15 @@ def _build_aux_write_meta(self, task: AuxSendTask, req_info: RecvReqInfo) -> Wri
             peer_slot = req_info.aux_slot
             assert peer_slot is not None, f"aux_slot is None for request {req_info.unique_rid}"
             assert task._slot is not None
-            src_ptrs = [
-                ptr + item_size * task._slot
-                for ptr, item_size in zip(src_aux_meta.ptrs, src_aux_meta.item_sizes)
-            ]
-            dst_ptrs = [
-                ptr + item_size * peer_slot
-                for ptr, item_size in zip(peer_aux_meta.ptrs, peer_aux_meta.item_sizes)
-            ]
-            sizes = list(src_aux_meta.item_sizes)
+            src_ptrs = src_aux_meta.ptrs + src_aux_meta.item_sizes * task._slot
+            dst_ptrs = peer_aux_meta.ptrs + peer_aux_meta.item_sizes * peer_slot
+            sizes = src_aux_meta.item_sizes.astype(np.int64, copy=False)
 
         if timer:
             timer.record_prepare_args_end(peer_ri.instance_rank)
-            timer.record_transfer_sizes(peer_ri.instance_rank, sum(sizes), len(src_ptrs))
+            timer.record_transfer_sizes(
+                peer_ri.instance_rank, int(sizes.sum()) if sizes.size > 0 else 0, src_ptrs.size
+            )
 
         return WriteMeta(
             task_future=task.future,
diff --git a/tensorrt_llm/_torch/disaggregation/resource/kv_extractor.py b/tensorrt_llm/_torch/disaggregation/resource/kv_extractor.py
index 3d062d88a626..447a820be2a0 100644
--- a/tensorrt_llm/_torch/disaggregation/resource/kv_extractor.py
+++ b/tensorrt_llm/_torch/disaggregation/resource/kv_extractor.py
@@ -20,7 +20,7 @@
 )
 from tensorrt_llm._torch.disaggregation.resource.utils import get_physical_pool
 from tensorrt_llm._torch.pyexecutor.resource_manager import KVCacheManager
-from tensorrt_llm._utils import get_size_in_bytes
+from tensorrt_llm._utils import get_size_in_bytes, nvtx_range
 from tensorrt_llm.bindings import DataType
 
 
@@ -44,9 +44,10 @@ def __init__(self, kv_arg):
     def page_table(self) -> KVCachePageTable:
         return self._page_table
 
+    @nvtx_range("KVRegionExtractorV1.extract")
     def extract(
         self,
-        region_ids: List[int],
+        region_ids: np.ndarray,
         layer_group_id: int = 0,
         pool_idx: int = 0,
     ) -> SpecRegion:
@@ -71,7 +72,8 @@ def extract(
         block_size = pool.slot_bytes
 
         # KV cache: filter out invalid block_ids (BAD_PAGE_INDEX = -1)
-        ptrs = [base_ptr + block_size * int(bid) for bid in region_ids if bid >= 0]
+        valid = region_ids >= 0
+        ptrs = base_ptr + block_size * region_ids[valid]
         memory = MemRegionGroup(ptrs=ptrs, bytes_per_region=block_size)
         return SpecRegion(memory=memory)
 
diff --git a/tensorrt_llm/_torch/disaggregation/transceiver.py b/tensorrt_llm/_torch/disaggregation/transceiver.py
index d0df89bbac39..09c9932a1b11 100644
--- a/tensorrt_llm/_torch/disaggregation/transceiver.py
+++ b/tensorrt_llm/_torch/disaggregation/transceiver.py
@@ -3,6 +3,7 @@
 from itertools import chain
 from typing import Any, Callable, Dict, List, Optional, cast
 
+import numpy as np
 import torch
 
 from tensorrt_llm import logger
@@ -19,6 +20,7 @@
 from tensorrt_llm._torch.pyexecutor.kv_cache_transceiver import KvCacheTransceiver
 from tensorrt_llm._torch.pyexecutor.llm_request import LlmRequest
 from tensorrt_llm._torch.pyexecutor.resource_manager import KVCacheManager, KVCacheManagerV2
+from tensorrt_llm._utils import nvtx_range
 from tensorrt_llm.bindings import LlmRequestState
 from tensorrt_llm.bindings.executor import ContextPhaseParams
 from tensorrt_llm.disaggregated_params import DisaggScheduleStyle
@@ -134,19 +136,24 @@ def shutdown(self):
         self._recv_reqs.clear()
         self._transfer_worker.shutdown()
 
-    def _get_block_ids(self, req: LlmRequest, group_idx: int, lg) -> list:
+    def _get_block_ids(self, req: LlmRequest, group_idx: int, lg) -> np.ndarray:
         if self._is_v2_manager:
             kv_cache_map = getattr(self._kv_cache_manager, "kv_cache_map")
-            return list(
+            # Returns Iterator[int], consume directly into ndarray
+            return np.fromiter(
                 kv_cache_map[req.py_request_id].get_aggregated_page_indices(
                     group_idx, valid_only=True
-                )
+                ),
+                dtype=np.int64,
             )
         else:
             first_layer = get_global_layer_ids(lg)[0]
-            return self._kv_cache_manager.get_batch_cache_indices(
-                [req.py_request_id], layer_idx=first_layer
-            )[0]
+            return np.asarray(
+                self._kv_cache_manager.get_batch_cache_indices(
+                    [req.py_request_id], layer_idx=first_layer
+                )[0],
+                dtype=np.int64,
+            )
 
     def _create_kv_slice(self, req: LlmRequest) -> KVSlice:
         tpb = self._kv_cache_manager.tokens_per_block
@@ -169,11 +176,11 @@ def _create_kv_slice(self, req: LlmRequest) -> KVSlice:
                 stale_end = max(0, (req.prompt_len + 1 - window_size) // tpb)
                 expected_valid = total_blocks - stale_end
                 if expected_valid <= 0:
-                    block_ids = []
-                elif len(block_ids) > expected_valid:
+                    block_ids = np.array([], dtype=np.int64)
+                elif block_ids.size > expected_valid:
                     block_ids = block_ids[-expected_valid:]
 
-            groups.append(list(block_ids))
+            groups.append(block_ids)
 
         return KVSlice(is_last_slice=True, block_ids_per_layer_groups=groups)
 
@@ -243,6 +250,7 @@ def _apply_aux(self, session, req: LlmRequest):
             req.context_phase_params.first_gen_tokens = first_gen_tokens
             req.context_phase_params.draft_tokens = draft_tokens
 
+    @nvtx_range("KvCacheTransceiverV2.respond_and_send_async")
     def respond_and_send_async(self, req: LlmRequest):
         rid = get_unique_rid(req)
         assert rid is not None
@@ -267,6 +275,7 @@ def respond_and_send_async(self, req: LlmRequest):
     def request_and_receive_sync(self, req: LlmRequest):
         raise NotImplementedError("request_and_receive_sync is not implemented")
 
+    @nvtx_range("KvCacheTransceiverV2.request_and_receive_async")
     def request_and_receive_async(self, req: LlmRequest):
         rid = get_unique_rid(req)
         if rid in self._recv_sessions:
diff --git a/tests/unittest/disaggregated/region/test_block.py b/tests/unittest/disaggregated/region/test_block.py
index bf5f02f55104..cc6cc3323608 100644
--- a/tests/unittest/disaggregated/region/test_block.py
+++ b/tests/unittest/disaggregated/region/test_block.py
@@ -1,3 +1,5 @@
+import numpy as np
+
 from tensorrt_llm._torch.disaggregation.base.region import (
     MemRegionGroup,
     SpecRegion,
@@ -57,38 +59,38 @@ def make_rankinfo(
 
 
 def test_mem_region_group():
-    ptrs = [11, 22, 33]
+    ptrs = np.array([11, 22, 33], dtype=np.int64)
     bytes_per_region = 16
     region = MemRegionGroup(ptrs=ptrs, bytes_per_region=bytes_per_region)
-    assert list(region.ptrs) == ptrs
+    np.testing.assert_array_equal(region.ptrs, ptrs)
     assert region.bytes_per_region == bytes_per_region
 
 
 def test_spec_region_and_spec_region_pair():
-    group_src = MemRegionGroup(ptrs=[101, 202], bytes_per_region=8)
-    group_dst = MemRegionGroup(ptrs=[303, 404], bytes_per_region=8)
+    group_src = MemRegionGroup(ptrs=np.array([101, 202], dtype=np.int64), bytes_per_region=8)
+    group_dst = MemRegionGroup(ptrs=np.array([303, 404], dtype=np.int64), bytes_per_region=8)
     spec_src = SpecRegion(memory=group_src, spec="spec_src")
     spec_dst = SpecRegion(memory=group_dst, spec="spec_dst")
     assert isinstance(spec_src, SpecRegion)
     assert isinstance(spec_dst, SpecRegion)
     pair = SpecRegionPair(src=spec_src, dst=spec_dst)
     assert isinstance(pair, SpecRegionPair)
-    assert pair.src.memory.ptrs == [101, 202]
-    assert pair.dst.memory.ptrs == [303, 404]
+    np.testing.assert_array_equal(pair.src.memory.ptrs, [101, 202])
+    np.testing.assert_array_equal(pair.dst.memory.ptrs, [303, 404])
     assert pair.src.spec == "spec_src"
     assert pair.dst.spec == "spec_dst"
 
 
 def test_identity_mapper():
-    src_group = MemRegionGroup(ptrs=[100, 200], bytes_per_region=32)
-    dst_group = MemRegionGroup(ptrs=[300, 400], bytes_per_region=32)
+    src_group = MemRegionGroup(ptrs=np.array([100, 200], dtype=np.int64), bytes_per_region=32)
+    dst_group = MemRegionGroup(ptrs=np.array([300, 400], dtype=np.int64), bytes_per_region=32)
     src_spec = SpecRegion(memory=src_group, spec="a")
     dst_spec = SpecRegion(memory=dst_group, spec="b")
     mapper = IdentityMapper()
     result = mapper.map(src_spec, dst_spec)
     assert isinstance(result, SpecRegionPair)
-    assert list(result.src.memory.ptrs) == [100, 200]
-    assert list(result.dst.memory.ptrs) == [300, 400]
+    np.testing.assert_array_equal(result.src.memory.ptrs, [100, 200])
+    np.testing.assert_array_equal(result.dst.memory.ptrs, [300, 400])
     assert result.src.memory.bytes_per_region == 32
     assert result.dst.memory.bytes_per_region == 32
 
@@ -107,8 +109,8 @@ def test_head_match_mapper():
         * self_ri.attention.dims_per_head
         * self_ri.attention.element_bytes
     )
-    src_group = MemRegionGroup(ptrs=[10, 20], bytes_per_region=1)
-    dst_group = MemRegionGroup(ptrs=[30, 40], bytes_per_region=1)
+    src_group = MemRegionGroup(ptrs=np.array([10, 20], dtype=np.int64), bytes_per_region=1)
+    dst_group = MemRegionGroup(ptrs=np.array([30, 40], dtype=np.int64), bytes_per_region=1)
     src_spec = SpecRegion(memory=src_group, spec="srcspec")
     dst_spec = SpecRegion(memory=dst_group, spec="dstspec")
     mapper = HeadMatchMapper(
@@ -121,8 +123,12 @@ def test_head_match_mapper():
     )
     result = mapper.map(src_spec, dst_spec)
     expected_off = transfer_layers * slot_size_per_layer
-    assert list(result.src.memory.ptrs) == [10 + mapper._src_block_off, 20 + mapper._src_block_off]
-    assert list(result.dst.memory.ptrs) == [30 + mapper._dst_block_off, 40 + mapper._dst_block_off]
+    np.testing.assert_array_equal(
+        result.src.memory.ptrs, [10 + mapper._src_block_off, 20 + mapper._src_block_off]
+    )
+    np.testing.assert_array_equal(
+        result.dst.memory.ptrs, [30 + mapper._dst_block_off, 40 + mapper._dst_block_off]
+    )
     assert result.src.memory.bytes_per_region == expected_off
     assert result.dst.memory.bytes_per_region == expected_off
 
@@ -133,8 +139,8 @@ def test_head_mismatch_mapper():
     transfer_layers = 1
     src_layer_off = 0
     peer_layer_off = 1
-    src_group = MemRegionGroup(ptrs=[111], bytes_per_region=32)
-    dst_group = MemRegionGroup(ptrs=[222], bytes_per_region=32)
+    src_group = MemRegionGroup(ptrs=np.array([111], dtype=np.int64), bytes_per_region=32)
+    dst_group = MemRegionGroup(ptrs=np.array([222], dtype=np.int64), bytes_per_region=32)
     src_spec = SpecRegion(memory=src_group, spec="srcspec")
     dst_spec = SpecRegion(memory=dst_group, spec="dstspec")
     mapper = HeadMismatchMapper(transfer_layers, src_layer_off, peer_layer_off, self_ri, peer_ri)
@@ -143,8 +149,8 @@ def test_head_mismatch_mapper():
     assert isinstance(result, SpecRegionPair)
     assert len(result.src.memory.ptrs) == expected_frag_count
     assert len(result.dst.memory.ptrs) == expected_frag_count
-    assert all(isinstance(x, int) for x in result.src.memory.ptrs)
-    assert all(isinstance(x, int) for x in result.dst.memory.ptrs)
+    assert isinstance(result.src.memory.ptrs, np.ndarray)
+    assert isinstance(result.dst.memory.ptrs, np.ndarray)
     assert result.src.memory.bytes_per_region == mapper._bytes_cont_heads
     assert result.dst.memory.bytes_per_region == mapper._bytes_cont_heads
 
diff --git a/tests/unittest/disaggregated/test_extractor.py b/tests/unittest/disaggregated/test_extractor.py
index 3899427869a6..493ae440e4dc 100644
--- a/tests/unittest/disaggregated/test_extractor.py
+++ b/tests/unittest/disaggregated/test_extractor.py
@@ -1,3 +1,4 @@
+import numpy as np
 import pytest
 
 from tensorrt_llm._torch.disaggregation.base.region import DataRole, MemRegionGroup, SpecRegion
@@ -89,7 +90,7 @@ def test_extract():
     )
 
     extractor = KVRegionExtractorV1(manager)
-    region_ids = [0, 1]
+    region_ids = np.array([0, 1], dtype=np.int64)
     spec_region = extractor.extract(region_ids)
 
     assert isinstance(spec_region, SpecRegion)
@@ -108,9 +109,10 @@ def test_extract():
         pool_base_ptr = (
             int(pool_ptrs.data_ptr()) if hasattr(pool_ptrs, "data_ptr") else int(pool_ptrs)
         )
+    assert isinstance(memory.ptrs, np.ndarray)
     expected_block_bytes = memory.bytes_per_region
     expected_ptrs = [pool_base_ptr + block_id * expected_block_bytes for block_id in region_ids]
-    assert list(memory.ptrs) == expected_ptrs
+    np.testing.assert_array_equal(memory.ptrs, expected_ptrs)
 
     manager.shutdown()
 
diff --git a/tests/unittest/disaggregated/test_kv_transfer.py b/tests/unittest/disaggregated/test_kv_transfer.py
index 73f7a4411841..f36316f4e8c5 100644
--- a/tests/unittest/disaggregated/test_kv_transfer.py
+++ b/tests/unittest/disaggregated/test_kv_transfer.py
@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from typing import List, Optional
 
+import numpy as np
 import pytest
 import torch
 
@@ -542,7 +543,7 @@ def get_block_ids_per_layer_groups(
             if len(block_ids) > max_blocks_in_window:
                 block_ids = block_ids[-max_blocks_in_window:]
 
-        block_ids_per_layer_groups.append(list(block_ids))
+        block_ids_per_layer_groups.append(np.asarray(block_ids, dtype=np.int64))
 
     return block_ids_per_layer_groups
 
diff --git a/tests/unittest/disaggregated/test_kv_transfer_mp.py b/tests/unittest/disaggregated/test_kv_transfer_mp.py
index d4f12e97da71..6c77c162ba0d 100644
--- a/tests/unittest/disaggregated/test_kv_transfer_mp.py
+++ b/tests/unittest/disaggregated/test_kv_transfer_mp.py
@@ -1,6 +1,7 @@
 import os
 import random
 
+import numpy as np
 import pytest
 import torch
 import torch.distributed as dist
@@ -326,7 +327,10 @@ def process_and_verify_request(
             sender_session = transfer_worker.create_tx_session(ctx_request)
 
             # Get block ids and send
-            block_ids = kv_cache_manager.get_batch_cache_indices([ctx_request.py_request_id])[0]
+            block_ids = np.asarray(
+                kv_cache_manager.get_batch_cache_indices([ctx_request.py_request_id])[0],
+                dtype=np.int64,
+            )
             send_kv_slice = KVSlice(is_last_slice=True, block_ids_per_layer_groups=[block_ids])
             send_future = sender_session.send(send_kv_slice)
 
@@ -365,7 +369,10 @@ def process_and_verify_request(
             receiver_session = transfer_worker.create_rx_session(gen_request)
 
             # Get block ids and receive
-            block_ids = kv_cache_manager.get_batch_cache_indices([gen_request.py_request_id])[0]
+            block_ids = np.asarray(
+                kv_cache_manager.get_batch_cache_indices([gen_request.py_request_id])[0],
+                dtype=np.int64,
+            )
             recv_kv_slice = KVSlice(is_last_slice=True, block_ids_per_layer_groups=[block_ids])
             recv_future = receiver_session.receive(recv_kv_slice)