File size: 6,122 Bytes
75bb6d0 660fb68 25d025b 3a5ee58 660fb68 25d025b 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 25d025b 3a5ee58 25d025b 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 660fb68 75bb6d0 3a5ee58 |
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 |
import asyncio
from phi.tools import Toolkit
from phi.utils.log import logger
from src.libs.rpc_client import rpc_call
class CryptoEVMWalletTools(Toolkit):
def __init__(self):
super().__init__(name="crypto_evm_wallet_tools")
# Registering methods to make them accessible via the toolkit
self.register(self.get_supported_evm_chains)
self.register(self.get_evm_smart_wallet_address)
self.register(self.get_evm_smart_wallet_balances)
self.register(self.get_all_evm_smart_wallet_address)
self.register(self.get_all_evm_smart_wallet_balances)
# Fetch the list of supported EVM chains
self.chains = self.get_supported_evm_chains()
def get_supported_evm_chains(self) -> list[str]:
"""
Fetches the list of supported EVM chains.
Returns:
str: A string representation of the response from the RPC call containing the list of supported EVM chains.
Raises:
None
"""
logger.info("Fetching supported EVM chains")
params = {}
response = asyncio.run(rpc_call(method_name="getEVMSupportedChains", params=params))
return f"{response}"
def get_evm_smart_wallet_address(
self,
user_email: str,
chain: str,
gasless: bool = True,
testnet: bool = True
) -> str:
"""
Fetches a smart EVM wallet address for the given user email and supported chain.
Creates and returns a smart EVM wallet address for the given user email and supported chain.
Parameters:
- user_email (str): The email of the user for whom the wallet is being fetched.
- chain (ethereum | binance | base | polygon): The EVM chain for which the wallet is being fetched.
- gasless (bool, optional): A flag indicating whether the wallet should be gasless. Defaults to `True`.
- testnet (bool, optional): A flag indicating whether the wallet should be on the testnet. Defaults to `True`.
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"Fetching crypto wallet account for {user_email}")
params = {
'chain': chain,
'gasless': gasless,
'testnet': testnet,
'userEmail': user_email,
}
response = asyncio.run(rpc_call(method_name="getEVMSmartWallet", params=params))
return f"{response}"
def get_all_evm_smart_wallet_address(
self,
user_email: str,
testnet: bool = True
) -> str:
"""
Fetches all EVM smart wallet addresses for a user.
This method takes a user's email and an optional testnet flag, constructs the
necessary parameters, and makes a remote procedure call (RPC) to fetch all EVM
smart wallet addresses associated with the user.
Args:
user_email (str): The email of the user whose wallet addresses are to be fetched.
testnet (bool, optional): A flag indicating whether to fetch addresses from the testnet. Defaults to True.
Returns:
str: The response from the RPC call containing the wallet addresses.
"""
logger.info(f"Fetching all crypto evm smart wallet addresses for {user_email}")
params = {
'testnet': testnet,
'userEmail': user_email,
}
response = asyncio.run(rpc_call(method_name="getAllEVMSmartWallets", params=params))
return f"{response}"
def get_evm_smart_wallet_balances(
self,
address: str,
user_email: str,
chain: str,
testnet: bool = True
) -> str:
"""
Fetches the balances of a smart EVM wallet address for the given user email and supported chain.
Parameters:
- address (str): The address of the smart EVM wallet for which the balances are being fetched.
- user_email (str): The email of the user for whom the wallet is being fetched.
- chain (str): The EVM chain for which the wallet is being fetched.
- testnet (bool, optional): A flag indicating whether the wallet should be on the testnet. Defaults to `True`.
Returns:
- str: A string representation of the response from the RPC call containing the balances of the smart EVM wallet address.
Raises:
None
Note:
This method uses `asyncio.run()` to run the asynchronous RPC call.
"""
logger.info(f"Fetching crypto wallet account for {user_email}")
params = {
'chain': chain,
'address': address,
'testnet': testnet,
'userEmail': user_email,
}
response = asyncio.run(rpc_call(method_name="getEVMSmartWalletAddressBalances", params=params))
return f"{response}"
def get_all_evm_smart_wallet_balances(
self,
user_email: str,
testnet: bool = True
) -> str:
"""
Fetches the balances of all smart EVM wallet addresses for the given user email.
This method takes a user's email and an optional testnet flag, constructs the
necessary parameters, and makes a remote procedure call (RPC) to fetch all EVM
smart wallet addresses associated with the user.
Args:
user_email (str): The email of the user whose wallet addresses are to be fetched.
testnet (bool, optional): A flag indicating whether to fetch addresses from the testnet. Defaults to True.
Returns:
str: The response from the RPC call containing the wallet addresses.
"""
logger.info(f"Fetching crypto wallet account for {user_email}")
params = {
'testnet': testnet,
'userEmail': user_email,
}
response = asyncio.run(rpc_call(method_name="getAllEVMSmartWalletsAddressBalances", params=params))
return f"{response}"
|