File size: 5,305 Bytes
431e9df |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
from typing import TypedDict, Any
import asyncio
from phi.tools import Toolkit
from phi.utils.log import logger
from src.libs.rpc_client import rpc_call
# Define TypedDict for wallet
class WalletParams(TypedDict):
userEmail: str
chain: str
testnet: bool
gasless: bool
# Define TypedDict for tokenIn and tokenOut (since they have identical structures)
class TokenParams(TypedDict):
chainId: int
address: str
decimals: int
symbol: str
name: str
class CrossChainSwapTools(Toolkit):
def __init__(self):
super().__init__(name="cross_chain_swap_tools")
# Registering methods to make them accessible via the toolkit
self.register(self.cross_chain_swap)
self.register(self.get_bridge_chains)
self.register(self.get_cross_chain_swap_token_list)
def cross_chain_swap(
self,
chainId: int,
amountLD: str,
recipient: str,
wallet: Any,
tokenIn: Any,
tokenOut: Any
) -> str:
"""
Performs a cross-chain token swap.
Parameters:
- chainId (int): The ID of the chain.
- amountLD (str): The amount to be swapped, in lowest denomination (e.g., wei for ETH).
- recipient (str): The recipient address.
- wallet (WalletParams): The wallet parameters.
- tokenIn (TokenParams): The input token parameters.
- tokenOut (TokenParams): The output token parameters.
Returns:
- str: A string representation of the response from the RPC call.
Raises:
None
Note:
This method uses `asyncio.run()` to run the asynchronous RPC call.
"""
logger.info(f"Performing cross-chain swap on chain {chainId} for recipient {recipient}")
params = {
'chainId': chainId,
'amountLD': amountLD,
'recipient': recipient,
'wallet': wallet,
'tokenIn': tokenIn,
'tokenOut': tokenOut,
}
response = asyncio.run(rpc_call(method_name="crossChainSwap", params=params))
return self._format_response(response)
def get_bridge_chains(self, chainId: int) -> str:
"""
Fetches the list of supported bridge chains.
Parameters:
- chainId (int): The ID of the chain. Supported values:
- 1: Ethereum
- 10: Optimism
- 56: BNB Chain
- 100: Gnosis Chain
- 137: Polygon
- 324: zkSync Era
- 1088: Metis
- 8453: Base
- 42161: Arbitrum
- 43114: Avalanche
"""
logger.info(f"Fetching supported bridge chains for chain ID {chainId}")
params = {'chainId': chainId}
response = asyncio.run(rpc_call(method_name="getBridgeChains", params=params))
return self._format_response(response)
def get_cross_chain_swap_token_list(self, chainId: int) -> str:
"""
Fetches the list of supported cross-chain swap tokens on the specified chainId.
Parameters:
- chainId (int): The ID of the chain.
Returns:
- str: A string representation of the response from the RPC call containing the list of cryptocurrency tokens,
each with details about their source token and the corresponding destination
tokens on different blockchain networks.
Raises:
None
Note:
This method uses `asyncio.run()` to run the asynchronous RPC call.
present data in proper nice readable format
"""
logger.info(f"Fetching cross-chain swap token list for chain ID {chainId}")
params = {'chainId': chainId}
response = asyncio.run(rpc_call(method_name="getCrossChainSwapTokenList", params=params))
return self._format_response(response)
def _format_response(self, response: dict) -> str:
"""
Formats the response from the RPC call into a readable string.
Parameters:
- response (dict): The response from the RPC call.
Returns:
- str: A formatted string representation of the response.
"""
if 'error' in response:
logger.error(f"RPC call failed with error: {response['error']}")
return f"Error: {response['error']}"
return f"Response: {response}"
# Example usage:
# toolkit = CrossChainSwapTools()
# print(toolkit.get_bridge_chains(1))
# print(toolkit.get_cross_chain_swap_token_list(1))
# swap_params = {
# "chainId": 1,
# "amountLD": "100",
# "recipient": "0xRecipientAddress",
# "wallet": {
# "userEmail": "user@example.com",
# "chain": "ethereum",
# "testnet": True,
# "gasless": True
# },
# "tokenIn": {
# "chainId": 1,
# "address": "0xInputTokenAddress",
# "decimals": 18,
# "symbol": "TOKEN_IN",
# "name": "Input Token",
# "logoUri": "https://example.com/logo_in.png"
# },
# "tokenOut": {
# "chainId": 137,
# "address": "0xOutputTokenAddress",
# "decimals": 18,
# "symbol": "TOKEN_OUT",
# "name": "Output Token",
# "logoUri": "https://example.com/logo_out.png"
# }
# }
# print(toolkit.cross_chain_swap(**swap_params))
|