src/tools/crypto_bridge_toolkit.py

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))