Paloma DEX MCP Server

Instructions

Implementation Reference

• padex.py:786-830 (handler)

  Handler function that executes the get_address_balance_single_chain tool. Validates inputs, retrieves Web3 client for the specified chain, calls helper _get_chain_balance, and returns formatted JSON response with native and PUSD balances.
  async def get_address_balance_single_chain(ctx: Context, address: str, chain_id: str) -> str: """Get balance for a specific address on a single chain (faster). Args: address: Ethereum address to check balances for chain_id: Chain ID (1, 10, 56, 100, 137, 8453, 42161) Returns: JSON string with balance information for the specified chain. """ try: paloma_ctx = ctx.request_context.lifespan_context # Validate address if not Web3.is_address(address): return f"Error: Invalid address format: {address}" if chain_id not in CHAIN_CONFIGS: available_chains = [str(k) for k in CHAIN_CONFIGS.keys()] return f"Error: Unsupported chain ID '{chain_id}'. Available: {available_chains}" if chain_id not in paloma_ctx.web3_clients: config = CHAIN_CONFIGS[chain_id] return f"Error: Web3 client not available for {config.name}" address = Web3.to_checksum_address(address) config = CHAIN_CONFIGS[chain_id] web3 = paloma_ctx.web3_clients[chain_id] # Get balance for single chain chain_balance = await _get_chain_balance(web3, address, config, chain_id) result = { "address": address, "chain": config.name, "chain_id": config.chain_id, "balance": chain_balance[config.name], "timestamp": asyncio.get_event_loop().time() } return json.dumps(result, indent=2) except Exception as e: logger.error(f"Error getting single chain balance: {e}") return f"Error getting single chain balance: {str(e)}"
• padex.py:677-727 (helper)

  Helper function used by the handler to fetch native token and PUSD balances for a single chain with per-call timeouts.
  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)}}