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}"