Paloma DEX MCP Server

Instructions

Args:
    chain_id: Chain ID (1, 10, 56, 100, 137, 8453, 42161)
    input_token_address: Address of token to trade from
    output_token_address: Address of token to trade to
    input_amount: Amount of input token in wei format

Returns:
    JSON string with price estimate and trading information.

Implementation Reference

• padex.py:1351-1480 (handler)

  Main handler function decorated with @mcp.tool(). Computes token swap price estimates using PalomaDEXAPI.get_token_estimate and AMM utilities, returning formatted JSON with exchange rates, price impact, etc.

  async def get_token_price_estimate(ctx: Context, chain_id: str, input_token_address: str, output_token_address: str, input_amount: str) -> str:
      """Get real-time price estimate for token swap.
      
      Args:
          chain_id: Chain ID (1, 10, 56, 100, 137, 8453, 42161)
          input_token_address: Address of token to trade from
          output_token_address: Address of token to trade to
          input_amount: Amount of input token in wei format
      
      Returns:
          JSON string with price estimate and trading information.
      """
      try:
          paloma_ctx = ctx.request_context.lifespan_context
          
          if chain_id not in CHAIN_CONFIGS:
              return f"Error: Unsupported chain ID {chain_id}"
          
          config = CHAIN_CONFIGS[chain_id]
          chain_name = get_chain_name_for_api(chain_id)
          
          if not chain_name:
              return f"Error: Chain name mapping not found for chain ID {chain_id}"
          
          # Validate addresses
          if not Web3.is_address(input_token_address):
              return f"Error: Invalid input token address: {input_token_address}"
          
          if not Web3.is_address(output_token_address):
              return f"Error: Invalid output token address: {output_token_address}"
          
          try:
              input_amount_int = int(input_amount)
              if input_amount_int <= 0:
                  raise ValueError("Amount must be positive")
          except ValueError:
              return f"Error: Invalid input amount: {input_amount}"
          
          # Use our Paloma-based API implementation
          try:
              estimate_data = await paloma_ctx.palomadex_api.get_token_estimate(
                  input_token_address, output_token_address, chain_id, input_amount
              )
              
              if not estimate_data.get('exist', False):
                  return f"Error: Trading pair does not exist for these tokens"
              
              if estimate_data.get('empty', True):
                  return f"Error: Pool has no liquidity for this trading pair"
              
              # Get token information from blockchain
              web3 = paloma_ctx.web3_clients.get(chain_id)
              if web3:
                  try:
                      input_contract = web3.eth.contract(address=input_token_address, abi=ERC20_ABI)
                      output_contract = web3.eth.contract(address=output_token_address, abi=ERC20_ABI)
                      
                      input_symbol = input_contract.functions.symbol().call()
                      output_symbol = output_contract.functions.symbol().call()
                      input_decimals = input_contract.functions.decimals().call()
                      output_decimals = output_contract.functions.decimals().call()
                      
                      # Convert amounts for display
                      input_amount_display = float(input_amount_int) / (10 ** input_decimals)
                      output_amount_wei = int(estimate_data.get('estimated_amount', '0'))
                      output_amount_display = float(output_amount_wei) / (10 ** output_decimals)
                      
                      # Calculate exchange rate and price impact
                      exchange_rate = output_amount_display / input_amount_display if input_amount_display > 0 else 0
                      
                      # Calculate price impact using AMM math
                      price_impact = AMM.calculate_price_impact(
                          input_amount_int, 
                          1000000 * 10**input_decimals,  # Mock reserve
                          1000000 * 10**output_decimals   # Mock reserve
                      )
                      
                  except Exception as e:
                      logger.warning(f"Failed to get token info from blockchain: {e}")
                      input_symbol = "Unknown"
                      output_symbol = "Unknown"
                      input_decimals = 18
                      output_decimals = 18
                      input_amount_display = float(input_amount_int) / 1e18
                      output_amount_display = float(estimate_data.get('estimated_amount', '0')) / 1e18
                      exchange_rate = 0
                      price_impact = 0
              else:
                  input_symbol = "Unknown"
                  output_symbol = "Unknown"
                  input_amount_display = float(input_amount_int) / 1e18
                  output_amount_display = float(estimate_data.get('estimated_amount', '0')) / 1e18
                  exchange_rate = 0
                  price_impact = 0
              
              result = {
                  "chain": config.name,
                  "chain_id": config.chain_id,
                  "input_token": {
                      "address": input_token_address,
                      "symbol": input_symbol,
                      "amount_wei": input_amount,
                      "amount_display": str(input_amount_display)
                  },
                  "output_token": {
                      "address": output_token_address,
                      "symbol": output_symbol,
                      "estimated_amount_wei": estimate_data.get('estimated_amount', '0'),
                      "estimated_amount_display": str(output_amount_display)
                  },
                  "trading_info": {
                      "exchange_rate": f"1 {input_symbol} = {exchange_rate:.6f} {output_symbol}",
                      "price_impact": f"{price_impact:.2f}%",
                      "trading_fee": "0.3%"
                  },
                  "pool_exists": estimate_data.get('exist', False),
                  "has_liquidity": not estimate_data.get('empty', True),
                  "data_source": "paloma_amm_calculation",
                  "raw_api_response": estimate_data
              }
              
              return json.dumps(result, indent=2)
              
          except Exception as api_error:
              logger.error(f"Price estimation failed: {api_error}")
              return f"Error: Failed to get price estimate: {str(api_error)}"
                  
      except Exception as e:
          logger.error(f"Error getting token price estimate: {e}")
          return f"Error getting token price estimate: {str(e)}"

• padex.py:439-472 (helper)

  PalomaDEXAPI.get_token_estimate: Core helper that performs the actual AMM swap calculation using mock pool reserves and returns estimation data used by the main handler.

  async def get_token_estimate(self, input_token_address: str, output_token_address: str, 
                             chain_id: str, input_amount: str) -> Dict:
      """Get token swap estimation using AMM math."""
      try:
          # Create token denoms for Paloma
          input_denom = create_token_denom(chain_id, input_token_address, "input")
          output_denom = create_token_denom(chain_id, output_token_address, "output")
          
          if not input_denom or not output_denom:
              return {"exist": False, "empty": True, "estimated_amount": "0"}
          
          # Mock pool reserves (in real implementation, query from Paloma)
          input_reserve = 1000000 * 10**18  # 1M tokens
          output_reserve = 1000000 * 10**18  # 1M tokens
          
          input_amount_int = int(input_amount)
          
          # Calculate swap output using AMM
          estimated_output = AMM.calculate_swap_output(
              input_amount_int, input_reserve, output_reserve
          )
          
          return {
              "amount0": str(input_amount),
              "amount1": str(estimated_output),
              "estimated_amount": str(estimated_output),
              "exist": True,
              "empty": False
          }
          
      except Exception as e:
          logger.error(f"Error in token estimation: {e}")
          return {"exist": False, "empty": True, "estimated_amount": "0"}

• padex.py:365-414 (helper)

  AMM utility class providing calculate_swap_output and calculate_price_impact functions used in token price estimation.

  class AMM:
      """AMM calculation utilities."""
      
      @staticmethod
      def calculate_swap_output(input_amount: int, input_reserve: int, output_reserve: int, fee_rate: float = 0.003) -> int:
          """Calculate swap output using constant product formula."""
          if input_reserve <= 0 or output_reserve <= 0:
              return 0
          
          # Apply fee to input amount
          input_amount_with_fee = int(input_amount * (1 - fee_rate))
          
          # Constant product formula: (x + dx) * (y - dy) = x * y
          # dy = (y * dx) / (x + dx)
          numerator = output_reserve * input_amount_with_fee
          denominator = input_reserve + input_amount_with_fee
          
          if denominator <= 0:
              return 0
          
          return numerator // denominator
      
      @staticmethod
      def calculate_price_impact(input_amount: int, input_reserve: int, output_reserve: int) -> float:
          """Calculate price impact percentage."""
          if input_reserve <= 0 or output_reserve <= 0:
              return 0.0
          
          # Current price (before swap)
          current_price = output_reserve / input_reserve
          
          # Price after swap
          output_amount = AMM.calculate_swap_output(input_amount, input_reserve, output_reserve)
          if input_amount <= 0:
              return 0.0
          
          new_price = output_amount / input_amount
          
          # Price impact as percentage
          if current_price <= 0:
              return 0.0
          
          price_impact = (current_price - new_price) / current_price * 100
          return max(0.0, price_impact)
      
      @staticmethod
      def apply_slippage_tolerance(amount: int, slippage_tolerance: float) -> int:
          """Apply slippage tolerance to get minimum received amount."""
          return int(amount * (1 - slippage_tolerance / 100))

• padex.py:1351-1351 (registration)

  @mcp.tool() decorator registers the get_token_price_estimate function with the FastMCP server.

  async def get_token_price_estimate(ctx: Context, chain_id: str, input_token_address: str, output_token_address: str, input_amount: str) -> str: