Cross-chain bridges are the backbone of DeFi interoperability, but monitoring token movements across chains remains surprisingly difficult for developers. In this tutorial, I will walk you through building a Python-based monitor that tracks lock-and-mint events across multiple EVM networks in real time.
The Problem
When users bridge tokens from Ethereum to Polygon (or any L2), the bridge contract locks tokens on the source chain and mints wrapped equivalents on the destination. If you are building a dApp that needs to react to these events, you need a reliable way to listen for them across chains.
What We Will Build
We will create a cross-chain event listener that:
- Connects to multiple RPC endpoints (Ethereum, Polygon, Arbitrum, Optimism)
- Monitors known bridge contracts for
TokensLockedandTokensMintedevents - Correlates lock events with their corresponding mint events using transaction hashes and user addresses
- Stores results in a local SQLite database for querying
Prerequisites
pip install web3 aiohttp sqlite-utils
Step 1: Define Bridge Contract Addresses
BRIDGES = {
"polygon_pos": {
"ethereum": "0x8484Ef722627bf18ca5Ae6BcF031c23E6e922B30",
"polygon": "0x8484Ef722627bf18ca5Ae6BcF031c23E6e922B30"
},
"arbitrum": {
"ethereum": "0x72Ce9c846789fdB6fC1f34aC4AD25Dd9ef7031ef",
"arbitrum": "0x72Ce9c846789fdB6fC1f34aC4AD25Dd9ef7031ef"
}
}
Step 2: Connect to Multiple RPCs
from web3 import Web3
RPCS = {
"ethereum": "https://eth.llamarpc.com",
"polygon": "https://polygon.llamarpc.com",
"arbitrum": "https://arbitrum.llamarpc.com"
}
connections = {name: Web3(Web3.HTTPProvider(url)) for name, url in RPCS.items()}
Step 3: Listen for Events
BRIDGE_ABI = '[
{"anonymous":false,"inputs":[
{"indexed":true,"name":"user","type":"address"},
{"indexed":true,"name":"token","type":"address"},
{"indexed":false,"name":"amount","type":"uint256"},
{"indexed":false,"name":"destChainId","type":"uint256"}
],"name":"TokensLocked","type":"event"}
]'
def scan_events(w3, contract_addr, from_block, to_block):
contract = w3.eth.contract(address=contract_addr, abi=BRIDGE_ABI)
return contract.events.TokensLocked.get_logs(
fromBlock=from_block,
toBlock=to_block
)
# Scan the last 1000 blocks on each chain
for chain_name, w3 in connections.items():
latest = w3.eth.block_number
events = scan_events(w3, BRIDGES["polygon_pos"][chain_name], latest - 1000, latest)
for evt in events:
print(f"[{chain_name}] Lock: {evt.args.user} locked {evt.args.amount} of {evt.args.token}")
Step 4: Correlate Cross-Chain Events
import hashlib
def generate_correlation_id(user, amount, chain_id):
raw = f"{user}-{amount}-{chain_id}".encode()
return hashlib.sha256(raw).hexdigest()[:16]
This correlation ID lets you match a lock event on Ethereum with its corresponding mint event on Polygon, even when the transaction hashes differ.
Full Source Code
The complete project with async support, database persistence, and Docker setup is available at github.com/Byaigo.
Why This Matters
Cross-chain monitoring is essential for:
- Bridge security auditing
- Automated cross-chain arbitrage bots
- Real-time bridge health dashboards
- Off-chain reconciliation systems
Final Thoughts
With just a few hundred lines of Python, you can build a production-ready cross-chain event monitor. The Web3.py library handles all the complex RLP encoding, and Python's async capabilities let you monitor dozens of chains simultaneously without breaking a sweat.
If you found this useful, consider supporting my open-source work:
ETH: 0x18da907cb9d981bc798acb87ac27b03a2dc3cbb7
Happy bridging! 🚀
Top comments (0)