From be0650d94ec23e90bb916b258e21c5542fb477ba Mon Sep 17 00:00:00 2001 From: Nick Ninov Date: Wed, 24 Jun 2026 18:08:58 +0300 Subject: [PATCH] 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 | 3 +- examples/agents/main.py | 72 +++++++++++++++++---- examples/agents/models.py | 31 ++++++++- examples/agents/tools.py | 125 +++++++++++++++++++++++++++++++++--- 4 files changed, 204 insertions(+), 27 deletions(-) diff --git a/examples/agents/env.example b/examples/agents/env.example index e9d04c7..2b772f3 100644 --- a/examples/agents/env.example +++ b/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" diff --git a/examples/agents/main.py b/examples/agents/main.py index 892b1a9..03b9253 100644 --- a/examples/agents/main.py +++ b/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 } ] }) diff --git a/examples/agents/models.py b/examples/agents/models.py index c1e3e64..7becd7e 100644 --- a/examples/agents/models.py +++ b/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 diff --git a/examples/agents/tools.py b/examples/agents/tools.py index 9afa2d0..fff2e06 100644 --- a/examples/agents/tools.py +++ b/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}"