app / src /tools /crypto_evm_wallet_toolkit.py
lemdaddy's picture
add smart wallet balances rpc method
660fb68
raw
history blame
5.09 kB
import asyncio
from enum import Enum
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_evm_wallet_address)
self.register(self.get_supported_evm_chains)
self.register(self.get_evm_smart_wallet_address)
self.register(self.get_evm_smart_wallet_balances)
# Fetching 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_wallet_address(self, user_email: str, chain: str, testnet: bool = True) -> str:
# """
# Fetches an EVM wallet address for the given user email and supported chain.
# Creates and returns an 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 created.
# - chain (ethereum | binance | base | polygon): 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.
# Raises:
# None
# Note:
# This method uses asyncio.run() to run the asynchronous RPC call.
# """
# logger.info(f"Creating crypto wallet account for {user_email}")
# params = {
# 'chain': chain,
# 'testnet': testnet,
# 'userEmail': user_email,
# }
# response = asyncio.run(rpc_call(method_name="getEVMWallet", 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_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}"