diff --git a/src/pynapse/pdp/verifier.py b/src/pynapse/pdp/verifier.py
index 5c6cf69..7e81b02 100644
--- a/src/pynapse/pdp/verifier.py
+++ b/src/pynapse/pdp/verifier.py
@@ -1,9 +1,12 @@
 from __future__ import annotations
 
+from typing import List
+
 from web3 import AsyncWeb3, Web3
 
 from pynapse.contracts import PDP_VERIFIER_ABI
 from pynapse.core.chains import Chain
+from pynapse.core.typed_data import _piece_cid_bytes
 
 
 class SyncPDPVerifier:
@@ -21,6 +24,24 @@ def get_active_piece_count(self, data_set_id: int) -> int:
     def get_active_pieces(self, data_set_id: int, offset: int, limit: int):
         return self._contract.functions.getActivePieces(data_set_id, offset, limit).call()
 
+    def find_piece_ids_by_cid(
+        self,
+        data_set_id: int,
+        piece_cid: str,
+        start_piece_id: int = 0,
+        limit: int = 1,
+    ) -> List[int]:
+        """Direct on-chain CID→piece-id lookup.
+
+        Mirrors the upstream ``findPieceIdsByCid`` helper — replaces an
+        O(n) scan through ``getActivePieces``.
+        """
+        cid_tuple = (_piece_cid_bytes(piece_cid),)
+        ids = self._contract.functions.findPieceIdsByCid(
+            data_set_id, cid_tuple, start_piece_id, limit
+        ).call()
+        return [int(pid) for pid in ids]
+
     def get_data_set_leaf_count(self, data_set_id: int) -> int:
         return int(self._contract.functions.getDataSetLeafCount(data_set_id).call())
 
@@ -59,6 +80,24 @@ async def get_active_piece_count(self, data_set_id: int) -> int:
     async def get_active_pieces(self, data_set_id: int, offset: int, limit: int):
         return await self._contract.functions.getActivePieces(data_set_id, offset, limit).call()
 
+    async def find_piece_ids_by_cid(
+        self,
+        data_set_id: int,
+        piece_cid: str,
+        start_piece_id: int = 0,
+        limit: int = 1,
+    ) -> List[int]:
+        """Direct on-chain CID→piece-id lookup.
+
+        Mirrors the upstream ``findPieceIdsByCid`` helper — replaces an
+        O(n) scan through ``getActivePieces``.
+        """
+        cid_tuple = (_piece_cid_bytes(piece_cid),)
+        ids = await self._contract.functions.findPieceIdsByCid(
+            data_set_id, cid_tuple, start_piece_id, limit
+        ).call()
+        return [int(pid) for pid in ids]
+
     async def get_data_set_leaf_count(self, data_set_id: int) -> int:
         return int(await self._contract.functions.getDataSetLeafCount(data_set_id).call())
 
diff --git a/tests/test_pdp_verifier_find_piece.py b/tests/test_pdp_verifier_find_piece.py
new file mode 100644
index 0000000..791b743
--- /dev/null
+++ b/tests/test_pdp_verifier_find_piece.py
@@ -0,0 +1,25 @@
+"""Tests for PDPVerifier.find_piece_ids_by_cid (#718)."""
+
+from __future__ import annotations
+
+from pynapse.core.typed_data import _piece_cid_bytes
+from pynapse.pdp.verifier import AsyncPDPVerifier, SyncPDPVerifier
+
+
+def test_find_piece_ids_by_cid_method_exists():
+    assert hasattr(SyncPDPVerifier, "find_piece_ids_by_cid")
+    assert hasattr(AsyncPDPVerifier, "find_piece_ids_by_cid")
+
+
+def test_piece_cid_bytes_roundtrip_known_v1():
+    # Sample CommP PieceCIDv1 from upstream fixtures. The important property
+    # is that the helper can decode it into the bytes payload expected by
+    # the on-chain Cids.Cid tuple without throwing.
+    piece_cid_v1 = "baga6ea4seaqlwzed5tgjxpcnlxz2ilhpgitfhuodvgfhc6e6kroivbxmsjpesbi"
+    data = _piece_cid_bytes(piece_cid_v1)
+    # CIDv1 with fil-commitment-unsealed codec and 32-byte SHA256: always
+    # starts with 0x01 version, 0x81 0xe2 0x03 codec varint.
+    assert data[0] == 0x01
+    assert data[1:4] == b"\x81\xe2\x03"
+    # 32-byte digest at the tail
+    assert len(data) >= 32