Paloma DEX MCP Server

Instructions

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.

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