Paloma DEX MCP Server

Instructions

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.

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)}}