File size: 6,092 Bytes
431e9df a36d1ff 431e9df a36d1ff 431e9df 0ad2f68 a36d1ff 431e9df a36d1ff 431e9df a36d1ff 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 169 170 171 172 173 174 175 176 177 |
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,
user_email: str,
tokenIn: str,
tokenOut: str,
chain: str = "base",
) -> str:
"""
Performs a cross-chain token swap. this is a sensitive function, show a preview and allow user to give final permission to execute function
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.
- user_email (str): The email of the user for whom the wallet is being fetched and used to sign the transaction. get enail from user toolkit
- tokenIn (TokenParams): The input token parameters as structured from: get_cross_chain_swap_token_list. stringify object
- tokenOut (TokenParams): The output token parameters as structured from: get_cross_chain_swap_token_list. stringify object
- chain (str): The EVM chain for which the wallet is being fetched and used to execute this transaction. use chain base as default
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': {
"userEmail": user_email,
"chain": chain,
"testnet": True,
"gasless": True,
"connected": True
},
'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. always call before executing
cross_chain_swap to get accurate tokenIn and tokenOut data
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))
|