Paloma DEX MCP Server

Instructions

Implementation Reference

• padex.py:729-783 (handler)

  The primary handler function for the 'get_address_balances' tool. It validates the input address, creates concurrent tasks to fetch native and PUSD balances across all supported chains using Web3 clients, combines results, and returns formatted JSON.
  @mcp.tool() async def get_address_balances(ctx: Context, address: str, timeout_seconds: float = 30.0) -> str: """Get balances for a specific address across all chains (concurrent execution). Args: address: Ethereum address to check balances for timeout_seconds: Timeout for the entire operation (default: 30 seconds) Returns: JSON string with balance information across all chains. """ try: paloma_ctx = ctx.request_context.lifespan_context # Validate address if not Web3.is_address(address): return f"Error: Invalid address format: {address}" address = Web3.to_checksum_address(address) # Create tasks for concurrent execution tasks = [] chain_names = [] for chain_id, config in CHAIN_CONFIGS.items(): if chain_id in paloma_ctx.web3_clients: web3 = paloma_ctx.web3_clients[chain_id] task = _get_chain_balance(web3, address, config, chain_id) tasks.append(task) chain_names.append(config.name) # Execute all balance checks concurrently (no overall timeout, individual chains handle their own timeouts) results = await asyncio.gather(*tasks, return_exceptions=True) # Combine results balances = {} for result in results: if isinstance(result, dict): balances.update(result) elif isinstance(result, Exception): logger.error(f"Chain balance check failed: {result}") result = { "address": address, "balances": balances, "chains_checked": len(tasks), "timeout_seconds": timeout_seconds, "timestamp": asyncio.get_event_loop().time() } return json.dumps(result, indent=2) except Exception as e: logger.error(f"Error getting address balances: {e}") return f"Error getting address balances: {str(e)}"
• padex.py:677-727 (helper)

  Supporting helper function used by get_address_balances to retrieve native token balance and optional PUSD token balance for a single chain, with per-chain timeouts to prevent hanging.
  async def _get_chain_balance(web3: Web3, address: str, config: ChainConfig, chain_id: str) -> Dict[str, Any]: """Helper function to get balance for a single chain with individual timeout.""" try: # Get native balance with individual timeout (5 seconds per chain) native_balance_wei = await asyncio.wait_for( asyncio.get_event_loop().run_in_executor( None, lambda: web3.eth.get_balance(address) ), timeout=5.0 ) native_balance = web3.from_wei(native_balance_wei, 'ether') chain_balances = { "native_balance": str(native_balance), "native_symbol": "ETH" if chain_id == ChainID.ETHEREUM_MAIN else config.name.split()[0] } # Get PUSD balance if configured (with individual timeout) if config.pusd_token: try: pusd_contract = web3.eth.contract( address=config.pusd_token, abi=ERC20_ABI ) # Wrap contract calls in timeout pusd_balance_wei = await asyncio.wait_for( asyncio.get_event_loop().run_in_executor( None, lambda: pusd_contract.functions.balanceOf(address).call() ), timeout=5.0 ) pusd_decimals = await asyncio.wait_for( asyncio.get_event_loop().run_in_executor( None, lambda: pusd_contract.functions.decimals().call() ), timeout=5.0 ) pusd_balance = pusd_balance_wei / (10 ** pusd_decimals) chain_balances["pusd_balance"] = str(pusd_balance) except asyncio.TimeoutError: chain_balances["pusd_balance"] = "Timeout" except Exception as e: chain_balances["pusd_balance"] = f"Error: {str(e)}" return {config.name: chain_balances} except asyncio.TimeoutError: return {config.name: {"error": "Timeout (5s)"}} except Exception as e: return {config.name: {"error": str(e)}}