example: Agent onchain

New features:
- Swap Tokens
- Send Transactions
- Search Transactions
- Add payment requests to messages
- Add payment requests to listening messages

Related to:
- https://github.com/status-im/status-python-sdk/issues/3

examples/agents/env.example

@@ -1,9 +1,10 @@
 # Status Account setup
 PASSWORD = "your-password-here"
-DISPLAY_NAME = "status-display-name"
+NAME = "status-display-name"
 MNEMONIC = "phrase_1 phrase_2 phrase_3 phrase_4 phrase_5 phrase_6 phrase_7 phrase_8 phrase_9 phrase_10 phrase_11 phrase_12"
 ALCHEMY_TOKEN = "your-alchemy-token"
 COINGECKO_API_KEY = "your-coingecko-api-key"
+INFURA_TOKEN = "your-infura-token"
 
 # LLM setup
 GROQ_API_KEY = "your-groq-api-key"

examples/agents/main.py

@@ -1,8 +1,9 @@
 from langchain_groq import ChatGroq
 from langchain.agents import create_agent
 from dotenv import load_dotenv
-import os
-import sys
+from typing import Optional
+import os, sys
+import pandas as pd
 
 # Temp solution until repo it turned into a PyPI library
 # Add the repo root to sys.path so `bot` is importable when running this
@@ -13,14 +14,15 @@ from bot import Account, launch_docker_container
 
 class StatusToolKit:
 
-    def __init__(self, password: str, display_name: str, mnemonic: str, alchemy_token: str, coingecko_api_key: str):
-        self.account = Account()
+    def __init__(self, password: str, display_name: str, mnemonic: str, alchemy_token: str, coingecko_api_key: str, infura_token: str, backup_folder: Optional[str] = None):
+        self.account = Account(backup_folder=backup_folder)
         self.account.login(
             password=password,
-            display_name=display_name,
+            name=display_name,
             mnemonic=mnemonic,
             alchemy_token=alchemy_token,
-            coingecko_api_key=coingecko_api_key
+            coingecko_api_key=coingecko_api_key,
+            infura_token=infura_token
         )
         self.display_name = self.account.display_name
         self.tools = [
@@ -31,13 +33,33 @@ class StatusToolKit:
             tools.SearchTokenTool(account=self.account),
             tools.SearchExternalBalanceTool(account=self.account),
             tools.SearchMessagesTool(account=self.account),
-            tools.SendMessagesTool(account=self.account)
+            tools.SearchTransactionsTool(account=self.account),
+            tools.SendMessagesTool(account=self.account),
+            tools.SendTransactionTool(account=self.account),
+            tools.SwapTokensTool(account=self.account)
         ]
 
 
     def get_tools(self) -> list:
         return self.tools
 
+    def normalize_amount(self, amount: str, token_key: str) -> float:
+        """
+        Convert the WEI amount from Payment requests to a regular amount.
+
+        Parameters:
+            - `amount` - the amount in WEI
+            - `token_key` - the Chain ID and Token Address
+
+        Output:
+            - The regular amount
+        """
+        chain_id, address = token_key.split("-")
+        tokens = self.account.get_tokens()
+        query = (tokens["address"].str.lower() == address.lower()) & (tokens["chain_id"] == int(chain_id))
+        decimals = tokens.loc[query, "decimals"].iloc[0]
+        raw_amount = int(amount) / (10**int(decimals))
+        return float(raw_amount)
 
 if __name__ == "__main__":
 
@@ -58,10 +80,11 @@ if __name__ == "__main__":
 
     status_toolkit = StatusToolKit(
         os.environ["PASSWORD"],
-        os.environ["DISPLAY_NAME"],
+        os.environ["NAME"],
         os.environ["MNEMONIC"],
         os.environ["ALCHEMY_TOKEN"],
-        os.environ["COINGECKO_API_KEY"]
+        os.environ["COINGECKO_API_KEY"],
+        os.environ["INFURA_TOKEN"]
     )
     agent = create_agent(
         model=llm,
@@ -74,24 +97,45 @@ if __name__ == "__main__":
     )
 
     for message in status_toolkit.account.listen_messages():
-        latest_message = None
+        content = None
         for chat in message["event"]["chats"]:
 
-            from_public_key = chat.get("lastMessage", {}).get("from")
+            latest_message: dict = chat.get("lastMessage", {})
+            if not latest_message:
+                continue
+
+            from_public_key = latest_message.get("from")
             if from_public_key != PUBLIC_KEY:
                 continue
 
-            latest_message = chat["lastMessage"]["text"]
+            content = chat["lastMessage"]["text"]
+            payment_requests: list[dict] = latest_message.get("paymentRequests", [])
+            if payment_requests:
+                payment_request = payment_requests[0]
+                amount = status_toolkit.normalize_amount(payment_request["amount"], payment_request["tokenKey"])
+                chain_id, token_address = payment_request["tokenKey"].split("-")
+                payment_content = {
+                    "Receiver Wallet": payment_request['receiver'],
+                    "Token Symbol": payment_request['symbol'],
+                    "Token Address": token_address,
+                    "Amount": amount,
+                    "Chain ID": chain_id
+                }
+                content += f"\n---\n# Payment request\n" + "\n".join([
+                    f"{name}: {value}"
+                    for name, value in payment_content.items()
+                ])
+
             break
 
-        if not latest_message:
+        if not content:
             continue
 
         result = agent.invoke({
             "messages": [
                 {
                     "role": "user",
-                    "content": latest_message
+                    "content": content
                 }
             ]
         })

examples/agents/models.py

@@ -18,9 +18,13 @@ class TokenSearchInput(BaseModel):
     token_symbols: Optional[list[str]] = Field(description="Token Symbol for the given Chain ID", default=None)
 
 class BalanceSearchInput(BaseModel):
-    chain_id: int = Field(description="Chain ID where the token exists.", default=1)
-    token_addresses: list[str] = Field(description="Token Addresses for the given Chain ID")
-    wallet_address: str = Field(description="The wallet address that will be looked up")
+    chain_id: int = Field(description="Chain ID where the tokens exist.", default=1)
+    token_addresses: list[str] = Field(description=(
+        "Required. Non-empty list of token contract addresses to look up on the given chain. "
+        "This cannot be null or omitted. Use `get_token_info` first to resolve token "
+        "symbols (e.g. ETH, SNT) into addresses for the chain."
+    ))
+    wallet_address: str = Field(description="The external wallet address (or ENS name) whose balance will be looked up.")
     ccy: str = Field(description="ISO 4217 alpha code to represent the fiat currency", default="USD")
 
 class AccountContactInput(BaseModel):
@@ -48,6 +52,27 @@ class MessageInput(BaseModel):
     start_date: Optional[DateStr] = Field(description="Required to fetch chat messages from the specified date. Date should be in YYYY-MM-DD format.", default=None)
     end_date: Optional[DateStr] = Field(description="Required to fetch chat messages to the specified date. Date should be in YYYY-MM-DD format.", default=None)
 
+class TransactionSearchInput(BaseModel):
+    chain_ids: Optional[list[int]] = Field(description="Chain IDs where the token exists.", default=None)
+    token_symbols: Optional[list[str]] = Field(description="Token Symbol for the given Chain ID", default=None)
+    refresh: bool = Field(description="If `True` then the data will be refetched from scratch. If `False` then the data will be cached after the first call.")
+    start_date: Optional[DateStr] = Field(description="Required to fetch chat messages from the specified date. Date should be in YYYY-MM-DD format.", default=None)
+    end_date: Optional[DateStr] = Field(description="Required to fetch chat messages to the specified date. Date should be in YYYY-MM-DD format.", default=None)
+
+class SendTransactionInput(BaseModel):
+    address: str = Field(description="The wallet address of the receiver")
+    symbol: str = Field(description="Either a valid Status token symbol (e.g. `ETH`, `SNT`) or its token address")
+    amount: float = Field(description="The amount of the token that will be sent to the receiver")
+    chain_id: int = Field(description="Chain ID where the token exists.", default=1)
+
+class SwapTokensInput(BaseModel):
+    from_token: str = Field(description="The token to swap from. Either a valid Status token symbol (e.g. `ETH`, `SNT`) or its token address.")
+    to_token: str = Field(description="The token to swap to. Either a valid Status token symbol (e.g. `ETH`, `SNT`) or its token address.")
+    amount: float = Field(description="The amount of `from_token` to swap.")
+    chain_id: int = Field(description=(
+        "Chain ID where both tokens exist. The swap happens on a single chain, so `from_token` and `to_token` "
+        "must be on the same chain."
+    ), default=1)
 
 class NoArgs(BaseModel):
     pass

examples/agents/tools.py

@@ -5,7 +5,7 @@ import pandas as pd
 import datetime
 
 import models
-from bot import Account
+from bot import Account, exceptions
 
 
 class StatusBaseTool(BaseTool):
@@ -18,7 +18,7 @@ class StatusBaseTool(BaseTool):
         self.account = account
 
     def to_datetime(self, value: str) -> datetime.datetime:
-        return datetime.datetime.strftime(value, "%Y-%m-%d") if value else None
+        return datetime.datetime.strptime(value, "%Y-%m-%d") if value else None
 
 
 
@@ -89,11 +89,19 @@ class SearchTokenTool(StatusBaseTool):
 class SearchExternalBalanceTool(StatusBaseTool):
 
     name: str = "search_external_balance"
-    description: str = "Get the balance for an external address."
+    description: str = (
+        "Get the token balances for an external wallet address. "
+        "Requires explicit token addresses — call `get_token_info` first to obtain them for the chain."
+    )
     args_schema: Type[BaseModel] = models.BalanceSearchInput
 
     def _run(self, chain_id: int, token_addresses: list[str], wallet_address: str, ccy: str) -> str:
-        balance = self.account.get_balance(token_addresses, chain_id, wallet_address, ccy)
+        balance = self.account.get_balance(
+            token_addresses,
+            chain_id,
+            wallet_address,
+            ccy
+        )
         return balance.to_markdown(index=False) if len(balance) > 0 else "No balance found..."
 
 
@@ -143,6 +151,7 @@ class AccountContactManagementTool(StatusBaseTool):
 
         return f"Executed {action}"
 
+
 class SearchMessagesTool(StatusBaseTool):
 
     name: str = "search_messages"
@@ -150,15 +159,63 @@ class SearchMessagesTool(StatusBaseTool):
     args_schema: Type[BaseModel] = models.MessageInput
 
     def _run(self, chat_id: str, message: Optional[str], start_date: Optional[models.DateStr], end_date: Optional[models.DateStr]) -> str:
-        to_datetime = lambda value: datetime.datetime.strptime(value, "%Y-%m-%d") if value else None
-        messages = self.account.get_messages(chat_id, to_datetime(start_date), to_datetime(end_date))
+        messages = self.account.get_messages(chat_id, self.to_datetime(start_date), self.to_datetime(end_date))
         markdown = f"# Chat\nStart date: {start_date}\nEnd date: {end_date}"
-        if messages:
-            messages_markdown = [f"[{message['whisper_timestamp']}] {'Me' if message['from'] == self.account.info['public_key'] else 'Contact'}: {message['text']}" for message in messages]
-            markdown = f"{markdown}\nMessages:\n{messages_markdown}"
+        if not messages:
+            return f"{markdown}\nNo messages found..."
 
+        messages_markdown = []
+        for message in messages:
+            text = f"[{message['whisper_timestamp']}] {'Me' if message['from'] == self.account.info['public_key'] else 'Contact'}: {message['text']}"
+
+            payment_requests: list[dict] = message.get("payment_requests", [])
+            payments = []
+            for payment_request in payment_requests:
+                tokens = self.account.get_tokens()
+                chain_id, address = payment_request["tokenKey"].split("-")
+                query = (tokens["address"].str.lower() == address.lower()) & (tokens["chain_id"] == int(chain_id))
+                decimals = int(tokens.loc[query, "decimals"].drop_duplicates().iloc[0])
+                payments.append({
+                    "Receiver Wallet Address": payment_request["receiver"],
+                    "Token Symbol": payment_request["symbol"],
+                    "Requested Amount": int(payment_request["amount"]) / (10 ** decimals),
+                    "Token Address": address,
+                    "Chain ID": chain_id,
+                })
+
+            if payments:
+                text += f"\n\n{pd.DataFrame(payments).to_markdown(index=False)}"
+
+            messages_markdown.append(text)
+
+        markdown = f"{markdown}\nMessages:\n\n{'\n'.join(messages_markdown)}"
         return markdown
 
+class SearchTransactionsTool(StatusBaseTool):
+
+    name: str = "search_transactions"
+    description: str = "Get historical wallet transactions (regular, internal and ERC-20 transfers) for the given chain IDs, token symbols and date range. Set `refresh=True` to force a fresh fetch from Alchemy instead of returning the cached history."
+    args_schema: Type[BaseModel] = models.TransactionSearchInput
+
+    def _run(self, chain_ids: Optional[list[int]], token_symbols: Optional[list[str]], refresh: bool, start_date: Optional[models.DateStr], end_date: Optional[models.DateStr]):
+
+        transactions = self.account.get_transactions(refresh)
+
+        if chain_ids:
+            transactions = transactions.loc[transactions["chain_id"].isin(chain_ids)].reset_index(drop=True)
+
+        if token_symbols:
+            transactions = transactions.loc[transactions["symbol"].isin(token_symbols)].reset_index(drop=True)
+
+        start_date = self.to_datetime(start_date)
+        if start_date:
+            transactions = transactions.loc[transactions["timestamp"] >= start_date].reset_index(drop=True)
+
+        end_date = self.to_datetime(end_date)
+        if start_date:
+            transactions = transactions.loc[transactions["timestamp"] <= end_date].reset_index(drop=True)
+
+        return transactions.to_markdown(index=False)
 
 
 class SendMessagesTool(StatusBaseTool):
@@ -170,3 +227,53 @@ class SendMessagesTool(StatusBaseTool):
     def _run(self, chat_id: str, message: Optional[str], start_date: Optional[models.DateStr], end_date: Optional[models.DateStr]) -> str:
         self.account.send_message(chat_id, message)
         return f"Message was sent successfully in chat ID {chat_id}!"
+
+
+class SendTransactionTool(StatusBaseTool):
+
+    name: str = "send_transaction"
+    description: str = "Send crypto (ETH or an ERC-20 token) from the account's wallet to a receiver address on the given chain. Returns the transaction hash to monitor its progress."
+    args_schema: Type[BaseModel] = models.SendTransactionInput
+
+    def _run(self, address: str, symbol: str, amount: float, chain_id: int) -> str:
+        try:
+            transaction_hash = self.account.send_transaction(address, symbol, amount, chain_id)
+        except exceptions.InvalidTokenError as error:
+            return f"Could not send {amount} {symbol} on chain ID {chain_id}: {error}"
+        except exceptions.WalletNotConfiguredError as error:
+            return f"Wallet is not configured for transactions: {error}"
+        except exceptions.NotLoggedInError as error:
+            return f"Cannot send a transaction: {error}"
+
+        if not transaction_hash:
+            return f"Transaction for {amount} {symbol} to {address} on chain ID {chain_id} could not be sent..."
+
+        return f"Sent {amount} {symbol} to {address} on chain ID {chain_id}!\nTransaction hash: {transaction_hash}\nMonitor at: http://etherscan.io/tx/{transaction_hash}"
+
+
+class SwapTokensTool(StatusBaseTool):
+
+    name: str = "swap_tokens"
+    description: str = (
+        "Swap tokens in the account's wallet on a single chain and return the transaction hash to monitor its progress. "
+        "Only ETH <-> ERC-20 swaps are supported (e.g. ETH -> SNT or SNT -> ETH); either `from_token` or `to_token` must be ETH. "
+        "ERC-20 <-> ERC-20 swaps (e.g. SNT -> USDT) are not supported."
+    )
+    args_schema: Type[BaseModel] = models.SwapTokensInput
+
+    def _run(self, from_token: str, to_token: str, amount: float, chain_id: int) -> str:
+        try:
+            transaction_hash = self.account.swap_tokens(from_token, to_token, amount, chain_id)
+        except exceptions.InvalidTokenError as error:
+            return f"Could not swap {amount} {from_token} to {to_token} on chain ID {chain_id}: {error}"
+        except exceptions.WalletNotConfiguredError as error:
+            return f"Wallet is not configured for swaps: {error}"
+        except exceptions.NotLoggedInError as error:
+            return f"Cannot perform a swap: {error}"
+        except exceptions.BackendError as error:
+            return f"Could not build a swap route for {from_token} -> {to_token} on chain ID {chain_id}: {error}"
+
+        if not transaction_hash:
+            return f"Swap of {amount} {from_token} to {to_token} on chain ID {chain_id} could not be sent..."
+
+        return f"Swapped {amount} {from_token} to {to_token} on chain ID {chain_id}!\nTransaction hash: {transaction_hash}\nMonitor at: http://etherscan.io/tx/{transaction_hash}"