Merge pull request #38 from pattern-tech/feat/adding-moralis

feat[api]: adding moralis agent

Author: yasinfakhar

api/src/agentflow/agents/hub.py

@@ -1,4 +1,4 @@
-from src.agentflow.agents import ether_scan_agent, goldrush_agent
+from src.agentflow.agents import ether_scan_agent, goldrush_agent, moralis_agent
 from typing import Any, List
 
 
@@ -17,7 +17,8 @@ class AgentHub:
     def __init__(self):
         self.agents = {
             "etherscan": ether_scan_agent.etherscan_agent,
-            "goldrush": goldrush_agent.goldrush_agent
+            "goldrush": goldrush_agent.goldrush_agent,
+            "moralis": moralis_agent.moralis_agent
         }
 
     def get_agents(self, agent_names: List[str]) -> List[Any]:

api/src/agentflow/agents/moralis_agent.py

@@ -0,0 +1,62 @@
+from langchain.tools import tool
+from langchain.agents import AgentExecutor
+
+from src.util.configuration import Config
+from src.agentflow.utils.enum import AgentType
+from src.agentflow.utils.tools_index import get_all_tools
+from src.agentflow.utils.shared_tools import handle_exceptions
+from src.agentflow.utils.shared_tools import init_llm, init_agent, init_prompt
+
+
+@tool
+@handle_exceptions
+def moralis_agent(query: str):
+    """
+    An agent for handling Ethereum blockchain-related queries and tasks.
+    This agent can perform the following tasks:
+
+    - Get active chains for a wallet address across all chains
+    - Get token balances for a specific wallet address and their token prices in USD. (paginated)
+    - Get the stats for a wallet address.
+    - Retrieve the full transaction history of a specified wallet address, including sends, receives, token and NFT transfers and contract interactions.
+    - Get the contents of a transaction by the given transaction hash.
+    - Get ERC20 approvals for one or many wallet addresses and/or contract addresses, ordered by block number in descending order.
+
+    Args:
+        query (str): query about Ethereum blockchain tasks.
+
+    Returns:
+        str: Response containing the requested Ethereum blockchain information
+    """
+    config = Config.get_config()
+
+    llm = init_llm(service=config["llm"]["provider"],
+                   model_name=config["llm"]["model"],
+                   api_key=config["llm"]["api_key"],
+                   stream=False)
+
+    tools = get_all_tools(tools_path="moralis_tools")
+
+    prompt = init_prompt(llm, AgentType.BLOCKCHAIN_AGENT)
+
+    agent = init_agent(llm, tools, prompt)
+
+    agent_executor = AgentExecutor(
+        agent=agent,
+        tools=tools,
+        return_intermediate_steps=True,
+        verbose=True)
+
+    response = agent_executor.invoke({"input": query})
+
+    try:
+        agent_steps = []
+        for step in response["intermediate_steps"]:
+            agent_steps.append({
+                "function_name": step[0].tool,
+                "function_args": step[0].tool_input,
+                "function_output": step[-1]
+            })
+        return {"agent_steps": agent_steps}
+    except:
+        return "no tools called inside agent"

api/src/agentflow/providers/moralis_tools.py

@@ -1,90 +1,257 @@
-import os
+import requests
 
-from web3 import Web3
+from typing import Any
 from moralis import evm_api
 from langchain.tools import tool
-from typing import List, Dict, Any
 
+from src.util.configuration import Config
 from src.agentflow.utils.shared_tools import handle_exceptions
 
+_config = Config.get_config()
+_moralis_config = Config.get_service_config(_config, "moralis")
 
-def _get_api_key() -> str:
+
+@tool
+@handle_exceptions
+def get_wallet_active_chains(wallet_address: str, output_include: list[str]) -> list[dict[str, Any]]:
+    """
+    Get active chains for a wallet address across all chains
+
+    Args:
+        wallet_address (str): Ethereum wallet address
+        output_include (list[str]):
+            A list of field names to include in in the output.
+
+
+    Returns:
+        List[dict[str, Any]]:
+            A list of dictionaries where each dictionary only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
+
+            - chain, chain_id, first_transaction, last_transaction
+    """
+    params = {
+        "address": wallet_address
+    }
+
+    result = evm_api.wallets.get_wallet_active_chains(
+        api_key=_moralis_config["api_key"],
+        params=params,
+    )
+
+    results = result["active_chains"]
+    final_results = []
+    for result in results:
+        final_results.append({item: result[item]
+                              for item in result.keys() if item in output_include})
+    return final_results
+
+
+@tool
+@handle_exceptions
+def get_wallet_token_balances(wallet_address: str, output_include: list[str], cursor: str = None) -> dict:
+    """
+    Get token balances for a specific wallet address and their token prices in USD. (paginated)
+
+    Args:
+        wallet_address (str): Ethereum wallet address
+        output_include (list[str]): A list of field names to include in the output.
+        cursor (str): The cursor returned in the previous response (used for getting the next page). end of page cursor is None
+
+
+    Returns:
+        List[dict[str, Any]]:
+            A list of dictionaries where each dictionary only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
+
+            - token_address, symbol, name, logo, thumbnail, decimals, balance, balance_formatted,
+              usd_price, usd_price_24hr_percent_change, usd_price_24hr_usd_change, usd_value,
+              usd_value_24hr_usd_change, native_token, portfolio_percentage
+    """
+    params = {
+        "chain": "eth",
+        "address": wallet_address
+    }
+
+    if cursor:
+        params["cursor"] = cursor
+
+    api_result = evm_api.wallets.get_wallet_token_balances_price(
+        api_key=_moralis_config["api_key"],
+        params=params,
+    )
+
+    results = api_result["result"]
+    final_results = []
+    for result in results:
+        final_results.append({item: result[item]
+                              for item in result.keys() if item in output_include})
+
+    return {"cursor": api_result["cursor"],
+            "results": final_results}
+
+
+@tool
+@handle_exceptions
+def get_wallet_stats(wallet_address: str, output_include: list[str]) -> dict:
     """
-    Retrieve the Moralis API key from the environment.
+    Get the stats for a wallet address.
+
+    Args:
+        wallet_address (str): Ethereum wallet address
 
     Returns:
-        str: The Moralis API key.
+        List[dict[str, Any]]:
+            A list of dictionaries where each dictionary only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
 
-    Raises:
-        Exception: If the API key is not found.
+            - nfts, collections, transactions, nft_transfers, token_transfers
     """
-    api_key = os.getenv("MORALIS_API_KEY")
-    if not api_key:
-        raise Exception("No API key found for Moralis.")
-    return api_key
+    params = {
+        "chain": "eth",
+        "address": wallet_address
+    }
+
+    result = evm_api.wallets.get_wallet_stats(
+        api_key=_moralis_config["api_key"],
+        params=params,
+    )
+
+    return {item: result[item]
+            for item in result.keys() if item in output_include}
 
 
 @tool
 @handle_exceptions
-def get_contract_transactions(contract_address: str) -> List[Dict[str, Any]]:
+def get_wallet_history(wallet_address: str, output_include: list[str], cursor: str = None) -> dict:
     """
-    Fetch the latest transactions for a specified contract address using the Moralis API 
-    and decode their function inputs.
+    Retrieve the full transaction history of a specified wallet address, including sends, receives, token and NFT transfers
+    and contract interactions. (paginated & in descending order)
 
     Args:
-        contract_address (str): The contract address to fetch transactions for.
+        wallet_address (str): Ethereum wallet address
+        output_include (list[str]): A list of field names to include in the output.
+        cursor (str): The cursor returned in the previous response (used for getting the next page). end of page cursor is None
 
     Returns:
-        List[Dict[str, Any]]: A list of decoded transaction details, limited to 20 entries.
+        dict[str, Any]:
+            A dictionary where each key-value pair only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
+
+            - hash, from_address_entity, from_address_entity_logo, from_address,
+              from_address_label, to_address_entity, to_address_entity_logo, to_address, to_address_label,
+              value, receipt_contract_address, block_timestamp, block_number, block_hash, internal_transactions,
+              nft_transfers, erc20_transfer, native_transfers
+    """
+    params = {
+        "chain": "eth",
+        "order": "DESC",
+        "address": wallet_address
+    }
+
+    if cursor:
+        params["cursor"] = cursor
+
+    api_result = evm_api.wallets.get_wallet_history(
+        api_key=_moralis_config["api_key"],
+        params=params,
+    )
+
+    final_results = []
+    for result in api_result["result"]:
+        final_results.append({item: result[item]
+                              for item in result.keys() if item in output_include})
+
+    return {"cursor": api_result["cursor"],
+            "results": final_results}
+
+
+@tool
+@handle_exceptions
+def get_transaction_detail(transaction_hash: str, output_include: list[str]) -> dict:
     """
-    # Retrieve API key
-    api_key = _get_api_key()
+    Get the contents of a transaction by the given transaction hash.
+
+    Args:
+        transaction_hash (str): transaction hash to be decoded
+        output_include (list[str]): A list of field names to include in the output.
+
+    Returns:
+        dict[str, Any]:
+            A dictionary where each key-value pair only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
+
+            - hash, from_address_entity, from_address_entity_logo, from_address,
+              from_address_label, to_address_entity, to_address_entity_logo, to_address, to_address_label,
+              value, receipt_gas_used, receipt_contract_address, receipt_root, receipt_status, block_timestamp,
+              block_number, block_hash, decoded_call, decoded_event
+
 
-    # Set up parameters for the Moralis API call
+    """
     params = {
-        "address": contract_address,
-        "chain": "eth"
+        "chain": "eth",
+        "transaction_hash": transaction_hash
     }
 
-    # Call the Moralis API to get wallet transactions
-    results = evm_api.transaction.get_wallet_transactions(
-        api_key=api_key,
+    result = evm_api.transaction.get_transaction_verbose(
+        api_key=_moralis_config["api_key"],
         params=params,
     )
 
-    # Set up a Web3 provider
-    eth_rpc = os.getenv("ETH_RPC")
-    if not eth_rpc:
-        raise Exception("ETH_RPC environment variable is not set.")
-    w3 = Web3(Web3.HTTPProvider(eth_rpc))
-
-    # Retrieve the contract ABI.
-    # NOTE: Adjust the import path for get_contract_abi as needed.
-    from src.agent.tools.some_module import get_contract_abi
-    contract_abi = get_contract_abi(contract_address)
-
-    contract = w3.eth.contract(address=contract_address, abi=contract_abi)
-
-    decoded_transactions: List[Dict[str, Any]] = []
-    for tx in results.get('result', []):
-        try:
-            # Attempt to decode the function input.
-            # If the input is empty or invalid, an exception may be raised.
-            decoded_input = contract.decode_function_input(
-                tx.get('input', '0x'))
-            function_name = decoded_input[0].fn_name if decoded_input else None
-        except Exception:
-            function_name = None
-
-        decoded_transactions.append({
-            'transaction_hash': tx.get('hash'),
-            'block_number': tx.get('block_number'),
-            'from': tx.get('from_address'),
-            'to': tx.get('to_address'),
-            'value': tx.get('value'),
-            'function_name': function_name,
-        })
-
-    return decoded_transactions[:20]
+    return {item: result[item]
+            for item in result.keys() if item in output_include}
+
+
+@tool
+@handle_exceptions
+def get_token_approvals(wallet_address: str, output_include: list[str], cursor: str = None) -> dict:
+    """
+    Get ERC20 approvals for one or many wallet addresses and/or contract addresses, ordered by block number in descending order.
+
+    Args:
+        wallet_address (str): Ethereum wallet address
+        output_include (list[str]): A list of field names to include in the output.
+        cursor (str): The cursor returned in the previous response (used for getting the next page). end of page cursor is None
+
+    Returns:
+        List[dict[str, Any]]:
+            A list of dictionaries where each dictionary only contains the keys
+            listed in `output_include` (if they exist in the source data).
+            Possible fields include:
+
+            - block_number, block_timestamp, transaction_hash, value, value_formatted, token, spender
+    """
+    base_url = _moralis_config["url"]
+    api_url = f"{base_url}/wallets/{wallet_address}/approvals"
+
+    params = {'chain': 'eth'}
+
+    if cursor:
+        params["cursor"] = cursor
+
+    headers = {
+        'accept': 'application/json',
+        'X-API-Key': _moralis_config["api_key"]
+    }
+
+    response = requests.get(api_url, headers=headers, params=params)
+
+    if response.status_code != 200:
+        raise Exception(
+            f"Failed to get token approvals. Status code: {response.status_code}")
+
+    api_result = response.json()
 
+    results = api_result["result"]
+    final_results = []
+    for result in results:
+        final_results.append({item: result[item]
+                              for item in result.keys() if item in output_include})
 
+    return {"cursor": api_result["cursor"],
+            "results": final_results}