diff --git a/src/net.h b/src/net.h
--- a/src/net.h
+++ b/src/net.h
@@ -11,6 +11,7 @@
 #include <addrman.h>
 #include <amount.h>
 #include <avalanche/delegation.h>
+#include <avalanche/proofradixtreeadapter.h>
 #include <bloom.h>
 #include <chainparams.h>
 #include <compat.h>
@@ -651,6 +652,10 @@
         CRollingBloomFilter filterProofKnown GUARDED_BY(cs_proof_inventory){
             10000, 0.000001};
         std::chrono::microseconds nextInvSend{0};
+
+        std::shared_ptr<
+            RadixTree<const avalanche::Proof, avalanche::ProofRadixTreeAdapter>>
+            sharedProofs;
     };
 
     // m_proof_relay == nullptr if we're not relaying proofs with this peer
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
--- a/src/net_processing.cpp
+++ b/src/net_processing.cpp
@@ -7,6 +7,7 @@
 
 #include <addrman.h>
 #include <avalanche/avalanche.h>
+#include <avalanche/compactproofs.h>
 #include <avalanche/peermanager.h>
 #include <avalanche/processor.h>
 #include <avalanche/proof.h>
@@ -3414,7 +3415,8 @@
            msg_type == NetMsgType::AVAPOLL ||
            msg_type == NetMsgType::AVARESPONSE ||
            msg_type == NetMsgType::AVAPROOF ||
-           msg_type == NetMsgType::GETAVAADDR;
+           msg_type == NetMsgType::GETAVAADDR ||
+           msg_type == NetMsgType::GETAVAPROOFS;
 }
 
 /**
@@ -5345,6 +5347,24 @@
         return;
     }
 
+    if (msg_type == NetMsgType::GETAVAPROOFS) {
+        if (pfrom.m_proof_relay == nullptr) {
+            return;
+        }
+
+        pfrom.m_proof_relay->sharedProofs =
+            g_avalanche->withPeerManager([&](const avalanche::PeerManager &pm) {
+                return pm.getShareableProofsSnapshot();
+            });
+
+        avalanche::CompactProofs compactProofs(
+            pfrom.m_proof_relay->sharedProofs);
+        m_connman.PushMessage(
+            &pfrom, msgMaker.Make(NetMsgType::AVAPROOFS, compactProofs));
+
+        return;
+    }
+
     if (msg_type == NetMsgType::GETADDR) {
         // This asymmetric behavior for inbound and outbound connections was
         // introduced to prevent a fingerprinting attack: an attacker can send
diff --git a/src/protocol.h b/src/protocol.h
--- a/src/protocol.h
+++ b/src/protocol.h
@@ -303,6 +303,18 @@
  */
 extern const char *GETAVAADDR;
 
+/**
+ * The getavaproofs message requests an avaproofs message that provides
+ * the proof short ids of all the valid proofs known by our peer.
+ */
+extern const char *GETAVAPROOFS;
+
+/**
+ * The avaproofs message the proof short ids of all the valid proofs that we
+ * know.
+ */
+extern const char *AVAPROOFS;
+
 /**
  * Indicate if the message is used to transmit the content of a block.
  * These messages can be significantly larger than usual messages and therefore
diff --git a/src/protocol.cpp b/src/protocol.cpp
--- a/src/protocol.cpp
+++ b/src/protocol.cpp
@@ -52,6 +52,8 @@
 const char *AVARESPONSE = "avaresponse";
 const char *AVAPROOF = "avaproof";
 const char *GETAVAADDR = "getavaaddr";
+const char *GETAVAPROOFS = "getavaproofs";
+const char *AVAPROOFS = "avaproofs";
 
 bool IsBlockLike(const std::string &strCommand) {
     return strCommand == NetMsgType::BLOCK ||
diff --git a/test/functional/abc_p2p_proof_inventory.py b/test/functional/abc_p2p_proof_inventory.py
--- a/test/functional/abc_p2p_proof_inventory.py
+++ b/test/functional/abc_p2p_proof_inventory.py
@@ -10,6 +10,7 @@
 
 from test_framework.address import ADDRESS_ECREG_UNSPENDABLE
 from test_framework.avatools import (
+    AvaP2PInterface,
     avalanche_proof_from_hex,
     gen_proof,
     get_proof_ids,
@@ -21,9 +22,11 @@
     MSG_TYPE_MASK,
     CInv,
     msg_avaproof,
+    msg_getavaproofs,
     msg_getdata,
 )
 from test_framework.p2p import P2PInterface, p2p_lock
+from test_framework.siphash import siphash256
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal, assert_greater_than
 from test_framework.wallet_util import bytes_to_wif
@@ -187,14 +190,46 @@
         node0.sendavalancheproof(proof.serialize().hex())
         self.sync_proofs()
 
+    def test_respond_getavaproofs(self):
+        self.log.info("Check the node responds to getavaproofs messages")
+        node = self.nodes[0]
+
+        sending_peer = node.add_p2p_connection(AvaP2PInterface())
+        for _ in range(50):
+            _, proof = gen_proof(node)
+            sending_peer.send_avaproof(proof)
+            wait_for_proof(node, f"{proof.proofid:0{64}x}")
+
+        proofids = get_proof_ids(node)
+
+        receiving_peer = node.add_p2p_connection(AvaP2PInterface())
+
+        def received_avaproofs(peer):
+            with p2p_lock:
+                return peer.last_message.get("avaproofs")
+
+        receiving_peer.send_message(msg_getavaproofs())
+        self.wait_until(lambda: received_avaproofs(receiving_peer), timeout=1)
+
+        avaproofs = received_avaproofs(receiving_peer)
+        assert_equal(len(avaproofs.shortids), len(proofids))
+
+        expected_shortids = [
+            siphash256(
+                avaproofs.key0,
+                avaproofs.key1,
+                proofid) & 0x0000ffffffffffff for proofid in sorted(proofids)]
+        assert_equal(expected_shortids, avaproofs.shortids)
+
     def test_unbroadcast(self):
         self.log.info("Test broadcasting proofs")
 
         node = self.nodes[0]
 
-        # Disconnect the other nodes, or they will request the proof and
+        # Disconnect the other nodes/peers, or they will request the proof and
         # invalidate the test
-        [node.stop_node() for node in self.nodes[1:]]
+        [n.stop_node() for n in self.nodes[1:]]
+        node.disconnect_p2ps()
 
         def add_peers(count):
             peers = []
@@ -289,6 +324,7 @@
         self.test_receive_proof()
         self.test_proof_relay()
         self.test_manually_sent_proof()
+        self.test_respond_getavaproofs()
 
         # Run these tests last because they need to disconnect the nodes
         self.test_unbroadcast()
diff --git a/test/functional/test_framework/messages.py b/test/functional/test_framework/messages.py
--- a/test/functional/test_framework/messages.py
+++ b/test/functional/test_framework/messages.py
@@ -2118,6 +2118,57 @@
         return "msg_getavaaddr()"
 
 
+class msg_getavaproofs:
+    __slots__ = ()
+    msgtype = b"getavaproofs"
+
+    def __init__(self):
+        pass
+
+    def deserialize(self, f):
+        pass
+
+    def serialize(self):
+        return b""
+
+    def __repr__(self):
+        return "msg_getavaproofs()"
+
+
+class msg_avaproofs:
+    __slots__ = ("key0", "key1", "shortids")
+    msgtype = b"avaproofs"
+
+    def __init__(self):
+        self.key0 = 0
+        self.key1 = 0
+        self.shortids = []
+
+    def deserialize(self, f):
+        self.key0 = struct.unpack("<Q", f.read(8))[0]
+        self.key1 = struct.unpack("<Q", f.read(8))[0]
+        shortids_length = deser_compact_size(f)
+        for _ in range(shortids_length):
+            # shortids are defined to be 6 bytes in the spec, so append
+            # two zero bytes and read it in as an 8-byte number
+            self.shortids.append(
+                struct.unpack("<Q", f.read(6) + b'\x00\x00')[0])
+
+    def serialize(self):
+        r = b""
+        r += struct.pack("<Q", self.key0)
+        r += struct.pack("<Q", self.key1)
+        r += ser_compact_size(len(self.shortids))
+        for shortid in self.shortids:
+            # We only want the first 6 bytes
+            r += struct.pack("<Q", shortid)[0:6]
+        return r
+
+    def __repr__(self):
+        return "msg_avaproofs(key0={}, key1={}, len(shortids)={}, shortids={})".format(
+            self.key0, self.key1, len(self.shortids), self.shortids)
+
+
 class TestFrameworkMessages(unittest.TestCase):
     def test_legacy_avalanche_proof_serialization_round_trip(self):
         """Verify that a LegacyAvalancheProof object is unchanged after a
diff --git a/test/functional/test_framework/p2p.py b/test/functional/test_framework/p2p.py
--- a/test/functional/test_framework/p2p.py
+++ b/test/functional/test_framework/p2p.py
@@ -40,6 +40,7 @@
     msg_avahello,
     msg_avapoll,
     msg_avaproof,
+    msg_avaproofs,
     msg_block,
     msg_blocktxn,
     msg_cfcheckpt,
@@ -52,6 +53,7 @@
     msg_filterload,
     msg_getaddr,
     msg_getavaaddr,
+    msg_getavaproofs,
     msg_getblocks,
     msg_getblocktxn,
     msg_getdata,
@@ -94,6 +96,7 @@
     b"addrv2": msg_addrv2,
     b"avapoll": msg_avapoll,
     b"avaproof": msg_avaproof,
+    b"avaproofs": msg_avaproofs,
     b"avaresponse": msg_tcpavaresponse,
     b"avahello": msg_avahello,
     b"block": msg_block,
@@ -108,6 +111,7 @@
     b"filterload": msg_filterload,
     b"getaddr": msg_getaddr,
     b"getavaaddr": msg_getavaaddr,
+    b"getavaproofs": msg_getavaproofs,
     b"getblocks": msg_getblocks,
     b"getblocktxn": msg_getblocktxn,
     b"getdata": msg_getdata,
@@ -432,6 +436,8 @@
 
     def on_avaproof(self, message): pass
 
+    def on_avaproofs(self, message): pass
+
     def on_avaresponse(self, message): pass
 
     def on_avahello(self, message): pass
@@ -460,6 +466,8 @@
 
     def on_getavaaddr(self, message): pass
 
+    def on_getavaproofs(self, message): pass
+
     def on_getblocks(self, message): pass
 
     def on_getblocktxn(self, message): pass