get_token_price_estimate
Calculate real-time token swap price estimates on Paloma DEX across multiple EVM chains to inform trading decisions before execution.
Instructions
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.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| chain_id | Yes | ||
| input_token_address | Yes | ||
| output_token_address | Yes | ||
| input_amount | Yes |
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: